ObsiGate/tests/test_hub_reglages.py

# tests/test_hub_reglages.py
# Hub HUB-21 (propagation) — Réglages généraux du Hub consommés par
# ObsiGate : client (parsing de la forme figée, cache 60 s, repli
# intégral), bannière injectée côté serveur (page principale et page
# /recherche, texte échappé) et drapeau ``recherche_partout`` rendu
# en attribut de la page /recherche. Patron posé par l'hôte pilote
# TikTak 0.61.22. ``domaine_defaut`` n'est pas appliqué : ObsiGate
# n'a pas de notion de domaine propre (voir backend/hub_reglages.py).

import json
import time
import urllib.error

import pytest

from backend import hub_reglages


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


def _reponse(corps=None, statut=200):
    """Fausse réponse urlopen portant un payload du contrat D6(a)."""
    if corps is None:
        corps = {"recherche_partout": True, "banniere": "",
                 "domaine_defaut": None, "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_consultation(monkeypatch, reglages, appels=None):
    """Remplace la consultation HTTP par des réglages fixes."""
    def _faux(url):
        if appels is not None:
            appels.append(url)
        return reglages
    monkeypatch.setattr(hub_reglages, "_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({"recherche_partout": False,
                         "banniere": "Maintenance dimanche",
                         "domaine_defaut": "famille", "ttl_secondes": 60})

    monkeypatch.setattr(hub_reglages.urllib.request, "urlopen", _faux_urlopen)
    reglages = hub_reglages.reglages_effectifs()
    assert reglages == {"recherche_partout": False,
                        "banniere": "Maintenance dimanche",
                        "domaine_defaut": "famille"}
    assert vu["url"] == "https://hub-test.dracodev.net/api/reglages/effectifs"
    assert vu["cookie"] is None            # endpoint public : aucun cookie
    assert vu["accept"] == "application/json"
    assert vu["timeout"] == 2              # délai du contrat


@pytest.mark.parametrize("corps", [
    {"recherche_partout": "oui", "banniere": "", "domaine_defaut": None},
    {"recherche_partout": True, "banniere": 42, "domaine_defaut": None},
    {"recherche_partout": True, "banniere": "", "domaine_defaut": "travail"},
    {"banniere": "", "domaine_defaut": None},
    ["pas", "un", "dict"],
])
def test_consultation_forme_inattendue_repli(monkeypatch, corps):
    monkeypatch.setattr(hub_reglages.urllib.request, "urlopen",
                        lambda requete, timeout=None: _reponse(corps))
    assert hub_reglages.reglages_effectifs() == hub_reglages.DEFAUTS


def test_consultation_erreur_http_repli(monkeypatch):
    monkeypatch.setattr(hub_reglages.urllib.request, "urlopen",
                        lambda requete, timeout=None: _reponse(statut=500))
    assert hub_reglages.reglages_effectifs() == hub_reglages.DEFAUTS
    assert hub_reglages.banniere() == ""
    assert hub_reglages.recherche_partout() is True


def test_consultation_hub_injoignable_repli(monkeypatch):
    def _leve(requete, timeout=None):
        raise urllib.error.URLError("injoignable")
    monkeypatch.setattr(hub_reglages.urllib.request, "urlopen", _leve)
    assert hub_reglages.reglages_effectifs() == hub_reglages.DEFAUTS


def test_consultation_json_illisible_repli(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_reglages.urllib.request, "urlopen",
                        lambda requete, timeout=None: _Resp())
    assert hub_reglages.reglages_effectifs() == hub_reglages.DEFAUTS


# ── Client : cache 60 s, ancien cache, pas de rafale ────────────────

def test_cache_60_s(monkeypatch):
    appels = []
    _patch_consultation(monkeypatch, {
        "recherche_partout": False, "banniere": "Annonce A",
        "domaine_defaut": None}, appels)
    assert hub_reglages.banniere() == "Annonce A"
    assert hub_reglages.recherche_partout() is False
    assert len(appels) == 1                       # cache 60 s
    hub_reglages._cache_at = time.monotonic() - 61
    assert hub_reglages.banniere() == "Annonce A"
    assert len(appels) == 2                       # expiration → rappel


def test_echec_ancien_cache_sans_rafale(monkeypatch):
    appels = []
    _patch_consultation(monkeypatch, {
        "recherche_partout": True, "banniere": "Annonce A",
        "domaine_defaut": None}, appels)
    assert hub_reglages.banniere() == "Annonce A"
    hub_reglages._cache_at = time.monotonic() - 61
    _patch_consultation(monkeypatch, None, appels)   # Hub tombe
    assert hub_reglages.banniere() == "Annonce A"    # ancien cache
    assert len(appels) == 2                          # pas de rafale
    assert hub_reglages.banniere() == "Annonce A"
    assert len(appels) == 2


# ── Intégration : bannière et drapeau dans les pages servies ────────

@pytest.fixture()
def hub_banniere(monkeypatch):
    etat = {"reglages": None}
    monkeypatch.setattr(hub_reglages, "_consulter_hub",
                        lambda url: etat["reglages"])
    hub_reglages.reinitialiser_cache()
    return etat


def test_banniere_presente_et_echappee_index(client, hub_banniere):
    hub_banniere["reglages"] = {
        "recherche_partout": True,
        "banniere": "Maintenance dimanche <b>gras</b> & cie",
        "domaine_defaut": None}
    r = client.get("/")
    assert r.status_code == 200
    assert 'id="banniere-hub"' in r.text
    assert "Maintenance dimanche &lt;b&gt;gras&lt;/b&gt; &amp; cie" in r.text
    assert "obsigate_banniere_masquee" in r.text


def test_banniere_absente_si_vide_ou_hub_muet(client, hub_banniere):
    hub_banniere["reglages"] = {
        "recherche_partout": True, "banniere": "", "domaine_defaut": "pro"}
    r = client.get("/")
    assert r.status_code == 200
    assert "banniere-hub" not in r.text
    hub_banniere["reglages"] = None
    hub_reglages.reinitialiser_cache()
    r = client.get("/")
    assert r.status_code == 200
    assert "banniere-hub" not in r.text


def test_recherche_banniere_et_drapeau(client, hub_banniere):
    hub_banniere["reglages"] = {
        "recherche_partout": False, "banniere": "Annonce recherche",
        "domaine_defaut": None}
    r = client.get("/recherche")
    assert r.status_code == 200
    assert 'data-recherche-partout="false"' in r.text
    assert 'id="banniere-hub"' in r.text
    assert "Annonce recherche" in r.text
    hub_banniere["reglages"] = {
        "recherche_partout": True, "banniere": "", "domaine_defaut": None}
    hub_reglages.reinitialiser_cache()
    r = client.get("/recherche")
    assert 'data-recherche-partout="true"' in r.text
    assert "banniere-hub" not in r.text
    hub_banniere["reglages"] = None              # Hub muet → défaut actif
    hub_reglages.reinitialiser_cache()
    r = client.get("/recherche")
    assert 'data-recherche-partout="true"' in r.text