"""Falsification CI for the environment-suspicion detector and its lane wiring.

A detector that has never been shown to fire is not evidence of a clean run.

These tests inject known-bad conditions — synthetic boundary samples

everywhere, a real stress-ng cpu50 tier where the optional binary exists —

and assert ``environment_suspect`` actually fires, stays quiet on clean

windows, and is monotonic (more injected badness never yields fewer kinds of

suspicion). Lane wiring is pinned too: both CLIs expose --load-tier, and a

non-none tier without stress-ng produces explicit skip rows.

"""

from __future__ import annotations

import os

import shutil

import subprocess

import sys

import tempfile

import time

import unittest

from pathlib import Path

from typing import Any

import bench

import workflow_ab

from oakbench.envwatch import environment_suspect, sample_environment

from oakbench.loadgen import LoadGenerator, load_tier_unavailable_reason

SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"

STRESS_NG_PRESENT = shutil.which("stress-ng") is not None

def clean_sample() -> dict[str, Any]:

    return {

        "loadavg": [0.2, 0.2, 0.2],

        "mem_available_bytes": 8 * 1024**3,

        "cpu_speed_limit_pct": 100,

        "thermal_zone_max_c": None,

        "cpu_governor": None,

    }

def reason_kinds(reasons: list[str]) -> set[tuple[str, str]]:

    """Stable (kind, boundary) categories so monotonicity compares signal

    kinds, not exact human-readable strings (which embed magnitudes)."""

    kinds: set[tuple[str, str]] = set()

    for reason in reasons:

        if reason.startswith("load_per_core"):

            kind = "load_per_core"

        elif "thermal throttling" in reason:

            kind = "thermal_throttle"

        else:

            kind = "other"

        boundary = "before" if "before" in reason else ("after" if "after" in reason else "window")

        kinds.add((kind, boundary))

    return kinds

class SyntheticFalsificationTest(unittest.TestCase):

    """Pure-python injected-load samples: runs everywhere, no stress-ng."""

    def test_clean_window_is_ran_and_clean_empty_list(self) -> None:

        reasons = environment_suspect(clean_sample(), clean_sample(), cpu_count=8)

        self.assertEqual(reasons, [])  # [] is ran-and-clean, never None (ADR-0002)

    def test_injected_loadavg_above_per_core_threshold_fires(self) -> None:

        loaded = dict(clean_sample(), loadavg=[16.0, 12.0, 8.0])  # 2.0/core on 8 cores

        reasons = environment_suspect(clean_sample(), loaded, cpu_count=8)

        self.assertTrue(reasons)

        self.assertTrue(any(r.startswith("load_per_core") for r in reasons))

    def test_injected_thermal_throttle_fires(self) -> None:

        throttled = dict(clean_sample(), cpu_speed_limit_pct=60)

        reasons = environment_suspect(throttled, clean_sample(), cpu_count=8)

        self.assertTrue(reasons)

        self.assertTrue(any("thermal throttling" in r for r in reasons))

    def test_combined_injection_fires_both_signals(self) -> None:

        bad = dict(clean_sample(), loadavg=[20.0, 20.0, 20.0], cpu_speed_limit_pct=55)

        reasons = environment_suspect(bad, bad, cpu_count=8)

        kinds = {kind for kind, _boundary in reason_kinds(reasons)}

        self.assertIn("load_per_core", kinds)

        self.assertIn("thermal_throttle", kinds)

    def test_monotonic_more_load_never_yields_fewer_reason_kinds(self) -> None:

        # Escalating injected badness: clean -> high load after -> high load

        # both boundaries -> high load + thermal throttle. Each step's reason

        # kinds must be a superset-or-equal of the previous step's.

        eight = 8

        clean = clean_sample()

        loaded = dict(clean_sample(), loadavg=[14.0, 10.0, 6.0])

        hotter = dict(loaded, cpu_speed_limit_pct=70)

        windows = [

            (clean, clean),

            (clean, loaded),

            (loaded, loaded),

            (loaded, hotter),

            (hotter, hotter),

        ]

        previous: set[tuple[str, str]] = set()

        for before, after in windows:

            current = reason_kinds(environment_suspect(before, after, cpu_count=eight))

            self.assertTrue(

                previous.issubset(current),

                f"suspicion regressed: {previous} not subset of {current}",

            )

            previous = current

        self.assertTrue(previous)  # the worst window did fire

    def test_higher_load_magnitude_keeps_superset_of_reasons(self) -> None:

        mild = dict(clean_sample(), loadavg=[13.0, 0.0, 0.0])   # 1.625/core

        heavy = dict(clean_sample(), loadavg=[40.0, 0.0, 0.0])  # 5.0/core

        mild_kinds = reason_kinds(environment_suspect(mild, mild, cpu_count=8))

        heavy_kinds = reason_kinds(environment_suspect(heavy, heavy, cpu_count=8))

        self.assertTrue(mild_kinds)

        self.assertTrue(mild_kinds.issubset(heavy_kinds))

class LiveStressNgFalsificationTest(unittest.TestCase):

    """Optional live falsification: only runs where stress-ng is installed."""

    DEADLINE_S = 20.0

    @unittest.skipUnless(STRESS_NG_PRESENT, "stress-ng not installed (optional perturbation binary)")

    def test_cpu50_load_makes_environment_suspect_fire(self) -> None:

        started_at = time.monotonic()

        before = sample_environment()

        self.assertIsNotNone(before.get("loadavg"), "host cannot report loadavg; cannot falsify here")

        before_per_core = before["loadavg"][0] / max(1, os.cpu_count() or 1)

        # Detect the injected load relative to the pre-load baseline: any

        # measurable rise in per-core load while stress-ng runs must trip a

        # threshold pinned just above the quiet baseline.

        threshold = before_per_core + 0.01

        with tempfile.TemporaryDirectory() as tmp:

            generator = LoadGenerator("cpu50", Path(tmp) / "load-temp", timeout_s=int(self.DEADLINE_S))

            start = generator.start()

            self.assertTrue(start["started"], start)

            try:

                time.sleep(3)

                fired: list[str] = []

                while time.monotonic() - started_at < self.DEADLINE_S - 2:

                    during = sample_environment()

                    fired = environment_suspect(before, during, load_per_core_threshold=threshold)

                    if any(r.startswith("load_per_core") for r in fired):

                        break

                    time.sleep(1)

            finally:

                stop = generator.stop()

            self.assertTrue(

                any(r.startswith("load_per_core") for r in fired),

                f"environment_suspect never fired under live cpu50 load (stop={stop})",

            )

            self.assertLess(time.monotonic() - started_at, self.DEADLINE_S)

class LaneWiringTest(unittest.TestCase):

    """Both lanes expose --load-tier; unappliable tiers become skip rows."""

    def _help_text(self, script: str) -> str:

        proc = subprocess.run(

            [sys.executable, str(SCRIPTS / script), "--help"],

            capture_output=True,

            text=True,

            timeout=60,

            check=False,

        )

        self.assertEqual(proc.returncode, 0, proc.stderr)

        return proc.stdout

    def test_bench_cli_has_load_tier_flag(self) -> None:

        text = self._help_text("bench.py")

        self.assertIn("--load-tier", text)

        for tier in ("none", "cpu50", "io", "mixed"):

            self.assertIn(tier, text)

    def test_workflow_ab_cli_has_load_tier_flag(self) -> None:

        text = self._help_text("workflow_ab.py")

        self.assertIn("--load-tier", text)

        for tier in ("none", "cpu50", "io", "mixed"):

            self.assertIn(tier, text)

    def test_bench_skip_rows_when_stress_ng_missing(self) -> None:

        # bench.py needs real VCS binaries/fixtures for a smoke run, so the

        # skip-row builder is unit-tested directly (it is exactly what main()

        # emits when LoadGenerator.start reports stress_ng_missing).

        from oakbench.subjects import Subject

        subjects = [

            Subject(name="git", kind="git", label="Git", bin=Path("git")),

            Subject(name="oak_main", kind="oak", label="Oak", bin=Path("oak")),

        ]

        scenarios = [bench.Scenario("tiny_text", 24, 1024, 4, False, 2)]

        reason = load_tier_unavailable_reason("stress_ng_missing")

        rows = bench.load_tier_skip_rows(subjects, scenarios, {"bench_id": "t"}, reason, "cpu50")

        self.assertEqual(len(rows), 2 * 2)  # runs x subjects

        for row in rows:

            self.assertTrue(row["skipped"])

            self.assertEqual(row["skip_reason"], "load_tier_unavailable:stress_ng_missing")

            self.assertEqual(row["returncode"], 77)

            self.assertEqual(row["load_tier"], "cpu50")  # requested tier, recorded explicitly

            self.assertIsNone(row["environment_suspect"])  # machinery never ran (ADR-0002)

            self.assertEqual(row["operation"], "scenario.load_tier_skipped")

            self.assertEqual(row["tool_call_count"], 0)

    def test_workflow_ab_skip_rows_when_stress_ng_missing(self) -> None:

        from oakbench.subjects import Subject

        subjects = [Subject(name="git", kind="git", label="Git", bin=Path("git"))]

        workflows = [workflow_ab.WORKFLOWS["bugfix_test_loop"]]

        reason = load_tier_unavailable_reason("stress_ng_missing")

        rows = workflow_ab.load_tier_skip_rows(subjects, workflows, 3, {"bench_id": "t"}, reason, "io")

        self.assertEqual(len(rows), 3)

        for row in rows:

            self.assertTrue(row["skipped"])

            self.assertEqual(row["skip_reason"], "load_tier_unavailable:stress_ng_missing")

            self.assertEqual(row["returncode"], 77)

            self.assertEqual(row["load_tier"], "io")

            self.assertIsNone(row["environment_suspect"])

            self.assertEqual(row["operation"], "workflow.skipped")

    def test_none_tier_rows_still_record_the_explicit_string(self) -> None:

        # The wiring attaches load_tier="none" (never null) to measured rows

        # so a row without load is distinguishable from a row never wired.

        from oakbench.loadgen import environment_boundary_fields

        sample = {"loadavg": [0.1, 0.1, 0.1], "cpu_speed_limit_pct": 100, "cpu_governor": None}

        fields = environment_boundary_fields("none", sample, sample, cpu_count=8)

        self.assertEqual(fields["load_tier"], "none")

        self.assertEqual(fields["environment_suspect"], [])

if __name__ == "__main__":

    unittest.main()