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