kanbante/tests/test_agent_journal.py

"""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()