"""Parallel-run hygiene: branch names, fixture builds, and latest.* copies."""

from __future__ import annotations

import importlib.util

import json

import os

import shutil

import sys

import tempfile

import unittest

from pathlib import Path

from unittest.mock import patch

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

from oakbench import results as oakbench_results  # noqa: E402

from oakbench.results import ResultsStore  # noqa: E402

from oakbench.rows import RowContractViolation  # noqa: E402

def load_script(name: str):

    path = Path(__file__).resolve().parents[1] / "scripts" / f"{name}.py"

    spec = importlib.util.spec_from_file_location(name, path)

    module = importlib.util.module_from_spec(spec)

    sys.modules[name] = module

    spec.loader.exec_module(module)

    return module

class BranchNameUniquenessTest(unittest.TestCase):

    def test_same_inputs_differ_by_uniquifier(self) -> None:

        bench = load_script("bench")

        shared = {

            "bench_id": "20260610T120000",

            "scenario_name": "tiny_text",

            "subject_name": "git",

            "run_index": 0,

        }

        first = bench.oak_network_branch_name(**shared, pid=1001, token="aabb")

        second = bench.oak_network_branch_name(**shared, pid=1002, token="aabb")

        third = bench.oak_network_branch_name(**shared, pid=1001, token="ccdd")

        self.assertNotEqual(first, second)

        self.assertNotEqual(first, third)

        self.assertTrue(first.startswith("bench-20260610T120000-tiny_text-git-r0-"))

class EnsureFixtureRaceTest(unittest.TestCase):

    def setUp(self) -> None:

        self.bench = load_script("bench")

        self.tmp = Path(tempfile.mkdtemp(prefix="oakbench-parallel-fixture-"))

    def tearDown(self) -> None:

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

    def test_lost_race_reuses_valid_dest_and_cleans_temp(self) -> None:

        scenario = self.bench.Scenario("tiny_text", 3, 64, 1, False, 1)

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

        expected_files = sum(1 for _ in dest.rglob("*.txt"))

        marker_checks = {"count": 0}

        real_exists = Path.exists

        def exists(self: Path) -> bool:

            if self == dest / ".fixture-ready":

                marker_checks["count"] += 1

                if marker_checks["count"] == 1:

                    return False

            return real_exists(self)

        with patch.object(Path, "exists", exists):

            with patch.object(os, "rename", side_effect=OSError(17, "exists")):

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

        self.assertEqual(result, dest)

        self.assertEqual(sum(1 for _ in dest.rglob("*.txt")), expected_files)

        building_dirs = list(dest.parent.glob(f"{dest.name}.building-*"))

        self.assertEqual(building_dirs, [])

    def test_marker_mismatch_on_lost_race_raises(self) -> None:

        scenario = self.bench.Scenario("race", 2, 64, 1, False, 1)

        dest = self.bench.fixture_path(self.tmp, scenario)

        dest.mkdir(parents=True)

        (dest / ".fixture-ready").write_text("wrong-shape\n")

        real_rmtree = shutil.rmtree

        def skip_dest_rmtree(path: Path, *args, **kwargs) -> None:

            if path == dest:

                return

            real_rmtree(path, *args, **kwargs)

        with patch.object(shutil, "rmtree", side_effect=skip_dest_rmtree):

            with patch.object(os, "rename", side_effect=OSError(17, "exists")):

                with self.assertRaises(RuntimeError) as ctx:

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

        self.assertIn("wrong shape marker", str(ctx.exception))

class ResultsStoreAtomicLatestTest(unittest.TestCase):

    def setUp(self) -> None:

        self.tmp = Path(tempfile.mkdtemp(prefix="oakbench-results-atomic-"))

    def tearDown(self) -> None:

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

    def test_latest_jsonl_matches_raw_without_tmp_leftovers(self) -> None:

        store = ResultsStore(self.tmp, "core")

        rows = [

            {

                "bench_id": "parallel-test",

                "profile": "micro",

                "scenario": "tiny_text",

                "operation": "status.clean",

                "run": 0,

                "subject": "git",

                "subject_kind": "git",

                "elapsed_ms": 1.0,

                "returncode": 0,

                "command": ["git", "status"],

            }

        ]

        raw_path, _ = store.write("parallel-test", rows, summary_text="# summary\n")

        latest_path = self.tmp / "latest.jsonl"

        self.assertEqual(latest_path.read_text(), raw_path.read_text())

        tmp_leftovers = list(self.tmp.glob("*.tmp-*"))

        self.assertEqual(tmp_leftovers, [])

    def test_contract_violation_keeps_raw_but_never_touches_latest(self) -> None:

        store = ResultsStore(self.tmp, "core")

        valid_row = {

            "bench_id": "old-run",

            "profile": "micro",

            "scenario": "tiny_text",

            "operation": "status.clean",

            "run": 0,

            "subject": "git",

            "subject_kind": "git",

            "elapsed_ms": 1.0,

            "returncode": 0,

            "command": ["git", "status"],

        }

        store.write("old-run", [valid_row], summary_text="# old summary\n")

        latest_path = self.tmp / "latest.jsonl"

        latest_summary_path = self.tmp / "latest.summary.md"

        old_latest = latest_path.read_text()

        old_latest_summary = latest_summary_path.read_text()

        invalid_row = dict(valid_row, bench_id="new-run", returncode=None)

        with self.assertRaises(RowContractViolation):

            store.write("new-run", [invalid_row], summary_text="# new summary\n")

        # latest.* pointers still reference the previous, contract-clean run.

        self.assertEqual(latest_path.read_text(), old_latest)

        self.assertEqual(json.loads(latest_path.read_text())["bench_id"], "old-run")

        self.assertEqual(latest_summary_path.read_text(), old_latest_summary)

        # The raw bench_id file keeps the invalid rows for emitter debugging.

        raw_path = self.tmp / "new-run.jsonl"

        self.assertEqual(json.loads(raw_path.read_text()), invalid_row)

    def test_write_all_retries_partial_os_write(self) -> None:

        path = self.tmp / "partial-write.bin"

        fd = os.open(path, os.O_WRONLY | os.O_CREAT, 0o644)

        real_write = os.write

        def partial_write(target_fd: int, data) -> int:

            size = len(data)

            return real_write(target_fd, data[: max(1, size // 2)])

        try:

            with patch.object(oakbench_results.os, "write", side_effect=partial_write):

                oakbench_results._write_all(fd, b"abcdef")

        finally:

            os.close(fd)

        self.assertEqual(path.read_bytes(), b"abcdef")

class ParallelContentionPayloadPrefixTest(unittest.TestCase):

    def test_prefix_includes_subject_worker_level_and_run(self) -> None:

        contention = load_script("parallel_contention")

        prefixes = {

            contention.contention_payload_prefix("bench", subject, workers, 0)

            for subject in ("git", "oak_installed")

            for workers in (2, 8)

        }

        self.assertEqual(len(prefixes), 4)

        self.assertIn("workers/bench/git/w2/r0", prefixes)

        self.assertIn("workers/bench/oak_installed/w8/r0", prefixes)

if __name__ == "__main__":

    unittest.main()