"""Exercise deployment failure boundaries without Docker or network mutations.

Run with: python3 -m unittest discover -s scripts/tests -v
"""

import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest


ROOT = Path(__file__).resolve().parents[2]
RUNTIME_KEYS = (
    "IMAGE_PREFIX", "DEPLOY_TAG", "REGISTRY_USERNAME", "REGISTRY_PASSWORD",
    "WQ_EMAIL", "WQ_PASSWORD", "DATABASE_URL", "ADMIN_PASSWORD", "ENCRYPTION_KEY",
    "PUBLIC_ORIGIN",
)

DOCKER_STUB = r'''
import json, os, pathlib, sys
a = sys.argv[1:]
if a[0] == "compose":
    if "config" in a: op = "config"
    elif "pull" in a: op = "pull"
    elif "run" in a: op = "migrate" if "migrate" in a else "preflight"
    elif "images" in a: op = "images"
    elif "stop" in a: op = "stop"
    elif "up" in a: op = "up"
    elif "ps" in a: op = "ps"
    else: raise SystemExit("Unexpected compose invocation")
else:
    op = {"create": "lock", "rm": "unlock", "image": "inspect"}.get(a[0], a[0])
entry = {"op": op, "args": a, "auth_dir": os.environ.get("DOCKER_CONFIG"),
         "env": {k: os.environ.get(k) for k in
                 ("ADMIN_PASSWORD", "WQ_PASSWORD", "DATABASE_URL", "DEPLOY_TAG")}}
if op == "login":
    entry["password"] = sys.stdin.read()
    auth = pathlib.Path(os.environ["DOCKER_CONFIG"])
    (auth / "config.json").write_text("test credentials")
with open(os.environ["TEST_DOCKER_LOG"], "a") as out:
    out.write(json.dumps(entry) + "\n")
if os.environ.get("TEST_FAIL") == op:
    raise SystemExit(23)
if op == "lock": print("owned-lock-id")
if op == "images": print("sha256:previous-backend\nsha256:previous-web")
'''

SSH_STUB = r'''
import json, os, pathlib, stat, subprocess, sys
a = sys.argv[1:]
key = pathlib.Path(a[a.index("-i") + 1])
hosts = pathlib.Path(next(x.split("=", 1)[1] for x in a if x.startswith("UserKnownHostsFile=")))
pathlib.Path(os.environ["TEST_SSH_LOG"]).write_text(json.dumps({
    "args": a, "key": key.read_text(), "hosts": hosts.read_text(),
    "key_mode": stat.S_IMODE(key.stat().st_mode), "key_path": str(key),
}))
if os.environ.get("TEST_FAIL") == "ssh":
    raise SystemExit(255)
# Simulate a different host: none of the deployment variables are inherited.
env = {k: v for k, v in os.environ.items() if k.startswith("TEST_") or k in ("PATH", "HOME", "TMPDIR")}
result = subprocess.run(["bash", "--noprofile", "--norc", "-se"],
                        input=sys.stdin.buffer.read(), env=env)
raise SystemExit(result.returncode)
'''


class DeploymentTest(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.directory = Path(self.temp.name)
        self.bin = self.directory / "bin"
        self.bin.mkdir()
        for name, code in (("docker", DOCKER_STUB), ("ssh", SSH_STUB)):
            path = self.bin / name
            path.write_text(f"#!{sys.executable}\n" + code)
            path.chmod(0o700)
        self.log = self.directory / "docker.jsonl"
        self.ssh_log = self.directory / "ssh.json"
        self.env = {
            "PATH": str(self.bin) + os.pathsep + os.environ["PATH"],
            "HOME": str(self.directory), "TMPDIR": str(self.directory),
            "TEST_DOCKER_LOG": str(self.log), "TEST_SSH_LOG": str(self.ssh_log),
            "IMAGE_PREFIX": "registry.example.test/team/wq", "DEPLOY_TAG": "abc123",
            "REGISTRY_USERNAME": "publisher", "REGISTRY_PASSWORD": "fake-registry-token",
            "WQ_EMAIL": "test@example.test", "WQ_PASSWORD": "fake-worldquant-password",
            "DATABASE_URL": "postgresql+asyncpg://wq:fake@10.0.0.30:5432/wq",
            "ADMIN_PASSWORD": "fake-admin-password", "ENCRYPTION_KEY": "fake-key",
            "PUBLIC_ORIGIN": "https://alpha.example.test",
            "PROD_HOST": "server-b.example.test", "PROD_USER": "deploy", "PROD_PORT": "2222",
            "PROD_SSH_KEY": "fake-private-key\nsecond-line",
            "PROD_KNOWN_HOSTS": "[server-b.example.test]:2222 ssh-ed25519 fake-public-key",
            "DEPLOY_PATH": str(self.directory / "server B's project"),
        }

    def run_script(self, script="deploy-production.sh", **overrides):
        self.log.unlink(missing_ok=True)
        self.ssh_log.unlink(missing_ok=True)
        return subprocess.run(
            ["bash", str(ROOT / "scripts" / script)],
            env=self.env | overrides, capture_output=True, text=True, timeout=20,
        )

    def calls(self):
        return [json.loads(line) for line in self.log.read_text().splitlines()] if self.log.exists() else []

    def assert_auth_cleaned(self):
        for call in self.calls():
            if call["auth_dir"]:
                self.assertFalse(Path(call["auth_dir"]).exists())

    def test_release_pulls_before_stopping_and_uses_only_local_images_afterward(self):
        result = self.run_script()
        self.assertEqual(result.returncode, 0, result.stderr)
        calls = self.calls()
        ops = [c["op"] for c in calls]
        important = [op for op in ops if op in {"pull", "lock", "preflight", "stop", "migrate", "up", "unlock"}]
        self.assertEqual(important, ["pull", "lock", "preflight", "stop", "migrate", "up", "unlock"])
        for call in calls:
            args = call["args"]
            if call["op"] in {"preflight", "migrate", "up"}:
                self.assertEqual(args[args.index("--pull") + 1], "never")
                self.assertEqual(call["env"]["DATABASE_URL"], self.env["DATABASE_URL"])
            self.assertNotIn("build", args)
            if call["op"] == "unlock":
                self.assertEqual(args, ["rm", "owned-lock-id"])
        self.assertIn("Production is healthy", result.stdout)
        self.assert_auth_cleaned()

    def test_failures_before_switch_do_not_stop_running_services(self):
        for stage in ("config", "login", "pull", "lock", "preflight"):
            with self.subTest(stage=stage):
                result = self.run_script(TEST_FAIL=stage)
                self.assertNotEqual(result.returncode, 0)
                ops = [c["op"] for c in self.calls()]
                self.assertNotIn("stop", ops)
                self.assertNotIn("migrate", ops)
                self.assertNotIn("up", ops)
                self.assertNotIn("Production is healthy", result.stdout)
                self.assertEqual("unlock" in ops, stage == "preflight")
                self.assert_auth_cleaned()

    def test_migration_failure_does_not_start_application(self):
        result = self.run_script(TEST_FAIL="migrate")
        self.assertEqual(result.returncode, 23)
        ops = [c["op"] for c in self.calls()]
        self.assertIn("stop", ops)
        self.assertNotIn("up", ops)
        self.assertIn("unlock", ops)
        self.assertNotIn("Production is healthy", result.stdout)
        self.assert_auth_cleaned()

    def test_health_failure_is_not_reported_as_success(self):
        result = self.run_script(TEST_FAIL="up")
        self.assertEqual(result.returncode, 23)
        self.assertNotIn("Production is healthy", result.stdout)
        self.assertIn("unlock", [c["op"] for c in self.calls()])
        self.assert_auth_cleaned()

    def test_missing_runtime_configuration_fails_before_docker(self):
        for key in RUNTIME_KEYS:
            with self.subTest(key=key):
                result = self.run_script(**{key: ""})
                self.assertNotEqual(result.returncode, 0)
                self.assertIn(key, result.stderr)
                self.assertEqual(self.calls(), [])

    def test_ssh_transmits_literal_secrets_and_only_deployment_files(self):
        marker = self.directory / "MUST_NOT_EXIST"
        secret = f"literal $value 'double\" slash\\; $(touch {marker})\nsecond line"
        result = self.run_script("deploy-remote.sh", ADMIN_PASSWORD=secret, WQ_PASSWORD=secret, REGISTRY_PASSWORD=secret)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertFalse(marker.exists())
        self.assertNotIn(secret, result.stdout + result.stderr)
        login = next(c for c in self.calls() if c["op"] == "login")
        self.assertEqual(login["password"], secret)
        preflight = next(c for c in self.calls() if c["op"] == "preflight")
        self.assertEqual(preflight["env"]["ADMIN_PASSWORD"], secret)
        self.assertEqual(preflight["env"]["WQ_PASSWORD"], secret)
        for call in self.calls():
            if call["op"] in {"preflight", "migrate", "up"}:
                self.assertEqual(call["env"]["DATABASE_URL"], self.env["DATABASE_URL"])
        ssh = json.loads(self.ssh_log.read_text())
        self.assertIn("StrictHostKeyChecking=yes", ssh["args"])
        self.assertIn("BatchMode=yes", ssh["args"])
        self.assertEqual(ssh["args"][ssh["args"].index("-p") + 1], "2222")
        self.assertEqual(ssh["key"], self.env["PROD_SSH_KEY"] + "\n")
        self.assertEqual(ssh["key_mode"], 0o600)
        self.assertNotIn(secret, " ".join(ssh["args"]))
        self.assertFalse(Path(ssh["key_path"]).exists())
        releases = list((Path(self.env["DEPLOY_PATH"]) / "releases").iterdir())
        self.assertEqual(len(releases), 1)
        files = {str(p.relative_to(releases[0])) for p in releases[0].rglob("*") if p.is_file()}
        self.assertEqual(files, {"compose.production.yaml", "scripts/deploy-production.sh"})
        self.assert_auth_cleaned()

    def test_ssh_and_remote_failures_propagate_and_clean_key(self):
        for stage in ("ssh", "pull", "migrate", "up"):
            with self.subTest(stage=stage):
                result = self.run_script("deploy-remote.sh", TEST_FAIL=stage)
                self.assertEqual(result.returncode, 255 if stage == "ssh" else 23, result.stderr)
                self.assertNotIn("Production is healthy", result.stdout)
                ssh = json.loads(self.ssh_log.read_text())
                self.assertFalse(Path(ssh["key_path"]).exists())
                self.assert_auth_cleaned()

    def test_missing_ssh_configuration_fails_before_connecting(self):
        for key in ("PROD_HOST", "PROD_USER", "PROD_SSH_KEY", "PROD_KNOWN_HOSTS", "DEPLOY_PATH"):
            with self.subTest(key=key):
                result = self.run_script("deploy-remote.sh", **{key: ""})
                self.assertNotEqual(result.returncode, 0)
                self.assertFalse(self.ssh_log.exists())


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