228 lines
7.8 KiB
Python
228 lines
7.8 KiB
Python
"""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 "")
|