"""Release boundary and rollback regression tests; no SSH or real PM2 changes.""" import importlib.util import io import json from pathlib import Path import shutil import subprocess import tarfile import tempfile import unittest from unittest.mock import patch ROOT = Path(__file__).resolve().parents[2] spec = importlib.util.spec_from_file_location("release_host", ROOT / "scripts/release/host.py") host = importlib.util.module_from_spec(spec) spec.loader.exec_module(host) class ArchiveTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.root = Path(self.temporary.name).resolve() self.archive = self.root / "artifact.tar.gz" self.destination = self.root / "release" def tearDown(self): self.temporary.cleanup() def archive_with(self, name, link=None, kind=None): with tarfile.open(self.archive, "w:gz") as bundle: member = tarfile.TarInfo(name) if link is not None: member.type, member.linkname = tarfile.SYMTYPE, link elif kind is not None: member.type = kind else: member.size = 4 bundle.addfile(member, None if link is not None or kind is not None else io.BytesIO(b"test")) def test_extracts_a_regular_file(self): self.archive_with("app/version.txt") host.safe_extract(self.archive, self.destination) self.assertEqual((self.destination / "app/version.txt").read_bytes(), b"test") def test_refuses_parent_traversal(self): self.archive_with("../outside") with self.assertRaisesRegex(RuntimeError, "Unsafe archive"): host.safe_extract(self.archive, self.destination) self.assertFalse((self.root / "outside").exists()) def test_refuses_absolute_paths(self): self.archive_with("/outside") with self.assertRaisesRegex(RuntimeError, "Unsafe archive"): host.safe_extract(self.archive, self.destination) def test_refuses_symlink_escape(self): self.archive_with("app/link", "../../outside") with self.assertRaisesRegex(RuntimeError, "escapes"): host.safe_extract(self.archive, self.destination) def test_refuses_absolute_symlink(self): self.archive_with("app/link", "/etc/passwd") with self.assertRaisesRegex(RuntimeError, "Absolute archive"): host.safe_extract(self.archive, self.destination) def test_refuses_special_files(self): self.archive_with("device", kind=tarfile.FIFOTYPE) with self.assertRaisesRegex(RuntimeError, "Special files"): host.safe_extract(self.archive, self.destination) def test_refuses_existing_release(self): self.destination.mkdir() self.archive_with("file") with self.assertRaisesRegex(RuntimeError, "already exists"): host.safe_extract(self.archive, self.destination) def test_keeps_disk_reserve_without_deleting_anything(self): self.archive_with("file") with patch.object(host.shutil, "disk_usage", return_value=shutil._ntuple_diskusage(10, 9, 1)): with self.assertRaisesRegex(RuntimeError, "Insufficient space"): host.safe_extract(self.archive, self.destination) self.assertFalse(self.destination.exists()) class ActivationTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.root = Path(self.temporary.name).resolve() self.releases = self.root / "releases" self.release_id = "20261030T120000Z-0123456789ab" self.target = self.releases / self.release_id self.target.mkdir(parents=True) self.previous = self.root / "old" (self.previous / ".next-old").mkdir(parents=True) (self.previous / ".next-old/BUILD_ID").write_text("old-build") (self.root / ".pm2").mkdir() (self.root / ".pm2/dump.pm2").write_text("[]") host.json_write(self.target / "release.json", {"sourceCommit": "0123456789ab" + "0" * 28, "buildId": "new-build"}) host.json_write(self.target / "staged.json", {"previousCwd": str(self.previous), "previousDistDir": ".next-old", "artifactSha256": "artifact-hash"}) self.app = {"pid": 100, "pm2_env": {"pm_exec_path": "/old/node_modules/next/dist/bin/next", "args": ["start", "-p", "3125", "-H", "127.0.0.1"], "CANVAS_ZCODE_STATE_DIR": "/private/state", "NEXT_DIST_DIR": ".next-old", "pm_err_log_path": "/logs/error", "pm_out_log_path": "/logs/out"}} self.patches = [patch.object(host, "RELEASES", self.releases), patch.object(host, "verify_shared"), patch.object(host, "app_state", return_value=(self.app, self.previous, ".next-old")), patch.object(host.Path, "home", return_value=self.root), patch.object(host, "run", return_value=""), patch.object(host, "restart"), patch.object(host, "job_status", return_value={"CanvasAgentRun": []})] self.mocks = [p.start() for p in self.patches] self.restart = self.mocks[-2] def tearDown(self): for p in reversed(self.patches): p.stop() self.temporary.cleanup() def test_active_job_blocks_before_restart(self): self.mocks[-1].return_value = {"CanvasAgentRun": [{"count": 1, "recent": 1}]} with self.assertRaisesRegex(RuntimeError, "Recent active"): host.activate(self.release_id) self.restart.assert_not_called() self.assertFalse((self.target / "activation.json").exists()) def test_new_release_health_failure_restores_previous_build(self): with patch.object(host, "healthy", side_effect=[False, True]): with self.assertRaisesRegex(RuntimeError, "Candidate health"): host.activate(self.release_id) self.assertEqual([call.args[0].name for call in self.restart.call_args_list], ["candidate.pm2.json", "previous.pm2.json"]) self.assertEqual(json.loads((self.target / "activation.json").read_text())["phase"], "rolled-back") def test_success_records_provenance_and_keeps_native_runtime_state(self): with patch.object(host, "healthy", return_value=True): result = host.activate(self.release_id) self.assertEqual(result["phase"], "active") config = json.loads((self.target / "candidate.pm2.json").read_text())["apps"][0] self.assertEqual(config["name"], "ai-chatbot") self.assertEqual(config["cwd"], str(self.target)) self.assertEqual(config["env"]["CANVAS_ZCODE_STATE_DIR"], "/private/state") self.assertNotIn("/private/state", json.dumps(result)) def test_changed_production_blocks_activation(self): self.mocks[2].return_value = (self.app, self.root / "other", ".next") with self.assertRaisesRegex(RuntimeError, "changed after staging"): host.activate(self.release_id) self.restart.assert_not_called() def test_never_retries_an_activation_without_inspection(self): host.json_write(self.target / "activation.json", {"phase": "activating"}) with self.assertRaisesRegex(RuntimeError, "Already activated"): host.activate(self.release_id) self.restart.assert_not_called() def test_rollback_failure_is_reported(self): with patch.object(host, "healthy", return_value=False): with self.assertRaisesRegex(RuntimeError, "Rollback health check failed"): host.activate(self.release_id) class HelperTests(unittest.TestCase): def test_production_e2e_switch_blocks_release(self): app = {"name": "ai-chatbot", "pm2_env": { "pm_cwd": "/var/www/ai-chatbot", "pm_exec_path": str(host.APP / "node_modules/next/dist/bin/next"), "watch": False, "CANVAS_E2E_LOCAL": "true"}} with patch.object(host.socket, "gethostname", return_value="SfSTE0ttmsycO"), \ patch.object(host, "run", return_value=json.dumps([app])): with self.assertRaisesRegex(RuntimeError, "Production E2E mode"): host.app_state() def test_release_ids_cannot_escape_root(self): for value in ["../app", "/etc", "20261030T120000Z-0123456789ab/../../outside"]: with self.assertRaisesRegex(RuntimeError, "Invalid release"): host.release_path(value) def test_task_query_javascript_is_valid(self): sources = [] with patch.object(host, "run", side_effect=lambda *a, **k: sources.append(k["stdin"]) or "{}"): host.job_status() result = subprocess.run(["node", "--check"], input=sources[0], text=True, capture_output=True) self.assertEqual(result.returncode, 0, result.stderr) if __name__ == "__main__": unittest.main()