import contextlib
import io
import json
import os
import stat
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from harness.lib import verification_root


class VerificationRootTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name) / "cache"
        self.issue = self.root / "repositories" / "alice" / "ledger" / "worktrees" / "issue-7"
        self.issue.mkdir(parents=True, mode=0o700)
        for path in (
            self.root / "repositories",
            self.root / "repositories/alice",
            self.root / "repositories/alice/ledger",
            self.root / "repositories/alice/ledger/worktrees",
            self.issue,
        ):
            os.chmod(path, 0o700)
        # The trusted harness/NFS root is not required to be role-owned 0700.
        os.chmod(self.root, 0o770)
        self.handle = self.root / "runtime" / "handle.json"
        self.handle.parent.mkdir(mode=0o700)
        self.meta = {
            "root": str(self.root),
            "owner": "alice",
            "repository": "ledger",
            "issue": 7,
            "role": "gitea-impl-agent",
        }
        self.status = {
            "status": "held",
            "root": str(self.root),
            "owner": "alice",
            "repository": "ledger",
            "issue": 7,
            "role": "gitea-impl-agent",
            "deadline": 10**20,
        }

    def tearDown(self):
        self.tmp.cleanup()

    @contextlib.contextmanager
    def broker(self, *, metadata=None, status=None):
        with (
            patch.object(
                verification_root.broker,
                "read_handle_metadata",
                return_value=self.meta if metadata is None else metadata,
            ),
            patch.object(
                verification_root.broker,
                "request_handle",
                return_value=self.status if status is None else status,
            ),
        ):
            yield

    def test_umask_does_not_make_verification_parent_world_readable(self):
        old = os.umask(0o022)
        try:
            with self.broker():
                result = verification_root.prepare(
                    self.root, "alice", "ledger", 7, "run-1", self.handle
                )
        finally:
            os.umask(old)
        verification = self.issue / "verification"
        self.assertEqual(result, verification / "run-1")
        self.assertEqual(stat.S_IMODE(verification.stat().st_mode), 0o700)
        self.assertEqual(stat.S_IMODE((verification / "run-1").stat().st_mode), 0o700)

    def test_existing_secure_parent_accepts_a_fresh_run(self):
        verification = self.issue / "verification"
        verification.mkdir(mode=0o700)
        (verification / "prior-run").mkdir(mode=0o700)
        with self.broker():
            result = verification_root.prepare(
                self.root, "alice", "ledger", 7, "new-run", self.handle
            )
        self.assertEqual(result, verification / "new-run")
        self.assertTrue((verification / "prior-run").is_dir())
        self.assertEqual(stat.S_IMODE(result.stat().st_mode), 0o700)

    def test_bad_lease_is_rejected_before_filesystem_mutation(self):
        before = list(self.issue.iterdir())
        with self.broker(status={"error": "expired"}):
            with self.assertRaises(verification_root.VerificationError):
                verification_root.prepare(
                    self.root, "alice", "ledger", 7, "run-1", self.handle
                )
        self.assertEqual(list(self.issue.iterdir()), before)

    def test_missing_deadline_and_role_mismatch_fail_before_mutation(self):
        status = dict(self.status)
        status.pop("deadline")
        with self.broker(status=status):
            with self.assertRaisesRegex(verification_root.VerificationError, "expired"):
                verification_root.prepare(
                    self.root, "alice", "ledger", 7, "run-1", self.handle
                )
        status = dict(self.status, role="gitea-review-agent")
        with self.broker(status=status):
            with self.assertRaisesRegex(verification_root.VerificationError, "role"):
                verification_root.prepare(
                    self.root, "alice", "ledger", 7, "run-1", self.handle
                )
        self.assertFalse((self.issue / "verification").exists())

    def test_existing_insecure_verification_is_not_repaired(self):
        verification = self.issue / "verification"
        verification.mkdir(mode=0o755)
        with self.broker():
            with self.assertRaises(verification_root.VerificationError):
                verification_root.prepare(
                    self.root, "alice", "ledger", 7, "run-1", self.handle
                )
        self.assertEqual(stat.S_IMODE(verification.stat().st_mode), 0o755)
        self.assertFalse((verification / "run-1").exists())

    def test_insecure_issue_directory_is_rejected_without_repair(self):
        os.chmod(self.issue, 0o755)
        with self.broker():
            with self.assertRaisesRegex(verification_root.VerificationError, "Issue directory"):
                verification_root.prepare(
                    self.root, "alice", "ledger", 7, "run-1", self.handle
                )
        self.assertEqual(stat.S_IMODE(self.issue.stat().st_mode), 0o755)
        self.assertFalse((self.issue / "verification").exists())

    def test_symlink_and_existing_run_are_rejected_without_mutation(self):
        verification = self.issue / "verification"
        verification.mkdir(mode=0o700)
        existing = verification / "run-1"
        existing.mkdir(mode=0o700)
        with self.broker():
            with self.assertRaisesRegex(verification_root.VerificationError, "existing run"):
                verification_root.prepare(
                    self.root, "alice", "ledger", 7, "run-1", self.handle
                )
        self.assertTrue(existing.is_dir())
        existing.rmdir()
        verification.rmdir()
        verification.symlink_to(self.issue, target_is_directory=True)
        with self.broker():
            with self.assertRaises(verification_root.VerificationError):
                verification_root.prepare(
                    self.root, "alice", "ledger", 7, "run-2", self.handle
                )
        self.assertTrue(verification.is_symlink())

    def test_binding_and_identifier_validation(self):
        bad = dict(self.meta, owner="mallory")
        with self.broker(metadata=bad):
            with self.assertRaisesRegex(verification_root.VerificationError, "binding"):
                verification_root.prepare(
                    self.root, "alice", "ledger", 7, "run-1", self.handle
                )
        wrong_root = dict(self.status, root=str(self.root / "other"))
        with self.broker(status=wrong_root):
            with self.assertRaisesRegex(verification_root.VerificationError, "binding"):
                verification_root.prepare(
                    self.root, "alice", "ledger", 7, "run-1", self.handle
                )
        for value in ("../x", "a/b", "", ".", "..", ".hidden", "x" * 129):
            with self.subTest(value=value), self.assertRaises(verification_root.VerificationError):
                verification_root.prepare(
                    self.root, "alice", "ledger", 7, value, self.handle
                )

    def test_real_broker_cli_prepares_canonical_private_root(self):
        locks = self.root / "repositories/alice/ledger/locks"
        locks.mkdir(mode=0o700)
        (locks / "repository.lock").touch(mode=0o600)
        broker_script = Path(verification_root.broker.__file__)
        socket_path = self.handle.parent / "broker.sock"
        command = [
            sys.executable,
            str(broker_script),
            "serve",
            "--root",
            str(self.root),
            "--owner",
            "alice",
            "--repository",
            "ledger",
            "--issue",
            "7",
            "--role",
            "gitea-impl-agent",
            "--run-id",
            "broker-run",
            "--handle",
            str(self.handle),
            "--socket",
            str(socket_path),
            "--ttl-seconds",
            "30",
            "--acquire-timeout",
            "1",
        ]
        proc = subprocess.Popen(
            command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
        )
        assert proc.stdout is not None and proc.stderr is not None
        try:
            ready = json.loads(proc.stdout.readline())
            self.assertEqual(ready["status"], "held")
            helper = subprocess.run(
                [
                    sys.executable,
                    str(Path(verification_root.__file__)),
                    "prepare",
                    "--root",
                    str(self.root),
                    "--owner",
                    "alice",
                    "--repository",
                    "ledger",
                    "--issue",
                    "7",
                    "--run-id",
                    "real-run",
                    "--handle",
                    str(self.handle),
                ],
                text=True,
                capture_output=True,
            )
            self.assertEqual(helper.returncode, 0, helper.stderr)
            record = json.loads(helper.stdout)
            run = Path(record["path"])
            self.assertEqual(run, self.issue / "verification/real-run")
            self.assertEqual(stat.S_IMODE(run.parent.stat().st_mode), 0o700)
            self.assertEqual(stat.S_IMODE(run.stat().st_mode), 0o700)
            self.assertEqual(
                verification_root.broker.request_handle(self.handle, "release")["status"],
                "released",
            )
            proc.wait(timeout=3)
        finally:
            if proc.poll() is None:
                proc.kill()
                proc.wait()
            proc.stdout.close()
            proc.stderr.close()

    def test_created_inode_replacement_fails_without_removing_replacement(self):
        verification = self.issue / "verification"
        verification.mkdir(mode=0o700)
        real_open = verification_root._open_private_child

        def replace_before_open(parent_fd, name, label):
            os.rename(name, f"{name}.original", src_dir_fd=parent_fd, dst_dir_fd=parent_fd)
            os.mkdir(name, 0o700, dir_fd=parent_fd)
            return real_open(parent_fd, name, label)

        with self.broker(), patch.object(
            verification_root, "_open_private_child", side_effect=replace_before_open
        ):
            with self.assertRaisesRegex(verification_root.VerificationError, "changed run directory"):
                verification_root.prepare(
                    self.root, "alice", "ledger", 7, "race-run", self.handle
                )
        self.assertTrue((verification / "race-run").is_dir())

    def test_replaced_new_verification_parent_is_not_removed_on_run_failure(self):
        real_create = verification_root._create_private_child
        calls = 0

        def replace_parent_before_run(parent_fd, name, label):
            nonlocal calls
            calls += 1
            if calls == 1:
                return real_create(parent_fd, name, label)
            os.rename(
                self.issue / "verification",
                self.issue / "verification.original",
            )
            (self.issue / "verification").mkdir(mode=0o700)
            raise verification_root.VerificationError("forced run failure")

        with self.broker(), patch.object(
            verification_root, "_create_private_child", side_effect=replace_parent_before_run
        ):
            with self.assertRaisesRegex(verification_root.VerificationError, "forced run failure"):
                verification_root.prepare(
                    self.root, "alice", "ledger", 7, "race-parent", self.handle
                )
        self.assertTrue((self.issue / "verification").is_dir())
        self.assertTrue((self.issue / "verification.original").is_dir())

    def test_cli_is_directly_executable_and_emits_verified_modes(self):
        script = Path(verification_root.__file__)
        help_result = subprocess.run(
            [str(script), "prepare", "--help"],
            text=True,
            capture_output=True,
        )
        self.assertEqual(help_result.returncode, 0, help_result.stderr)
        for option in ("--root", "--owner", "--repository", "--issue", "--run-id", "--handle"):
            self.assertIn(option, help_result.stdout)

        output = io.StringIO()
        with self.broker(), contextlib.redirect_stdout(output):
            code = verification_root.main(
                [
                    "prepare",
                    "--root",
                    str(self.root),
                    "--owner",
                    "alice",
                    "--repository",
                    "ledger",
                    "--issue",
                    "7",
                    "--run-id",
                    "run-cli",
                    "--handle",
                    str(self.handle),
                ]
            )
        self.assertEqual(code, 0)
        record = json.loads(output.getvalue())
        self.assertEqual(record["verification_mode"], "0700")
        self.assertEqual(record["run_mode"], "0700")
        self.assertEqual(Path(record["path"]), self.issue / "verification/run-cli")


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