"""Garde-fous du workflow CI Gitea (BUG-082, BUG-083, BUG-091, BUG-093).

Sans dépendance (pas de PyYAML) : analyse ligne à ligne de
`.gitea/workflows/ci.yml`, suffisante pour les conventions de ce fichier.
"""
from __future__ import annotations

import re
from pathlib import Path

CI_YML = Path(__file__).resolve().parent.parent / ".gitea" / "workflows" / "ci.yml"
REQUIREMENTS = Path(__file__).resolve().parent.parent / "backend" / "requirements.txt"
REPO_ROOT = Path(__file__).resolve().parent.parent


def _job_text(job: str) -> str:
    """Corps YAML du job `job` (jusqu'au job suivant ou à la fin du fichier)."""
    text = CI_YML.read_text(encoding="utf-8")
    start = text.index(f"\n  {job}:")
    rest = text[start + 1 :]
    nxt = re.search(r"\n {2}[A-Za-z][A-Za-z0-9_-]*:\s*\n", rest)
    return rest[: nxt.start()] if nxt else rest


def _steps(job: str) -> list[tuple[str, str]]:
    """[(nom d'étape, corps YAML)] pour un job donné."""
    chunks = re.split(r"\n {6}- name: ", "\n" + _job_text(job))[1:]
    steps = []
    for chunk in chunks:
        name, _, body = chunk.partition("\n")
        steps.append((name.strip(), body))
    return steps


def _run_bodies() -> list[tuple[int, str]]:
    """Toutes les lignes shell de chaque bloc `run:` → [(n° ligne, code)]."""
    lines = CI_YML.read_text(encoding="utf-8").splitlines()
    bodies: list[tuple[int, str]] = []
    i = 0
    while i < len(lines):
        m = re.match(r"^(\s*)run:(?:\s*\|\s*)?$", lines[i])
        inline = re.match(r"^(\s*)run:\s+(\S.*)$", lines[i])
        if m:
            base = len(m.group(1))
            i += 1
            while i < len(lines):
                cur = lines[i]
                if not cur.strip():
                    i += 1
                    continue
                if len(cur) - len(cur.lstrip()) <= base:
                    break
                bodies.append((i + 1, cur.strip()))
                i += 1
        elif inline:
            bodies.append((i + 1, inline.group(2).strip()))
            i += 1
        else:
            i += 1
    return bodies


class TestRunnerProofScripts:
    def test_no_hash_inside_run_bodies(self):
        """BUG-083 : aucun `#` dans le code shell des `run:`.

        Le runner Gitea Act tronque naïvement au premier `#` (même entre
        guillemets) : `echo "... see #87)"` devenait une citation non
        fermée → `unexpected EOF while looking for matching '"'` (job
        `security` rouge). Les lignes-commentaires shell (`# ...`) restent
        autorisées : leur troncature est sémantiquement neutre.
        """
        offenders = [
            f"L{n}: {code}"
            for n, code in _run_bodies()
            if not code.startswith("#") and "#" in code
        ]
        assert not offenders, (
            "BUG-083 : `#` interdit dans le code des `run:` "
            f"(tronqué par le runner) :\n" + "\n".join(offenders)
        )


class TestSemgrepStep:
    """BUG-091 : semgrep-core est inexécutable sur le runner (exit 127).

    L'étape est donc désactivée (avertissement, non bloquante) au lieu d'être
    supprimée : elle documente pourquoi, et se réactive telle quelle dès que le
    runner dispose d'un CPU x86-64-v2.
    """

    SEMGREP_STEP_PREFIX = "Semgrep"

    def _semgrep_step(self) -> tuple[str, str]:
        matches = [
            (n, b) for n, b in _steps("security") if n.startswith(self.SEMGREP_STEP_PREFIX)
        ]
        assert len(matches) == 1, (
            "BUG-091 : une unique étape Semgrep (désactivée) attendue dans le "
            f"job security, trouvé {len(matches)}"
        )
        return matches[0]

    @staticmethod
    def _run_commands(body: str) -> list[str]:
        """Commandes shell du bloc `run:` de l'étape (hors lignes vides)."""
        m = re.search(r"^\s*run:\s*\|?\s*$", body, re.M)
        assert m, "étape sans bloc `run:`"
        rest = body[m.end() :]
        lines: list[str] = []
        for line in rest.splitlines():
            if not line.strip():
                continue
            # le bloc run: est indenté de 2 spaces de plus que la clef
            if len(line) - len(line.lstrip()) <= 8:
                break
            lines.append(line.strip())
        return lines

    def test_semgrep_step_does_not_execute_core(self):
        """Le core natif ne doit plus être lancé (exit 127 bloquant le job).

        Seule commande admise : l'avertissement d'activation. Le message
        mentionne voluntaryirement « semgrep » — c'est l'**exécution** qui
        est interdite, pas le mot.
        """
        name, body = self._semgrep_step()
        commands = self._run_commands(body)
        assert commands, f"BUG-091 : l'étape « {name} » n'a plus de commande"
        for cmd in commands:
            assert cmd.startswith('echo "::warning::'), (
                f"BUG-091 : l'étape « {name} » ne doit exécuter qu'un avertissement, "
                f"trouvé : {cmd!r}"
            )

    def test_semgrep_step_is_non_blocking_and_explains_itself(self):
        """Désactivée = `continue-on-error` + avertissement explicite."""
        name, body = self._semgrep_step()
        assert re.search(r"^\s*continue-on-error:\s*true\s*$", body, re.M), (
            f"BUG-091 : l'étape « {name} » doit porter continue-on-error: true"
        )
        assert "::warning::" in body, (
            f"BUG-091 : l'étape « {name} » doit émettre un ::warning:: "
            "expliquant la désactivation"
        )
        assert "BUG-091" in body, (
            f"BUG-091 : l'étape « {name} » doit référencer BUG-091"
        )

    def test_bandit_and_pip_audit_stay_blocking(self):
        """La désactivation de semgrep ne doit rien dégraver d'autre (#87)."""
        found = {}
        for name, body in _steps("security"):
            low = name.lower()
            if low.startswith("bandit"):
                found["bandit"] = body
            elif low.startswith("pip-audit"):
                found["pip-audit"] = body
        assert set(found) == {"bandit", "pip-audit"}, (
            f"étapes Bandit et Pip-audit attendues dans le job security, "
            f"trouvé {sorted(found)}"
        )
        for tool, body in found.items():
            assert "continue-on-error: true" not in body, (
                f"BUG-091 : l'étape {tool} doit rester bloquante (#87)"
            )

    def test_semgrep_rules_still_shipped_and_documented(self):
        """Les règles locales restent versionnées et documentées (#87 T7)."""
        rules = REPO_ROOT / "semgrep-rules" / "obsigate-python.yaml"
        assert rules.exists(), "ruleset semgrep manquant"
        text = CI_YML.read_text(encoding="utf-8")
        # La commande locale est documentée (commentaire de l'étape), pas exécutée.
        assert re.search(r"semgrep --config semgrep-rules/\s*\n?\s*#?\s*backend/", text), (
            "#87 T7 : commande locale `semgrep --config semgrep-rules/ backend/` "
            "attendue en commentaire dans le workflow"
        )


class TestFrontendStepsHaveTheirDeps:
    @staticmethod
    def _root_step_files() -> list[str]:
        """Fichiers `node tests/frontend/<f>` de l'étape racine (sans jsdom)."""
        text = CI_YML.read_text(encoding="utf-8")
        root_part = text.split("Frontend JSDOM tests", 1)[0]
        root_steps = root_part.split("Frontend unit tests", 1)[1]
        return re.findall(r"node tests/frontend/(\S+\.mjs)", root_steps)

    @staticmethod
    def _has_static_jsdom_import(rel: str) -> bool:
        path = REPO_ROOT / "tests" / "frontend" / rel
        return any(
            re.match(r"^\s*import\b.*\bfrom\s+['\"]jsdom['\"]", line)
            or re.match(r"""\brequire\(\s*['"]jsdom['"]\s*\)""", line)
            for line in path.read_text(encoding="utf-8").splitlines()
        )

    def test_root_step_files_need_no_jsdom(self):
        """BUG-082 : l'étape racine tourne sans `tests/frontend/node_modules`
        (installé seulement par l'étape JSDOM) : aucun de ses fichiers ne
        doit importer `jsdom` statiquement — sinon `ERR_MODULE_NOT_FOUND`
        et `lint` rouge (cas `upload.test.mjs`, puis `config-ai-keys.test.mjs`).
        """
        offenders = [f for f in self._root_step_files() if self._has_static_jsdom_import(f)]
        assert not offenders, (
            "BUG-082 : ces fichiers importent `jsdom` mais tournent dans "
            "l'étape racine (sans node_modules) — les déplacer dans l'étape "
            f"JSDOM :\n" + "\n".join(offenders)
        )

    def test_jsdom_dependent_tests_run_in_jsdom_step(self):
        """BUG-082 : les suites à import statique `jsdom` tournent bien dans
        l'étape JSDOM (les deux branches)."""
        text = CI_YML.read_text(encoding="utf-8")
        jsdom_part = text.split("Frontend JSDOM tests", 1)[1]
        for suite in ("node upload.test.mjs", "node config-ai-keys.test.mjs"):
            assert jsdom_part.count(suite) >= 2, (
                f"BUG-082 : `{suite}` attendu dans les deux branches de "
                "l'étape JSDOM"
            )


_SPEC_RE = re.compile(
    r"^([A-Za-z0-9._-]+)\s*(?:\[[^\]]*\])?\s*(>=|==|~=|>|<)\s*([0-9][^\s;#]*)"
)


def _floor(pkg: str) -> tuple[int, ...] | None:
    """Plancher `>=` déclaré pour `pkg` dans backend/requirements.txt."""
    for raw in REQUIREMENTS.read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        m = _SPEC_RE.match(line)
        if not m or m.group(1).lower() != pkg or m.group(2) != ">=":
            continue
        return tuple(int(p) for p in re.match(r"[0-9]+(?:\.[0-9]+)*", m.group(3)).group(0).split("."))
    return None


class TestDependencySecurityFloors:
    """Planchers de sécurité des dépendances (#87, BUG-091, BUG-093).

    `pip-audit` est bloquant dans le job `security`. Comme l'image du runner
    (`catthehacker/ubuntu:act-latest`) embarque des paquets *préinstallés* dans
    sa toolcache Python, un plancher trop bas est « already satisfied » et
    n'est jamais mis à niveau : c'est exactement ce qui a fait échouer le
    job sur pypdf 6.16.0 (PYSEC-2026-3910 / PYSEC-2026-3911, DoS de ressources
    atteignables via backend/pdf_reader.py).
    """

    #: (paquet, plancher minimal, advisories corrigées au-dessus)
    FLOORS = {
        "pypdf": (6, 16, 1),  # PYSEC-2026-3910, PYSEC-2026-3911 (fix 6.16.1)
        # BUG-095 : 2.13.0 a son propre advisory (CVE-2026-102274, fix 2.14.0).
        "pyjwt": (2, 14, 0),  # PYSEC-2026-178 (2.13.0) puis CVE-2026-102274 (2.14.0)
    }

    def test_security_floors_are_declared(self):
        missing = [p for p in self.FLOORS if _floor(p) is None]
        assert not missing, (
            "plancher `>=` manquant dans backend/requirements.txt pour : "
            f"{missing}"
        )

    def test_security_floors_are_high_enough(self):
        too_low = {
            p: (_floor(p), minimum)
            for p, minimum in self.FLOORS.items()
            if (_floor(p) or ()) < minimum
        }
        assert not too_low, (
            "BUG-093 : plancher(s) sous le correctif de sécurité, "
            f"le job `security` (pip-audit bloquant) échouerait : {too_low}"
        )