from __future__ import annotations

import contextlib

import os

import selectors

import subprocess

import sys

import tempfile

import textwrap

import time

import unittest

from pathlib import Path

from unittest import mock

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

from oakbench.runlock import measurement_lock

from oakbench import runlock

try:

    import fcntl  # noqa: F401

except ImportError:  # pragma: no cover - Windows

    HAS_FCNTL = False

else:

    HAS_FCNTL = True

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

@contextlib.contextmanager

def isolated_lock_env(lock_path: Path, lock_value: str | None = None):

    updates = {"OAK_BENCH_LOCK_PATH": str(lock_path)}

    if lock_value is not None:

        updates["OAK_BENCH_LOCK"] = lock_value

    with mock.patch.dict(os.environ, updates, clear=False):

        if lock_value is None:

            os.environ.pop("OAK_BENCH_LOCK", None)

        yield

def subprocess_env(lock_path: Path) -> dict[str, str]:

    env = os.environ.copy()

    env["OAK_BENCH_LOCK_PATH"] = str(lock_path)

    env.pop("OAK_BENCH_LOCK", None)

    env["PYTHONDONTWRITEBYTECODE"] = "1"

    pythonpath = env.get("PYTHONPATH")

    env["PYTHONPATH"] = str(SCRIPTS_DIR) if not pythonpath else str(SCRIPTS_DIR) + os.pathsep + pythonpath

    return env

def readline_with_timeout(stream, timeout_s: float) -> str | None:

    selector = selectors.DefaultSelector()

    try:

        selector.register(stream, selectors.EVENT_READ)

        if not selector.select(timeout_s):

            return None

        return stream.readline()

    finally:

        selector.close()

class MeasurementLockTest(unittest.TestCase):

    def test_configured_lock_identity_is_canonical_and_recorded(self):

        with tempfile.TemporaryDirectory() as tmp:

            root = Path(tmp).resolve()

            actual = root / "actual"

            actual.mkdir()

            alias = root / "alias"

            alias.symlink_to(actual, target_is_directory=True)

            with isolated_lock_env(alias / "measure.lock"):

                self.assertEqual(runlock._lock_path(), actual / "measure.lock")

                with measurement_lock("canonical-identity") as info:

                    self.assertEqual(info.path, str(actual / "measure.lock"))

    @unittest.skipUnless(HAS_FCNTL, "fcntl not available")

    def test_acquiring_while_another_process_holds_lock_blocks_until_release(self) -> None:

        with tempfile.TemporaryDirectory(prefix="oakbench-runlock-test-") as tmp:

            tmp_path = Path(tmp)

            lock_path = tmp_path / "measurement.lock"

            holder_script = tmp_path / "holder.py"

            holder_script.write_text(

                textwrap.dedent(

                    f"""

                    import sys

                    import time

                    sys.path.insert(0, {str(SCRIPTS_DIR)!r})

                    from oakbench.runlock import measurement_lock

                    with measurement_lock("holder-process"):

                        print("ready", flush=True)

                        time.sleep(0.75)

                    """

                )

            )

            proc = subprocess.Popen(

                [sys.executable, str(holder_script)],

                stdout=subprocess.PIPE,

                stderr=subprocess.PIPE,

                text=True,

                env=subprocess_env(lock_path),

            )

            try:

                ready = readline_with_timeout(proc.stdout, 2.0) if proc.stdout is not None else None

                if ready != "ready\n":

                    proc.kill()

                    _, stderr = proc.communicate(timeout=2.0)

                    self.fail(f"holder process did not acquire lock: ready={ready!r} stderr={stderr}")

                start = time.perf_counter()

                with isolated_lock_env(lock_path):

                    with measurement_lock("waiter") as lock_info:

                        pass

                elapsed_ms = (time.perf_counter() - start) * 1000

                self.assertTrue(lock_info.enabled)

                self.assertIsNotNone(lock_info.wait_ms)

                self.assertGreaterEqual(lock_info.wait_ms, 400.0)

                self.assertGreaterEqual(elapsed_ms, 400.0)

                _, stderr = proc.communicate(timeout=2.0)

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

            finally:

                if proc.poll() is None:

                    proc.kill()

                    proc.communicate(timeout=2.0)

    def test_oak_bench_lock_off_bypasses(self) -> None:

        with tempfile.TemporaryDirectory(prefix="oakbench-runlock-test-") as tmp:

            lock_path = Path(tmp) / "measurement.lock"

            with isolated_lock_env(lock_path, "off"):

                with measurement_lock("disabled") as lock_info:

                    pass

            self.assertFalse(lock_info.enabled)

            self.assertIsNone(lock_info.wait_ms)

            self.assertFalse(lock_path.exists())

    @unittest.skipUnless(HAS_FCNTL, "fcntl not available")

    def test_wait_ms_is_near_zero_when_uncontended(self) -> None:

        with tempfile.TemporaryDirectory(prefix="oakbench-runlock-test-") as tmp:

            lock_path = Path(tmp) / "measurement.lock"

            with isolated_lock_env(lock_path):

                with measurement_lock("uncontended") as lock_info:

                    pass

            self.assertTrue(lock_info.enabled)

            self.assertIsNotNone(lock_info.wait_ms)

            self.assertLess(lock_info.wait_ms, 100.0)

    @unittest.skipUnless(HAS_FCNTL, "fcntl not available")

    def test_lock_file_content_contains_holder_label(self) -> None:

        with tempfile.TemporaryDirectory(prefix="oakbench-runlock-test-") as tmp:

            lock_path = Path(tmp) / "measurement.lock"

            with isolated_lock_env(lock_path):

                with measurement_lock("content-label"):

                    content = lock_path.read_text()

            self.assertIn("content-label", content)

            self.assertIn(f"pid={os.getpid()}", content)

if __name__ == "__main__":

    unittest.main()