import importlib.util import json import os from pathlib import Path import shutil import subprocess import sys from tempfile import TemporaryDirectory from unittest import TestCase ROOT = Path(__file__).resolve().parents[2] SCRIPT_PATH = ROOT / "deploy" / "configure_phase1_publisher.py" SPEC = importlib.util.spec_from_file_location("configure_phase1_publisher", SCRIPT_PATH) MODULE = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(MODULE) class Phase1DeploymentGateTests(TestCase): def run_fake_remote_gate( self, *, fail_check=False, bootstrap_resources=False, revise_resource_projection=False, ): temporary_directory = TemporaryDirectory() self.addCleanup(temporary_directory.cleanup) app_dir = Path(temporary_directory.name) (app_dir / "deploy").mkdir() (app_dir / "venv" / "bin").mkdir(parents=True) (app_dir / "venv" / "bin" / "python").symlink_to(sys.executable) for name in ( "configure_phase1_publisher.py", "resolve_phase1_pm2.sh", "phase1-publisher-deployment-gate.sh", ): shutil.copy2(ROOT / "deploy" / name, app_dir / "deploy" / name) manage_log = app_dir / "manage.log" (app_dir / "manage.py").write_text( """import os import sys with open(os.environ["FAKE_MANAGE_LOG"], "a") as log: log.write(" ".join(sys.argv[1:]) + "\\n") command = sys.argv[1] if len(sys.argv) > 1 else "" if command == "check" and os.environ.get("FAKE_MANAGE_FAIL") == "true": raise SystemExit(7) if command == "showmigrations": print("[X] 0102_alter_integrationoutbox_dead_letter_at") """ ) pm2_log = app_dir / "pm2.log" fake_pm2 = app_dir / "pm2" fake_pm2.write_text( """#!/usr/bin/env bash set -eu printf '%s\\n' "$*" >> "$FAKE_PM2_LOG" if [[ $1 == --version ]]; then printf '5.4.0\\n' elif [[ $1 == jlist ]]; then printf '[]\\n' fi """ ) fake_pm2.chmod(0o700) env_file = app_dir / ".env" env_file.write_text( "RB_INTEGRATION_CAPTURE_ENABLED=True\n" "RB_INTEGRATION_PUBLISH_ENABLED=False\n" "RB_INTEGRATION_ACTIVATION_APPROVED=False\n" "RB_INTEGRATION_REVERSE_DELIVERY_ENABLED=False\n" ) environment = os.environ.copy() environment.update( { "TT_PHASE1_APP_DIR": str(app_dir), "TT_PHASE1_PM2_BIN": str(fake_pm2), "TT_PHASE1_MAIN_PM2_APP": "timetabler-api", "FAKE_PM2_LOG": str(pm2_log), "FAKE_MANAGE_LOG": str(manage_log), "FAKE_MANAGE_FAIL": "true" if fail_check else "false", "RB_INTEGRATION_PUBLISH_ENABLED": "False", "RB_INTEGRATION_ACTIVATION_APPROVED": "False", "RB_INTEGRATION_REVERSE_DELIVERY_ENABLED": "False", "TT_PHASE1_RESOURCE_BOOTSTRAP_ENABLED": ( "True" if bootstrap_resources else "False" ), "TT_PHASE1_RESOURCE_BOOTSTRAP_BATCH_SIZE": "25", "TT_PHASE1_RESOURCE_BOOTSTRAP_RUN_ID": "bootstrap-gate-test", "TT_PHASE1_RESOURCE_PROJECTION_REVISION_ENABLED": ( "True" if revise_resource_projection else "False" ), "TT_PHASE1_RESOURCE_PROJECTION_OPERATION_KEY": ( "rb-resource-projection-v2-test" ), } ) result = subprocess.run( ["bash", str(app_dir / "deploy" / "phase1-publisher-deployment-gate.sh")], check=False, capture_output=True, text=True, env=environment, ) return result, env_file.read_text(), pm2_log.read_text(), manage_log.read_text() def test_default_configuration_is_capture_on_but_publish_and_reverse_off(self): values = MODULE.requested_configuration({}) attestation = MODULE.validate(values) self.assertTrue(attestation["capture_enabled"]) self.assertFalse(attestation["publish_enabled"]) self.assertFalse(attestation["activation_approved"]) self.assertFalse(attestation["reverse_delivery_enabled"]) self.assertEqual(attestation["phase"], "phase1") self.assertEqual(attestation["schema_version"], "2") def test_activation_fails_without_broker_topic_snapshot_auth_and_joint_approval(self): values = MODULE.requested_configuration( { "RB_INTEGRATION_PUBLISH_ENABLED": "True", "RB_INTEGRATION_ACTIVATION_APPROVED": "False", } ) with self.assertRaisesRegex(ValueError, "enabled/disabled together"): MODULE.validate(values) values["RB_INTEGRATION_ACTIVATION_APPROVED"] = "True" values["RB_INTEGRATION_TRANSPORT"] = "kafka" values["RB_INTEGRATION_DEPLOYMENT_ID"] = "timetabler-staging" values["RB_INTEGRATION_SOURCE_SCOPE"] = "timetabler/staging" with self.assertRaisesRegex(ValueError, "missing"): MODULE.validate(values) def test_sasl_activation_attests_presence_without_returning_secrets(self): environment = { "RB_INTEGRATION_PUBLISH_ENABLED": "True", "RB_INTEGRATION_ACTIVATION_APPROVED": "True", "RB_INTEGRATION_TRANSPORT": "kafka", "RB_INTEGRATION_KAFKA_TOPIC": "tt-phase1", "RB_INTEGRATION_KAFKA_BOOTSTRAP_SERVERS": "broker.invalid:9093", "RB_INTEGRATION_KAFKA_SECURITY_PROTOCOL": "SASL_SSL", "RB_INTEGRATION_KAFKA_SASL_MECHANISM": "SCRAM-SHA-512", "RB_INTEGRATION_KAFKA_SASL_USERNAME": "producer-user", "RB_INTEGRATION_KAFKA_SASL_PASSWORD": "do-not-print-password", "RB_INTEGRATION_SNAPSHOT_SERVICE_TOKEN": "do-not-print-token", "RB_INTEGRATION_DEPLOYMENT_ID": "timetabler-staging", "RB_INTEGRATION_SOURCE_SCOPE": "timetabler/staging", } values = MODULE.requested_configuration(environment) attestation = MODULE.validate(values) rendered = str(attestation) self.assertTrue(attestation["publish_enabled"]) self.assertTrue(attestation["sasl_identity_configured"]) self.assertTrue(attestation["snapshot_token_configured"]) self.assertNotIn(environment["RB_INTEGRATION_KAFKA_SASL_PASSWORD"], rendered) self.assertNotIn(environment["RB_INTEGRATION_SNAPSHOT_SERVICE_TOKEN"], rendered) def test_activation_rejects_empty_deployment_and_source_scope(self): values = MODULE.requested_configuration( { "RB_INTEGRATION_PUBLISH_ENABLED": "True", "RB_INTEGRATION_ACTIVATION_APPROVED": "True", "RB_INTEGRATION_TRANSPORT": "kafka", "RB_INTEGRATION_KAFKA_TOPIC": "tt-phase1", "RB_INTEGRATION_KAFKA_BOOTSTRAP_SERVERS": "broker.invalid:9092", "RB_INTEGRATION_SNAPSHOT_SERVICE_TOKEN": "token", "RB_INTEGRATION_DEPLOYMENT_ID": "", "RB_INTEGRATION_SOURCE_SCOPE": "", } ) with self.assertRaisesRegex(ValueError, "deployment identity is required"): MODULE.validate(values) def test_environment_update_is_atomic_private_and_disable_only_preserves_other_values(self): with TemporaryDirectory() as directory: path = Path(directory) / ".env" path.write_text("OTHER_SETTING=kept\nRB_INTEGRATION_PUBLISH_ENABLED=True\n") MODULE.update_environment_file( path, { "RB_INTEGRATION_PUBLISH_ENABLED": "False", "RB_INTEGRATION_ACTIVATION_APPROVED": "False", "RB_INTEGRATION_REVERSE_DELIVERY_ENABLED": "False", }, ) content = path.read_text() self.assertIn("OTHER_SETTING=kept", content) self.assertIn("RB_INTEGRATION_PUBLISH_ENABLED=False", content) self.assertIn("RB_INTEGRATION_ACTIVATION_APPROVED=False", content) self.assertEqual(os.stat(path).st_mode & 0o777, 0o600) def test_shell_unsafe_legacy_dotenv_is_preserved_without_execution(self): with TemporaryDirectory() as directory: path = Path(directory) / ".env" path.write_text("LEGACY_VALUE=dotenv value & not shell syntax\n") MODULE.update_environment_file( path, { "RB_INTEGRATION_SNAPSHOT_SERVICE_TOKEN": "token & DP8qz", "RB_INTEGRATION_PUBLISH_ENABLED": "False", }, ) content = path.read_text() self.assertIn("LEGACY_VALUE=dotenv value & not shell syntax", content) self.assertIn( "RB_INTEGRATION_SNAPSHOT_SERVICE_TOKEN='token & DP8qz'", content ) def test_workflow_forces_phase1_reverse_off_schema2_and_runs_remote_gate(self): workflow = (ROOT / ".github" / "workflows" / "deploy.yml").read_text() gate = (ROOT / "deploy" / "phase1-publisher-deployment-gate.sh").read_text() self.assertIn("RB_INTEGRATION_PHASE: phase1", workflow) self.assertIn("RB_INTEGRATION_REVERSE_DELIVERY_ENABLED: 'False'", workflow) self.assertIn("RB_INTEGRATION_SCHEMA_VERSION: '2'", workflow) self.assertIn("phase1-publisher-deployment-gate.sh", workflow) self.assertIn("--disable-only", workflow) self.assertIn("trap rollback_workflow EXIT", workflow) self.assertIn("resolve_phase1_pm2.sh", workflow) self.assertIn("Cannot safely stop an enabled prior publisher", workflow) self.assertIn("TT_PHASE1_DEFAULT_OFF_PRELOADED=True", workflow) self.assertIn("workflow_dispatch:", workflow) self.assertIn("bootstrap_resources:", workflow) self.assertIn("TT_PHASE1_RESOURCE_BOOTSTRAP_ENABLED", workflow) self.assertIn("revise_resource_projection:", workflow) self.assertIn("resource_projection_operation_key:", workflow) self.assertIn("TT_PHASE1_RESOURCE_PROJECTION_REVISION_ENABLED", workflow) self.assertIn('"$pm2_bin" restart "$main_app" --update-env', workflow) self.assertIn("TT_PHASE1_ENGINE_RESPONSE_CONSUMER_NAME", workflow) self.assertIn('--exclude-name "$engine_consumer_name"', workflow) self.assertIn( 'bash "$resolved_checkout/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 ) self.assertNotIn('"$pm2_bin" delete "$engine_consumer_name"', workflow) self.assertNotIn('"$pm2_bin" start "$engine_consumer_script"', workflow) self.assertIn("RB_INTEGRATION_KAFKA_BOOTSTRAP_SERVERS", workflow) self.assertIn("RB_INTEGRATION_SNAPSHOT_SERVICE_TOKEN", workflow) self.assertIn("RB_INTEGRATION_PUBLISHER_CONCURRENCY: '1'", workflow) self.assertIn("source \"$APP_DIR/deploy/resolve_phase1_pm2.sh\"", gate) self.assertIn('"$PM2_BIN" delete "$PUBLISHER_NAME"', gate) self.assertIn('"$PM2_BIN" start "$PYTHON_BIN"', gate) self.assertIn("probe_resource_booking_publisher --liveness", gate) self.assertIn("0102_alter_integrationoutbox_dead_letter_at", gate) self.assertIn("publisher_count=$(PM2_SILENT=true", gate) self.assertIn("forcing publication and approval off", gate) self.assertIn("snapshot-only default-off mode", gate) self.assertIn("phase1_attest_no_publisher_processes /proc", gate) self.assertNotIn('source "$ENV_FILE"', gate) self.assertIn("parse_pm2_jlist.py", gate) self.assertIn('--exclude-name "$CONSUMER_NAME"', gate) self.assertIn('PM2_SILENT=true "$PM2_BIN" jlist --silent', gate) self.assertIn("RB_INTEGRATION_PUBLISH_ENABLED=False", gate) self.assertIn("bootstrap_resource_booking_resources", gate) self.assertIn("--confirm-resource-bootstrap", gate) self.assertIn("Resource bootstrap is allowed only while", gate) self.assertIn("revise_resource_booking_resource_projection", gate) self.assertIn("--confirm-resource-projection-revision", gate) self.assertIn("projection revision cannot run together", gate) def test_pm2_jlist_parser_ignores_cli_warning_prefix(self): parser = ROOT / "deploy" / "parse_pm2_jlist.py" raw = ( ">>>> In-memory PM2 is out-of-date\n" "[PM2] warning text\n" '[{"name":"api","pm2_env":{"name":"api","pm_cwd":"/srv/tt"}},' '{"name":"publisher","pm2_env":{"name":"publisher","pm_cwd":"/srv/tt"}},' '{"name":"tt_response","pm2_env":{"name":"tt_response","pm_cwd":"/srv/tt"}}]\n' ) main = subprocess.run( [ sys.executable, str(parser), "find-main", "--app-dir", "/srv/tt", "--exclude-name", "publisher", "--exclude-name", "tt_response", ], input=raw, check=True, capture_output=True, text=True, ) self.assertEqual(main.stdout.strip(), "api") def test_pm2_jlist_parser_attests_exact_online_script(self): parser = ROOT / "deploy" / "parse_pm2_jlist.py" with TemporaryDirectory() as directory: script = Path(directory) / "kafka_consumer" / "tt_response.py" script.parent.mkdir() script.write_text("# consumer\n") document = json.dumps( [ { "name": "tt_response", "pid": 4321, "pm2_env": { "name": "tt_response", "status": "online", "pm_exec_path": str(script), "exec_interpreter": sys.executable, "pm_cwd": str(script.parent.parent), }, } ] ) result = subprocess.run( [ sys.executable, str(parser), "attest-script", "--name", "tt_response", "--script-path", str(script), "--interpreter", sys.executable, "--cwd", str(script.parent.parent), ], input=f"warning\n{document}", text=True, capture_output=True, check=False, ) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout.strip(), "tt_response") wrong_interpreter = subprocess.run( [ sys.executable, str(parser), "attest-script", "--name", "tt_response", "--script-path", str(script), "--interpreter", str(script.parent / "wrong-python"), "--cwd", str(script.parent.parent), ], input=document, text=True, capture_output=True, check=False, ) self.assertNotEqual(wrong_interpreter.returncode, 0) wrong = subprocess.run( [ sys.executable, str(parser), "attest-script", "--name", "tt_response", "--script-path", str(script.parent / "other.py"), ], input=document, text=True, capture_output=True, check=False, ) self.assertNotEqual(wrong.returncode, 0) configured = subprocess.run( [ sys.executable, str(parser), "attest-script-config", "--name", "tt_response", "--script-path", str(script), ], input=json.dumps( [ { "name": "tt_response", "pid": 0, "pm2_env": { "name": "tt_response", "status": "errored", "pm_exec_path": str(script), }, } ] ), text=True, capture_output=True, check=False, ) self.assertEqual(configured.returncode, 0, configured.stderr) def test_pm2_jlist_parser_emits_bounded_engine_failure_diagnostic(self): parser = ROOT / "deploy" / "parse_pm2_jlist.py" with TemporaryDirectory() as directory: root = Path(directory) script = root / "kafka_consumer" / "tt_response.py" script.parent.mkdir() script.write_text("# consumer\n") error_log = root / "tt-response-error.log" error_log.write_text( "Traceback (most recent call last):\n" f' File "{script}", line 84, in response_consumer\n' "ModuleNotFoundError: No module named 'safe_package.submodule'\n" "RuntimeError: secret-value-must-not-appear\n" ) document = json.dumps( [ { "name": "tt_response", "pid": 0, "pm2_env": { "name": "tt_response", "status": "errored", "pm_exec_path": str(script), "pm_err_log_path": str(error_log), "restart_time": 12, "unstable_restarts": 3, "exit_code": 1, "DATABASE_PASSWORD": "must-not-appear", }, } ] ) result = subprocess.run( [ sys.executable, str(parser), "diagnose-script", "--name", "tt_response", "--script-path", str(script), ], input=document, text=True, capture_output=True, check=True, ) evidence = json.loads(result.stdout) self.assertEqual(evidence["status"], "errored") self.assertEqual(evidence["restart_count"], 12) self.assertEqual( evidence["exception_class_counts"], {"ModuleNotFoundError": 1, "RuntimeError": 1}, ) self.assertEqual(evidence["last_exception_class"], "RuntimeError") self.assertEqual(evidence["missing_modules"], ["safe_package.submodule"]) self.assertEqual( evidence["traceback_frames"], [{"file": "tt_response.py", "function": "response_consumer", "line": 84}], ) self.assertFalse(evidence["raw_log_emitted"]) self.assertNotIn("secret-value", result.stdout) self.assertNotIn("must-not-appear", result.stdout) self.assertNotIn(str(root), result.stdout) def test_pm2_jlist_parser_rejects_symlink_error_log(self): parser = ROOT / "deploy" / "parse_pm2_jlist.py" with TemporaryDirectory() as directory: root = Path(directory) script = root / "tt_response.py" script.write_text("# consumer\n") target = root / "target.log" target.write_text("RuntimeError: hidden\n") link = root / "error.log" link.symlink_to(target) document = json.dumps( [ { "name": "tt_response", "pid": 0, "pm2_env": { "name": "tt_response", "status": "errored", "pm_exec_path": str(script), "pm_err_log_path": str(link), }, } ] ) result = subprocess.run( [ sys.executable, str(parser), "diagnose-script", "--name", "tt_response", "--script-path", str(script), ], input=document, text=True, capture_output=True, check=False, ) self.assertNotEqual(result.returncode, 0) self.assertNotIn("hidden", result.stdout + result.stderr) def test_workflow_exposes_read_only_engine_consumer_diagnostic(self): workflow = (ROOT / ".github" / "workflows" / "deploy.yml").read_text() self.assertIn("- diagnose-engine-consumer", workflow) self.assertIn('= "diagnose-engine-consumer"', workflow) self.assertIn("diagnose-script", workflow) self.assertIn("TT_PHASE1_LOAD_MANIFEST_JSON=''", workflow) def test_pm2_resolver_accepts_only_an_absolute_probed_executable(self): resolver = ROOT / "deploy" / "resolve_phase1_pm2.sh" with TemporaryDirectory() as directory: fake_pm2 = Path(directory) / "pm2" fake_pm2.write_text("#!/usr/bin/env bash\n[[ $1 == --version ]]\n") fake_pm2.chmod(0o700) environment = os.environ.copy() environment["TT_PHASE1_PM2_BIN"] = str(fake_pm2) result = subprocess.run( ["bash", str(resolver)], check=True, capture_output=True, text=True, env=environment, ) self.assertEqual(result.stdout.strip(), str(fake_pm2)) environment["TT_PHASE1_PM2_BIN"] = "pm2" rejected = subprocess.run( ["bash", str(resolver)], check=False, capture_output=True, text=True, env=environment, ) self.assertNotEqual(rejected.returncode, 0) self.assertIn("no absolute executable PM2", rejected.stderr) def test_pm2_resolver_sources_owner_controlled_home_nvm_runtime(self): resolver = ROOT / "deploy" / "resolve_phase1_pm2.sh" with TemporaryDirectory() as directory: home = Path(directory) nvm_dir = home / ".nvm" node_bin = home / "fake-node" / "bin" nvm_dir.mkdir() node_bin.mkdir(parents=True) fake_pm2 = node_bin / "pm2" fake_pm2.write_text( "#!/bin/bash\n[[ $1 == --version ]] && printf '7.0.3\\n'\n" ) fake_pm2.chmod(0o700) (nvm_dir / "nvm.sh").write_text( 'nvm() { export PATH="$HOME/fake-node/bin:$PATH"; }\n' ) environment = os.environ.copy() environment.pop("TT_PHASE1_PM2_BIN", None) environment.pop("TT_PHASE1_RUNTIME_BOOTSTRAP", None) environment["HOME"] = str(home) environment["PATH"] = "/usr/bin:/bin" result = subprocess.run( ["/bin/bash", str(resolver)], check=True, capture_output=True, text=True, env=environment, ) self.assertEqual(result.stdout.strip(), str(fake_pm2)) def test_pm2_resolver_attests_single_current_user_daemon_runtime(self): resolver = ROOT / "deploy" / "resolve_phase1_pm2.sh" with TemporaryDirectory() as directory: root = Path(directory) process_dir = root / "proc" / "123" node_bin = root / "node" / "bin" pm2_bin = root / "supervisor-tools" / "bin" process_dir.mkdir(parents=True) node_bin.mkdir(parents=True) pm2_bin.mkdir(parents=True) node = node_bin / "node" node.write_text("#!/bin/bash\nexit 0\n") node.chmod(0o700) fake_pm2 = pm2_bin / "pm2" fake_pm2.write_text( "#!/bin/bash\n[[ $1 == --version ]] && printf '7.0.3\\n'\n" ) fake_pm2.chmod(0o700) (process_dir / "exe").symlink_to(node) uid = os.geteuid() (process_dir / "status").write_text( f"Name:\tPM2 v7.0.1\nUid:\t{uid}\t{uid}\t{uid}\t{uid}\n" ) (process_dir / "cmdline").write_bytes( b"PM2 v7.0.1: God Daemon (/owner/.pm2)\x00" ) (process_dir / "environ").write_bytes( b"HOME=/owner\x00PATH=" + str(pm2_bin).encode() + b":/usr/bin\x00" ) result = subprocess.run( [ "/bin/bash", "-c", 'source "$1"; phase1_resolve_pm2_from_proc "$2"; printf "%s\\n" "$PHASE1_PROC_PM2_BIN"', "phase1-proc-test", str(resolver), str(root / "proc"), ], check=True, capture_output=True, text=True, ) self.assertEqual(Path(result.stdout.strip()), fake_pm2.resolve()) second_process = root / "proc" / "456" second_process.mkdir() (second_process / "exe").symlink_to(node) shutil.copy2(process_dir / "status", second_process / "status") shutil.copy2(process_dir / "cmdline", second_process / "cmdline") rejected = subprocess.run( [ "/bin/bash", "-c", 'source "$1"; phase1_resolve_pm2_from_proc "$2"', "phase1-proc-test", str(resolver), str(root / "proc"), ], check=False, capture_output=True, text=True, ) self.assertNotEqual(rejected.returncode, 0) self.assertIn("expected one current-user PM2 God daemon", rejected.stderr) def test_default_off_process_attestation_rejects_current_user_publisher(self): resolver = ROOT / "deploy" / "resolve_phase1_pm2.sh" with TemporaryDirectory() as directory: proc_root = Path(directory) / "proc" proc_root.mkdir() empty = subprocess.run( [ "/bin/bash", "-c", 'source "$1"; phase1_attest_no_publisher_processes "$2"', "phase1-process-test", str(resolver), str(proc_root), ], check=False, capture_output=True, text=True, ) self.assertEqual(empty.returncode, 0, empty.stderr) process_dir = proc_root / "789" process_dir.mkdir() uid = os.geteuid() (process_dir / "status").write_text( f"Name:\tpython\nUid:\t{uid}\t{uid}\t{uid}\t{uid}\n" ) (process_dir / "cmdline").write_bytes( b"/app/venv/bin/python\x00manage.py\x00" b"publish_resource_booking_outbox\x00--interval\x001\x00" ) rejected = subprocess.run( [ "/bin/bash", "-c", 'source "$1"; phase1_attest_no_publisher_processes "$2"', "phase1-process-test", str(resolver), str(proc_root), ], check=False, capture_output=True, text=True, ) self.assertNotEqual(rejected.returncode, 0) self.assertIn("found 1 publisher process", rejected.stderr) def test_default_off_remote_gate_stops_publisher_and_attests_without_starting_it(self): result, environment, pm2_log, manage_log = self.run_fake_remote_gate() self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("installed/configured but stopped", result.stdout) self.assertIn("delete timetabler-rb-phase1-publisher", pm2_log) self.assertIn("restart timetabler-api --update-env", pm2_log) self.assertIn("save --force", pm2_log) self.assertFalse(any(line.startswith("start ") for line in pm2_log.splitlines())) self.assertIn("RB_INTEGRATION_PUBLISH_ENABLED=False", environment) self.assertIn("RB_INTEGRATION_ACTIVATION_APPROVED=False", environment) self.assertIn("check", manage_log) self.assertIn("preflight_resource_booking_phase1", manage_log) self.assertNotIn("bootstrap_resource_booking_resources", manage_log) def test_default_off_remote_gate_runs_explicit_resource_bootstrap_once(self): result, _environment, _pm2_log, manage_log = self.run_fake_remote_gate( bootstrap_resources=True ) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(manage_log.count("bootstrap_resource_booking_resources"), 1) self.assertIn( "bootstrap_resource_booking_resources --confirm-resource-bootstrap " "--batch-size 25 --run-id bootstrap-gate-test", manage_log, ) def test_default_off_remote_gate_runs_projection_revision_once(self): result, _environment, _pm2_log, manage_log = self.run_fake_remote_gate( revise_resource_projection=True ) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual( manage_log.count("revise_resource_booking_resource_projection"), 1 ) self.assertIn( "revise_resource_booking_resource_projection " "--confirm-resource-projection-revision " "--operation-key rb-resource-projection-v2-test --batch-size 25", manage_log, ) def test_remote_gate_rejects_simultaneous_bootstrap_and_projection_revision(self): result, _environment, _pm2_log, manage_log = self.run_fake_remote_gate( bootstrap_resources=True, revise_resource_projection=True, ) self.assertNotEqual(result.returncode, 0) self.assertIn("cannot run together", result.stderr) self.assertNotIn("bootstrap_resource_booking_resources", manage_log) self.assertNotIn("revise_resource_booking_resource_projection", manage_log) def test_remote_gate_failure_rolls_back_flags_and_never_starts_publisher(self): result, environment, pm2_log, _manage_log = self.run_fake_remote_gate(fail_check=True) self.assertNotEqual(result.returncode, 0) self.assertIn("forcing publication and approval off", result.stderr) self.assertIn("RB_INTEGRATION_PUBLISH_ENABLED=False", environment) self.assertIn("RB_INTEGRATION_ACTIVATION_APPROVED=False", environment) self.assertIn("RB_INTEGRATION_REVERSE_DELIVERY_ENABLED=False", environment) self.assertFalse(any(line.startswith("start ") for line in pm2_log.splitlines())) self.assertIn("restart timetabler-api --update-env", pm2_log) self.assertIn("save --force", pm2_log)