7.80.0 — Émission d'événements vers le journal du service d'agents (Lot 2)

tests/test_evenements_journal.py

@@ -0,0 +1,282 @@
+"""Émission d'événements vers le journal du service d'agents
+(Fusion, Lot 2 — contrat ACC-0.2).
+
+Un faux récepteur local capture les POST signés HMAC et les vérifie
+comme le fait le service (journal.verifier_signature) ; on prouve
+aussi l'inactivité sans clé et la non-régression métier quand le
+récepteur est en panne. Surfaces couvertes : API de session (board
+et collections), API publique v2 et MCP (acteur « agent:mcp »).
+"""
+
+import hashlib
+import hmac
+import json
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+import pytest
+
+from app.db import get_conn
+
+CLE = "cle-flowdeck-de-test"
+
+
+class _Recepteur(BaseHTTPRequestHandler):
+    recus = []
+
+    def do_POST(self):
+        n = int(self.headers.get("Content-Length") or 0)
+        corps = self.rfile.read(n)
+        type(self).recus.append(
+            {"headers": {k.lower(): v for k, v in self.headers.items()},
+             "corps": corps})
+        self.send_response(201)
+        self.end_headers()
+
+    def log_message(self, *a):
+        pass
+
+
+def _demarrer(monkeypatch):
+    _Recepteur.recus = []
+    serveur = HTTPServer(("127.0.0.1", 0), _Recepteur)
+    port = serveur.server_address[1]
+    threading.Thread(target=serveur.serve_forever, daemon=True).start()
+    monkeypatch.setenv("AGENT_EVENEMENTS_URL",
+                       f"http://127.0.0.1:{port}/api/v1/evenements")
+    monkeypatch.setenv("AGENT_EVENEMENTS_CLE", CLE)
+    return serveur
+
+
+def _attendre(nombre, delai=6.0):
+    """Attend `nombre` événements ; vérifie la signature HMAC de
+    chacun (comme le service) et retourne la liste des enveloppes."""
+    fin = time.time() + delai
+    while time.time() < fin and len(_Recepteur.recus) < nombre:
+        time.sleep(0.05)
+    assert len(_Recepteur.recus) >= nombre, \
+        f"{len(_Recepteur.recus)} événement(s) reçu(s), {nombre} attendus"
+    envs = []
+    for recu in _Recepteur.recus:
+        sig = recu["headers"].get("x-draco-signature", "")
+        attendu = hmac.new(CLE.encode(), recu["corps"],
+                           hashlib.sha256).hexdigest()
+        assert hmac.compare_digest(sig.lower(), attendu), \
+            "signature HMAC invalide"
+        env = json.loads(recu["corps"].decode("utf-8"))
+        assert env["source_app"] == "flowdeck"
+        assert env["id"] and env["occurred_at"]
+        envs.append(env)
+    return envs
+
+
+def _par_type(envs, type_):
+    return [e for e in envs if e["type"] == type_]
+
+
+def test_enveloppe_et_signature():
+    from app.services import agent_evenements
+
+    env, corps = agent_evenements.construire_enveloppe(
+        "flowdeck.page.created", "page:42", "Page de test",
+        url="/board/pages/42", actor="user")
+    assert env["source_app"] == "flowdeck"
+    assert env["domain"] == "pro"
+    assert env["actor"] == "user"
+    assert env["object"] == {"ref": "page:42", "label": "Page de test",
+                             "url": "/board/pages/42"}
+    assert json.loads(corps.decode("utf-8")) == env
+    sig = hmac.new(CLE.encode(), corps, hashlib.sha256).hexdigest()
+    assert len(sig) == 64
+
+
+def test_cycle_de_vie_api_session(client, monkeypatch):
+    serveur = _demarrer(monkeypatch)
+    try:
+        # Page d'espace de travail (board) : création, màj, suppression.
+        r = client.post("/board/api/pages", params={"title": "Page journal"})
+        assert r.status_code == 200, r.text
+        page_id = r.json()["id"]
+        r = client.put(f"/board/api/pages/{page_id}",
+                       params={"title": "Page journal (éditée)"})
+        assert r.status_code == 200, r.text
+        r = client.delete(f"/board/api/pages/{page_id}")
+        assert r.status_code == 200, r.text
+
+        # Collection puis sa page : création, màj, suppression.
+        r = client.post("/db/api", json={"name": "Collection journal"})
+        assert r.status_code == 200, r.text
+        cid = r.json()["id"]
+        r = client.post(f"/db/{cid}/pages/api",
+                        json={"title": "Ligne journal"})
+        assert r.status_code == 200, r.text
+        ligne_id = r.json()["id"]
+        r = client.put(f"/db/pages/{ligne_id}/api",
+                       json={"title": "Ligne journal (éditée)"})
+        assert r.status_code == 200, r.text
+        r = client.delete(f"/db/pages/{ligne_id}/api")
+        assert r.status_code == 200, r.text
+
+        envs = _attendre(7)
+
+        crees = _par_type(envs, "flowdeck.page.created")
+        assert len(crees) == 2
+        urls = {e["object"]["url"] for e in crees}
+        assert f"/board/pages/{page_id}" in urls
+        assert f"/db/{cid}/view/table" in urls
+        board = next(e for e in crees
+                     if e["object"]["url"] == f"/board/pages/{page_id}")
+        assert board["object"]["ref"] == f"page:{page_id}"
+        assert board["object"]["label"] == "Page journal"
+        assert board["actor"] == "user"
+
+        maj = _par_type(envs, "flowdeck.page.updated")
+        assert len(maj) == 2
+        labels = {e["object"]["label"] for e in maj}
+        assert labels == {"Page journal (éditée)", "Ligne journal (éditée)"}
+
+        suppr = _par_type(envs, "flowdeck.page.deleted")
+        assert len(suppr) == 2
+        refs = {e["object"]["ref"] for e in suppr}
+        assert refs == {f"page:{page_id}", f"page:{ligne_id}"}
+
+        coll = _par_type(envs, "flowdeck.collection.created")
+        assert len(coll) == 1
+        assert coll[0]["object"] == {
+            "ref": f"collection:{cid}", "label": "Collection journal",
+            "url": f"/db/{cid}/view/table"}
+        assert coll[0]["actor"] == "user"
+    finally:
+        serveur.shutdown()
+
+
+def test_api_publique_v2(client, monkeypatch):
+    serveur = _demarrer(monkeypatch)
+    try:
+        r = client.post("/auth/register", json={
+            "email": "[email protected]", "password": "secret123",
+            "name": "journal"})
+        assert r.status_code == 200, r.text
+        tok = client.post("/api/v1/token").json()["token"]
+        h1 = {"Authorization": f"Bearer {tok}"}
+        r = client.post("/api/v2/tokens",
+                        json={"name": "ci", "scopes": "read,write"},
+                        headers=h1)
+        assert r.status_code == 200, r.text
+        h = {"Authorization": f"Bearer {r.json()['token']}"}
+
+        r = client.post("/api/v2/collections",
+                        json={"name": "Collection v2"}, headers=h)
+        assert r.status_code == 201, r.text
+        cid = r.json()["id"]
+        r = client.post(f"/api/v2/collections/{cid}/pages",
+                        json={"title": "Page v2"}, headers=h)
+        assert r.status_code == 201, r.text
+        pid = r.json()["id"]
+        r = client.patch(f"/api/v2/pages/{pid}",
+                         json={"title": "Page v2 (éditée)"}, headers=h)
+        assert r.status_code == 200, r.text
+        r = client.delete(f"/api/v2/pages/{pid}", headers=h)
+        assert r.status_code == 200, r.text
+
+        envs = _attendre(4)
+        types = [e["type"] for e in envs]
+        assert types.count("flowdeck.collection.created") == 1
+        assert types.count("flowdeck.page.created") == 1
+        assert types.count("flowdeck.page.updated") == 1
+        assert types.count("flowdeck.page.deleted") == 1
+        assert all(e["actor"] == "user" for e in envs)
+        coll = _par_type(envs, "flowdeck.collection.created")[0]
+        assert coll["object"]["ref"] == f"collection:{cid}"
+        page = _par_type(envs, "flowdeck.page.created")[0]
+        assert page["object"]["ref"] == f"page:{pid}"
+        assert page["object"]["url"] == f"/db/{cid}/view/table"
+    finally:
+        serveur.shutdown()
+
+
+@pytest.fixture()
+def jeton_mcp(client):
+    brut = "fd_test_token_journal"
+    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
+
+
+def _appel_mcp(client, jeton, rid, nom, args=None):
+    r = client.post(
+        "/mcp",
+        json={"jsonrpc": "2.0", "id": rid, "method": "tools/call",
+              "params": {"name": nom, "arguments": args or {}}},
+        headers={"Authorization": f"Bearer {jeton}"})
+    assert r.status_code == 200, r.text
+    return json.loads(r.json()["result"]["content"][0]["text"])
+
+
+def test_mcp_acteur_agent(client, monkeypatch, jeton_mcp):
+    serveur = _demarrer(monkeypatch)
+    try:
+        coll = _appel_mcp(client, jeton_mcp, 1, "create_collection",
+                          {"name": "Collection MCP"})
+        cid = coll["id"]
+        page = _appel_mcp(client, jeton_mcp, 2, "create_page",
+                          {"collection_id": cid, "title": "Page MCP"})
+        pid = page["id"]
+        _appel_mcp(client, jeton_mcp, 3, "update_page",
+                   {"id": pid, "title": "Page MCP (éditée)"})
+
+        envs = _attendre(3)
+        assert all(e["actor"] == "agent:mcp" for e in envs)
+        coll_env = _par_type(envs, "flowdeck.collection.created")
+        assert len(coll_env) == 1
+        assert coll_env[0]["object"] == {
+            "ref": f"collection:{cid}", "label": "Collection MCP",
+            "url": f"/db/{cid}/view/table"}
+        creee = _par_type(envs, "flowdeck.page.created")
+        assert len(creee) == 1
+        assert creee[0]["object"]["ref"] == f"page:{pid}"
+        assert creee[0]["object"]["label"] == "Page MCP"
+        maj = _par_type(envs, "flowdeck.page.updated")
+        assert len(maj) == 1
+        assert maj[0]["object"]["label"] == "Page MCP (éditée)"
+    finally:
+        serveur.shutdown()
+
+
+def test_inactif_sans_cle(client, monkeypatch):
+    serveur = _demarrer(monkeypatch)
+    monkeypatch.delenv("AGENT_EVENEMENTS_CLE")
+    try:
+        from app.services import agent_evenements
+
+        assert agent_evenements.actif() is False
+        assert agent_evenements.emettre(
+            "flowdeck.page.created", "page:1", "x") is False
+        r = client.post("/board/api/pages", params={"title": "Sans journal"})
+        assert r.status_code == 200, r.text
+        r = client.post("/db/api", json={"name": "Collection sans journal"})
+        assert r.status_code == 200, r.text
+        time.sleep(0.6)
+        assert _Recepteur.recus == []
+    finally:
+        serveur.shutdown()
+
+
+def test_recepteur_en_panne_ne_casse_rien(client, monkeypatch):
+    # Port fermé : connexion refusée dans le fil d'arrière-plan.
+    monkeypatch.setenv("AGENT_EVENEMENTS_URL",
+                       "http://127.0.0.1:59998/api/v1/evenements")
+    monkeypatch.setenv("AGENT_EVENEMENTS_CLE", CLE)
+    r = client.post("/board/api/pages", params={"title": "Malgré la panne"})
+    assert r.status_code == 200, r.text
+    page_id = r.json()["id"]
+    r = client.get(f"/board/api/pages/{page_id}")
+    assert r.status_code == 200, r.text
+    assert r.json()["title"] == "Malgré la panne"
+    r = client.post("/db/api", json={"name": "Collection malgré la panne"})
+    assert r.status_code == 200, r.text