"""Instrument tests for agent_loop_hosted_v1 (no network, no credentials).

The counting proxy is exercised against a loopback plaintext HTTP server

(``use_tls=False``) so request counting, keep-alive framing, chunked bodies,

Host rewriting, header injection and CONNECT tunnels are all checked locally.

"""

from __future__ import annotations

import http.client

import socket

import subprocess

import sys

import tempfile

import threading

import unittest

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from pathlib import Path

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

import agent_loop_hosted as alh  # noqa: E402

from oakbench.httpcount import HttpCountingProxy, RequestRecord, route_shape  # noqa: E402

class _Handler(BaseHTTPRequestHandler):

    protocol_version = "HTTP/1.1"

    seen_hosts: list = []

    seen_injected: list = []

    def log_message(self, *a):  # noqa: D401

        pass

    def do_GET(self):  # noqa: N802

        _Handler.seen_hosts.append(self.headers.get("Host"))

        _Handler.seen_injected.append(self.headers.get("X-Bench-Auth"))

        if self.path.startswith("/chunked"):

            self.send_response(200)

            self.send_header("Transfer-Encoding", "chunked")

            self.end_headers()

            for part in (b"hello ", b"world"):

                self.wfile.write(b"%x\r\n%s\r\n" % (len(part), part))

            self.wfile.write(b"0\r\n\r\n")

            return

        body = b"x" * 1000

        self.send_response(200)

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

        self.end_headers()

        self.wfile.write(body)

    def do_POST(self):  # noqa: N802

        n = int(self.headers.get("Content-Length", "0"))

        data = self.rfile.read(n)

        self.send_response(200)

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

        self.end_headers()

        self.wfile.write(data)

class HttpCountingProxyTest(unittest.TestCase):

    def setUp(self):

        _Handler.seen_hosts = []

        _Handler.seen_injected = []

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

        self.port = self.srv.server_address[1]

        threading.Thread(target=self.srv.serve_forever, daemon=True).start()

        self.proxy = HttpCountingProxy("127.0.0.1", upstream_port=self.port, use_tls=False,

                                       inject_headers={"X-Bench-Auth": "secret"}).start()

    def tearDown(self):

        self.proxy.stop()

        self.srv.shutdown()

    def test_counts_keepalive_requests_and_bodies(self):

        host, port = self.proxy.address

        c = http.client.HTTPConnection(host, port, timeout=5)

        c.request("GET", "/api/o/r/branches?token=abc")

        self.assertEqual(len(c.getresponse().read()), 1000)

        c.request("POST", "/api/o/r/commits/info", body=b"y" * 300)

        self.assertEqual(c.getresponse().read(), b"y" * 300)

        c.request("GET", "/chunked")

        self.assertEqual(c.getresponse().read(), b"hello world")

        c.close()

        s = self.proxy.snapshot()

        self.assertEqual(s.requests, 3)

        self.assertEqual(s.connections, 1)

        self.assertEqual([r.method for r in s.records], ["GET", "POST", "GET"])

        # query strings are never recorded (tokens can live there)

        self.assertEqual(s.records[0].path, "/api/o/r/branches")

        self.assertGreaterEqual(s.records[1].request_bytes, 300)

        self.assertGreater(s.response_bytes, 1300)

        # Host rewritten to the upstream; injected header delivered, never recorded

        self.assertTrue(all(h.startswith("127.0.0.1:") for h in _Handler.seen_hosts))

        self.assertEqual(_Handler.seen_injected, ["secret", "secret"])

        self.assertNotIn("secret", repr(s.records))

    def test_connect_tunnel_counts_bytes_only(self):

        host, port = self.proxy.address

        sock = socket.create_connection((host, port), timeout=5)

        sock.sendall(f"CONNECT 127.0.0.1:{self.port} HTTP/1.1\r\nHost: x\r\n\r\n".encode())

        self.assertIn(b"200", sock.recv(1024))

        sock.sendall(b"GET /plain HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n")

        data = b""

        while True:

            b = sock.recv(65536)

            if not b:

                break

            data += b

        sock.close()

        self.assertTrue(data.endswith(b"x" * 1000))

        import time

        time.sleep(0.1)

        s = self.proxy.snapshot()

        self.assertEqual(s.requests, 0)  # tunnel contents are opaque: no request count

        self.assertEqual(len(s.tunnels), 1)

        self.assertEqual(s.tunnel_hosts()["127.0.0.1"]["connections"], 1)

        self.assertGreater(s.tunnel_bytes, 1000)

    def test_route_shape_collapses_hashes(self):

        self.assertEqual(route_shape("/api/o/r/commits/" + "ab" * 32), "/api/o/r/commits/:hash")

        self.assertEqual(route_shape("/api/o/r/runs/12"), "/api/o/r/runs/:n")

class SerialRoundsTest(unittest.TestCase):

    def _r(self, s, e):

        return RequestRecord("GET", "/", 200, 0, 0, s, 1, e, 1)

    def test_parallel_requests_share_a_round(self):

        self.assertEqual(alh.serial_rounds([self._r(0, 10), self._r(1, 9), self._r(2, 12)]), 1)

    def test_sequential_requests_each_cost_a_round(self):

        self.assertEqual(alh.serial_rounds([self._r(0, 10), self._r(11, 20), self._r(21, 30)]), 3)

    def test_empty_is_zero_rounds(self):

        self.assertEqual(alh.serial_rounds([]), 0)

class GitTraceParseTest(unittest.TestCase):

    def test_http2_request_counted_once(self):

        # Real libcurl HTTP/2 trace shape: info lines AND a Send-header request

        # line for the same request (regression: these were counted twice).

        text = (

            "== Info: [HTTP/2] [1] [:method: GET]\n"

            "== Info: [HTTP/2] [1] [:path: /o/r.git/info/refs?service=git-upload-pack]\n"

            "=> Send header: GET /o/r.git/info/refs?service=git-upload-pack HTTP/2\n"

            "<= Recv header: HTTP/2 401\n"

            "== Info: [HTTP/2] [3] [authorization: Basic <redacted>]\n"

            "== Info: [HTTP/2] [3] [:method: GET]\n"

            "== Info: [HTTP/2] [3] [:path: /o/r.git/info/refs?service=git-upload-pack]\n"

            "=> Send header: GET /o/r.git/info/refs?service=git-upload-pack HTTP/2\n"

            "== Info: [HTTP/2] [5] [:method: POST]\n"

            "== Info: [HTTP/2] [5] [:path: /o/r.git/git-upload-pack]\n"

            "=> Send header: POST /o/r.git/git-upload-pack HTTP/2\n"

            "=> Send header: POST /o/r.git/git-receive-pack HTTP/1.1\n"

        )

        with tempfile.NamedTemporaryFile("w", delete=False) as fh:

            fh.write(text)

        n, routes = alh.parse_git_trace(fh.name)

        self.assertEqual(n, 4)

        self.assertEqual(routes, {"GET /info/refs": 2, "POST /git-upload-pack": 1, "POST /git-receive-pack": 1})

class OracleTest(unittest.TestCase):

    def test_workspace_digest_matches_git_blob_ids_and_ignores_vcs_dirs(self):

        with tempfile.TemporaryDirectory() as d:

            root = Path(d)

            subprocess.run(["git", "init", "-q", str(root)], check=True)

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

            (root / "sub").mkdir()

            (root / "sub" / "b.txt").write_text("beta\n")

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

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

            env = {"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@x", "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@x", "PATH": "/usr/bin:/bin:/opt/homebrew/bin"}

            subprocess.run(["git", "-C", d, "add", "a.txt", "sub"], check=True)

            subprocess.run(["git", "-C", d, "commit", "-q", "-m", "x"], check=True, env=env)

            self.assertEqual(alh.workspace_digest(root), alh.fixture_tree(root, "HEAD"))

            ok, _ = alh.acquire_oracle(root, root)

            self.assertTrue(ok)

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

            ok, detail = alh.acquire_oracle(root, root)

            self.assertFalse(ok)

            self.assertIn("differ=1", detail)

    def test_task_patch_is_three_files_about_fifty_lines(self):

        p = alh.task_patch("t")

        self.assertEqual(len(p), 3)

        lines = sum(text.count("\n") for _k, text in p.values())

        self.assertTrue(45 <= lines <= 55, lines)

        self.assertTrue(set(p) <= alh.TASK_PATHS)

class SeedFixtureTest(unittest.TestCase):

    def test_seed_is_deterministic_sliced_and_has_merges(self):

        import seed_agent_loop_fixture as seed

        with tempfile.TemporaryDirectory() as d:

            src = Path(d) / "src"

            for i in range(230):

                p = src / f"dir{i % 7}" / f"f{i}.txt"

                p.parent.mkdir(parents=True, exist_ok=True)

                p.write_text(f"file {i}\n")

            (src / ".github").mkdir()

            (src / ".github" / "ci.yml").write_text("excluded\n")

            heads = []

            for name in ("a", "b"):

                g = Path(d) / name

                seed.seed_git(src, g, commits=40, seed=7, merge_every=5, files=5)

                out = subprocess.run(["git", "-C", str(g), "rev-parse", "main"], capture_output=True, text=True).stdout.strip()

                heads.append(out)

                count = int(subprocess.run(["git", "-C", str(g), "rev-list", "--count", "main"], capture_output=True, text=True).stdout)

                merges = int(subprocess.run(["git", "-C", str(g), "rev-list", "--count", "--merges", "main"], capture_output=True, text=True).stdout)

                self.assertEqual(count, 40)

                self.assertGreater(merges, 0)

                files = subprocess.run(["git", "-C", str(g), "ls-files"], capture_output=True, text=True).stdout.split()

                self.assertNotIn(".github/ci.yml", files)

                # every commit in the import slices adds <= IMPORT_SLICE files

                first = subprocess.run(["git", "-C", str(g), "rev-list", "--reverse", "main"], capture_output=True, text=True).stdout.split()[0]

                n0 = len(subprocess.run(["git", "-C", str(g), "ls-tree", "-r", "--name-only", first], capture_output=True, text=True).stdout.split())

                self.assertLessEqual(n0, seed.IMPORT_SLICE)

            self.assertEqual(heads[0], heads[1])

if __name__ == "__main__":

    unittest.main()