import json import os from pathlib import Path import shutil import subprocess import sys from tempfile import TemporaryDirectory from unittest import TestCase import venv ROOT = Path(__file__).resolve().parents[2] PARSER = ROOT / "deploy" / "parse_pm2_jlist.py" class TtResponseDeploymentTests(TestCase): def run_parser(self, *arguments: str, document: list[dict]): return subprocess.run( [sys.executable, str(PARSER), *arguments], input=f"PM2 warning\n{json.dumps(document)}\n", check=False, capture_output=True, text=True, ) def test_runtime_attestation_checks_pm2_and_live_process_environment(self): with TemporaryDirectory() as directory: root = Path(directory) app_dir = root / "timetabler" script = app_dir / "kafka_consumer" / "tt_response.py" python = app_dir / "venv" / "bin" / "python" script.parent.mkdir(parents=True) python.parent.mkdir(parents=True) script.write_text("# worker\n") python.write_text("# interpreter marker\n") proc_dir = root / "proc" / "4322" proc_dir.mkdir(parents=True) (proc_dir / "cwd").symlink_to(app_dir) (proc_dir / "cmdline").write_bytes( str(python).encode() + b"\0" + str(script).encode() + b"\0" ) (proc_dir / "environ").write_bytes( b"VIRTUAL_ENV=" + str(app_dir / "venv").encode() + b"\0PATH=" + str(python.parent).encode() + b":/usr/bin\0" ) process = { "name": "tt_response", "pm_id": 4, "pid": 4322, "pm2_env": { "name": "tt_response", "status": "online", "pm_exec_path": str(script), "pm_cwd": str(app_dir), "exec_interpreter": str(python), "VIRTUAL_ENV": str(app_dir / "venv"), "PATH": f"{python.parent}:/usr/bin", "restart_time": 6, }, } arguments = ( "attest-script", "--name", "tt_response", "--script-path", str(script), "--interpreter-path", str(python), "--cwd", str(app_dir), "--virtual-env", str(app_dir / "venv"), "--path-prefix", str(python.parent), "--proc-root", str(root / "proc"), "--expected-pm-id", "4", "--expected-pid", "4322", "--expected-restart-count", "6", "--print-state", ) result = self.run_parser(*arguments, document=[process]) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout.strip(), "4:4322:6") process["pm2_env"]["exec_interpreter"] = "/usr/bin/python3" rejected = self.run_parser(*arguments, document=[process]) self.assertNotEqual(rejected.returncode, 0) self.assertIn("virtualenv Python", rejected.stderr) def test_runtime_preflight_uses_exact_venv_without_importing_worker(self): preflight = ROOT / "deploy" / "preflight_tt_response_runtime.py" with TemporaryDirectory() as directory: app_dir = Path(directory).resolve() / "timetabler" (app_dir / "deploy").mkdir(parents=True) (app_dir / "kafka_consumer").mkdir() (app_dir / "backend").mkdir() shutil.copy2(preflight, app_dir / "deploy" / preflight.name) (app_dir / "kafka_consumer" / "tt_response.py").write_text( "raise RuntimeError('worker must not be imported by preflight')\n" ) (app_dir / "backend" / "__init__.py").write_text("") (app_dir / "backend" / "settings.py").write_text("") for module_name in ( "Crypto", "confluent_kafka", "django", "psycopg2", "redis", "requests", "rest_framework", ): (app_dir / f"{module_name}.py").write_text("") venv.EnvBuilder(with_pip=False).create(app_dir / "venv") result = subprocess.run( [ str(app_dir / "venv" / "bin" / "python"), str(app_dir / "deploy" / preflight.name), "--app-dir", str(app_dir), "--virtual-env", str(app_dir / "venv"), ], check=False, capture_output=True, text=True, ) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("preflight passed", result.stdout) wrong_venv = subprocess.run( [ sys.executable, str(app_dir / "deploy" / preflight.name), "--app-dir", str(app_dir), "--virtual-env", str(app_dir / "venv"), ], check=False, capture_output=True, text=True, ) self.assertNotEqual(wrong_venv.returncode, 0) self.assertIn("selected virtualenv", wrong_venv.stderr) def test_runtime_attestation_rejects_duplicate_consumer_names(self): document = [ { "name": "tt_response", "pid": pid, "pm2_env": { "name": "tt_response", "status": "online", "pm_exec_path": script, }, } for pid, script in ((10, "/srv/tt_response.py"), (11, "/srv/other.py")) ] result = self.run_parser( "attest-script", "--name", "tt_response", "--script-path", "/srv/tt_response.py", document=document, ) self.assertNotEqual(result.returncode, 0) self.assertIn("exactly one PM2 app", result.stderr) def test_non_target_fingerprint_ignores_only_tt_response(self): target = { "name": "tt_response", "pm_id": 4, "pid": 10, "pm2_env": {"name": "tt_response", "restart_time": 1}, } api = { "name": "api", "pm_id": 1, "pid": 20, "pm2_env": { "name": "api", "status": "online", "pm_exec_path": "/srv/api.py", "restart_time": 3, }, } first = self.run_parser( "fingerprint-excluding", "--name", "tt_response", document=[target, api], ) self.assertEqual(first.returncode, 0, first.stderr) target["pid"] = 11 target["pm2_env"]["restart_time"] = 2 target_changed = self.run_parser( "fingerprint-excluding", "--name", "tt_response", document=[target, api], ) self.assertEqual(first.stdout, target_changed.stdout) api["pid"] = 21 other_changed = self.run_parser( "fingerprint-excluding", "--name", "tt_response", document=[target, api], ) self.assertNotEqual(first.stdout, other_changed.stdout) def test_guarded_restart_rebinds_existing_pm2_app_once_and_saves_it(self): with TemporaryDirectory() as directory: app_dir = Path(directory).resolve() / "timetabler" for relative in ( "deploy", "kafka_consumer", "venv/bin", ): (app_dir / relative).mkdir(parents=True, exist_ok=True) for name in ( "parse_pm2_jlist.py", "preflight_tt_response_runtime.py", "resolve_phase1_pm2.sh", "restart_tt_response.sh", "tt_response.ecosystem.config.cjs", ): shutil.copy2(ROOT / "deploy" / name, app_dir / "deploy" / name) shutil.rmtree(app_dir / "venv") venv.EnvBuilder(with_pip=False).create(app_dir / "venv") (app_dir / "kafka_consumer" / "tt_response.py").write_text("# worker\n") (app_dir / "backend").mkdir() (app_dir / "backend" / "__init__.py").write_text("") (app_dir / "backend" / "settings.py").write_text("") for module_name in ( "Crypto", "confluent_kafka", "django", "psycopg2", "redis", "requests", "rest_framework", ): (app_dir / f"{module_name}.py").write_text("") python = app_dir / "venv" / "bin" / "python" script = app_dir / "kafka_consumer" / "tt_response.py" initial_state = [ { "name": "timetabler-api", "pm_id": 1, "pid": 3001, "pm2_env": { "name": "timetabler-api", "status": "online", "pm_exec_path": str(app_dir / "manage.py"), "pm_cwd": str(app_dir), "exec_interpreter": "/usr/bin/python3", "restart_time": 2, }, }, { "name": "tt_response", "pm_id": 4, "pid": 4321, "pm2_env": { "name": "tt_response", "status": "online", "pm_exec_path": str(script), "pm_cwd": str(app_dir), "exec_interpreter": "/usr/bin/python3", "restart_time": 5, }, }, ] state_path = app_dir / "pm2-state.json" state_path.write_text(json.dumps(initial_state)) log_path = app_dir / "pm2.log" fake_pm2 = app_dir / "pm2" fake_pm2.write_text( f"""#!{sys.executable} import json import os from pathlib import Path import sys state_path = Path(os.environ["FAKE_PM2_STATE"]) log_path = Path(os.environ["FAKE_PM2_LOG"]) arguments = sys.argv[1:] with log_path.open("a") as log: log.write(" ".join(arguments) + "\\n") if arguments == ["--version"]: print("7.0.1") elif arguments[:2] == ["jlist", "--silent"]: print(state_path.read_text()) elif arguments and arguments[0] == "startOrRestart": state = json.loads(state_path.read_text()) app_dir = Path(os.environ["TT_ENGINE_RESPONSE_APP_DIR"]) for process in state: if process["name"] == os.environ["TT_ENGINE_RESPONSE_CONSUMER_NAME"]: process["pid"] = 4322 process["pm2_env"].update({{ "pm_cwd": str(app_dir), "exec_interpreter": str(app_dir / "venv" / "bin" / "python"), "VIRTUAL_ENV": str(app_dir / "venv"), "PATH": os.environ["PATH"], "restart_time": process["pm2_env"]["restart_time"] + 1, }}) state_path.write_text(json.dumps(state)) elif arguments[:2] == ["save", "--force"]: pass else: raise SystemExit(9) """ ) fake_pm2.chmod(0o700) proc_dir = app_dir / "proc" / "4322" proc_dir.mkdir(parents=True) (proc_dir / "cwd").symlink_to(app_dir) (proc_dir / "cmdline").write_bytes( str(python).encode() + b"\0" + str(script).encode() + b"\0" ) expected_path = f"{python.parent}:/usr/bin:/bin" (proc_dir / "environ").write_bytes( b"VIRTUAL_ENV=" + str(app_dir / "venv").encode() + b"\0PATH=" + expected_path.encode() + b"\0" ) environment = os.environ.copy() environment.update( { "PATH": "/usr/bin:/bin", "TT_ENGINE_RESPONSE_APP_DIR": str(app_dir), "TT_ENGINE_RESPONSE_PROC_ROOT": str(app_dir / "proc"), "TT_ENGINE_RESPONSE_STABILITY_SECONDS": "0", "TT_PHASE1_PM2_BIN": str(fake_pm2), "FAKE_PM2_STATE": str(state_path), "FAKE_PM2_LOG": str(log_path), } ) result = subprocess.run( ["bash", str(app_dir / "deploy" / "restart_tt_response.sh")], check=False, capture_output=True, text=True, env=environment, ) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("remained stable", result.stdout) commands = log_path.read_text().splitlines() self.assertEqual( sum(command.startswith("startOrRestart ") for command in commands), 1 ) self.assertEqual(commands.count("save --force"), 1) final_state = json.loads(state_path.read_text()) worker = next(row for row in final_state if row["name"] == "tt_response") self.assertEqual(worker["pm_id"], 4) self.assertEqual(worker["pm2_env"]["exec_interpreter"], str(python)) self.assertEqual(worker["pm2_env"]["VIRTUAL_ENV"], str(app_dir / "venv")) self.assertEqual(worker["pm2_env"]["PATH"], expected_path) self.assertEqual(worker["pm2_env"]["restart_time"], 6) def test_workflow_uses_guarded_restart_instead_of_name_only_restart(self): workflow = (ROOT / ".github" / "workflows" / "deploy.yml").read_text() self.assertIn("deploy/restart_tt_response.sh", workflow) self.assertIn('TT_ENGINE_RESPONSE_APP_DIR="$resolved_checkout"', workflow) self.assertIn('TT_PHASE1_PM2_BIN="$pm2_bin"', workflow) self.assertNotIn( '"$pm2_bin" restart "$engine_consumer_name" --update-env', workflow )