"""Tests for the opt-in dirty-tree spectrum (--dirty-spectrum).

Covers: deterministic seeded file selection in dirty_fraction, byte-identical

restore, flag plumbing (argparse default off), operation-name expansion, and

an integration-lite run proving the flag-off operation sequence is unchanged

when the flag is on (spectrum rows are purely additive).

"""

from __future__ import annotations

import sys

from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))

import hashlib

import shutil

import subprocess

import tempfile

import unittest

from unittest import mock

import bench

from oakbench.rows import validate_rows

from oakbench.subjects import Subject

GIT = shutil.which("git")

SPECTRUM_OPS_FULL = [

    "status.dirty.p01",

    "diff.dirty.p01",

    "status.dirty.p10",

    "diff.dirty.p10",

    "status.dirty.p50",

    "diff.dirty.p50",

]

def tree_digest(root: Path) -> dict[str, str]:

    return {

        path.relative_to(root).as_posix(): hashlib.sha256(path.read_bytes()).hexdigest()

        for path in bench.tracked_fixture_files(root)

    }

@unittest.skipIf(GIT is None, "git not installed")

class DirtyFractionTest(unittest.TestCase):

    def setUp(self) -> None:

        self.tmp = Path(tempfile.mkdtemp(prefix="oakbench-dirty-spectrum-"))

        self.repo = self.tmp / "repo"

        self.repo.mkdir()

        for index in range(100):

            subdir = self.repo / f"group-{index // 10:03d}"

            subdir.mkdir(exist_ok=True)

            (subdir / f"file-{index:04d}.txt").write_text(f"content {index}\n")

        env = bench.base_env()

        for command in (

            [GIT, "init", "-q"],

            [GIT, "add", "-A"],

            [GIT, "commit", "-q", "-m", "initial"],

        ):

            subprocess.run(command, cwd=self.repo, env=env, check=True, capture_output=True)

    def tearDown(self) -> None:

        shutil.rmtree(self.tmp, ignore_errors=True)

    def test_fraction_counts_use_ceil(self) -> None:

        for fraction, expected in ((0.01, 1), (0.10, 10), (0.50, 50)):

            changed = bench.dirty_fraction(self.repo, fraction, seed=f"seed:{fraction}")

            self.assertEqual(len(changed), expected)

            bench.restore_dirty_fraction(self.repo, changed, seed=f"seed:{fraction}")

    def test_same_seed_same_file_set(self) -> None:

        first = bench.dirty_fraction(self.repo, 0.10, seed="stable-seed")

        bench.restore_dirty_fraction(self.repo, first, seed="stable-seed")

        second = bench.dirty_fraction(self.repo, 0.10, seed="stable-seed")

        bench.restore_dirty_fraction(self.repo, second, seed="stable-seed")

        self.assertEqual(first, second)

    def test_selection_is_sorted_and_tracked_only(self) -> None:

        changed = bench.dirty_fraction(self.repo, 0.50, seed="sorted-seed")

        self.assertEqual(changed, sorted(changed))

        self.assertFalse(any(rel.startswith(".git") for rel in changed))

        bench.restore_dirty_fraction(self.repo, changed, seed="sorted-seed")

    def test_dirty_appends_fixed_line_and_restore_is_byte_identical(self) -> None:

        before = tree_digest(self.repo)

        changed = bench.dirty_fraction(self.repo, 0.01, seed="roundtrip-seed")

        self.assertEqual(len(changed), 1)

        dirtied = self.repo / changed[0]

        marker = bench.dirty_spectrum_marker("roundtrip-seed")

        self.assertTrue(dirtied.read_bytes().endswith(marker))

        status = subprocess.run(

            [GIT, "status", "--porcelain"],

            cwd=self.repo,

            env=bench.base_env(),

            check=True,

            capture_output=True,

            text=True,

        )

        self.assertIn(changed[0], status.stdout)

        bench.restore_dirty_fraction(self.repo, changed, seed="roundtrip-seed")

        self.assertEqual(tree_digest(self.repo), before)

    def test_restore_refuses_unexpected_tail(self) -> None:

        changed = bench.dirty_fraction(self.repo, 0.01, seed="tamper-seed")

        path = self.repo / changed[0]

        with path.open("ab") as fh:

            fh.write(b"extra\n")

        with self.assertRaises(RuntimeError):

            bench.restore_dirty_fraction(self.repo, changed, seed="tamper-seed")

class FlagPlumbingTest(unittest.TestCase):

    def test_argparse_accepts_dirty_spectrum(self) -> None:

        with mock.patch.object(sys, "argv", ["bench.py", "--dirty-spectrum"]):

            args = bench.parse_args()

        self.assertTrue(args.dirty_spectrum)

    def test_flag_defaults_off(self) -> None:

        with mock.patch.object(sys, "argv", ["bench.py"]):

            args = bench.parse_args()

        self.assertFalse(args.dirty_spectrum)

    def test_op_list_unchanged_when_off(self) -> None:

        self.assertEqual(bench.dirty_spectrum_operations(False), [])

        self.assertEqual(bench.dirty_spectrum_operations(False, skip_diff=True), [])

    def test_op_list_expands_when_on(self) -> None:

        self.assertEqual(bench.dirty_spectrum_operations(True), SPECTRUM_OPS_FULL)

        self.assertEqual(

            bench.dirty_spectrum_operations(True, skip_diff=True),

            ["status.dirty.p01", "status.dirty.p10", "status.dirty.p50"],

        )

    def test_run_subject_scenario_defaults_spectrum_off(self) -> None:

        # The flag-off path must not depend on callers passing the new

        # parameter: existing call sites stay byte-identical in behavior.

        import inspect

        signature = inspect.signature(bench.run_subject_scenario)

        self.assertIs(signature.parameters["dirty_spectrum"].default, False)

@unittest.skipIf(GIT is None, "git not installed")

class DirtySpectrumIntegrationTest(unittest.TestCase):

    """Integration-lite: drive run_subject_scenario on a tiny git fixture."""

    def setUp(self) -> None:

        self.tmp = Path(tempfile.mkdtemp(prefix="oakbench-spectrum-lane-"))

    def tearDown(self) -> None:

        shutil.rmtree(self.tmp, ignore_errors=True)

    def run_scenario(self, dirty_spectrum: bool) -> list[dict]:

        scenario = bench.Scenario("tiny_text", 8, 256, 2, False, 1)

        fixture = bench.ensure_fixture(self.tmp, scenario)

        metadata = {

            "bench_id": "test",

            "profile": "micro",

            "benchmark_track": "agent-default",

            "command_semantics_version": bench.COMMAND_SEMANTICS_VERSION,

            "timestamp_utc": "test",

        }

        return bench.run_subject_scenario(

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

            scenario,

            fixture,

            run_index=0,

            run_root=self.tmp / ("runs-on" if dirty_spectrum else "runs-off"),

            metadata=metadata,

            skip_diff=False,

            admitted_output_chars=20_000,

            track="agent-default",

            skip_determinism_probe=True,

            skip_remote=True,

            dirty_spectrum=dirty_spectrum,

        )

    def test_spectrum_rows_additive_and_flag_off_sequence_unchanged(self) -> None:

        rows_off = self.run_scenario(dirty_spectrum=False)

        rows_on = self.run_scenario(dirty_spectrum=True)

        ops_off = [row["operation"] for row in rows_off]

        ops_on = [row["operation"] for row in rows_on]

        # Flag off: no spectrum operation names anywhere.

        for op in SPECTRUM_OPS_FULL:

            self.assertNotIn(op, ops_off)

        # Flag on: every spectrum row present, valid, and successful.

        for op in SPECTRUM_OPS_FULL:

            self.assertIn(op, ops_on)

        self.assertEqual(validate_rows(rows_on, "core"), [])

        failures = [row for row in rows_on if row["returncode"] != 0]

        self.assertEqual(failures, [])

        # Purely additive: removing spectrum rows reproduces the flag-off

        # operation sequence exactly, in order.

        self.assertEqual([op for op in ops_on if op not in SPECTRUM_OPS_FULL], ops_off)

        # Scale annotations: 8 tracked files -> ceil gives 1, 1, 4.

        by_op = {row["operation"]: row for row in rows_on}

        self.assertEqual(by_op["status.dirty.p01"]["dirty_file_count"], 1)

        self.assertEqual(by_op["status.dirty.p10"]["dirty_file_count"], 1)

        self.assertEqual(by_op["status.dirty.p50"]["dirty_file_count"], 4)

        self.assertEqual(by_op["diff.dirty.p50"]["dirty_fraction"], 0.50)

        # Spectrum status rows must observe exactly the dirtied files: the

        # measured command runs on the dirty tree, after untimed setup.

        status_row = by_op["status.dirty.p50"]

        self.assertEqual(status_row["command"], by_op["status.dirty"]["command"])

if __name__ == "__main__":

    unittest.main()