#!/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 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",), 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_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_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_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_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()