"""Tests for shallow_worker_branch_refresh_v1 instruments: fixture, request counter, oracle."""

from __future__ import annotations

import os

import shutil

import socket

import sqlite3

import sys

import tempfile

import threading

import time

import unittest

import http.client

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from pathlib import Path

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

import task_loop as lane

from oakbench import merge_history_fixture as fixture

from oakbench import shallow_refresh

from oakbench.httpcount import HttpFramingError, HttpRequestStreamParser, RequestCountingProxy

from oakbench.subjects import Subject

GIT = shutil.which("git")

def git_env() -> dict[str, str]:

    env = {key: value for key, value in os.environ.items() if key in {"PATH", "HOME", "TMPDIR"}}

    env.update({"GIT_CONFIG_GLOBAL": os.devnull, "GIT_CONFIG_NOSYSTEM": "1"})

    return env

class MergeHistoryFixtureTest(unittest.TestCase):

    def test_stream_is_deterministic_and_seed_sensitive(self) -> None:

        first, manifest_a = fixture.build_stream(300, 7)

        second, manifest_b = fixture.build_stream(300, 7)

        other, _ = fixture.build_stream(300, 8)

        self.assertEqual(first, second)

        self.assertEqual(manifest_a.tree_content_hash, manifest_b.tree_content_hash)

        self.assertNotEqual(first, other)

    def test_exact_commit_count_and_merge_ratio(self) -> None:

        for commits, seed in ((1000, 1), (2000, 2)):

            _stream, manifest = fixture.build_stream(commits, seed)

            self.assertEqual(manifest.commits, commits)

            self.assertGreater(manifest.merge_commits, 0)

            self.assertTrue(0.05 <= manifest.merge_fraction <= 0.10, manifest.merge_fraction)

    def test_as_dict_satisfies_row_contract_types(self) -> None:

        _stream, manifest = fixture.build_stream(50, 3)

        fields = manifest.as_dict()

        for key in ("fixture_id", "fixture_version", "fixture_seed", "fixture_generator"):

            self.assertIsInstance(fields[key], str)

        self.assertIsNone(fields["fixture_git_head"])  # unobserved until imported

    @unittest.skipUnless(GIT, "git required")

    def test_generated_history_matches_golden_head_and_model_tree(self) -> None:

        with tempfile.TemporaryDirectory() as tmp:

            manifest = fixture.generate(Path(tmp) / "src.git", 1000, 1, git_bin=GIT, env=git_env())

            # Golden identity: changing this is a FIXTURE_SCHEMA bump (ADR-0005).

            self.assertEqual(manifest.head, "ed3b72471b73274cdf80f542e7b5835faf6badc2")

            self.assertEqual(manifest.merge_commits, 66)

            checkout = fixture.tree_content_hash(fixture.worktree_files(Path(tmp) / "src.git"))

            self.assertEqual(checkout, manifest.tree_content_hash)

            merges = lane.subprocess.run(

                [GIT, "rev-list", "--merges", "--count", "main"], cwd=Path(tmp) / "src.git",

                env=git_env(), text=True, stdout=lane.subprocess.PIPE, check=True,

            ).stdout.strip()

            self.assertEqual(int(merges), 66)

REQUESTS = (

    b"POST /api/o/r/commits/info HTTP/1.1\r\nContent-Length: 5\r\n\r\nhello"

    b"GET /api/o/r?x=1 HTTP/1.1\r\nHost: h\r\n\r\n"

    b"POST /api/o/r/push HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\n3\r\nabc\r\n0\r\n\r\n"

)

class HttpRequestStreamParserTest(unittest.TestCase):

    def test_pipelined_content_length_and_chunked_requests(self) -> None:

        parser = HttpRequestStreamParser()

        self.assertEqual(parser.feed(REQUESTS), 3)

        self.assertEqual(

            [(item.method, item.path) for item in parser.requests],

            [("POST", "/api/o/r/commits/info"), ("GET", "/api/o/r"), ("POST", "/api/o/r/push")],

        )

        self.assertTrue(parser.at_request_boundary)

    def test_byte_at_a_time_framing_is_identical(self) -> None:

        parser = HttpRequestStreamParser()

        started = sum(parser.feed(REQUESTS[index : index + 1]) for index in range(len(REQUESTS)))

        self.assertEqual(started, 3)

        self.assertEqual(len(parser.requests), 3)

    def test_body_bytes_resembling_a_request_are_not_counted(self) -> None:

        body = b"GET /fake HTTP/1.1\r\n\r\n"

        parser = HttpRequestStreamParser()

        parser.feed(b"POST /x HTTP/1.1\r\nContent-Length: %d\r\n\r\n" % len(body) + body)

        self.assertEqual([item.path for item in parser.requests], ["/x"])

    def test_non_http_stream_is_rejected(self) -> None:

        with self.assertRaises(HttpFramingError):

            HttpRequestStreamParser().feed(b"\x16\x03\x01garbage\r\n\r\n")

class _Handler(BaseHTTPRequestHandler):

    protocol_version = "HTTP/1.1"

    def _reply(self) -> None:

        length = int(self.headers.get("Content-Length") or 0)

        if length:

            self.rfile.read(length)

        body = b"ok"

        self.send_response(200)

        self.send_header("Content-Length", str(len(body)))

        self.end_headers()

        self.wfile.write(body)

    do_GET = _reply

    do_POST = _reply

    def log_message(self, format: str, *args) -> None:  # noqa: A002

        pass

class RequestCountingProxyTest(unittest.TestCase):

    def setUp(self) -> None:

        self.httpd = ThreadingHTTPServer(("127.0.0.1", 0), _Handler)

        self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)

        self.thread.start()

    def tearDown(self) -> None:

        self.httpd.shutdown()

        self.httpd.server_close()

    def _proxy(self, **kwargs) -> RequestCountingProxy:  # type: ignore[no-untyped-def]

        proxy = RequestCountingProxy(

            "127.0.0.1", self.httpd.server_address[1], path_suffixes=("/commits/info",), **kwargs

        ).start()

        self.addCleanup(proxy.stop)

        return proxy

    def _wait_for(self, proxy: RequestCountingProxy, count: int) -> None:

        deadline = time.monotonic() + 2.0

        while time.monotonic() < deadline and (proxy.request_stats().request_count or 0) < count:

            time.sleep(0.01)

    def test_counts_keepalive_requests_per_path_suffix(self) -> None:

        proxy = self._proxy()

        connection = http.client.HTTPConnection("127.0.0.1", proxy.address[1], timeout=5)

        for path in ("/api/o/r/commits/info", "/api/o/r/commits/info", "/api/o/r"):

            connection.request("POST", path, body=b"{}")

            self.assertEqual(connection.getresponse().read(), b"ok")

        connection.close()

        self._wait_for(proxy, 3)

        stats = proxy.request_stats()

        self.assertEqual(stats.request_count, 3)

        self.assertEqual(stats.count_suffix("/commits/info"), 2)

        self.assertEqual(stats.methods, {"POST": 3})

        self.assertEqual(proxy.snapshot().connections, 1)

        proxy.reset_counts()

        self.assertEqual(proxy.request_stats().request_count, 0)  # measured zero after reset

    def test_unframable_stream_reports_null_not_zero(self) -> None:

        proxy = self._proxy()

        with socket.create_connection(proxy.address, timeout=5) as sock:

            sock.sendall(b"\x16\x03\x01not http\r\n\r\n")

            sock.shutdown(socket.SHUT_WR)

            sock.settimeout(2)

            try:

                sock.recv(1024)

            except OSError:

                pass

        deadline = time.monotonic() + 2.0

        while time.monotonic() < deadline and proxy.request_stats().framing_errors == 0:

            time.sleep(0.01)

        stats = proxy.request_stats()

        self.assertIsNone(stats.request_count)

        self.assertIsNone(stats.count_suffix("/commits/info"))

        fields = shallow_refresh.request_metric_fields(stats, proxy.snapshot(), "/commits/info")

        self.assertIsNone(fields["http_request_count"])

        self.assertIsNone(fields["http_commits_info_request_count"])

    def test_injected_delay_is_applied_per_request_and_labeled(self) -> None:

        proxy = self._proxy(injected_request_delay_ms=50)

        connection = http.client.HTTPConnection("127.0.0.1", proxy.address[1], timeout=5)

        started = time.perf_counter()

        for _ in range(3):

            connection.request("GET", "/x")

            connection.getresponse().read()

        elapsed = time.perf_counter() - started

        connection.close()

        self.assertGreaterEqual(elapsed, 0.15)

        self.assertEqual(proxy.request_stats().injected_request_delay_ms, 50.0)

    def test_negative_delay_rejected(self) -> None:

        with self.assertRaises(ValueError):

            RequestCountingProxy("127.0.0.1", 1, injected_request_delay_ms=-1)

class OracleTest(unittest.TestCase):

    HEAD = "a" * 64

    TREE = "t" * 64

    def evaluate(self, **overrides):  # type: ignore[no-untyped-def]

        observed = {

            "observed_branch": "task",

            "observed_branch_head": self.HEAD,

            "observed_head": self.HEAD,

            "worktree_tree_hash": self.TREE,

        }

        observed.update(overrides)

        return shallow_refresh.evaluate_oracle(

            expected_branch="task", expected_head=self.HEAD, expected_tree_hash=self.TREE, **observed

        )

    def test_accepts_exact_final_state(self) -> None:

        self.assertTrue(self.evaluate()["oracle_passed"])

    def test_rejects_wrong_head(self) -> None:

        result = self.evaluate(observed_branch_head="b" * 64, observed_head="b" * 64)

        self.assertFalse(result["oracle_passed"])

        self.assertFalse(result["oracle_branch_head_is_server_main"])

    def test_rejects_wrong_tree(self) -> None:

        result = self.evaluate(worktree_tree_hash="x" * 64)

        self.assertFalse(result["oracle_passed"])

        self.assertFalse(result["oracle_worktree_tree_matches_fixture"])

    def test_rejects_wrong_branch(self) -> None:

        self.assertFalse(self.evaluate(observed_branch="main")["oracle_passed"])

    def test_unobserved_server_head_never_passes(self) -> None:

        result = shallow_refresh.evaluate_oracle(

            expected_branch="task", expected_head=None, expected_tree_hash=self.TREE,

            observed_branch="task", observed_branch_head=None, observed_head=None,

            worktree_tree_hash=self.TREE,

        )

        self.assertFalse(result["oracle_passed"])

    def test_worktree_tree_hash_detects_content_change(self) -> None:

        with tempfile.TemporaryDirectory() as tmp:

            root = Path(tmp)

            (root / "a.txt").write_text("one\n")

            (root / ".oak").mkdir()

            (root / ".oak" / "oak.db").write_text("ignored")

            before = shallow_refresh.worktree_tree_hash(root)

            (root / "a.txt").write_text("two\n")

            self.assertNotEqual(before, shallow_refresh.worktree_tree_hash(root))

        self.assertIsNone(shallow_refresh.worktree_tree_hash(Path(tmp) / "missing"))

class OakDiskStateTest(unittest.TestCase):

    def test_missing_database_is_unobserved_not_zero(self) -> None:

        with tempfile.TemporaryDirectory() as tmp:

            state = shallow_refresh.oak_disk_state(Path(tmp), "task")

        self.assertEqual(state["error"], "oak_db_missing")

        self.assertIsNone(state["commit_rows"])

        self.assertIsNone(state["ancestry_missing_edges"])

        self.assertIsNone(shallow_refresh.serve_depth_behavior(state["commit_rows"]))

    def test_reads_branch_head_and_ancestry_gaps(self) -> None:

        with tempfile.TemporaryDirectory() as tmp:

            db_dir = Path(tmp) / ".oak"

            db_dir.mkdir()

            connection = sqlite3.connect(str(db_dir / "oak.db"))

            connection.executescript(

                "CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT);"

                "CREATE TABLE branch_heads (branch_name TEXT PRIMARY KEY, head_hash TEXT);"

                "CREATE TABLE commits (hash TEXT PRIMARY KEY, parent_hash TEXT, merge_parent_hash TEXT);"

                "INSERT INTO metadata VALUES ('current_branch','task'),('head','h3');"

                "INSERT INTO branch_heads VALUES ('task','h3');"

                "INSERT INTO commits VALUES ('h3','h2','m1'),('h2','h1',NULL);"

            )

            connection.commit()

            connection.close()

            state = shallow_refresh.oak_disk_state(Path(tmp), "task")

        self.assertEqual((state["current_branch"], state["head"], state["branch_head"]), ("task", "h3", "h3"))

        self.assertEqual(state["commit_rows"], 2)

        self.assertEqual(state["ancestry_missing_edges"], 2)  # h1 and m1 absent

    def test_serve_depth_classification(self) -> None:

        self.assertEqual(shallow_refresh.serve_depth_behavior(1), shallow_refresh.SERVE_DEPTH_HONORED)

        self.assertEqual(shallow_refresh.serve_depth_behavior(1000), shallow_refresh.SERVE_DEPTH_IGNORED)

class ShallowRefreshLaneTest(unittest.TestCase):

    def test_fixture_spec_parsing(self) -> None:

        self.assertEqual(lane.parse_shallow_refresh_fixtures("1000:1, 2000:2"), [(1000, 1), (2000, 2)])

        for bad in ("", "1000", "x:1", "0:1"):

            with self.assertRaises(ValueError):

                lane.parse_shallow_refresh_fixtures(bad)

    def test_summed_request_counts_stay_null_when_any_step_unobserved(self) -> None:

        self.assertIsNone(lane._summed_or_none([{"k": 1}, {"k": None}], "k"))

        self.assertEqual(lane._summed_or_none([{"k": 1}, {"k": 0}], "k"), 1)

    def test_missing_serve_binary_emits_skip_rows(self) -> None:

        subjects = [Subject("git", "git", "Git", Path("/usr/bin/git"))]

        with tempfile.TemporaryDirectory() as tmp:

            rows = lane.run_shallow_refresh_campaign(

                subjects, None, None, [(10, 1)], Path(tmp), 1, {"bench_id": "t"}, 1000, False, 1, 0.0

            )

        self.assertEqual(len(rows), 1)

        self.assertEqual(rows[0]["returncode"], lane.SKIP_RETURNCODE)

        self.assertEqual(rows[0]["scenario"], lane.SHALLOW_REFRESH_SCENARIO)

        self.assertIn("--oak-serve-bin", rows[0]["skip_reason"])

    def test_oak_cells_route_refresh_command_and_repeat_row(self) -> None:

        from unittest.mock import patch

        calls: list[tuple[str, list[str]]] = []

        def fake_step(subject, proxy, operation, command, *args, **kwargs):  # type: ignore[no-untyped-def]

            calls.append((operation, command))

            return {"operation": operation, "command": command, "returncode": 0, "elapsed_ms": 1.0,

                    "http_request_count": 1, "http_commits_info_request_count": 0,

                    "network_wire_bytes_observed": 10, "network_connections_observed": 1}

        def fake_disk(root, branch):  # type: ignore[no-untyped-def]

            return {"current_branch": branch, "head": "h", "branch_head": "h", "commit_rows": 1,

                    "ancestry_missing_edges": 0, "error": None}

        subject = Subject("oak_x", "oak", "Oak", Path("/bin/oak"))

        with tempfile.TemporaryDirectory() as tmp, patch.object(lane, "_shallow_refresh_step", fake_step), \

                patch.object(lane.shallow_refresh, "oak_disk_state", fake_disk), \

                patch.object(lane.shallow_refresh, "worktree_tree_hash", return_value="T"):

            for cell in ("shallow", "shallow_pull"):

                rows = lane._shallow_refresh_oak_trial(

                    subject, None, "http://x", "qa/r", "h", "T", Path(tmp) / cell, 0,

                    {"bench_id": "t"}, 1000, {}, cell=cell,

                )

                operations = [row["operation"] for row in rows]

                self.assertIn(f"{cell}.total", operations)

                self.assertEqual(

                    f"{cell}.branch_create_repeat_local" in operations, cell == "shallow"

                )

        refresh = dict(calls)

        self.assertEqual(refresh["shallow.parent_refresh"][1:], ["fetch"])

        self.assertEqual(refresh["shallow_pull.parent_refresh"][1:], ["pull"])

        self.assertEqual(refresh["shallow_pull.branch_create"][1:], ["switch", "-c", "task"])

if __name__ == "__main__":

    unittest.main()