from __future__ import annotations

import hashlib
import json
from pathlib import Path
import shutil
import socket
import subprocess
import sys
import tempfile
import time
import unittest


REPO_ROOT = Path(__file__).resolve().parents[1]
SOURCE_SCRIPT = REPO_ROOT / "scripts" / "start-nas-preview-engine.ps1"
POWERSHELL = shutil.which("powershell.exe") or shutil.which("powershell")


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def run(command: list[str], *, cwd: Path, timeout: int = 30) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        command,
        cwd=cwd,
        text=True,
        encoding="utf-8",
        errors="replace",
        capture_output=True,
        timeout=timeout,
        check=False,
    )


def run_without_pipe_capture(
    command: list[str], *, cwd: Path, output_dir: Path, timeout: int = 60
) -> subprocess.CompletedProcess[str]:
    """Wait for the launcher process without waiting on inherited pipe EOF."""
    stdout_path = output_dir / "launcher.stdout.log"
    stderr_path = output_dir / "launcher.stderr.log"
    with stdout_path.open("w", encoding="utf-8") as stdout_handle, stderr_path.open(
        "w", encoding="utf-8"
    ) as stderr_handle:
        process = subprocess.Popen(
            command,
            cwd=cwd,
            text=True,
            encoding="utf-8",
            errors="replace",
            stdout=stdout_handle,
            stderr=stderr_handle,
        )
        returncode = process.wait(timeout=timeout)
    return subprocess.CompletedProcess(
        command,
        returncode,
        stdout_path.read_text(encoding="utf-8", errors="replace"),
        stderr_path.read_text(encoding="utf-8", errors="replace"),
    )


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


def wait_for_port_closed(port: int, *, timeout: float = 10.0) -> bool:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            sock.settimeout(0.2)
            if sock.connect_ex(("127.0.0.1", port)) != 0:
                return True
        time.sleep(0.1)
    return False


@unittest.skipUnless(POWERSHELL, "Windows PowerShell is required")
class NasPreviewEngineLauncherContractTests(unittest.TestCase):
    maxDiff = None

    def setUp(self) -> None:
        self.temp = tempfile.TemporaryDirectory(prefix="vignette-nas-engine-test-")
        self.root = Path(self.temp.name)
        self.repo = self.root / "release"
        self.script = self.repo / "scripts" / SOURCE_SCRIPT.name
        self.api = self.repo / "apps" / "api"
        self.env_file = self.root / "preview.env"
        self.runtime = self.root / "runtime"
        self.receipts = self.root / "receipts"
        self.receipts.mkdir()
        self.script.parent.mkdir(parents=True)
        self.api.mkdir(parents=True)
        shutil.copy2(SOURCE_SCRIPT, self.script)
        self.secret = "nas-preview-engine-test-secret-" + ("x" * 32)
        self.env_file.write_text(
            f"ENGINE_GATEWAY_SHARED_SECRET={self.secret}\n",
            encoding="utf-8",
        )
        self.started_pid: int | None = None

    def tearDown(self) -> None:
        if self.started_pid is not None:
            subprocess.run(
                ["taskkill.exe", "/PID", str(self.started_pid), "/T", "/F"],
                text=True,
                capture_output=True,
                check=False,
            )
        self.temp.cleanup()

    def _write_gateway(self, *, ready: bool = True) -> None:
        package = self.api / "engine_gateway"
        package.mkdir(parents=True, exist_ok=True)
        (package / "__init__.py").write_text("", encoding="utf-8")
        ready_literal = "True" if ready else "False"
        (package / "gateway.py").write_text(
            """
import os
from fastapi import FastAPI, Header, HTTPException

app = FastAPI()
secret = os.environ["ENGINE_GATEWAY_SHARED_SECRET"]

@app.get("/health")
async def health():
    return {"ok": True}

@app.get("/ready")
async def ready(force: bool = False, token: str | None = Header(default=None, alias="X-Vignette-Engine-Token")):
    if token != secret:
        raise HTTPException(status_code=401, detail="unauthorized")
    return {"ok": READY, "force": force}
""".replace("READY", ready_literal).lstrip(),
            encoding="utf-8",
        )

    def _commit_detached(self) -> tuple[str, str]:
        commands = [
            ["git", "init", "-q"],
            ["git", "config", "user.name", "Yun Chan"],
            ["git", "config", "user.email", "yunchan@twentyoz.kr"],
            ["git", "add", "--all"],
            ["git", "commit", "-q", "-m", "test fixture"],
        ]
        for command in commands:
            result = run(command, cwd=self.repo)
            self.assertEqual(0, result.returncode, result.stderr)
        head = run(["git", "rev-parse", "HEAD"], cwd=self.repo).stdout.strip()
        tree = run(["git", "rev-parse", "HEAD^{tree}"], cwd=self.repo).stdout.strip()
        result = run(["git", "checkout", "-q", "--detach", head], cwd=self.repo)
        self.assertEqual(0, result.returncode, result.stderr)
        return head, tree

    def _launcher_args(
        self,
        *,
        port: int,
        head: str,
        tree: str,
        check_only: bool = True,
        listen_address: str = "127.0.0.1",
        script_sha: str | None = None,
        env_sha: str | None = None,
        ready_timeout: int = 20,
    ) -> list[str]:
        receipt = self.receipts / f"receipt-{port}.json"
        args = [
            str(POWERSHELL),
            "-NoLogo",
            "-NoProfile",
            "-ExecutionPolicy",
            "Bypass",
            "-File",
            str(self.script),
            "-EnvFile",
            str(self.env_file),
            "-Workspace",
            str(self.repo),
            "-ListenAddress",
            listen_address,
            "-Port",
            str(port),
            "-Python",
            sys.executable,
            "-ExpectedCommit",
            head,
            "-ExpectedTree",
            tree,
            "-ExpectedScriptSha256",
            script_sha or sha256_file(self.script),
            "-ExpectedPythonSha256",
            sha256_file(Path(sys.executable)),
            "-ExpectedEnvSha256",
            env_sha or sha256_file(self.env_file),
            "-ExpectedConsumerEngineUrl",
            f"http://{listen_address}:{port}",
            "-RuntimeStateDir",
            str(self.runtime),
            "-ReceiptPath",
            str(receipt),
            "-ReadyTimeoutSec",
            str(ready_timeout),
        ]
        if check_only:
            args.append("-CheckOnly")
        return args

    def test_check_only_is_mutation_free_and_secret_free(self) -> None:
        self._write_gateway()
        head, tree = self._commit_detached()
        port = free_port()
        result = run(
            self._launcher_args(port=port, head=head, tree=tree),
            cwd=self.root,
        )
        self.assertEqual(0, result.returncode, result.stderr)
        payload = json.loads(result.stdout)
        self.assertEqual("preflight_passed", payload["status"])
        self.assertFalse(payload["mutation"])
        self.assertFalse(payload["binding"]["secret_value_emitted"])
        self.assertNotIn(self.secret, result.stdout + result.stderr)
        self.assertFalse(self.runtime.exists())
        self.assertEqual([], list(self.receipts.iterdir()))

    def test_attached_or_dirty_source_is_rejected(self) -> None:
        self._write_gateway()
        head, tree = self._commit_detached()
        port = free_port()
        branch = run(["git", "switch", "-q", "-c", "unsafe"], cwd=self.repo)
        self.assertEqual(0, branch.returncode, branch.stderr)
        attached = run(
            self._launcher_args(port=port, head=head, tree=tree),
            cwd=self.root,
        )
        self.assertNotEqual(0, attached.returncode)
        self.assertIn("detached HEAD", attached.stderr)

        detached = run(["git", "checkout", "-q", "--detach", head], cwd=self.repo)
        self.assertEqual(0, detached.returncode, detached.stderr)
        (self.repo / "untracked.txt").write_text("unsafe", encoding="utf-8")
        dirty = run(
            self._launcher_args(port=port, head=head, tree=tree),
            cwd=self.root,
        )
        self.assertNotEqual(0, dirty.returncode)
        self.assertIn("tracked-clean and untracked-clean", dirty.stderr)

    def test_hash_drift_is_rejected(self) -> None:
        self._write_gateway()
        head, tree = self._commit_detached()
        port = free_port()
        result = run(
            self._launcher_args(
                port=port,
                head=head,
                tree=tree,
                script_sha="0" * 64,
            ),
            cwd=self.root,
        )
        self.assertNotEqual(0, result.returncode)
        self.assertIn("Launcher SHA-256", result.stderr)

    def test_wildcard_and_unconfirmed_lan_bindings_are_rejected(self) -> None:
        self._write_gateway()
        head, tree = self._commit_detached()
        wildcard = run(
            self._launcher_args(
                port=free_port(),
                head=head,
                tree=tree,
                listen_address="0.0.0.0",
            ),
            cwd=self.root,
        )
        self.assertNotEqual(0, wildcard.returncode)
        self.assertIn("Wildcard and broadcast", wildcard.stderr)

        unconfirmed = run(
            self._launcher_args(
                port=free_port(),
                head=head,
                tree=tree,
                listen_address="192.0.2.1",
            ),
            cwd=self.root,
        )
        self.assertNotEqual(0, unconfirmed.returncode)
        self.assertIn("ConfirmNasPreviewLanExposure", unconfirmed.stderr)

    def test_existing_listener_is_never_reused(self) -> None:
        self._write_gateway()
        head, tree = self._commit_detached()
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
            listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            listener.bind(("127.0.0.1", 0))
            listener.listen()
            port = int(listener.getsockname()[1])
            result = run(
                self._launcher_args(port=port, head=head, tree=tree),
                cwd=self.root,
            )
        self.assertNotEqual(0, result.returncode)
        self.assertIn("already has a listener", result.stderr)

    def test_external_junction_cannot_redirect_receipt_into_source(self) -> None:
        self._write_gateway()
        source_receipts = self.repo / "source-receipts"
        source_receipts.mkdir()
        (source_receipts / ".keep").write_text("pinned\n", encoding="utf-8")
        head, tree = self._commit_detached()
        junction = self.root / "receipt-junction"
        created = run(
            ["cmd.exe", "/c", "mklink", "/J", str(junction), str(source_receipts)],
            cwd=self.root,
        )
        self.assertEqual(0, created.returncode, created.stderr)
        port = free_port()
        args = self._launcher_args(port=port, head=head, tree=tree)
        receipt_index = args.index("-ReceiptPath") + 1
        args[receipt_index] = str(junction / "receipt.json")
        result = run(args, cwd=self.root)
        self.assertNotEqual(0, result.returncode)
        self.assertIn("reparse-point ancestor", result.stderr)
        self.assertFalse((source_receipts / "receipt.json").exists())

    def test_actual_loopback_launch_binds_receipt_and_owner(self) -> None:
        self._write_gateway()
        head, tree = self._commit_detached()
        port = free_port()
        result = run_without_pipe_capture(
            self._launcher_args(
                port=port,
                head=head,
                tree=tree,
                check_only=False,
            ),
            cwd=self.root,
            output_dir=self.root,
            timeout=60,
        )
        self.assertEqual(0, result.returncode, result.stderr)
        receipt_path = self.receipts / f"receipt-{port}.json"
        payload = json.loads(receipt_path.read_text(encoding="utf-8"))
        self.started_pid = int(payload["pid"])
        self.assertEqual("passed", payload["status"])
        self.assertEqual(head, payload["binding"]["source_commit"])
        self.assertEqual(tree, payload["binding"]["source_tree"])
        self.assertTrue(payload["checks"]["generation_ready"])
        self.assertTrue(payload["checks"]["listener_owner_matches"])
        serialized = json.dumps(payload, ensure_ascii=False)
        self.assertNotIn(self.secret, serialized)
        self.assertNotIn(self.secret, result.stdout + result.stderr)

    def test_failed_readiness_removes_only_the_owned_process(self) -> None:
        self._write_gateway(ready=False)
        head, tree = self._commit_detached()
        port = free_port()
        result = run(
            self._launcher_args(
                port=port,
                head=head,
                tree=tree,
                check_only=False,
                ready_timeout=10,
            ),
            cwd=self.root,
            timeout=45,
        )
        self.assertNotEqual(0, result.returncode)
        self.assertIn("generation readiness did not pass", result.stderr)
        self.assertTrue(wait_for_port_closed(port))
        self.assertFalse((self.receipts / f"receipt-{port}.json").exists())


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