#!/usr/bin/env python3
"""Black-box tests for agentctl serve / claim new / integrate prepare / pi packet."""

from __future__ import annotations

import json
import pathlib
import shutil
import socket
import subprocess
import sys
import tempfile
import threading
import time
import unittest
import urllib.error
import urllib.request

SOURCE = pathlib.Path(__file__).resolve().parents[1] / "agentctl.py"
POLICY_SOURCE = SOURCE.parents[1] / "configs" / "agent-governance.json"
CONSOLE_STATIC = SOURCE.parent / "governance-console"


def _free_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]


class ConsoleServeTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix="agentctl-serve-")
        self.addCleanup(self.temp.cleanup)
        self.root = pathlib.Path(self.temp.name) / "repo"
        self.root.mkdir()
        self.git("init", "-b", "dev")
        self.git("config", "user.name", "Agentctl Test")
        self.git("config", "user.email", "agentctl-test@example.invalid")
        (self.root / "scripts").mkdir()
        for name in (
            "agentctl.py", "agentctl_w4.py", "agentctl_catalog.py",
            "agentctl_visibility.py", "agentctl_console.py", "pi_subagents.py",
            "agent_worktree.py",
        ):
            src = SOURCE.parent / name
            if src.exists():
                shutil.copy2(src, self.root / "scripts" / name)
        static_dest = self.root / "scripts" / "governance-console"
        static_dest.mkdir(parents=True, exist_ok=True)
        if (CONSOLE_STATIC / "index.html").exists():
            shutil.copy2(CONSOLE_STATIC / "index.html", static_dest / "index.html")
        else:
            (static_dest / "index.html").write_text("<html>ok</html>", encoding="utf-8")
        dest = self.root / "configs"
        dest.mkdir(parents=True, exist_ok=True)
        shutil.copy2(POLICY_SOURCE, dest / "agent-governance.json")
        (self.root / ".gitignore").write_text(".locks/\n.worktrees/\n", encoding="utf-8")
        for directory in ("src/a", ".doc/ops", ".doc/status", ".doc/test-reports"):
            (self.root / directory).mkdir(parents=True, exist_ok=True)
        (self.root / "src/a" / ".keep").write_text("keep\n", encoding="utf-8")
        self._write_current([])
        (self.root / ".doc/status/FEATURE-MATRIX.md").write_text(
            "# Features\n\n| record_id | 模块 | 状态 |\n| --- | --- | --- |\n| core | core | planned |\n",
            encoding="utf-8")
        self.git("add", ".")
        self.git("commit", "-m", "fixture")
        self.cli("init")
        self._server = None
        self._thread = None

    def tearDown(self):
        if self._server is not None:
            self._server.shutdown()
            self._server.server_close()
        if self._thread is not None:
            self._thread.join(timeout=5)

    def git(self, *args, cwd=None):
        return subprocess.run(
            ["git", *args], cwd=cwd or self.root,
            text=True, capture_output=True, check=True,
        ).stdout.strip()

    def cli(self, *args, cwd=None, check=True):
        result = subprocess.run(
            [sys.executable, "-B", str(self.root / "scripts/agentctl.py"), *args],
            cwd=cwd or self.root, text=True, capture_output=True, timeout=60,
        )
        if check:
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        return result

    def _last_json(self, text: str):
        lines = [ln for ln in text.splitlines() if ln.strip()]
        return json.loads(lines[-1])

    def _write_current(self, claim_rows):
        rows = [
            "| owner | task | branch | worktree | mutex | status |",
            "| --- | --- | --- | --- | --- | --- |",
            *claim_rows,
        ]
        body = (
            "# Coordination\n\n## 焦点\n\nfocus\n\n## 并行认领\n\n"
            + "\n".join(rows) + "\n\n"
            "## 清理接管与遗留现场\n\n"
            "| 路径/事项 | 分类与保留原因 | cleanup_owner | 复查触发 |\n"
            "| --- | --- | --- | --- |\n"
            "| none | none | - | - |\n\n"
            "## 集成槽位\n\n"
            "| holder | phase | since | scope | queue |\n"
            "|--------|-------|-------|-------|-------|\n"
            "| - | idle | 2026-09-24 | test | - |\n\n"
            "## 受控约束\n\n- c\n\n## 活跃入口\n\n- e\n"
        )
        (self.root / ".doc/ops/CURRENT.md").write_text(body, encoding="utf-8")

    def initialized_task(self, task="alpha", capabilities=("docs:publish",)):
        path = self.root / ".worktrees" / task
        if not path.exists():
            self.git("worktree", "add", "-b", "feature/" + task, str(path), "dev")
        args = ["task", "register", "--id", task, "--owner", "Test Agent",
                "--scope", "src/a", "--scope", ".doc"]
        for cap in capabilities:
            args.extend(("--capability", cap))
        self.cli(*args, cwd=path)
        return path

    def start_server(self, host="127.0.0.1", port=None):
        port = port or _free_port()
        sys.path.insert(0, str(self.root / "scripts"))
        import agentctl as ac  # noqa: E402
        import agentctl_console as console  # noqa: E402

        # Ensure policy loaded for this fixture repo.
        repo = ac.Repo(str(self.root))
        ac.require_policy(repo)
        args = type("A", (), {"host": host, "port": port, "static": "scripts/governance-console"})()
        # Build server without serve_forever so tests control lifecycle.
        if host not in ("127.0.0.1", "localhost", "::1"):
            with self.assertRaises(ac.Denied) as ctx:
                console.serve(repo, args)
            return None, ctx.exception

        from http.server import ThreadingHTTPServer
        # Re-enter serve setup by calling internal pattern: use CLI subprocess for smoke,
        # and for unit endpoints use Handler via temporary monkey of serve_forever.
        holder = {}

        def run():
            class Stoppable(ThreadingHTTPServer):
                allow_reuse_address = True

            # Reconstruct Handler by invoking console.serve until emit, then hijack.
            # Simpler: spawn `agentctl serve` subprocess.
            pass

        proc = subprocess.Popen(
            [sys.executable, "-B", str(self.root / "scripts/agentctl.py"),
             "serve", "--host", host, "--port", str(port)],
            cwd=str(self.root), text=True,
            stdout=subprocess.PIPE, stderr=subprocess.PIPE,
        )

        def _stop_proc(p=proc):
            if p.poll() is None:
                p.terminate()
                try:
                    p.wait(timeout=3)
                except subprocess.TimeoutExpired:
                    p.kill()
            if p.stdout:
                p.stdout.close()
            if p.stderr:
                p.stderr.close()
        self.addCleanup(_stop_proc)
        # Wait until port accepts connections.
        deadline = time.time() + 8
        while time.time() < deadline:
            if proc.poll() is not None:
                out = (proc.stdout.read() if proc.stdout else "") + (proc.stderr.read() if proc.stderr else "")
                raise AssertionError(f"serve exited early: {out}")
            try:
                with socket.create_connection(("127.0.0.1", port), timeout=0.2):
                    break
            except OSError:
                time.sleep(0.05)
        else:
            proc.kill()
            raise AssertionError("serve did not bind in time")
        return port, proc

    def http_json(self, port, path, method="GET", body=None, expect_status=200):
        data = None if body is None else json.dumps(body).encode("utf-8")
        req = urllib.request.Request(
            f"http://127.0.0.1:{port}{path}",
            data=data, method=method,
            headers={"Content-Type": "application/json"} if data is not None else {},
        )
        try:
            with urllib.request.urlopen(req, timeout=10) as resp:
                raw = resp.read().decode("utf-8")
                status = resp.status
                ctype = resp.headers.get("Content-Type", "")
        except urllib.error.HTTPError as exc:
            raw = exc.read().decode("utf-8")
            status = exc.code
            ctype = exc.headers.get("Content-Type", "")
        self.assertEqual(status, expect_status, raw)
        if "json" in ctype or raw[:1] in "{[":
            return json.loads(raw)
        return raw

    def test_non_loopback_host_denied(self):
        sys.path.insert(0, str(self.root / "scripts"))
        import agentctl as ac  # noqa: E402
        import agentctl_console as console  # noqa: E402
        repo = ac.Repo(str(self.root))
        ac.require_policy(repo)
        args = type("A", (), {"host": "0.0.0.0", "port": 7432, "static": "scripts/governance-console"})()
        with self.assertRaises(ac.Denied) as ctx:
            console.serve(repo, args)
        self.assertEqual(ctx.exception.code, "INVALID_REQUEST")

    def test_get_eight_endpoints_and_spa(self):
        port, _ = self.start_server()
        for path in (
            "/api/status", "/api/brief", "/api/queue", "/api/claims",
            "/api/cleanup", "/api/docs/drift", "/api/catalog/drift", "/api/daemon",
        ):
            payload = self.http_json(port, path)
            self.assertIsInstance(payload, (dict, str))
            if path == "/api/brief":
                self.assertTrue(isinstance(payload, str) and payload.startswith("#"))
            else:
                self.assertIsInstance(payload, dict)
        brief_json = self.http_json(port, "/api/brief?format=json")
        self.assertIn("brief", brief_json)
        html = self.http_json(port, "/", expect_status=200)
        self.assertIn("agentctl console", html if isinstance(html, str) else str(html))

    def test_unknown_post_shell_404(self):
        port, _ = self.start_server()
        payload = self.http_json(port, "/api/shell", method="POST", body={}, expect_status=404)
        self.assertEqual(payload["code"], "NOT_FOUND")

    def test_claim_new_missing_task_400(self):
        port, _ = self.start_server()
        payload = self.http_json(port, "/api/claim/new", method="POST", body={}, expect_status=400)
        self.assertEqual(payload["code"], "INVALID_REQUEST")

    def test_claim_new_without_docs_publish_403(self):
        self.initialized_task(capabilities=())
        port, _ = self.start_server()
        payload = self.http_json(
            port, "/api/claim/new", method="POST",
            body={"task": "alpha", "owner": "Test Agent", "mutex": "m1"},
            expect_status=403,
        )
        self.assertEqual(payload["code"], "CAPABILITY_REQUIRED")

    def test_claim_new_cli_create_replace_and_replay(self):
        task = self.initialized_task(capabilities=("docs:publish",))
        rid = "alpha-claim-new-test1"
        created = self._last_json(self.cli(
            "claim", "new", "--task", "alpha", "--owner", "Test Agent",
            "--mutex", "m1", "--request-id", rid, cwd=task,
        ).stdout)
        self.assertEqual(created["status"], "created")
        text = (task / ".doc/ops/CURRENT.md").read_text(encoding="utf-8")
        self.assertIn("| Test Agent | alpha |", text)
        self.assertIn("| m1 | in_progress |", text)
        # Identical row → idempotent (no second apply)
        again = self._last_json(self.cli(
            "claim", "new", "--task", "alpha", "--owner", "Test Agent",
            "--mutex", "m1", "--request-id", rid, cwd=task,
        ).stdout)
        self.assertEqual(again["status"], "idempotent")
        # Without replace while row differs → refuse
        denied = self.cli(
            "claim", "new", "--task", "alpha", "--owner", "Test Agent",
            "--mutex", "m2", "--request-id", "alpha-claim-new-test2", cwd=task, check=False,
        )
        self.assertNotEqual(denied.returncode, 0)
        self.assertIn("INVALID_REQUEST", denied.stderr + denied.stdout)
        replaced = self._last_json(self.cli(
            "claim", "new", "--task", "alpha", "--owner", "Test Agent",
            "--mutex", "m2", "--replace", "--request-id", "alpha-claim-new-test3", cwd=task,
        ).stdout)
        self.assertEqual(replaced["status"], "replaced")
        text2 = (task / ".doc/ops/CURRENT.md").read_text(encoding="utf-8")
        self.assertIn("| m2 | in_progress |", text2)

    def test_integrate_prepare_first_parent_and_dirty_refuse(self):
        feat = self.initialized_task(task="feat1", capabilities=("docs:publish",))
        (feat / "src/a").mkdir(parents=True, exist_ok=True)
        (feat / "src/a/note.txt").write_text("hello\n", encoding="utf-8")
        self.git("add", "src/a/note.txt", cwd=feat)
        self.git("commit", "-m", "feat change", cwd=feat)
        expected = self.git("rev-parse", "refs/heads/dev")
        # Prepare from main cwd is allowed by design.
        out = self._last_json(self.cli(
            "integrate", "prepare", "--task", "feat1", "--expected-dev", expected,
            "--integrate-id", "integrate-feat1",
        ).stdout)
        self.assertEqual(out["status"], "prepared")
        self.assertEqual(out["first_parent"], expected)
        integ = self.root / ".worktrees" / "integrate-feat1"
        self.assertTrue(integ.is_dir())
        # Idempotent on same base
        again = self._last_json(self.cli(
            "integrate", "prepare", "--task", "feat1", "--expected-dev", expected,
            "--integrate-id", "integrate-feat1",
        ).stdout)
        self.assertTrue(again.get("idempotent"))
        # Dirty refuse on different attempt id that reuses? Use existing dirty tree.
        (integ / "dirt.txt").write_text("x", encoding="utf-8")
        # Advance expected to force non-idempotent path that hits dirty check —
        # tip still same so idempotent returns early. Force by corrupting first-parent check:
        # Create second integrate with dirty pre-created empty? Create dirty empty dir as worktree path.
        dirty_id = "integrate-dirty"
        dirty_path = self.root / ".worktrees" / dirty_id
        # Manually add a fake dirty worktree at expected tip then make dirty.
        self.git("worktree", "add", "-b", f"feature/{dirty_id}", str(dirty_path), expected)
        (dirty_path / "dirt2.txt").write_text("y", encoding="utf-8")
        denied = self.cli(
            "integrate", "prepare", "--task", "feat1", "--expected-dev", expected,
            "--integrate-id", dirty_id, check=False,
        )
        self.assertNotEqual(denied.returncode, 0)
        self.assertIn("INVALID_REQUEST", denied.stderr + denied.stdout)

    def test_pi_packet_rejects_symbolic_and_writes(self):
        task = self.initialized_task(capabilities=("docs:publish",))
        # Ensure HEAD tracked file for freeze if needed
        denied = self.cli(
            "pi", "packet", "--task", "alpha", "--role", "research",
            "--goal", "g", "--authorization-ref", "user:test",
            "--source-sha", "HEAD", cwd=task, check=False,
        )
        self.assertNotEqual(denied.returncode, 0)
        blob = denied.stderr + denied.stdout
        self.assertTrue(
            "INVALID_SOURCE_SHA" in blob or "INVALID_REQUEST" in blob or "denied" in blob.lower(),
            blob,
        )
        head = self.git("rev-parse", "HEAD", cwd=task)
        ok = self._last_json(self.cli(
            "pi", "packet", "--task", "alpha", "--role", "research",
            "--goal", "research goal", "--authorization-ref", "user:test",
            "--source-sha", head, "--request-id", "pkt-1",
            "--approved-command", "python3 -c print(1)",
            "--write", cwd=task,
        ).stdout)
        self.assertEqual(ok["status"], "ok")
        self.assertEqual(ok["packet"]["source_sha"], head)
        self.assertEqual(ok["packet"]["role_id"], "research")
        rel = ok["path"]
        self.assertTrue(rel.endswith("packet.json"))
        written = json.loads((task / rel).read_text(encoding="utf-8"))
        self.assertEqual(written["request_id"], "pkt-1")
        self.assertEqual(written["approved_commands"], [["python3", "-c", "print(1)"]])


if __name__ == "__main__":
    unittest.main()