import importlib.util import io import json import os from datetime import date from pathlib import Path import sys from unittest import TestCase, mock ROOT = Path(__file__).resolve().parents[2] SCRIPT = ROOT / "deploy" / "phase1_e2e_entry_gate.py" SPEC = importlib.util.spec_from_file_location("phase1_e2e_entry_gate", SCRIPT) MODULE = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = MODULE SPEC.loader.exec_module(MODULE) def example_manifest(): return json.loads( ( ROOT / "tests/contracts/resource_booking_phase1_e2e/manifest.example.json" ).read_text() ) class Phase1E2EEntryGateTests(TestCase): def test_http_error_retains_only_safe_request_attribution(self): url = "https://staging.example.invalid/api/admin/unschedule" body = b'{"error":"internal server error"}' http_error = MODULE.HTTPError(url, 500, "failure", None, io.BytesIO(body)) opener = mock.Mock() opener.open.side_effect = http_error with mock.patch.object(MODULE, "build_opener", return_value=opener): with self.assertRaises(MODULE.EntryGateError) as raised: MODULE._request_json( "POST", url, headers={ "X-Request-ID": "load-02-000590", "Authorization": "Token must-not-survive", }, payload={"signature": "must-not-survive"}, ) details = raised.exception.safe_details self.assertEqual( details, { "method": "POST", "path": "/api/admin/unschedule", "request_id": "load-02-000590", "http_status": 500, "response_body_sha256": MODULE.hashlib.sha256(body).hexdigest(), }, ) serialized = json.dumps(details) self.assertNotIn("staging.example.invalid", serialized) self.assertNotIn("must-not-survive", serialized) def test_example_manifest_covers_every_scoped_case(self): manifest = example_manifest() result = MODULE.load_and_validate_manifest( json.dumps(manifest), environment="staging", source_scope="default", require_approval=False, ) self.assertEqual( {case["id"] for case in result["cases"]}, set(MODULE.REQUIRED_CASE_DRIVERS), ) def test_entry_rejects_unapproved_fixture(self): with self.assertRaisesRegex(MODULE.EntryGateError, "not approved"): MODULE.load_and_validate_manifest( json.dumps(example_manifest()), environment="staging", source_scope="default", ) def test_entry_rejects_missing_or_wrong_case_driver(self): manifest = example_manifest() manifest["approved"] = True manifest["approval_ref"] = "approval-record-1" manifest["cases"][0]["driver"] = "direct_database_write" with self.assertRaisesRegex(MODULE.EntryGateError, "must use driver"): MODULE.load_and_validate_manifest( json.dumps(manifest), environment="staging", source_scope="default" ) def test_missing_credentials_block_before_network_access(self): with mock.patch.dict(os.environ, {}, clear=True): with mock.patch.object(MODULE, "_request_json") as request: with self.assertRaisesRegex( MODULE.EntryGateError, "configuration is incomplete" ): MODULE.EntryConfig.from_environment() request.assert_not_called() def test_timetabler_health_requires_exact_watermark_and_reverse_false(self): healthy = { "ready": True, "errors": [], "phase": "phase1", "reverse_delivery_enabled": False, "capture_enabled": True, "publish_enabled": True, "activation_approved": True, "source_scope": "default", "schema_version": "2", "transport": "kafka", "dead_letter_count": 0, "publisher_live": True, "transport_watermark": 61, "publisher_last_sequence": 61, } self.assertEqual(MODULE.validate_timetabler_health(healthy, "default"), 61) healthy["reverse_delivery_enabled"] = True with self.assertRaisesRegex(MODULE.EntryGateError, "reverse_delivery_enabled"): MODULE.validate_timetabler_health(healthy, "default") def test_manifest_requires_an_exact_watermark_for_every_stage(self): manifest = example_manifest() del manifest["expected_stage_start_watermarks"]["conflict"] with self.assertRaisesRegex(MODULE.EntryGateError, "every expected stage"): MODULE.load_and_validate_manifest( json.dumps(manifest), environment="staging", source_scope="default", require_approval=False, ) def test_workflow_entry_is_read_only_and_isolated(self): workflow = (ROOT / ".github/workflows/deploy.yml").read_text() entry_job = workflow.split(" phase1-e2e:\n", 1)[1] self.assertIn("inputs.phase1_e2e_action", entry_job) self.assertIn("phase1_e2e_entry_gate.py entry", entry_job) self.assertIn("TT_PHASE1_E2E_FIXTURE_JSON", entry_job) self.assertIn("TT_PHASE1_E2E_EMAIL", entry_job) self.assertIn("TT_PHASE1_E2E_PASSWORD", entry_job) self.assertNotIn("TT_PHASE1_E2E_ADMIN_TOKEN", entry_job) self.assertNotIn("TT_PHASE1_E2E_API_SECRET_KEY", entry_job) self.assertNotIn("RB_PHASE1_E2E_ADMIN_TOKEN", entry_job) self.assertNotIn("RB_PHASE1_E2E_SIGNATURE_SECRET", entry_job) self.assertNotIn("RB_PHASE1_E2E_BASE_URL", entry_job) self.assertIn("appleboy/ssh-action", entry_job) self.assertNotIn("deploy-staging.sh", entry_job) self.assertNotIn("LOAD-", entry_job) def test_host_attestation_emits_no_host_or_fingerprint(self): host = "private-staging-host.example" expected = __import__("hashlib").sha256(host.encode()).hexdigest() with mock.patch.dict( os.environ, { "TT_STAGING_HOST_VALUE": host, "TT_PHASE1_E2E_HOST_SHA256": expected, }, clear=True, ): self.assertEqual( MODULE.attest_staging_host(), {"staging_host_attested": True} ) def test_dynamic_login_uses_deployed_crypto_and_signer(self): source = SCRIPT.read_text() self.assertIn("encrypt_aes_128_cbc(self.password)", source) self.assertIn( "AdminApiBase.sign(signed.copy(), settings.API_SECRET_KEY)", source ) self.assertNotIn('os.environ.get("TT_PHASE1_E2E_API_SECRET_KEY"', source) def test_conflict_occurrence_basis_is_timezone_aware_and_bounded(self): occurrence = MODULE._conflict_occurrence_window( week_start=date(2025, 6, 23), slot=20, slot_per_day=48, minute_per_slot=30, duration_minutes=30, timezone_name="Asia/Singapore", ) self.assertEqual(occurrence["utc_start"], "2025-06-23T02:00:00Z") self.assertEqual(occurrence["utc_end"], "2025-06-23T02:30:00Z") self.assertEqual(occurrence["timezone"], "Asia/Singapore") def test_post_conflict_entry_requires_exact_disarmed_allocation(self): manifest = example_manifest() manifest["conflict_armed"] = False manifest["expected_stage_start_watermarks"]["conflict"] = 113 manifest["expected_stage_start_watermarks"]["entry"] = 114 self.assertEqual( MODULE._validate_conflict_allocation_state( manifest, scheduled=True, scheduled_slot=21, actual_staff_ids={1758}, actual_location_ids={99}, expected_staff_id=1758, expected_location_id=99, normalized_slot=21, slot_per_week=336, ), "allocated", ) manifest["conflict_armed"] = True with self.assertRaisesRegex(MODULE.EntryGateError, "disarmed post-conflict"): MODULE._validate_conflict_allocation_state( manifest, scheduled=True, scheduled_slot=21, actual_staff_ids={1758}, actual_location_ids={99}, expected_staff_id=1758, expected_location_id=99, normalized_slot=21, slot_per_week=336, ) def test_post_conflict_entry_rejects_missing_or_mismatched_allocation(self): manifest = example_manifest() manifest["conflict_armed"] = False manifest["expected_stage_start_watermarks"]["conflict"] = 113 manifest["expected_stage_start_watermarks"]["entry"] = 114 with self.assertRaisesRegex(MODULE.EntryGateError, "expected.*allocation"): MODULE._validate_conflict_allocation_state( manifest, scheduled=False, scheduled_slot=None, actual_staff_ids=set(), actual_location_ids=set(), expected_staff_id=1758, expected_location_id=99, normalized_slot=21, slot_per_week=336, ) with self.assertRaisesRegex(MODULE.EntryGateError, "does not match"): MODULE._validate_conflict_allocation_state( manifest, scheduled=True, scheduled_slot=22, actual_staff_ids={1758}, actual_location_ids={99}, expected_staff_id=1758, expected_location_id=99, normalized_slot=21, slot_per_week=336, ) def test_source_reads_send_bounded_user_agent_and_report_stage(self): source = SCRIPT.read_text() self.assertIn( '"User-Agent": f"timetabler-phase1-e2e/{config.run_id}"', source ) self.assertIn("source health read failed", source) self.assertIn("fixed-watermark snapshot read failed", source) def test_final_fixture_selectors_are_bounded_and_run_owned(self): run_id = "phase1-e2e-example" manifest = { "reference_selectors": { "all_fixture_activity_ids": { "model": "activity", "filters": {"code__startswith": f"{run_id}-"}, "expected_count": 9, "dedicated_e2e": True, }, "all_fixture_staff_ids": { "model": "staff", "filters": { "code__startswith": f"{run_id}-fixture-staff-" }, "expected_count": 7, "dedicated_e2e": True, }, "all_fixture_location_ids": { "model": "location", "filters": { "code__startswith": f"{run_id}-fixture-location-" }, "expected_count": 7, "dedicated_e2e": True, }, } } validated = MODULE.validate_final_fixture_selectors(manifest, run_id) self.assertEqual(len(validated), 3) manifest["reference_selectors"]["all_fixture_staff_ids"]["filters"] = { "code__startswith": "unrelated-" } with self.assertRaisesRegex(MODULE.EntryGateError, "bounded and run-owned"): MODULE.validate_final_fixture_selectors(manifest, run_id) def test_final_action_uses_absence_gate_instead_of_conflict_fixture(self): source = SCRIPT.read_text() self.assertIn('if action == "final":', source) self.assertIn("fixture_absence = final_fixture_absence", source) self.assertIn('result["fixture_absence"] = fixture_absence', source)