"""Lot 2, vague 2 — émission d'événements de Kanbante vers le journal du service d'agents (contrat ACC-0.2). Couvre : l'enveloppe signée (HMAC recalculé côté test), l'inactivité sans clé, l'émission sur déplacement de carte (``kanbante.card.moved``) et sur entrée en Révision (``kanbante.card.review_requested``), et l'absence totale d'effet d'une panne du récepteur sur le métier. """ import hashlib import hmac import json import threading import time from datetime import datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import pytest from app import agent_journal, models from tests.conftest import csrf_from CLE = "cle-kanbante-de-test" class Recepteur: """Faux service d'agents : capture corps bruts et signatures.""" def __init__(self) -> None: self.recus: list[tuple[bytes, str]] = [] self._verrou = threading.Lock() recus = self.recus verrou = self._verrou class Mains(BaseHTTPRequestHandler): def log_message(self, *args): pass def do_POST(self): corps = self.rfile.read(int(self.headers.get("Content-Length", 0))) with verrou: recus.append((corps, self.headers.get("X-Draco-Signature", ""))) self.send_response(201) self.end_headers() self._serveur = ThreadingHTTPServer(("127.0.0.1", 0), Mains) threading.Thread(target=self._serveur.serve_forever, daemon=True).start() self.url = f"http://127.0.0.1:{self._serveur.server_address[1]}/api/v1/evenements" def enveloppes(self, type_: str | None = None, ref: str | None = None) -> list[dict]: with self._verrou: bruts = list(self.recus) envs = [json.loads(corps) for corps, _ in bruts] if type_ is not None: envs = [e for e in envs if e["type"] == type_] if ref is not None: envs = [e for e in envs if e["object"]["ref"] == ref] return envs def attendre(self, type_: str, ref: str, timeout: float = 5.0) -> dict: """Poll jusqu'à l'arrivée de l'événement voulu (envoi en arrière-plan : l'attente fait partie du contrat de test).""" limite = time.monotonic() + timeout while time.monotonic() < limite: trouves = self.enveloppes(type_, ref) if trouves: return trouves[0] time.sleep(0.05) raise AssertionError(f"Événement {type_} pour {ref} jamais reçu") @pytest.fixture def recepteur(): return Recepteur() @pytest.fixture def journal_actif(monkeypatch, recepteur): """Émission configurée vers le faux récepteur local.""" monkeypatch.setenv("AGENT_EVENEMENTS_CLE", CLE) monkeypatch.setenv("AGENT_EVENEMENTS_URL", recepteur.url) return recepteur def test_enveloppe_signee_verifiable(journal_actif): fil = agent_journal.emettre( "kanbante.card.moved", "issue:Projets/demo#1", "Tâche un", "http://gitea.test/Projets/demo/issues/1", actor="user", ) assert fil is not None fil.join(timeout=5) env = journal_actif.attendre("kanbante.card.moved", "issue:Projets/demo#1") corps, signature = journal_actif.recus[0] # La signature se revérifie indépendamment, comme le fait le service. assert signature == hmac.new(CLE.encode(), corps, hashlib.sha256).hexdigest() assert json.loads(corps) == env assert env["id"] assert env["source_app"] == "kanbante" assert env["domain"] == "pro" assert env["actor"] == "user" assert env["object"] == { "ref": "issue:Projets/demo#1", "label": "Tâche un", "url": "http://gitea.test/Projets/demo/issues/1", } # occurred_at : ISO 8601 lisible par le service. datetime.fromisoformat(env["occurred_at"]) def test_inactif_sans_cle(monkeypatch, recepteur): monkeypatch.delenv("AGENT_EVENEMENTS_CLE", raising=False) monkeypatch.setenv("AGENT_EVENEMENTS_URL", recepteur.url) assert agent_journal.actif() is False assert agent_journal.emettre( "kanbante.card.moved", "issue:Projets/demo#1", "Tâche un", "http://gitea.test/Projets/demo/issues/1", ) is None time.sleep(0.3) assert recepteur.recus == [] def _deplacer(client, numero: int, de: str, vers: str): token = csrf_from(client) r = client.post( "/api/move", data={ "repo_full_name": "Projets/demo", "issue_number": numero, "to_column": vers, "from_column": de, "csrf_token": token, }, ) assert r.status_code == 200, r.text assert r.json()["ok"] is True return r def test_deplacement_emet_moved(client, journal_actif): _deplacer(client, 1, "backlog", "in_progress") env = journal_actif.attendre("kanbante.card.moved", "issue:Projets/demo#1") assert env["actor"] == "user" assert env["object"]["label"] == "Tâche un" assert env["object"]["url"] == "http://gitea.test/Projets/demo/issues/1" # Un déplacement hors Révision n'émet pas de review_requested. time.sleep(0.3) assert journal_actif.enveloppes( "kanbante.card.review_requested", "issue:Projets/demo#1" ) == [] def test_entree_en_revision_emet_review_requested(client, journal_actif): _deplacer(client, 1, "backlog", "review") journal_actif.attendre("kanbante.card.moved", "issue:Projets/demo#1") env = journal_actif.attendre( "kanbante.card.review_requested", "issue:Projets/demo#1" ) assert env["object"]["label"] == "Tâche un" assert env["actor"] == "user" # Sortir de Révision ne redemande pas de révision. _deplacer(client, 1, "review", "done") journal_actif.attendre("kanbante.card.moved", "issue:Projets/demo#1") time.sleep(0.3) assert len(journal_actif.enveloppes( "kanbante.card.review_requested", "issue:Projets/demo#1" )) == 1 def test_panne_du_recepteur_sans_effet_sur_le_deplacement(client, monkeypatch): # Récepteur mort (port 1 fermé) mais émission configurée. monkeypatch.setenv("AGENT_EVENEMENTS_CLE", CLE) monkeypatch.setenv("AGENT_EVENEMENTS_URL", "http://127.0.0.1:1/api/v1/evenements") _deplacer(client, 1, "backlog", "in_progress") state = models.get_board_state("Projets/demo") assert [c["issue_number"] for c in state["in_progress"]] == [1] # L'échec d'envoi, lui, reste silencieux et borné dans le temps. fil = agent_journal.emettre( "kanbante.card.moved", "issue:Projets/demo#1", "Tâche un", "http://gitea.test/Projets/demo/issues/1", ) assert fil is not None fil.join(timeout=10) assert not fil.is_alive()