kanbante/tests/test_amorcage_pont.py

"""Fusion #145 (0.46.5) — adoption CONDITIONNÉE du compte d'amorçage.

Avant ce correctif, sur une instance fraîche, le premier visiteur
du pont draco adoptait le compte administrateur d'amorçage
(« liaison au premier passage ») — un simple utilisateur de la
suite (« dan », audit « compte partout » du 2026-11-06) l'a fait
en pile de contrôle. Désormais, l'adoption n'a lieu que si :

  (a) le ``sub`` entrant est « administrateur » pour kanbante au
      registre des rôles du Hub (repli propre : Hub muet = rôle
      inconnu = PAS d'adoption) ;
  (b) ou l'adoption est initiée depuis une session native déjà
      authentifiée (liaison explicite par l'utilisateur connecté).

Couvert ici, sur instance RÉELLEMENT fraîche (base SQLite
temporaire propre à chaque test : au démarrage, seul le compte
d'amorçage créé depuis l'environnement existe, sans liaison) :

  (i)   sub simple utilisateur → compte membre JIT propre,
        amorçage intact et toujours sans liaison ;
  (ii)  sub administrateur au registre → adoption de l'amorçage ;
  (ii-bis) Hub muet (rôle inconnu) → pas d'adoption, JIT membre ;
  (iii) liaison existante (cas de production) → résolution
        inchangée, quel que soit le rôle du registre ;
  (iv)  session native authentifiée → liaison explicite de
        l'amorçage possible (pont et résolution directe) ;

plus la part Kanbante de la fusion #144 : l'échec du formulaire
local suggère la voie « Se connecter avec ObsiGate (SSO) ».

Le faux Hub suit le patron de tests/test_hub_roles.py : l'appel
``httpx.get`` du client du registre est simulé, les rôles sont
lus dans le cookie relayé.
"""

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 accounts, draco, hub_roles, models

SECRET = "secret-amorcage-145"

SUB_SIMPLE = "fus145-simple"
SUB_ADMIN = "fus145-admin"
SUB_INCONNU = "fus145-inconnu"
SUB_LIE = "fus145-deja-lie"
SUB_NATIF = "fus145-natif"
SUB_NATIF_DIRECT = "fus145-natif-direct"


def _jeton(sub, nom="Visiteur 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):
        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``, lus dans le cookie relayé."""
    etat = {"roles": {}, "mode": "ok"}

    def faux_get(url, headers=None, timeout=None):
        if etat["mode"] == "panne":
            raise httpx.ConnectError("Hub injoignable (simulation)")
        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


@pytest.fixture()
def instance_fraiche(monkeypatch, tmp_path, fake_gitea, faux_hub):
    """Instance neuve : base temporaire, amorçage seul et non lié.

    Au démarrage (TestClient), ``ensure_bootstrap_admin`` crée
    l'unique compte « proprio » (admin) depuis l'environnement —
    c'est exactement l'état de l'audit : un seul compte, aucune
    liaison draco.
    """
    monkeypatch.setenv("KANBANTE_DB_PATH", str(tmp_path / "fraiche.db"))
    monkeypatch.setenv("KANBANTE_AUTH_USER", "proprio")
    monkeypatch.setenv("KANBANTE_AUTH_PASSWORD", "mdp-amorce")
    monkeypatch.setenv("DRACO_SESSION_SECRET", SECRET)
    monkeypatch.setattr(draco, "_refresh_revocations", lambda: None)
    draco._revoked_cache.clear()
    with TestClient(main.app) as client:
        users = models.list_users()
        assert len(users) == 1, users
        assert users[0]["username"] == "proprio"
        assert users[0]["role"] == "admin"
        assert not users[0].get("draco_sub")
        yield client, faux_hub, users[0]
    draco._revoked_cache.clear()


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


def _profil_username(client) -> str:
    r = client.get("/profile")
    assert r.status_code == 200
    return r.text.split("page-sub\">", 1)[1].split(" ·", 1)[0].strip()


def _csrf_via_login(client) -> str:
    client.get("/login")
    return client.cookies.get("kanbante_csrf") or ""


# ── (i) Simple utilisateur : JIT membre, amorçage intact ─────

def test_simple_utilisateur_jit_membre_amorcage_intact(instance_fraiche):
    client, hub, amorce = instance_fraiche
    hub["roles"][SUB_SIMPLE] = "utilisateur"
    r = _pont(client, SUB_SIMPLE, "Dan Simple")
    assert r.status_code == 303 and r.headers["location"] == "/"
    assert "kanbante_session" in r.headers.get("set-cookie", "")
    # L'amorçage est intact : toujours là, admin, TOUJOURS sans liaison.
    apres = models.get_user(amorce["id"])
    assert apres["role"] == "admin"
    assert not apres.get("draco_sub")
    # Le visiteur a reçu SON compte : membre, distinct, lié à son sub.
    cree = models.get_user_by_draco(SUB_SIMPLE)
    assert cree is not None and cree["id"] != amorce["id"]
    assert cree["role"] == "member"
    assert models.count_users() == 2
    assert _profil_username(client) == cree["username"]


# ── (ii) Administrateur au registre : adoption ───────────────

def test_administrateur_registre_adopte_amorcage(instance_fraiche):
    client, hub, amorce = instance_fraiche
    hub["roles"][SUB_ADMIN] = "administrateur"
    r = _pont(client, SUB_ADMIN, "Chef Suite")
    assert r.status_code == 303 and r.headers["location"] == "/"
    # Aucun compte créé : le sub s'est lié au compte d'amorçage.
    assert models.count_users() == 1
    lie = models.get_user_by_draco(SUB_ADMIN)
    assert lie is not None and lie["id"] == amorce["id"]
    assert lie["role"] == "admin"
    assert _profil_username(client) == "proprio"


def test_hub_muet_pas_d_adoption(instance_fraiche):
    """Repli propre : rôle inconnu (Hub injoignable) = pas d'adoption."""
    client, hub, amorce = instance_fraiche
    hub["mode"] = "panne"
    r = _pont(client, SUB_INCONNU, "Inconnu Exemple")
    assert r.status_code == 303 and r.headers["location"] == "/"
    assert not models.get_user(amorce["id"]).get("draco_sub")
    cree = models.get_user_by_draco(SUB_INCONNU)
    assert cree is not None and cree["role"] == "member"
    assert cree["id"] != amorce["id"]


# ── (iii) Liaison existante : résolution inchangée ───────────

def test_liaison_existante_resolution_inchangee(instance_fraiche):
    client, hub, amorce = instance_fraiche
    # Cas de production : l'amorçage est DÉJÀ lié (à Bruno).
    models.update_user(amorce["id"], draco_sub=SUB_LIE)
    # Même avec un rôle moindre au registre, la correspondance
    # existante désigne son compte, comme avant le correctif.
    hub["roles"][SUB_LIE] = "utilisateur"
    claims = {"sub": SUB_LIE, "name": "Bruno"}
    resolu = accounts.resolve_draco_user(claims)
    assert resolu is not None and resolu["id"] == amorce["id"]
    r = _pont(client, SUB_LIE, "Bruno")
    assert r.status_code == 303 and r.headers["location"] == "/"
    assert models.count_users() == 1
    assert _profil_username(client) == "proprio"


# ── (iv) Session native authentifiée : liaison explicite ─────

def test_session_native_lie_explicitement_via_le_pont(instance_fraiche):
    client, hub, amorce = instance_fraiche
    hub["roles"][SUB_NATIF] = "utilisateur"
    # Connexion native de l'amorçage (mot de passe du compte).
    token = _csrf_via_login(client)
    r = client.post(
        "/login",
        data={"username": "proprio", "password": "mdp-amorce",
              "next": "/", "csrf_token": token},
        follow_redirects=False,
    )
    assert r.status_code == 303
    assert "kanbante_session" in r.headers.get("set-cookie", "")
    # Le passage du pont, session native ouverte, lie le sub libre
    # au compte de la session — sans rôle administrateur au registre.
    r = _pont(client, SUB_NATIF, "Proprio Lui-Même")
    assert r.status_code == 303
    lie = models.get_user_by_draco(SUB_NATIF)
    assert lie is not None and lie["id"] == amorce["id"]
    assert models.count_users() == 1


def test_resolution_directe_avec_session_native(instance_fraiche):
    _client, _hub, amorce = instance_fraiche
    claims = {"sub": SUB_NATIF_DIRECT, "name": "Proprio"}
    # Sans drapeau : aucune adoption (résolution « à froid »).
    assert accounts.resolve_draco_user(claims) is None
    assert not models.get_user(amorce["id"]).get("draco_sub")
    # Avec le drapeau « session native authentifiée » : adoption.
    resolu = accounts.resolve_draco_user(
        claims, liaison_session_native=True
    )
    assert resolu is not None and resolu["id"] == amorce["id"]
    assert models.get_user_by_draco(SUB_NATIF_DIRECT)["id"] == amorce["id"]


# ── Fusion #144 : l'échec du formulaire suggère le SSO ───────

def test_echec_login_identifiant_sans_compte_local_suggere_sso(instance_fraiche):
    client, _hub, _amorce = instance_fraiche
    token = _csrf_via_login(client)
    r = client.post(
        "/login",
        data={"username": "compte-du-hub", "password": "peu-importe",
              "next": "/", "csrf_token": token},
        follow_redirects=False,
    )
    assert r.status_code == 401
    assert "aucun compte local" in r.text
    assert "Se connecter avec ObsiGate (SSO)" in r.text
    assert "kanbante_session" not in r.headers.get("set-cookie", "")


def test_echec_login_mauvais_mot_de_passe_suggere_sso(instance_fraiche):
    client, _hub, _amorce = instance_fraiche
    token = _csrf_via_login(client)
    r = client.post(
        "/login",
        data={"username": "proprio", "password": "mauvais",
              "next": "/", "csrf_token": token},
        follow_redirects=False,
    )
    assert r.status_code == 401
    assert "Identifiants incorrects" in r.text
    assert "Se connecter avec ObsiGate (SSO)" in r.text