timmy-talking-turd/tests/staging-deploy.test.py

#!/usr/bin/env python3
"""Filesystem and failure-path tests for immutable staging deployment."""
from __future__ import annotations

import hashlib
import importlib.util
import io
import json
from pathlib import Path
import subprocess
import sys
import tarfile
import tempfile
import unittest
from unittest import mock

ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts" / "deploy_staging.py"
COMMIT_A = "a" * 40
COMMIT_B = "b" * 40


def load_deploy():
    spec = importlib.util.spec_from_file_location("deploy_staging", SCRIPT)
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


def make_archive(path: Path, members: list[tuple[str, bytes, str]]) -> str:
    with tarfile.open(path, "w:gz") as archive:
        for name, body, kind in members:
            info = tarfile.TarInfo(name)
            if kind == "file":
                info.size = len(body)
                archive.addfile(info, io.BytesIO(body))
            elif kind == "dir":
                info.type = tarfile.DIRTYPE
                archive.addfile(info)
            elif kind == "symlink":
                info.type = tarfile.SYMTYPE
                info.linkname = "server.mjs"
                archive.addfile(info)
            elif kind == "hardlink":
                info.type = tarfile.LNKTYPE
                info.linkname = "server.mjs"
                archive.addfile(info)
            elif kind == "fifo":
                info.type = tarfile.FIFOTYPE
                archive.addfile(info)
            elif kind == "device":
                info.type = tarfile.CHRTYPE
                archive.addfile(info)
    return hashlib.sha256(path.read_bytes()).hexdigest()


class DeployTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name) / "opt"
        self.root.mkdir()
        self.deploy = load_deploy()
        self.config = self.deploy.DeploymentConfig(
            root=self.root,
            restart_command=("fixture-restart",),
            stop_command=("fixture-stop",),
            smoke_command=("fixture-smoke",),
            health_url="http://127.0.0.1:4174/api/healthz",
            command_timeout=1.0,
            health_timeout=1.0,
        )
        self.commands = []

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

    def runner(self, argv, **kwargs):
        self.commands.append((tuple(argv), kwargs))
        return subprocess.CompletedProcess(argv, 0, "", "")

    def healthy(self, url, commit, timeout):
        self.assertEqual(url, self.config.health_url)
        self.assertEqual(timeout, self.config.health_timeout)
        return {"ok": True, "commit": commit}

    def archive(self, name="release.tar.gz", members=None):
        path = Path(self.tmp.name) / name
        digest = make_archive(path, members or [
            ("timmy-release/", b"", "dir"),
            ("timmy-release/server.mjs", b"console.log('ok')\n", "file"),
            ("timmy-release/package.json", b"{}\n", "file"),
        ])
        return path, digest

    def seed_release(self, commit):
        release = self.root / "releases" / commit
        release.mkdir(parents=True)
        (release / "server.mjs").write_text("ok", encoding="utf-8")
        (release / ".timmy-release.json").write_text(json.dumps({"commit": commit, "tag": "old"}), encoding="utf-8")
        return release

    def point_current(self, commit):
        (self.root / "current").symlink_to(Path("releases") / commit)

    def promote(self, archive, digest, commit=COMMIT_B, **kwargs):
        return self.deploy.promote(
            config=self.config, tag="daily-test", archive=archive,
            expected_sha256=digest, commit=commit,
            run_command=kwargs.get("run_command", self.runner),
            health_check=kwargs.get("health_check", self.healthy),
        )

    def test_checksum_mismatch_fails_before_tar_is_opened(self):
        archive = Path(self.tmp.name) / "not-even-a-tar"
        archive.write_bytes(b"untrusted")
        with self.assertRaisesRegex(self.deploy.DeploymentError, "SHA-256 mismatch"):
            self.promote(archive, "0" * 64)
        self.assertFalse((self.root / "releases").exists())

    def test_verified_private_archive_copy_is_used_after_caller_archive_mutates(self):
        archive, digest = self.archive("mutable.tar.gz")
        evil = Path(self.tmp.name) / "evil.tar.gz"
        make_archive(evil, [
            ("timmy-release/", b"", "dir"),
            ("timmy-release/server.mjs", b"EVIL\n", "file"),
        ])
        real_inspect = self.deploy.inspect_archive
        inspected_paths = []

        def mutate_then_inspect(path, config):
            inspected_paths.append(Path(path))
            archive.write_bytes(evil.read_bytes())
            return real_inspect(path, config)

        with mock.patch.object(self.deploy, "inspect_archive", side_effect=mutate_then_inspect):
            self.promote(archive, digest)

        release = self.root / "releases" / COMMIT_B
        self.assertNotEqual(inspected_paths, [archive])
        self.assertEqual((release / "server.mjs").read_text(), "console.log('ok')\n")
        self.assertTrue(all(not path.exists() for path in inspected_paths), "verified temp archive must be cleaned")

    def test_archive_source_must_be_a_nonsymlink_regular_file(self):
        archive, digest = self.archive("regular.tar.gz")
        symlink = Path(self.tmp.name) / "archive-link.tar.gz"
        symlink.symlink_to(archive)
        for source in (symlink, Path(self.tmp.name)):
            with self.subTest(source=source), self.assertRaisesRegex(self.deploy.DeploymentError, "regular file|stage archive"):
                self.promote(source, digest)

    def test_absolute_and_traversal_paths_are_rejected(self):
        for index, unsafe in enumerate(("/etc/passwd", "root/../../escape", "../escape", "root//double")):
            archive, digest = self.archive(f"unsafe-{index}.tar.gz", [(unsafe, b"bad", "file")])
            with self.subTest(unsafe=unsafe), self.assertRaisesRegex(self.deploy.DeploymentError, "unsafe archive path"):
                self.promote(archive, digest)
            self.assertFalse((self.root / "releases" / COMMIT_B).exists())

    def test_links_devices_fifos_and_oversized_archives_are_rejected(self):
        for index, kind in enumerate(("symlink", "hardlink", "fifo", "device")):
            archive, digest = self.archive(f"special-{index}.tar.gz", [(f"root/bad-{kind}", b"", kind)])
            with self.subTest(kind=kind), self.assertRaisesRegex(self.deploy.DeploymentError, "unsupported archive member"):
                self.promote(archive, digest)
        archive, digest = self.archive("large.tar.gz", [("root/large", b"x" * 17, "file")])
        tiny = self.deploy.DeploymentConfig(**{**self.config.__dict__, "max_member_bytes": 16})
        with self.assertRaisesRegex(self.deploy.DeploymentError, "member size"):
            self.deploy.promote(config=tiny, tag="daily-test", archive=archive, expected_sha256=digest, commit=COMMIT_B, run_command=self.runner, health_check=self.healthy)

    def test_member_count_limit_is_enforced(self):
        archive, digest = self.archive("many.tar.gz", [(f"root/{i}", b"x", "file") for i in range(3)])
        tiny = self.deploy.DeploymentConfig(**{**self.config.__dict__, "max_members": 2})
        with self.assertRaisesRegex(self.deploy.DeploymentError, "member count"):
            self.deploy.promote(config=tiny, tag="daily-test", archive=archive, expected_sha256=digest, commit=COMMIT_B, run_command=self.runner, health_check=self.healthy)

    def test_secrets_env_and_forbidden_artifacts_are_rejected(self):
        forbidden = ("root/.env", "root/secrets/token.txt", "root/private.pem", "root/model.gguf", "root/.git/config", "root/video/raw.webm")
        for index, name in enumerate(forbidden):
            archive, digest = self.archive(f"secret-{index}.tar.gz", [(name, b"secret", "file")])
            with self.subTest(name=name), self.assertRaisesRegex(self.deploy.DeploymentError, "forbidden archive artifact"):
                self.promote(archive, digest)

    def test_forbidden_common_roots_are_rejected_case_insensitively_but_release_wrapper_is_allowed(self):
        for index, root in enumerate((".git", "ViDeO", "ARTIFACTS", "Credentials")):
            archive, _ = self.archive(f"forbidden-root-{index}.tar.gz", [
                (f"{root}/", b"", "dir"),
                (f"{root}/server.mjs", b"evil", "file"),
            ])
            with self.subTest(root=root), self.assertRaisesRegex(self.deploy.DeploymentError, "forbidden archive artifact"):
                self.deploy.inspect_archive(archive, self.config)
        archive, _ = self.archive("normal-wrapper.tar.gz")
        _, wrapper = self.deploy.inspect_archive(archive, self.config)
        self.assertEqual(wrapper, "timmy-release")

    def test_release_root_symlink_is_rejected(self):
        outside = Path(self.tmp.name) / "outside"
        outside.mkdir()
        (self.root / "releases").symlink_to(outside, target_is_directory=True)
        archive, digest = self.archive()
        with self.assertRaisesRegex(self.deploy.DeploymentError, "releases directory"):
            self.promote(archive, digest)
        self.assertEqual(list(outside.iterdir()), [])

    def test_deployment_root_symlink_is_rejected_before_writing_outside(self):
        outside = Path(self.tmp.name) / "outside-root"
        outside.mkdir()
        self.root.rmdir()
        self.root.symlink_to(outside, target_is_directory=True)
        archive, digest = self.archive()
        with self.assertRaisesRegex(self.deploy.DeploymentError, "deployment root"):
            self.promote(archive, digest)
        self.assertEqual(list(outside.iterdir()), [])

    def test_deployment_root_non_directory_is_rejected(self):
        self.root.rmdir()
        self.root.write_text("not a directory")
        archive, digest = self.archive()
        with self.assertRaisesRegex(self.deploy.DeploymentError, "deployment root is not a directory"):
            self.promote(archive, digest)

    def test_current_and_pending_symlink_boundaries_are_rejected_before_extraction(self):
        outside = Path(self.tmp.name) / "outside-boundary"
        outside.mkdir()
        (self.root / "releases").mkdir()
        (self.root / "current").symlink_to(outside, target_is_directory=True)
        archive, digest = self.archive("current-boundary.tar.gz")
        with self.assertRaisesRegex(self.deploy.DeploymentError, "current symlink escapes"):
            self.promote(archive, digest)
        self.assertEqual(list(outside.iterdir()), [])

        (self.root / "current").unlink()
        pending = self.root / "releases" / f".pending-{COMMIT_B}-fixed"
        pending.symlink_to(outside, target_is_directory=True)
        with mock.patch.object(self.deploy.secrets, "token_hex", return_value="fixed"):
            with self.assertRaisesRegex(self.deploy.DeploymentError, "pending release boundary"):
                self.promote(archive, digest)
        self.assertEqual(list(outside.iterdir()), [])

    def test_valid_archive_extracts_to_commit_release_and_is_never_overwritten(self):
        archive, digest = self.archive()
        result = self.promote(archive, digest)
        release = self.root / "releases" / COMMIT_B
        self.assertEqual(result["commit"], COMMIT_B)
        self.assertEqual((release / "server.mjs").read_text(), "console.log('ok')\n")
        self.assertEqual(json.loads((release / ".timmy-release.json").read_text())["sha256"], digest)
        with self.assertRaisesRegex(self.deploy.DeploymentError, "already exists"):
            self.promote(archive, digest)

    def test_success_atomically_swaps_current_and_uses_bounded_argv_commands(self):
        self.seed_release(COMMIT_A)
        self.point_current(COMMIT_A)
        archive, digest = self.archive()
        self.promote(archive, digest)
        self.assertEqual((self.root / "current").resolve(), self.root / "releases" / COMMIT_B)
        self.assertEqual([command for command, _ in self.commands], [("fixture-restart",), ("fixture-smoke",)])
        self.assertTrue(all(options["shell"] is False and options["timeout"] == 1.0 for _, options in self.commands))
        self.assertFalse(any(path.name.startswith(".current-") for path in self.root.iterdir()))

    def test_restart_health_and_smoke_failures_automatically_restore_prior_release(self):
        phases = ("restart", "health", "smoke")
        for phase in phases:
            with self.subTest(phase=phase):
                root = Path(self.tmp.name) / phase
                config = self.deploy.DeploymentConfig(**{**self.config.__dict__, "root": root})
                release = root / "releases" / COMMIT_A
                release.mkdir(parents=True)
                (release / "server.mjs").write_text("ok")
                (root / "current").symlink_to(Path("releases") / COMMIT_A)
                archive, digest = self.archive(f"{phase}.tar.gz")
                calls = []
                def run(argv, **kwargs):
                    calls.append(tuple(argv))
                    if (phase == "restart" and len(calls) == 1) or (phase == "smoke" and tuple(argv) == ("fixture-smoke",)):
                        raise subprocess.CalledProcessError(1, argv)
                    return subprocess.CompletedProcess(argv, 0, "", "")
                def health(url, commit, timeout):
                    if phase == "health" and commit == COMMIT_B:
                        raise self.deploy.DeploymentError("health failed")
                    return {"ok": True, "commit": commit}
                with self.assertRaisesRegex(self.deploy.DeploymentError, "promotion verification failed"):
                    self.deploy.promote(config=config, tag="daily-test", archive=archive, expected_sha256=digest, commit=COMMIT_B, run_command=run, health_check=health)
                self.assertEqual((root / "current").resolve(), root / "releases" / COMMIT_A)
                self.assertGreaterEqual(calls.count(("fixture-restart",)), 2)

    def test_first_promotion_failure_removes_current_and_stops_service_without_restart(self):
        archive, digest = self.archive("first-failure.tar.gz")
        calls = []

        def run(argv, **kwargs):
            calls.append(tuple(argv))
            if tuple(argv) == ("fixture-smoke",):
                raise subprocess.CalledProcessError(1, argv)
            return subprocess.CompletedProcess(argv, 0, "", "")

        with self.assertRaisesRegex(self.deploy.DeploymentError, "no prior release; service stopped"):
            self.promote(archive, digest, run_command=run)

        self.assertFalse((self.root / "current").exists())
        self.assertFalse((self.root / "current").is_symlink())
        self.assertEqual(calls, [("fixture-restart",), ("fixture-smoke",), ("fixture-stop",)])
        self.assertTrue((self.root / "releases" / COMMIT_B).is_dir(), "failed release remains inert evidence")

    def test_rollback_requires_valid_immutable_release_and_restarts_and_checks_health(self):
        self.seed_release(COMMIT_A)
        self.seed_release(COMMIT_B)
        self.point_current(COMMIT_B)
        result = self.deploy.rollback(config=self.config, commit=COMMIT_A, run_command=self.runner, health_check=self.healthy)
        self.assertEqual(result["commit"], COMMIT_A)
        self.assertEqual((self.root / "current").resolve(), self.root / "releases" / COMMIT_A)
        with self.assertRaisesRegex(self.deploy.DeploymentError, "does not exist"):
            self.deploy.rollback(config=self.config, commit="c" * 40, run_command=self.runner, health_check=self.healthy)
        tampered = self.seed_release("d" * 40)
        (tampered / ".timmy-release.json").write_text(json.dumps({"commit": COMMIT_A, "tag": "wrong"}))
        with self.assertRaisesRegex(self.deploy.DeploymentError, "metadata"):
            self.deploy.rollback(config=self.config, commit="d" * 40, run_command=self.runner, health_check=self.healthy)

    def test_default_command_adapter_accepts_verifier_kwargs_and_stays_bounded(self):
        result = self.deploy._run_argv(
            (sys.executable, "-c", "print('ok')"), check=True, text=True,
            capture_output=True, shell=False, timeout=1.0,
        )
        self.assertEqual(result.stdout.strip(), "ok")

    def test_status_and_cli_dry_run_are_rootless_and_machine_readable(self):
        self.seed_release(COMMIT_A)
        self.point_current(COMMIT_A)
        status = self.deploy.status(self.config)
        self.assertEqual(status["commit"], COMMIT_A)
        run = subprocess.run([
            sys.executable, str(SCRIPT), "--root", str(self.root), "--dry-run", "status"
        ], text=True, capture_output=True, check=False)
        self.assertEqual(run.returncode, 0, run.stderr)
        self.assertEqual(json.loads(run.stdout)["commit"], COMMIT_A)

    def test_commit_tag_and_checksum_arguments_are_strictly_validated(self):
        archive, digest = self.archive()
        for commit in ("abc", "A" * 40, "a" * 41, "../" + "a" * 40):
            with self.subTest(commit=commit), self.assertRaisesRegex(self.deploy.DeploymentError, "commit"):
                self.promote(archive, digest, commit=commit)
        with self.assertRaisesRegex(self.deploy.DeploymentError, "tag"):
            self.deploy.promote(config=self.config, tag="../bad", archive=archive, expected_sha256=digest, commit=COMMIT_B, run_command=self.runner, health_check=self.healthy)


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