coffre(obsigate#22): tests client de distribution (faux endpoint, valeurs fictives)

tests/test_coffre_client.py

@@ -0,0 +1,227 @@
+"""Tests du client de distribution coffre ObsiGate (Phase 4, issue ObsiGate #22).
+
+Aucune instance ObsiGate réelle n'est contactée : le « faux endpoint » est
+un ``httpx.MockTransport`` local, et toutes les valeurs distribuées dans
+ces tests sont fictives (``TEST-VALUE-XXXX``), jamais des secrets réels.
+"""
+
+import hashlib
+import hmac
+import json
+
+import httpx
+import pytest
+
+import app.main as main
+from app import models
+from app.coffre_client import (
+    CoffreAccesRefuse,
+    CoffreConfig,
+    CoffreContexteManquant,
+    CoffreIndisponible,
+    CoffreNonConfigure,
+    CoffreReferenceInvalide,
+    distribuer,
+    distribuer_async,
+    est_reference,
+    parse_reference,
+    usage_distribue,
+)
+
+# Configuration fictive, réservée aux tests.
+CONFIG_TEST = CoffreConfig(
+    base_url="https://obsigate.test",
+    jeton_consommateur="TEST-CONSUMER-TOKEN",
+)
+REFERENCE_TEST = "coffre:TEST-SLUG/TEST-NOM"
+VALEUR_TEST = "TEST-VALUE-XXXX"
+
+
+def _transport_faux(statut=200, corps=None, verifier=None):
+    def handler(request: httpx.Request) -> httpx.Response:
+        if verifier is not None:
+            verifier(request)
+        return httpx.Response(statut, json=corps if corps is not None else {})
+
+    return httpx.MockTransport(handler)
+
+
+def _verifier_requete(request: httpx.Request):
+    assert request.url.path == "/api/coffres/distribuer"
+    assert request.headers["Authorization"] == "Bearer TEST-CONSUMER-TOKEN"
+    corps = json.loads(request.content.decode())
+    assert corps["reference"] == REFERENCE_TEST
+    assert corps["contexte_type"] == "webhook"
+    assert corps["contexte_id"] == "gitea"
+
+
+@pytest.mark.parametrize(
+    "reference, attendu",
+    [
+        ("coffre:personnel/gitea", True),
+        ("coffre:TEST-SLUG/TEST-NOM", True),
+        ("coffre:a/b/c", False),  # un seul « / » après le préfixe
+        ("coffre:/nom", False),
+        ("coffre:slug/", False),
+        ("coffre:slug", False),
+        ("gitea", False),
+        ("", False),
+        (None, False),
+    ],
+)
+def test_reference_validation(reference, attendu):
+    assert est_reference(reference) is attendu
+    if attendu:
+        assert parse_reference(reference)[0]
+    else:
+        with pytest.raises(CoffreReferenceInvalide):
+            parse_reference(reference or "")
+
+
+def test_distribuer_succes_sync():
+    transport = _transport_faux(
+        200,
+        {"reference": REFERENCE_TEST, "valeur": VALEUR_TEST, "version": 1},
+        verifier=_verifier_requete,
+    )
+    assert (
+        distribuer(REFERENCE_TEST, "webhook", "gitea", config=CONFIG_TEST, transport=transport)
+        == VALEUR_TEST
+    )
+
+
+async def test_distribuer_succes_async():
+    transport = _transport_faux(
+        200,
+        {"reference": REFERENCE_TEST, "valeur": VALEUR_TEST, "version": 1},
+        verifier=_verifier_requete,
+    )
+    assert (
+        await distribuer_async(
+            REFERENCE_TEST, "webhook", "gitea", config=CONFIG_TEST, transport=transport
+        )
+        == VALEUR_TEST
+    )
+
+
+@pytest.mark.parametrize(
+    "statut, detail, message_attendu",
+    [
+        (403, "octroi_revoque", "ré-accorder"),
+        (403, "octroi_expire", "renouveler"),
+        (403, "octroi_absent", "octroi"),
+        (401, "consommateur_revoque", "ré-accorder"),
+        (401, "consommateur_invalide", "invalide"),
+        (410, "secret_expire", "expiré"),
+    ],
+)
+def test_distribuer_refus_sans_repli(statut, detail, message_attendu):
+    transport = _transport_faux(statut, {"detail": detail})
+    with pytest.raises(CoffreAccesRefuse) as excinfo:
+        distribuer(REFERENCE_TEST, "webhook", "gitea", config=CONFIG_TEST, transport=transport)
+    assert message_attendu in str(excinfo.value)
+    assert VALEUR_TEST not in str(excinfo.value)
+
+
+def test_distribuer_indisponible_503_et_reseau():
+    transport = _transport_faux(503, {"detail": "kek_absente"})
+    with pytest.raises(CoffreIndisponible):
+        distribuer(REFERENCE_TEST, "webhook", "gitea", config=CONFIG_TEST, transport=transport)
+
+    def _panne(request: httpx.Request):
+        raise httpx.ConnectError("faux ObsiGate injoignable", request=request)
+
+    with pytest.raises(CoffreIndisponible):
+        distribuer(
+            REFERENCE_TEST, "webhook", "gitea",
+            config=CONFIG_TEST, transport=httpx.MockTransport(_panne),
+        )
+
+
+def test_distribuer_sans_valeur_malgre_200():
+    transport = _transport_faux(200, {"reference": REFERENCE_TEST})
+    with pytest.raises(CoffreIndisponible):
+        distribuer(REFERENCE_TEST, "webhook", "gitea", config=CONFIG_TEST, transport=transport)
+
+
+def test_distribuer_reference_invalide_sans_appel():
+    def _interdit(request: httpx.Request):
+        raise AssertionError("aucun appel HTTP ne doit partir")
+
+    with pytest.raises(CoffreReferenceInvalide):
+        distribuer("valeur-brute", "webhook", "gitea",
+                   config=CONFIG_TEST, transport=httpx.MockTransport(_interdit))
+
+
+def test_distribuer_contexte_manquant_sans_appel():
+    with pytest.raises(CoffreContexteManquant):
+        distribuer(REFERENCE_TEST, "", "gitea", config=CONFIG_TEST)
+
+
+def test_distribuer_non_configure(monkeypatch):
+    monkeypatch.delenv("KANBANTE_OBSIGATE_URL", raising=False)
+    monkeypatch.delenv("KANBANTE_COFFRE_CONSOMMATEUR_TOKEN", raising=False)
+    with pytest.raises(CoffreNonConfigure):
+        distribuer(REFERENCE_TEST, "webhook", "gitea")
+
+
+def test_usage_distribue_ne_conserve_que_le_resultat():
+    transport = _transport_faux(200, {"valeur": VALEUR_TEST})
+    # Le bloc ne sert qu'à l'usage : ici, dériver un HMAC, comme le ferait
+    # la vérification d'un webhook. Seul le résultat de l'usage est
+    # conservé après le bloc, jamais la valeur distribuée.
+    with usage_distribue(
+        REFERENCE_TEST, "webhook", "gitea",
+        config=CONFIG_TEST, transport=transport,
+    ) as valeur:
+        signature = hmac.new(valeur.encode(), b"corps-test", hashlib.sha256).hexdigest()
+    assert len(signature) == 64
+
+
+def test_webhook_reference_octroi_revoque_sans_repli(client, monkeypatch):
+    """Référence webhook + octroi révoqué -> 401 explicite, sans repli."""
+    models.create_connection(
+        "coffre-webhook", "gitea", "Coffre webhook", "https://gitea.test",
+        token_encrypted="", account_login="", webhook_secret=REFERENCE_TEST,
+    )
+
+    async def _refuse(*args, **kwargs):
+        raise CoffreAccesRefuse("Accès révoqué dans ObsiGate pour cette référence.")
+
+    monkeypatch.setattr(main.coffre_client, "distribuer_async", _refuse)
+    reponse = client.post(
+        "/webhooks/coffre-webhook",
+        content=b"{}",
+        headers={"x-gitea-signature": "0" * 64},
+    )
+    assert reponse.status_code == 401
+    assert "ObsiGate" in reponse.json()["error"]
+
+
+def test_webhook_reference_distribuee_puis_oubliee(client, monkeypatch):
+    """Référence webhook distribuée -> HMAC vérifié, valeur non persistée."""
+    models.create_connection(
+        "coffre-webhook-ok", "gitea", "Coffre webhook OK", "https://gitea.test",
+        token_encrypted="", account_login="", webhook_secret=REFERENCE_TEST,
+    )
+
+    async def _distribue(reference, contexte_type, contexte_id):
+        assert reference == REFERENCE_TEST
+        assert contexte_type == "webhook"
+        assert contexte_id == "coffre-webhook-ok"
+        return VALEUR_TEST
+
+    monkeypatch.setattr(main.coffre_client, "distribuer_async", _distribue)
+    corps = b"{}"
+    signature = hmac.new(VALEUR_TEST.encode(), corps, hashlib.sha256).hexdigest()
+    reponse = client.post(
+        "/webhooks/coffre-webhook-ok",
+        content=corps,
+        headers={"x-gitea-signature": signature},
+    )
+    assert reponse.status_code == 200
+    # La valeur distribuée n'a jamais été persistée : la ligne stocke
+    # toujours la référence, jamais la valeur fictive.
+    ligne = models.get_connection_by_slug("coffre-webhook-ok")
+    assert ligne["webhook_secret"] == REFERENCE_TEST
+    assert VALEUR_TEST not in (ligne["webhook_secret"] or "")