"""Fusion — serveur MCP Flowdeck (branchement du service d'agents). POST /mcp, authentifié par les jetons d'API personnels existants (api_tokens). Le service d'agents lit et écrit (création/mise à jour de pages, création de collections — aucune suppression) au nom du propriétaire du jeton. """ import hashlib import json import pytest from app.db import get_conn @pytest.fixture() def jeton(client): brut = "fd_test_token_pour_mcp" with get_conn() as conn: conn.execute( "INSERT INTO api_tokens (user_id, name, token_hash, token_prefix, revoked) " "VALUES (1, 'service-agents', ?, ?, 0)", (hashlib.sha256(brut.encode()).hexdigest(), brut[:12])) conn.commit() return brut @pytest.fixture() def donnees(client): with get_conn() as conn: cur = conn.execute( "INSERT INTO collections (name, description, icon) VALUES ('Projet Alpha', 'Le projet de test', '🚀')") cid = cur.lastrowid conn.execute( "INSERT INTO collection_pages (collection_id, title, icon, position, property_values_json) " "VALUES (?, 'Page de cadrage', 'file', 1, ?)", (cid, json.dumps({"statut": "en cours"}))) conn.execute( "INSERT INTO collection_pages (collection_id, title, icon, position, property_values_json) " "VALUES (?, 'Revue hebdo', 'file', 2, '{}')", (cid,)) conn.commit() return cid def _rpc(client, jeton, rid, methode, params=None): h = {"Authorization": f"Bearer {jeton}"} if jeton else {} corps = {"jsonrpc": "2.0", "id": rid, "method": methode} if params is not None: corps["params"] = params return client.post("/mcp", json=corps, headers=h) def _appel(client, jeton, rid, nom, args=None): r = _rpc(client, jeton, rid, "tools/call", {"name": nom, "arguments": args or {}}) assert r.status_code == 200, r.text return json.loads(r.json()["result"]["content"][0]["text"]) def test_mcp_sans_jeton_401(client): r = _rpc(client, None, 1, "tools/list") assert r.status_code == 401 r = _rpc(client, "mauvais-jeton", 1, "tools/list") assert r.status_code == 401 def test_mcp_handshake_et_outils(client, jeton): r = _rpc(client, jeton, 1, "initialize", {}) assert r.json()["result"]["serverInfo"]["name"] == "flowdeck" r = _rpc(client, jeton, 2, "tools/list") noms = [t["name"] for t in r.json()["result"]["tools"]] assert noms == ["list_collections", "get_collection", "get_page", "search_pages", "list_repos", "create_page", "update_page", "create_collection"] def test_mcp_collections_et_pages(client, jeton, donnees): cols = _appel(client, jeton, 1, "list_collections") alpha = [c for c in cols if c["name"] == "Projet Alpha"] assert alpha and alpha[0]["page_count"] == 2 detail = _appel(client, jeton, 2, "get_collection", {"id": donnees}) assert detail["collection"]["name"] == "Projet Alpha" titres = [p["title"] for p in detail["pages"]] assert titres == ["Page de cadrage", "Revue hebdo"] assert detail["pages"][0]["properties"] == {"statut": "en cours"} res = _appel(client, jeton, 3, "search_pages", {"query": "cadrage"}) assert [p["title"] for p in res] == ["Page de cadrage"] res = _appel(client, jeton, 4, "get_collection", {"id": 99999}) assert "introuvable" in res["erreur"] def test_mcp_create_page_puis_relecture(client, jeton, donnees): creee = _appel(client, jeton, 10, "create_page", { "collection_id": donnees, "title": "Page créée via MCP", "icon": "file", "properties": {"statut": "à faire", "priorite": "haute"}}) assert creee["title"] == "Page créée via MCP" assert creee["collection_id"] == donnees assert creee["properties"] == {"statut": "à faire", "priorite": "haute"} detail = _appel(client, jeton, 11, "get_collection", {"id": donnees}) titres = [p["title"] for p in detail["pages"]] assert "Page créée via MCP" in titres relue = [p for p in detail["pages"] if p["id"] == creee["id"]][0] assert relue["properties"] == {"statut": "à faire", "priorite": "haute"} def test_mcp_update_page_titre_et_proprietes(client, jeton, donnees): creee = _appel(client, jeton, 20, "create_page", { "collection_id": donnees, "title": "Titre initial", "properties": {"statut": "brouillon"}}) maj = _appel(client, jeton, 21, "update_page", { "id": creee["id"], "title": "Titre mis à jour", "properties": {"statut": "publié"}}) assert maj["title"] == "Titre mis à jour" # properties remplace property_values_json (pas de fusion) assert maj["properties"] == {"statut": "publié"} relue = _appel(client, jeton, 22, "get_page", {"id": creee["id"]}) assert relue["title"] == "Titre mis à jour" assert relue["properties"] == {"statut": "publié"} def test_mcp_create_collection(client, jeton): creee = _appel(client, jeton, 30, "create_collection", { "name": "Projet Bêta", "description": "Créée via MCP", "icon": "🧪"}) assert creee["name"] == "Projet Bêta" assert creee["description"] == "Créée via MCP" assert creee["icon"] == "🧪" cols = _appel(client, jeton, 31, "list_collections") beta = [c for c in cols if c["name"] == "Projet Bêta"] assert beta and beta[0]["id"] == creee["id"] assert beta[0]["page_count"] == 0 detail = _appel(client, jeton, 32, "get_collection", {"id": creee["id"]}) assert detail["collection"]["name"] == "Projet Bêta" assert detail["pages"] == [] def test_mcp_erreurs_ecritures(client, jeton, donnees): res = _appel(client, jeton, 40, "create_page", { "collection_id": 99999, "title": "Orpheline"}) assert "introuvable" in res["erreur"] res = _appel(client, jeton, 41, "create_page", { "collection_id": donnees, "title": " "}) assert "erreur" in res res = _appel(client, jeton, 42, "update_page", { "id": 99999, "title": "Fantôme"}) assert "introuvable" in res["erreur"] res = _appel(client, jeton, 43, "create_collection", {"name": " "}) assert "erreur" in res def test_mcp_ecriture_sans_jeton_401(client, donnees): r = _rpc(client, None, 50, "tools/call", {"name": "create_page", "arguments": {"collection_id": donnees, "title": "Refusée"}}) assert r.status_code == 401 r = _rpc(client, "mauvais-jeton", 51, "tools/call", {"name": "create_collection", "arguments": {"name": "Refusée"}}) assert r.status_code == 401 def test_mcp_client_bearer_sans_session_ni_csrf(client, jeton, donnees): """Régression 7.76.1 : un vrai client MCP est machine-à-machine (Bearer seul, ni session ni jeton CSRF) — exactement le service d'agents. Le client de test habituel porte une session + un CSRF, ce qui masquait le 403 du middleware en production.""" from tests.conftest import anon anon(client) r = _rpc(client, jeton, 60, "initialize", {}) assert r.status_code == 200, r.text cols = _appel(client, jeton, 61, "list_collections") assert any(c["name"] == "Projet Alpha" for c in cols) r = _rpc(client, None, 62, "tools/list") assert r.status_code == 401, r.text def test_mcp_info_get(client, jeton): r = client.get("/mcp") assert r.status_code == 200 infos = r.json() assert infos["read_only"] is False assert "create_page" in infos["tools"] assert "update_page" in infos["tools"] assert "create_collection" in infos["tools"]