"""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_workspaces", "get_workspace", "list_collections", "get_collection", "get_page", "search_pages", "list_repos", "create_page", "update_page", "create_collection"] def test_mcp_workspaces(client, jeton): with get_conn() as conn: conn.execute("INSERT OR IGNORE INTO users (id, login) VALUES (1, 'tester')") cur = conn.execute( "INSERT INTO workspaces (name, owner_id) VALUES ('Techno', 1)") wid = cur.lastrowid conn.execute( "INSERT INTO pages (workspace, workspace_id, title, parent_section) " "VALUES ('Techno', ?, 'Page racine', 'Private')", (wid,)) conn.execute( "INSERT INTO pages (workspace, workspace_id, title, parent_section, deleted_at) " "VALUES ('Techno', ?, 'Page corbeille', 'Private', '2026-10-11 10:00:00')", (wid,)) conn.commit() ws = _appel(client, jeton, 70, "list_workspaces") techno = [w for w in ws if w["name"] == "Techno"] assert techno and techno[0]["page_count"] == 1 # la corbeille ne compte pas detail = _appel(client, jeton, 71, "get_workspace", {"id": wid}) assert detail["workspace"]["name"] == "Techno" assert [p["title"] for p in detail["pages"]] == ["Page racine"] res = _appel(client, jeton, 72, "get_workspace", {"id": 99999}) assert "introuvable" in res["erreur"] 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_perimetre_du_jeton(client, jeton, donnees): """Régression 7.78.0 (cas réel de Bruno) : list_workspaces rendait les 52 workspaces de la base, tous comptes confondus. Chaque jeton ne voit que les workspaces de son propriétaire (ou ceux dont il est membre) et les collections rattachées à ce périmètre.""" autre = "fd_token_autre_compte" with get_conn() as conn: conn.execute("INSERT OR IGNORE INTO users (id, login) VALUES (2, 'autre')") conn.execute( "INSERT INTO api_tokens (user_id, name, token_hash, token_prefix, revoked) " "VALUES (2, 'autre', ?, ?, 0)", (hashlib.sha256(autre.encode()).hexdigest(), autre[:12])) cur = conn.execute( "INSERT INTO workspaces (name, owner_id) VALUES ('Espace de lautre', 2)") wid_autre = cur.lastrowid cur = conn.execute( "INSERT INTO collections (name, description, icon, workspace_id) " "VALUES ('Base de lautre', '', '📋', ?)", (wid_autre,)) base_autre = cur.lastrowid conn.commit() # Bruno (jeton principal) ne voit ni le workspace ni la base de l'autre. noms = [w["name"] for w in _appel(client, jeton, 80, "list_workspaces")] assert "Espace de lautre" not in noms res = _appel(client, jeton, 81, "get_workspace", {"id": wid_autre}) assert "introuvable" in res["erreur"] cols = [c["name"] for c in _appel(client, jeton, 82, "list_collections")] assert "Base de lautre" not in cols and "Projet Alpha" in cols res = _appel(client, jeton, 83, "get_collection", {"id": base_autre}) assert "introuvable" in res["erreur"] # Et l'autre ne voit pas le workspace de Bruno, mais voit le sien. noms2 = [w["name"] for w in _appel(client, autre, 84, "list_workspaces")] assert noms2 == ["Espace de lautre"] 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"]