flowdeck/tests/test_hub_reglages.py

"""HUB-21 — réglages généraux du Hub consommés par Flowdeck.

Propagation du pilote TikTak 0.61.22 : client serveur
``app.services.hub_reglages`` (cache 60 s, horodatage posé avant
l'appel, repli intégral sur les défauts actuels), bannière
d'annonce dans le gabarit principal (texte échappé par
l'autoescape Jinja) et drapeau ``recherche_partout`` rendu en
attribut ``data-recherche-partout`` de la page /recherche (la
logique JS associée est couverte par tests/recherche.test.js).
``domaine_defaut`` n'est pas appliqué : Flowdeck n'a pas de
notion de domaine propre (voir le docstring du module).
"""

import time

import pytest

from app.services import hub_reglages


@pytest.fixture(autouse=True)
def _cache_propre():
    hub_reglages.reinitialiser_cache()
    yield
    hub_reglages.reinitialiser_cache()


def _pose_reglages(monkeypatch, reglages, compteur=None):
    """Simule la consultation du Hub (appel réseau évité)."""
    def faux_consulter():
        if compteur is not None:
            compteur.append(1)
        return reglages
    monkeypatch.setattr(hub_reglages, "_consulter_hub", faux_consulter)


# ── Client : validation, cache, repli ───────────────────────────────

def test_validation_forme_du_contrat():
    assert hub_reglages._valider({
        "recherche_partout": False, "banniere": "Maintenance dimanche",
        "domaine_defaut": "famille", "ttl_secondes": 60}) == {
        "recherche_partout": False, "banniere": "Maintenance dimanche",
        "domaine_defaut": "famille"}
    for charge in (
        None, "texte", ["x"], {},
        {"recherche_partout": "oui", "banniere": "", "domaine_defaut": None},
        {"recherche_partout": True, "banniere": 42, "domaine_defaut": None},
        {"recherche_partout": True, "banniere": "", "domaine_defaut": "travail"},
    ):
        assert hub_reglages._valider(charge) is None, charge


class _Reponse:
    def __init__(self, status_code, payload=None):
        self.status_code = status_code
        self._payload = payload

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


def _pose_http(monkeypatch, reponse_ou_exception, appels=None):
    import httpx

    from app.services import draco_urls
    monkeypatch.setattr(draco_urls, "hub_url",
                        lambda: "https://hub.example")

    def faux_get(url, headers=None, timeout=None):
        if appels is not None:
            appels.append((url, headers, timeout))
        if isinstance(reponse_ou_exception, Exception):
            raise reponse_ou_exception
        return reponse_ou_exception

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


def test_consultation_contrat_exact(monkeypatch):
    appels = []
    _pose_http(monkeypatch, _Reponse(200, {
        "recherche_partout": True, "banniere": "Annonce",
        "domaine_defaut": None, "ttl_secondes": 60}), appels)
    assert hub_reglages._consulter_hub() == {
        "recherche_partout": True, "banniere": "Annonce",
        "domaine_defaut": None}
    url, headers, timeout = appels[0]
    assert url == "https://hub.example/api/reglages/effectifs"
    assert headers["Accept"] == "application/json"
    assert "Cookie" not in headers          # endpoint public
    assert timeout == 2.0


@pytest.mark.parametrize("reponse", [
    _Reponse(500, {}),
    _Reponse(200, {"recherche_partout": "oui", "banniere": "",
                   "domaine_defaut": None}),
    _Reponse(200, ValueError("json illisible")),
    ConnectionError("hub injoignable"),
])
def test_consultation_erreurs_repli_none(monkeypatch, reponse):
    _pose_http(monkeypatch, reponse)
    assert hub_reglages._consulter_hub() is None


def test_hub_muet_repli_defauts(monkeypatch):
    _pose_reglages(monkeypatch, None)
    assert hub_reglages.reglages_effectifs() == hub_reglages.DEFAUTS
    assert hub_reglages.banniere() == ""
    assert hub_reglages.recherche_partout() is True


def test_cache_60_s(monkeypatch):
    appels = []
    _pose_reglages(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_en_vigueur(monkeypatch):
    appels = []
    _pose_reglages(monkeypatch, {
        "recherche_partout": True, "banniere": "Annonce A",
        "domaine_defaut": None}, appels)
    assert hub_reglages.banniere() == "Annonce A"
    hub_reglages._cache_at = time.monotonic() - 61
    _pose_reglages(monkeypatch, None, appels)     # Hub tombe
    assert hub_reglages.banniere() == "Annonce A"  # ancien cache
    assert len(appels) == 2                        # pas de rafale


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

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


def test_banniere_presente_et_echappee(client, hub_banniere):
    hub_banniere["reglages"] = {
        "recherche_partout": True,
        "banniere": "Maintenance dimanche <b>gras</b> & cie",
        "domaine_defaut": None}
    r = client.get("/recherche")
    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 "sessionStorage" in r.text
    assert "flowdeck_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("/recherche")
    assert r.status_code == 200
    assert "banniere-hub" not in r.text
    hub_banniere["reglages"] = None
    hub_reglages.reinitialiser_cache()
    r = client.get("/recherche")
    assert r.status_code == 200
    assert "banniere-hub" not in r.text


def test_recherche_drapeau_rendu_dans_la_coquille(client, hub_banniere):
    hub_banniere["reglages"] = {
        "recherche_partout": False, "banniere": "", "domaine_defaut": None}
    r = client.get("/recherche")
    assert r.status_code == 200
    assert 'data-recherche-partout="false"' 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
    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


def test_version_7_85_3():
    from pathlib import Path
    version = (Path(__file__).parent.parent / "VERSION").read_text().strip()
    assert version == "7.85.3"