# tests/test_hub_roles.py
# Hub HUB-19 (propagation, #228) — Rôles par application : client du
# registre du Hub (parsing, cache 60 s par sub, repli systématique) et
# effets sur le chemin draco d'ObsiGate (refus « aucun », garde
# administrateur des routes /api/auth/admin/* pour le canal draco).
# Contrat posé par l'hôte pilote ShaarIt 0.7.0 (décision D3(a)).

import io
import json
import time
import urllib.error

import pytest

from backend import draco_session, hub_roles

USER = {"id": "admin-1", "username": "admin", "display_name": "Admin"}
USER_SIMPLE = {"id": "user-1", "username": "normaluser", "display_name": "normal"}


@pytest.fixture(autouse=True)
def _hub_env(monkeypatch):
    """Cache vierge + Hub de test explicite pour chaque test."""
    hub_roles.reinitialiser_cache()
    monkeypatch.setenv("DRACO_HUB_URL", "https://hub-test.dracodev.net")
    yield
    hub_roles.reinitialiser_cache()


def _reponse(role, statut=200, corps=None):
    """Fausse réponse urlopen portant un payload du contrat D3(a)."""
    if corps is None:
        corps = {"app": "obsigate", "sub": "admin-1", "role": role,
                 "explicite": True, "ttl_secondes": 60}
    charge = json.dumps(corps).encode("utf-8")

    class _Resp:
        status = statut

        def read(self):
            return charge

        def __enter__(self):
            return self

        def __exit__(self, *exc):
            return False

    return _Resp()


def _patch_hub(monkeypatch, role, appels):
    """Remplace la consultation HTTP par un rôle fixe (compte les appels)."""
    def _faux(url, cookie):
        appels.append((url, cookie))
        return role
    monkeypatch.setattr(hub_roles, "_consulter_hub", _faux)


# ── Client : parsing du contrat ─────────────────────────────────────────

def test_consultation_parse_le_contrat(monkeypatch):
    vu = {}

    def _faux_urlopen(requete, timeout=None):
        vu["url"] = requete.full_url
        vu["cookie"] = requete.get_header("Cookie")
        vu["accept"] = requete.get_header("Accept")
        vu["timeout"] = timeout
        return _reponse("administrateur")

    monkeypatch.setattr(hub_roles.urllib.request, "urlopen", _faux_urlopen)
    role = hub_roles.role_pour("admin-1", "draco_session=jeton; autre=1")
    assert role == "administrateur"
    assert vu["url"] == "https://hub-test.dracodev.net/api/roles/obsigate"
    assert vu["cookie"] == "draco_session=jeton; autre=1"   # Cookie complet relayé
    assert vu["accept"] == "application/json"
    assert vu["timeout"] == 2                               # délai du contrat


@pytest.mark.parametrize("role", ["aucun", "utilisateur", "administrateur"])
def test_consultation_roles_reconnus(monkeypatch, role):
    monkeypatch.setattr(hub_roles.urllib.request, "urlopen",
                        lambda requete, timeout=None: _reponse(role))
    assert hub_roles.role_pour("admin-1", "c=1") == role


def test_consultation_role_hors_contrat_inconnu(monkeypatch):
    monkeypatch.setattr(hub_roles.urllib.request, "urlopen",
                        lambda requete, timeout=None: _reponse("super-admin"))
    assert hub_roles.role_pour("admin-1", "c=1") is None


def test_consultation_json_illisible_inconnu(monkeypatch):
    class _Resp:
        status = 200

        def read(self):
            return b"pas du json"

        def __enter__(self):
            return self

        def __exit__(self, *exc):
            return False

    monkeypatch.setattr(hub_roles.urllib.request, "urlopen",
                        lambda requete, timeout=None: _Resp())
    assert hub_roles.role_pour("admin-1", "c=1") is None


def test_consultation_erreur_http_inconnu(monkeypatch):
    def _erreur(requete, timeout=None):
        raise urllib.error.HTTPError(requete.full_url, 500, "boom", {}, io.BytesIO(b""))

    monkeypatch.setattr(hub_roles.urllib.request, "urlopen", _erreur)
    assert hub_roles.role_pour("admin-1", "c=1") is None


def test_consultation_hub_injoignable_inconnu(monkeypatch):
    def _panne(requete, timeout=None):
        raise urllib.error.URLError("injoignable")

    monkeypatch.setattr(hub_roles.urllib.request, "urlopen", _panne)
    assert hub_roles.role_pour("admin-1", "c=1") is None


def test_url_hub_derivee_par_hote_soeur(monkeypatch):
    # Sans variable explicite : dérivation depuis l'hôte de l'agent
    # (agent-dev → hub-dev), comme la barre commune (#223).
    monkeypatch.delenv("DRACO_HUB_URL")
    monkeypatch.setenv("DRACO_AGENT_URL", "https://agent-dev.dracodev.net")
    appels = []
    _patch_hub(monkeypatch, "utilisateur", appels)
    assert hub_roles.role_pour("admin-1", "c=1") == "utilisateur"
    assert appels[0][0] == "https://hub-dev.dracodev.net/api/roles/obsigate"


def test_url_hub_defaut_officiel(monkeypatch):
    monkeypatch.delenv("DRACO_HUB_URL")
    monkeypatch.delenv("DRACO_AGENT_URL", raising=False)
    monkeypatch.delenv("AGENT_EVENEMENTS_URL", raising=False)
    appels = []
    _patch_hub(monkeypatch, "utilisateur", appels)
    hub_roles.role_pour("admin-1", "c=1")
    assert appels[0][0] == "https://hub.dracodev.net/api/roles/obsigate"


# ── Client : cache 60 s par sub, horodatage avant l'appel ──────────────

def test_cache_par_sub_et_ttl(monkeypatch):
    appels = []
    _patch_hub(monkeypatch, "utilisateur", appels)
    assert hub_roles.role_pour("admin-1", "c=1") == "utilisateur"
    assert hub_roles.role_pour("admin-1", "c=1") == "utilisateur"
    assert len(appels) == 1                              # 2e lecture : cache
    assert hub_roles.role_pour("user-1", "c=2") == "utilisateur"
    assert len(appels) == 2                              # autre sub : nouvel appel
    # TTL expiré → nouvel appel.
    hub_roles._cache_at = time.monotonic() - hub_roles.CACHE_TTL_SECONDS - 1
    assert hub_roles.role_pour("admin-1", "c=1") == "utilisateur"
    assert len(appels) == 3


def test_echec_conserve_ancien_cache_sans_rafale(monkeypatch):
    appels = []
    _patch_hub(monkeypatch, "administrateur", appels)
    assert hub_roles.role_pour("admin-1", "c=1") == "administrateur"
    assert len(appels) == 1
    # Le Hub tombe ; le TTL expire : un appel est tenté, échoue, et
    # l'horodatage posé avant l'appel empêche la rafale — l'ancien
    # rôle connu n'est ni effacé ni réinterrogé aussitôt.
    hub_roles._cache_at = time.monotonic() - hub_roles.CACHE_TTL_SECONDS - 1
    _patch_hub(monkeypatch, None, appels)
    assert hub_roles.role_pour("user-1", "c=2") is None
    assert len(appels) == 2
    # Dans la fenêtre qui suit l'échec, l'ancien cache sert toujours.
    assert hub_roles.role_pour("admin-1", "c=1") == "administrateur"
    assert len(appels) == 2


def test_sub_vide_aucun_appel(monkeypatch):
    appels = []
    _patch_hub(monkeypatch, "utilisateur", appels)
    assert hub_roles.role_pour("", "c=1") is None
    assert appels == []


# ── Intégration : chemin draco (admin_client, secret draco posé) ──────

@pytest.fixture()
def _draco_env(monkeypatch, tmp_path):
    monkeypatch.setenv("DRACO_SESSION_SECRET", "secret-de-test")
    monkeypatch.setattr(draco_session, "REVOKED_FILE",
                        tmp_path / "draco_revoked.json")
    monkeypatch.setattr(draco_session, "ISSUED_FILE",
                        tmp_path / "draco_issued.json")
    draco_session._revoked = None
    draco_session._issued = None
    yield
    draco_session._revoked = None
    draco_session._issued = None


def _pose_cookie_draco(client, user):
    token, _jti, _exp = draco_session.issue_token(user)
    client.cookies.set("draco_session", token)
    return token


def test_draco_aucun_refuse_session_explicite(admin_client, _draco_env, monkeypatch):
    appels = []
    _patch_hub(monkeypatch, "aucun", appels)
    _pose_cookie_draco(admin_client, USER)
    r = admin_client.get("/api/auth/me")
    assert r.status_code == 403
    assert "niveau « aucun »" in r.json()["detail"]
    assert "registre des rôles du Hub" in r.json()["detail"]
    assert appels and appels[0][1].find("draco_session=") >= 0  # Cookie relayé


def test_draco_aucun_mais_connexion_native_intacte(admin_client, _draco_env, monkeypatch):
    # L'authentification native (émettrice, accès de secours) n'est
    # jamais touchée par le registre : le login par mot de passe du
    # même compte passe, même avec un niveau « aucun » au Hub.
    _patch_hub(monkeypatch, "aucun", [])
    r = admin_client.post("/api/auth/login",
                          json={"username": "admin", "password": "chab30"})
    assert r.status_code == 200, r.text
    # Et la session native ainsi établie ouvre /me (canal natif, le
    # rôle du Hub ne s'applique pas).
    r = admin_client.get("/api/auth/me")
    assert r.status_code == 200, r.text
    assert r.json()["username"] == "admin"


def test_draco_role_inconnu_comportement_actuel(admin_client, _draco_env, monkeypatch):
    _patch_hub(monkeypatch, None, [])                    # Hub injoignable
    _pose_cookie_draco(admin_client, USER)
    r = admin_client.get("/api/auth/me")
    assert r.status_code == 200, r.text
    assert r.json()["username"] == "admin"
    # Canal draco + rôle inconnu : les routes admin gardent les
    # règles actuelles (le rôle local admin suffit).
    r = admin_client.get("/api/auth/admin/users")
    assert r.status_code == 200, r.text


def test_draco_utilisateur_admin_local_refuse_admin(admin_client, _draco_env, monkeypatch):
    # Compte local admin authentifié par draco, mais niveau Hub
    # « utilisateur » : /api/auth/admin/* exige « administrateur ».
    _patch_hub(monkeypatch, "utilisateur", [])
    _pose_cookie_draco(admin_client, USER)
    r = admin_client.get("/api/auth/me")
    assert r.status_code == 200, r.text                  # session draco OK
    r = admin_client.get("/api/auth/admin/users")
    assert r.status_code == 403
    assert "administrateur" in r.json()["detail"]


def test_draco_administrateur_hub_admin_ok(admin_client, _draco_env, monkeypatch):
    _patch_hub(monkeypatch, "administrateur", [])
    _pose_cookie_draco(admin_client, USER)
    r = admin_client.get("/api/auth/admin/users")
    assert r.status_code == 200, r.text
    assert any(u["username"] == "admin" for u in r.json())


def test_draco_utilisateur_simple_admin_refuse_comme_avant(admin_client, _draco_env, monkeypatch):
    # Compte local non admin, niveau Hub « administrateur » : le
    # registre n'élève aucun privilège local — refus par la règle
    # actuelle (rôle local), comme avant HUB-19.
    _patch_hub(monkeypatch, "administrateur", [])
    _pose_cookie_draco(admin_client, USER_SIMPLE)
    r = admin_client.get("/api/auth/admin/users")
    assert r.status_code == 403
    assert r.json()["detail"] == "Accès admin requis"


def test_natif_admin_sans_draco_jamais_bloque_par_hub(admin_client, _draco_env, monkeypatch):
    # Session native (login) + niveau Hub « utilisateur » : le canal
    # n'est pas draco, la garde du registre ne s'applique pas.
    _patch_hub(monkeypatch, "utilisateur", [])
    r = admin_client.post("/api/auth/login",
                          json={"username": "admin", "password": "chab30"})
    assert r.status_code == 200, r.text
    # Le jeton natif valide prime : get_current_user ne passe pas par
    # le repli draco, le dict utilisateur n'a ni _draco_sub ni
    # _hub_role — la garde du registre ne peut pas s'appliquer.
    r = admin_client.get("/api/auth/admin/users")
    assert r.status_code == 200, r.text