"""HUB-19 (0.46.2) — rôles du Hub propagés à Kanbante (contrat D3(a)).

Portage du pilote ShaarIt 0.7.0. Couvert ici :

- client ``app.hub_roles`` : appel ``GET {hub}/api/roles/kanbante``
  (Cookie complet relayé, ``Accept: application/json``), URL du
  Hub résolue comme la barre (explicite > hôte de la requête >
  défaut officiel), cache 60 s par ``sub``, horodatage posé avant
  l'appel (échec = ancien cache conservé, pas de rafale), tout
  échec = rôle inconnu (``None``) ;
- pont ``/auth/draco`` : la consultation précède la résolution et
  la création JIT — rôle connu ``aucun`` → refus explicite 403,
  **aucune session ouverte, aucun compte créé**, aucune liaison à
  une session native préexistante ;
- ``/admin`` : session établie par draco + rôle connu ≠
  ``administrateur`` → refusé malgré le rôle local admin ; rôle
  ``administrateur`` → accepté ; rôle inconnu → repli local ;
  session **native** (même admin, même cookie draco présent) →
  règles locales inchangées.
"""

import base64
import hashlib
import hmac
import json
import time

import httpx
import pytest
from fastapi.testclient import TestClient

import app.main as main
from app import draco, hub_roles, models

SECRET = "secret-hub19-partage"

SUBS_TEST = (
    "hub19-refuse", "hub19-membre", "hub19-chef", "hub19-inconnu",
    "hub19-sans-liaison",
)


def _jeton(sub, nom="Alice Exemple", secret=SECRET):
    def b64(d: bytes) -> str:
        return base64.urlsafe_b64encode(d).rstrip(b"=").decode()

    header = b64(json.dumps({"alg": "HS256", "typ": "JWT"}).encode())
    payload = b64(json.dumps({
        "sub": sub, "name": nom, "iat": int(time.time()),
        "exp": int(time.time()) + 3600, "jti": f"jti-{sub}",
    }).encode())
    sig = hmac.new(secret.encode(), f"{header}.{payload}".encode(),
                   hashlib.sha256).digest()
    return f"{header}.{payload}.{b64(sig)}"


def _sub_depuis_cookie(cookie: str) -> str:
    for morceau in (cookie or "").split(";"):
        nom, _, valeur = morceau.strip().partition("=")
        if nom == "draco_session" and valeur.count(".") == 2:
            segment = valeur.split(".")[1]
            segment += "=" * (-len(segment) % 4)
            try:
                return json.loads(base64.urlsafe_b64decode(segment))["sub"]
            except Exception:
                return ""
    return ""


class _Reponse:
    def __init__(self, status_code=200, donnees=None):
        self.status_code = status_code
        self._donnees = donnees

    def json(self):
        if isinstance(self._donnees, Exception):
            raise self._donnees
        return self._donnees


@pytest.fixture(autouse=True)
def _cache_propre():
    hub_roles._reset_cache()
    yield
    hub_roles._reset_cache()


@pytest.fixture()
def faux_hub(monkeypatch):
    """Faux registre du Hub : rôles par ``sub``, lu dans le cookie relayé."""
    etat = {"roles": {}, "appels": [], "mode": "ok"}

    def faux_get(url, headers=None, timeout=None):
        etat["appels"].append(
            {"url": url, "headers": dict(headers or {}), "timeout": timeout}
        )
        if etat["mode"] == "panne":
            raise httpx.ConnectError("Hub injoignable (simulation)")
        if etat["mode"] == "erreur500":
            return _Reponse(500, {"detail": "erreur"})
        sub = _sub_depuis_cookie((headers or {}).get("Cookie", ""))
        role = etat["roles"].get(sub, "utilisateur")
        return _Reponse(200, {
            "app": "kanbante", "sub": sub, "role": role,
            "explicite": True, "ttl_secondes": 60,
        })

    monkeypatch.setattr(httpx, "get", faux_get)
    return etat


# ── Client : contrat d'appel et résolution d'URL ─────────────

def test_client_appel_contrat_et_cache(faux_hub):
    faux_hub["roles"]["sub-a"] = "administrateur"
    cookie = f"draco_session={_jeton('sub-a')}; autre=1"
    assert hub_roles.role_pour(cookie, "sub-a") == "administrateur"
    # Deuxième appel dans le TTL : servi par le cache, aucun appel.
    assert hub_roles.role_pour(cookie, "sub-a") == "administrateur"
    assert len(faux_hub["appels"]) == 1
    appel = faux_hub["appels"][0]
    assert appel["url"] == "https://hub.dracodev.net/api/roles/kanbante"
    assert appel["headers"]["Cookie"] == cookie
    assert appel["headers"]["Accept"] == "application/json"
    assert appel["timeout"] == 2.0


def test_client_url_hub_par_hote_dev(faux_hub):
    hub_roles.role_pour(
        f"draco_session={_jeton('sub-b')}", "sub-b",
        hote="kanbante-dev.dracodev.net",
    )
    assert faux_hub["appels"][0]["url"] == (
        "https://hub-dev.dracodev.net/api/roles/kanbante"
    )


def test_client_url_hub_explicite_prioritaire(faux_hub, monkeypatch):
    monkeypatch.setenv("DRACO_HUB_URL", "https://hub.exemple.test/")
    hub_roles.role_pour(
        f"draco_session={_jeton('sub-c')}", "sub-c",
        hote="kanbante-dev.dracodev.net",
    )
    assert faux_hub["appels"][0]["url"] == (
        "https://hub.exemple.test/api/roles/kanbante"
    )


def test_client_ttl_expire_reconsulte(faux_hub):
    cookie = f"draco_session={_jeton('sub-d')}"
    faux_hub["roles"]["sub-d"] = "utilisateur"
    assert hub_roles.role_pour(cookie, "sub-d") == "utilisateur"
    hub_roles._cache_fetched_at -= hub_roles.CACHE_TTL + 1
    faux_hub["roles"]["sub-d"] = "administrateur"
    assert hub_roles.role_pour(cookie, "sub-d") == "administrateur"
    assert len(faux_hub["appels"]) == 2


def test_client_echec_inconnu_et_ancien_cache_conserve(faux_hub):
    cookie = f"draco_session={_jeton('sub-e')}"
    faux_hub["roles"]["sub-e"] = "utilisateur"
    assert hub_roles.role_pour(cookie, "sub-e") == "utilisateur"
    # TTL expiré, Hub en panne : l'appel tenté rend « inconnu »…
    hub_roles._cache_fetched_at -= hub_roles.CACHE_TTL + 1
    faux_hub["mode"] = "panne"
    assert hub_roles.role_pour(cookie, "sub-e") is None
    appels = len(faux_hub["appels"])
    # …mais l'horodatage posé avant l'appel garde l'ancien cache :
    # l'appel suivant (dans le nouveau TTL) le ressert sans rafale.
    assert hub_roles.role_pour(cookie, "sub-e") == "utilisateur"
    assert len(faux_hub["appels"]) == appels


def test_client_reponses_illisibles_inconnues(monkeypatch):
    def faux_get(url, headers=None, timeout=None):
        return _Reponse(200, {"role": "super-utilisateur"})

    monkeypatch.setattr(httpx, "get", faux_get)
    assert hub_roles.role_pour("c=1", "sub-f") is None
    hub_roles._reset_cache()

    def faux_get_403(url, headers=None, timeout=None):
        return _Reponse(403, {"role": "aucun"})

    monkeypatch.setattr(httpx, "get", faux_get_403)
    assert hub_roles.role_pour("c=1", "sub-f") is None


# ── Intégration : pont draco et garde /admin ─────────────────

def _nettoyer():
    with models.get_db() as conn:
        ids = [
            r[0] for r in conn.execute(
                "SELECT id FROM users WHERE draco_sub IN ({})".format(
                    ",".join("?" * len(SUBS_TEST))
                ),
                SUBS_TEST,
            ).fetchall()
        ]
        for uid in ids:
            conn.execute("DELETE FROM api_tokens WHERE user_id = ?", (uid,))
            conn.execute(
                "DELETE FROM workspace_members WHERE user_id = ?", (uid,)
            )
            conn.execute("DELETE FROM users WHERE id = ?", (uid,))


@pytest.fixture()
def client_hub19(monkeypatch, fake_gitea, faux_hub):
    monkeypatch.setenv("KANBANTE_AUTH_USER", "bruno")
    monkeypatch.setenv("KANBANTE_AUTH_PASSWORD", "motdepasse")
    monkeypatch.setenv("DRACO_SESSION_SECRET", SECRET)
    monkeypatch.setattr(draco, "_refresh_revocations", lambda: None)
    draco._revoked_cache.clear()
    _nettoyer()
    with TestClient(main.app) as client:
        admin = next(u for u in models.list_users() if u["role"] == "admin")
        sub_avant = admin.get("draco_sub")
        try:
            yield client, faux_hub, admin
        finally:
            models.update_user(admin["id"], draco_sub=sub_avant)
    _nettoyer()
    draco._revoked_cache.clear()


def _neutraliser_amorcage(admin) -> bool:
    """Vrai si l'amorçage pur a été neutralisé (témoin lié à l'admin).

    Même patron que tests/test_draco_correspondance.py : les
    scénarios JIT exigent que l'unique admin sans liaison ne
    capture pas le premier sub de passage.
    """
    users = models.list_users()
    if len(users) == 1 and users[0]["role"] == "admin" \
            and not users[0].get("draco_sub"):
        models.update_user(users[0]["id"], draco_sub="hub19-temoin-amorce")
        return True
    return False


def _pont(client, sub, nom="Alice Exemple"):
    client.cookies.set("draco_session", _jeton(sub, nom))
    return client.get("/auth/draco", follow_redirects=False)


def test_pont_role_aucun_refuse_sans_creation(client_hub19):
    client, hub, _admin = client_hub19
    hub["roles"]["hub19-refuse"] = "aucun"
    avant = models.count_users()
    r = _pont(client, "hub19-refuse", "Refusée Exemple")
    assert r.status_code == 403
    assert "Accès refusé" in r.text
    assert "« aucun »" in r.text
    # Aucune session Kanbante ouverte par draco…
    assert "kanbante_session" not in r.headers.get("set-cookie", "")
    # …et aucun compte JIT créé pour ce sub.
    assert models.count_users() == avant
    assert models.get_user_by_draco("hub19-refuse") is None


def test_pont_role_aucun_ne_lie_pas_une_session_native(client_hub19):
    client, hub, admin = client_hub19
    hub["roles"]["hub19-sans-liaison"] = "aucun"
    # Session native préexistante (format historique, sans marqueur).
    client.cookies.set(
        "kanbante_session", main._make_session(admin["username"])
    )
    r = _pont(client, "hub19-sans-liaison")
    assert r.status_code == 403
    # La consultation a précédé la liaison : rien n'a été lié.
    assert models.get_user_by_draco("hub19-sans-liaison") is None
    # La session native elle-même n'est pas touchée.
    assert client.get("/").status_code == 200


def test_pont_role_utilisateur_ouvre_session_jit(client_hub19):
    client, hub, admin = client_hub19
    hub["roles"]["hub19-membre"] = "utilisateur"
    pur = _neutraliser_amorcage(admin)
    try:
        r = _pont(client, "hub19-membre", "Membre Exemple")
        assert r.status_code == 303 and r.headers["location"] == "/"
        assert "kanbante_session" in r.headers.get("set-cookie", "")
        cree = models.get_user_by_draco("hub19-membre")
        assert cree is not None and cree["role"] == "member"
        assert client.get("/").status_code == 200
    finally:
        if pur:
            models.update_user(admin["id"], draco_sub=None)


def test_pont_role_inconnu_comportement_actuel(client_hub19):
    client, hub, admin = client_hub19
    hub["mode"] = "panne"
    pur = _neutraliser_amorcage(admin)
    try:
        r = _pont(client, "hub19-inconnu", "Inconnu Exemple")
        assert r.status_code == 303 and r.headers["location"] == "/"
        assert models.get_user_by_draco("hub19-inconnu") is not None
    finally:
        if pur:
            models.update_user(admin["id"], draco_sub=None)


def _admin_lie_au_pont(client, hub, admin, role_hub):
    """Lie le compte admin local au sub chef, puis ouvre sa session par le pont."""
    models.update_user(admin["id"], draco_sub="hub19-chef")
    hub["roles"]["hub19-chef"] = role_hub
    hub_roles._reset_cache()
    r = _pont(client, "hub19-chef", "Chef Exemple")
    assert r.status_code == 303 and r.headers["location"] == "/"
    return r


def test_admin_session_draco_exige_administrateur_du_registre(client_hub19):
    client, hub, admin = client_hub19
    _admin_lie_au_pont(client, hub, admin, "utilisateur")
    # Rôle local admin, mais registre connu « utilisateur » : refusé.
    r = client.get("/admin")
    assert r.status_code == 403
    # Registre passé à « administrateur » : accepté.
    hub["roles"]["hub19-chef"] = "administrateur"
    hub_roles._reset_cache()
    r = client.get("/admin")
    assert r.status_code == 200


def test_admin_session_draco_role_inconnu_repli_local(client_hub19):
    client, hub, admin = client_hub19
    _admin_lie_au_pont(client, hub, admin, "administrateur")
    assert client.get("/admin").status_code == 200
    # Hub en panne et cache expiré : rôle inconnu → règles locales.
    hub["mode"] = "panne"
    hub_roles._cache_fetched_at -= hub_roles.CACHE_TTL + 1
    # L'appel de garde rend « inconnu » (l'ancien cache ne ressert
    # qu'aux appels suivants, patron du pilote) : le repli local joue.
    r = client.get("/admin")
    assert r.status_code == 200


def test_admin_session_native_non_soumise_au_registre(client_hub19):
    client, hub, admin = client_hub19
    models.update_user(admin["id"], draco_sub="hub19-chef")
    hub["roles"]["hub19-chef"] = "utilisateur"
    # Session NATIVE de l'admin (mot de passe / format historique) :
    # même avec un cookie draco présent et un rôle registre connu
    # « utilisateur », /admin reste régi par les règles locales.
    client.cookies.set(
        "kanbante_session", main._make_session(admin["username"])
    )
    client.cookies.set("draco_session", _jeton("hub19-chef", "Chef Exemple"))
    r = client.get("/admin")
    assert r.status_code == 200


def test_marqueur_session_draco_signe():
    """Le marqueur « établie par draco » est couvert par la signature."""
    from starlette.requests import Request

    def requete(valeur):
        scope = {
            "type": "http", "method": "GET", "path": "/",
            "headers": [(b"cookie", f"kanbante_session={valeur}".encode())],
        }
        return Request(scope)

    natif = main._make_session("bruno")
    assert main._session_info(requete(natif)) == ("bruno", False)
    via = main._make_session("bruno", via_draco=True)
    assert main._session_info(requete(via)) == ("bruno", True)
    # Marqueur ajouté à la main sur un jeton natif : signature fausse.
    pirate = natif.rsplit(".", 1)[0] + ".draco." + natif.rsplit(".", 1)[1]
    assert main._session_info(requete(pirate)) is None