flowdeck/tests/test_fusion_mcp_server.py

"""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_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"]