from __future__ import annotations

import functools
import json
import os
import pathlib
import signal
import socket
import subprocess
import sys
import tempfile
import threading
import time
import unittest
from concurrent.futures import ThreadPoolExecutor
from types import SimpleNamespace

from scripts import r10r2_replay_bundle


ROOT = pathlib.Path(__file__).resolve().parents[1]


class R10R2ProcessRegressionTests(unittest.TestCase):
    def _negative(self, mode: str, deadline: float, stall: float) -> dict:
        with tempfile.TemporaryDirectory() as directory:
            base = pathlib.Path(directory)
            command = base / "command.json"
            command.write_text(json.dumps([sys.executable, str(ROOT / "scripts" / "r10r2_hazard_child.py"), mode]), encoding="utf-8")
            event = base / "events.jsonl"; record = base / "record.json"
            env = {**os.environ, "R10_R2_RUN_ID": "negative-" + mode, "R10_R2_EVENT_JOURNAL_PATH": str(event), "R10_R2_SUPERVISOR_RECORD_PATH": str(record)}
            if mode == "signal": env["R10_R2_SELF_SIGNAL_AFTER_SECONDS"] = "0.4"
            completed = subprocess.run(
                [sys.executable, str(ROOT / "scripts" / "r10r2_supervisor.py"), "--deadline-seconds", str(deadline), "--heartbeat-stall-seconds", str(stall), "--child-command-json", str(command)],
                cwd=ROOT, env=env, capture_output=True, text=True, timeout=20, check=False,
            )
            payload = json.loads(record.read_text(encoding="utf-8")); payload["launcher_exit_code"] = completed.returncode
            return payload

    def test_r10r2_rg_41_timeout_final_record(self) -> None:
        record = self._negative("timeout", 1.2, 5.0)
        self.assertTrue(record["timed_out"]); self.assertTrue(record["final_record_written"]); self.assertTrue(record["cleanup_passed"])

    def test_r10r2_rg_42_stall_final_record(self) -> None:
        record = self._negative("stall", 4.0, 0.7)
        self.assertTrue(record["stall_detected"]); self.assertTrue(record["final_record_written"]); self.assertTrue(record["cleanup_passed"])

    def test_r10r2_rg_43_signal_final_record(self) -> None:
        started = time.monotonic()
        record = self._negative("signal", 4.0, 2.0)
        elapsed = time.monotonic() - started
        self.assertIsNotNone(record["signal_received"]); self.assertTrue(record["final_record_written"]); self.assertTrue(record["cleanup_passed"])
        self.assertLess(elapsed, 15.0, "signal finalization must remain below the independent 20 second watchdog")

    def test_r10r2_rg_44_subprocess_identity(self) -> None:
        completed = subprocess.run([sys.executable, "-c", "import os;print(os.getpid())"], capture_output=True, text=True, timeout=5, check=True)
        self.assertNotEqual(int(completed.stdout), os.getpid())

    def test_r10r2_rg_45_thread_cleanup(self) -> None:
        done = threading.Event(); worker = threading.Thread(target=done.set); worker.start(); worker.join(timeout=2)
        self.assertTrue(done.is_set()); self.assertFalse(worker.is_alive())

    def test_r10r2_rg_46_environment_restore(self) -> None:
        before = os.environ.get("R10_R2_TEST_ENV"); os.environ["R10_R2_TEST_ENV"] = "synthetic"; observed = os.environ.pop("R10_R2_TEST_ENV")
        if before is not None: os.environ["R10_R2_TEST_ENV"] = before
        self.assertEqual(observed, "synthetic")

    def test_r10r2_rg_47_chdir_restore(self) -> None:
        before = pathlib.Path.cwd()
        with tempfile.TemporaryDirectory() as directory:
            try:
                os.chdir(directory); self.assertEqual(pathlib.Path.cwd(), pathlib.Path(directory))
            finally:
                os.chdir(before)
        self.assertEqual(pathlib.Path.cwd(), before)

    def test_r10r2_rg_48_socket_pair_closed(self) -> None:
        left, right = socket.socketpair(); left.sendall(b"x"); self.assertEqual(right.recv(1), b"x"); left.close(); right.close()
        self.assertEqual(left.fileno(), -1); self.assertEqual(right.fileno(), -1)

    def test_r10r2_rg_49_executor_shutdown(self) -> None:
        with ThreadPoolExecutor(max_workers=1) as pool: self.assertEqual(pool.submit(lambda: 7).result(timeout=2), 7)

    def test_r10r2_rg_50_cache_cleared(self) -> None:
        @functools.lru_cache(maxsize=2)
        def value(number: int) -> int: return number * 2
        self.assertEqual(value(3), 6); value.cache_clear(); self.assertEqual(value.cache_info().currsize, 0)

    def test_r10r2_rg_51_cpu_linux_pair_runs_before_long_host_matrix(self) -> None:
        specs = r10r2_replay_bundle.pair_specs(
            SimpleNamespace(win311=pathlib.Path("py311"), win312=pathlib.Path("py312"), win313=pathlib.Path("py313"))
        )
        serial_first, concurrent = r10r2_replay_bundle.execution_phases(specs)
        self.assertEqual([spec["pair_id"] for spec in serial_first], ["l313-cpu"])
        self.assertNotIn("l313-cpu", {spec["pair_id"] for spec in concurrent})
        self.assertEqual(len(serial_first) + len(concurrent), len(specs))

    def test_r10r2_rg_52_reviewer_deadline_dominates_supervisor(self) -> None:
        reviewer_seconds = 1050.0
        command = r10r2_replay_bundle.command_for_supervisor(
            "python", "default", None, "normal", reviewer_seconds
        )
        supervisor_seconds = float(command[command.index("--deadline-seconds") + 1])
        self.assertLess(supervisor_seconds, reviewer_seconds)

    def test_r10r2_rg_53_dynamic_discovery_count_comes_from_raw_rows(self) -> None:
        discovered_ids = [f"tests.synthetic.test_{index:03d}" for index in range(416)]
        rows = [{"test_count": len(discovered_ids)} for _ in range(24)]
        policy = {
            "discovered_test_ids": discovered_ids,
            "manifest_test_ids": list(discovered_ids),
        }
        self.assertEqual(
            r10r2_replay_bundle.validated_discovered_test_count(rows, policy),
            416,
        )
        inconsistent = list(rows)
        inconsistent[-1] = {"test_count": 415}
        with self.assertRaises(ValueError):
            r10r2_replay_bundle.validated_discovered_test_count(inconsistent, policy)
