252 lines
12 KiB
Python
252 lines
12 KiB
Python
"""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 RuntimeSnapshotTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temporary.name).resolve()
|
|
self.current = self.root / "current"
|
|
(self.current / "public").mkdir(parents=True)
|
|
(self.current / "node_modules/pkg").mkdir(parents=True)
|
|
(self.current / "public/asset.txt").write_bytes(b"old-public")
|
|
(self.current / "public/untracked-font.ttf").write_bytes(b"runtime-font")
|
|
(self.current / "node_modules/pkg/index.js").write_bytes(b"same-package")
|
|
self.destination = self.root / "release"
|
|
self.archive = self.root / "artifact.tar.gz"
|
|
data = {"public/asset.txt": b"new-public",
|
|
"node_modules/pkg/index.js": b"same-package",
|
|
"node_modules/patched/index.js": b"new-patch"}
|
|
manifest = {"schema": 2, "runtimeFiles": {
|
|
name: {"sha256": host.hashlib.sha256(body).hexdigest(), "bytes": len(body)}
|
|
for name, body in data.items()}, "runtimeLinks": {}}
|
|
data["release.json"] = json.dumps(manifest).encode()
|
|
with tarfile.open(self.archive, "w:gz") as bundle:
|
|
for name, body in data.items():
|
|
info = tarfile.TarInfo("./" + name)
|
|
info.size = len(body)
|
|
bundle.addfile(info, io.BytesIO(body))
|
|
|
|
def tearDown(self):
|
|
self.temporary.cleanup()
|
|
|
|
def copy_public(self, args, **kwargs):
|
|
self.assertEqual(args[:3], ["cp", "-a", "--reflink=always"])
|
|
shutil.copytree(args[-2], args[-1])
|
|
return ""
|
|
|
|
def test_snapshot_keeps_untracked_assets_and_applies_patched_dependencies(self):
|
|
with patch.object(host, "run", side_effect=self.copy_public), \
|
|
patch.object(host, "clone_file", side_effect=shutil.copyfile) as clone:
|
|
host.extract_runtime(self.archive, self.destination, self.current)
|
|
clone.assert_called_once()
|
|
self.assertEqual((self.destination / "public/untracked-font.ttf").read_bytes(), b"runtime-font")
|
|
self.assertEqual((self.destination / "public/asset.txt").read_bytes(), b"new-public")
|
|
self.assertEqual((self.current / "public/asset.txt").read_bytes(), b"old-public")
|
|
self.assertEqual((self.destination / "node_modules/patched/index.js").read_bytes(), b"new-patch")
|
|
(self.destination / "node_modules/pkg/index.js").write_bytes(b"changed")
|
|
self.assertEqual((self.current / "node_modules/pkg/index.js").read_bytes(), b"same-package")
|
|
|
|
def test_space_guard_runs_before_copying_or_mutating(self):
|
|
with patch.object(host.shutil, "disk_usage", return_value=shutil._ntuple_diskusage(10, 9, 1)), \
|
|
patch.object(host, "run") as copy:
|
|
with self.assertRaisesRegex(RuntimeError, "Insufficient space"):
|
|
host.extract_runtime(self.archive, self.destination, self.current)
|
|
copy.assert_not_called()
|
|
self.assertFalse(self.destination.exists())
|
|
|
|
def test_unsupported_clone_stops_without_falling_back_to_full_copy(self):
|
|
with patch.object(host, "run", side_effect=self.copy_public), \
|
|
patch.object(host, "clone_file", side_effect=OSError("unsupported")):
|
|
with self.assertRaisesRegex(OSError, "unsupported"):
|
|
host.extract_runtime(self.archive, self.destination, self.current)
|
|
self.assertEqual((self.current / "node_modules/pkg/index.js").read_bytes(), b"same-package")
|
|
|
|
|
|
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_runtime"),
|
|
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()
|