Spark
133156580c
list_workspaces/get_workspace filtrés par user_workspace_ids ; collections filtrées par workspace effectif ; hors périmètre = introuvable, lectures et écritures. Test deux comptes.
240 lines
10 KiB
Python
240 lines
10 KiB
Python
"""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"]
|