Files
ObsiGate/tests/test_agent_journal.py
T
Spark f434f30286
CI / lint (push) Successful in 2m53s
CI / security (push) Failing after 1m46s
CI / test (push) Successful in 4m57s
CI / build (push) Successful in 1m36s
CI / e2e (push) Successful in 17m40s
feat: émission d'événements vers le journal du service d'agents #220
2026-10-22 19:58:25 +00:00

303 lines
11 KiB
Python

"""Émission d'événements vers le journal du service d'agents (#220,
Fusion Lot 2 vague 5 — contrat ACC-0.2).
Un faux récepteur local capture les POST signés HMAC et les vérifie
comme le fait le service d'agents ; on prouve aussi l'inactivité sans
clé, la non-régression métier quand le récepteur est en panne, et le
garde-fou central : le watcher / l'indexation n'émettent JAMAIS
(seules les écritures explicites via l'API ou les outils le font).
"""
import asyncio
import hashlib
import hmac
import json
import threading
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
from types import SimpleNamespace
import pytest
CLE = "cle-obsigate-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(types, delai=6.0):
"""Attend que chaque type demandé soit arrivé ; retourne
{type: enveloppe} — la signature HMAC de CHAQUE envoi est
revérifiée ici, comme le fait le service."""
vus = {}
fin = time.time() + delai
while time.time() < fin and not all(t in vus for t in types):
for recu in list(_Recepteur.recus):
env = json.loads(recu["corps"].decode("utf-8"))
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"
vus.setdefault(env["type"], env)
if not all(t in vus for t in types):
time.sleep(0.05)
manquants = [t for t in types if t not in vus]
assert not manquants, f"événements jamais reçus : {manquants}"
return vus
# ── Enveloppe ────────────────────────────────────────────────────────
def test_enveloppe_et_signature(monkeypatch):
from backend import agent_journal
monkeypatch.setenv("AGENT_EVENEMENTS_CLE", CLE)
env, corps = agent_journal.construire_enveloppe(
agent_journal.TYPE_CREE, "TestVault", "Journal/nouvelle.md")
assert env["source_app"] == "obsigate"
assert env["domain"] == "pro" # repli : aucun domaine modélisé (Q-A)
assert env["actor"] == "user"
assert env["object"] == {
"ref": "fichier:TestVault/Journal/nouvelle.md",
"label": "nouvelle.md",
"url": "/#file=TestVault:Journal/nouvelle.md",
}
assert json.loads(corps.decode("utf-8")) == env
assert env["id"] and env["occurred_at"]
sig = hmac.new(CLE.encode(), corps, hashlib.sha256).hexdigest()
assert len(sig) == 64
def test_domaine_par_vault(monkeypatch):
from backend import agent_journal
monkeypatch.setenv("AGENT_EVENEMENTS_DOMAINES",
json.dumps({"TestVault": "famille"}))
assert agent_journal.domaine_du_vault("TestVault") == "famille"
assert agent_journal.domaine_du_vault("AutreVault") == "pro"
# ── Écritures REST (acteur « user ») ─────────────────────────────────
def test_cycle_de_vie_rest(client, monkeypatch):
serveur = _demarrer(monkeypatch)
try:
r = client.post("/api/file/TestVault", json={
"path": "Journal/nouvelle.md", "content": "# Nouvelle\n"})
assert r.status_code == 200
r = client.put("/api/file/TestVault/save",
params={"path": "Journal/nouvelle.md"},
json={"content": "# Nouvelle v2\n"})
assert r.status_code == 200
r = client.patch("/api/file/TestVault", json={
"path": "Journal/nouvelle.md", "new_name": "renommee.md"})
assert r.status_code == 200
r = client.post("/api/move/TestVault", json={
"source_path": "Journal/renommee.md", "destination_dir": "Projets"})
assert r.status_code == 200
r = client.delete("/api/file/TestVault",
params={"path": "Projets/renommee.md"})
assert r.status_code == 200
vus = _attendre([
"obsigate.note.created", "obsigate.note.edited",
"obsigate.note.moved", "obsigate.note.deleted"])
env = vus["obsigate.note.created"]
assert env["actor"] == "user"
assert env["object"]["ref"] == "fichier:TestVault/Journal/nouvelle.md"
assert env["object"]["label"] == "nouvelle.md"
assert env["object"]["url"] == "/#file=TestVault:Journal/nouvelle.md"
env = vus["obsigate.note.edited"]
assert env["object"]["ref"] == "fichier:TestVault/Journal/nouvelle.md"
# moved : émis deux fois (renommage puis déplacement). Les
# envois partent en fils d'arrière-plan — l'ordre d'arrivée
# n'est pas garanti, on compare l'ensemble des références.
refs_deplacees = {
json.loads(recu["corps"].decode("utf-8"))["object"]["ref"]
for recu in _Recepteur.recus
if json.loads(recu["corps"].decode("utf-8"))["type"]
== "obsigate.note.moved"}
assert refs_deplacees == {
"fichier:TestVault/Journal/renommee.md",
"fichier:TestVault/Projets/renommee.md"}
env = vus["obsigate.note.deleted"]
assert env["object"]["ref"] == "fichier:TestVault/Projets/renommee.md"
finally:
serveur.shutdown()
# ── Écritures par les outils / MCP (acteur « agent:mcp ») ───────────
@pytest.fixture
def vault_outils(tmp_path, monkeypatch):
"""Vault minimal pour la couche outils (patron test_document_tools)."""
import backend.indexer
vault_dir = tmp_path / "Vault"
vault_dir.mkdir()
monkeypatch.setitem(
backend.indexer.index, "Vault",
{"name": "Vault", "path": str(vault_dir), "config": {}})
return vault_dir
def test_outils_mcp_emettent_agent_mcp(vault_outils, monkeypatch):
from backend.tools.api import ToolContext, ToolMode, call_tool
serveur = _demarrer(monkeypatch)
try:
ctx = ToolContext(
user={"username": "tester", "role": "admin", "vaults": ["*"]},
mode=ToolMode.MCP, audit_enabled=False)
out = call_tool("create_file", ctx, {
"vault": "Vault", "path": "par-mcp.md", "content": "# MCP\n",
}, confirm=True)
assert out.ok
out = call_tool("edit_file", ctx, {
"vault": "Vault", "path": "par-mcp.md", "content": "# MCP v2\n",
}, confirm=True)
assert out.ok
out = call_tool("rename_file", ctx, {
"vault": "Vault", "path": "par-mcp.md", "new_name": "renomme-mcp.md",
}, confirm=True)
assert out.ok
out = call_tool("delete_file", ctx, {
"vault": "Vault", "path": "renomme-mcp.md",
}, confirm=True)
assert out.ok
vus = _attendre([
"obsigate.note.created", "obsigate.note.edited",
"obsigate.note.moved", "obsigate.note.deleted"])
for env in vus.values():
assert env["actor"] == "agent:mcp"
assert vus["obsigate.note.created"]["object"]["ref"] == \
"fichier:Vault/par-mcp.md"
assert vus["obsigate.note.moved"]["object"]["ref"] == \
"fichier:Vault/renomme-mcp.md"
finally:
serveur.shutdown()
# ── Inactivité sans clé ──────────────────────────────────────────────
def test_inactif_sans_cle(client, monkeypatch):
serveur = _demarrer(monkeypatch)
monkeypatch.delenv("AGENT_EVENEMENTS_CLE")
try:
from backend import agent_journal
assert agent_journal.actif() is False
assert agent_journal.emettre(
agent_journal.TYPE_CREE, "TestVault", "x.md") is False
r = client.post("/api/file/TestVault", json={
"path": "sans-journal.md", "content": "# Rien\n"})
assert r.status_code == 200
time.sleep(0.6)
assert _Recepteur.recus == []
finally:
serveur.shutdown()
# ── Garde-fou : watcher et indexation n'émettent jamais ─────────────
def test_watcher_et_indexation_sans_evenement(client, test_vault_dir,
monkeypatch):
"""Une écriture disque externe (éditeur, synchro de fichiers…)
traitée par le watcher puis l'indexeur ne produit AUCUN
événement : seuls les appels explicites à l'API émettent."""
from pathlib import Path
appels = []
monkeypatch.setattr(
"backend.agent_journal.emettre",
lambda *a, **k: appels.append((a, k)) or False)
monkeypatch.setenv("AGENT_EVENEMENTS_CLE", CLE)
cible = Path(test_vault_dir) / "externe.md"
cible.write_text("# Écrit hors API\n", encoding="utf-8")
# Chemin watcher : les gestionnaires d'événements filesystem.
from backend.watcher import VaultEventHandler
boucle = asyncio.new_event_loop()
try:
file_attente: asyncio.Queue = asyncio.Queue()
handler = VaultEventHandler("TestVault", file_attente, boucle)
evt = SimpleNamespace(is_directory=False, src_path=str(cible),
dest_path=None)
handler.on_created(evt)
handler.on_modified(evt)
# Chemin indexation : la même fonction que le watcher appelle
# après debounce pour intégrer le fichier à l'index.
from backend.indexer import update_single_file
boucle.run_until_complete(update_single_file("TestVault", "externe.md"))
finally:
boucle.close()
assert appels == []
# ── Panne du récepteur : le métier n'est jamais affecté ─────────────
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:59999/api/v1/evenements")
monkeypatch.setenv("AGENT_EVENEMENTS_CLE", CLE)
r = client.post("/api/file/TestVault", json={
"path": "malgre-panne.md", "content": "# OK\n"})
assert r.status_code == 200
r = client.put("/api/file/TestVault/save",
params={"path": "malgre-panne.md"},
json={"content": "# OK v2\n"})
assert r.status_code == 200
r = client.get("/api/file/TestVault", params={"path": "malgre-panne.md"})
assert r.status_code == 200
assert "OK v2" in r.json()["html"] # le GET renvoie le Markdown rendu