import importlib.util import hashlib import json from pathlib import Path import sys from unittest import TestCase, mock ROOT = Path(__file__).resolve().parents[2] SCRIPT = ROOT / "deploy" / "phase1_e2e_harness.py" SPEC = importlib.util.spec_from_file_location("phase1_e2e_harness", 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 Phase1E2EHarnessTests(TestCase): def test_example_declares_all_scenarios_without_load_or_dr(self): manifest = example_manifest() MODULE.validate_execution_manifest(manifest, "phase1-e2e-example") self.assertFalse(manifest["load_enabled"]) self.assertFalse(manifest["dr_enabled"]) self.assertEqual( {case["id"] for case in manifest["cases"]}, set(MODULE.CASE_STAGES) ) def test_manifest_rejects_unsupported_domain_mutation_path(self): manifest = example_manifest() manifest["cases"][1]["steps"][0]["path"] = "/api/admin/database/reset" with self.assertRaisesRegex(MODULE.HarnessError, "unsupported Timetabler path"): MODULE.validate_execution_manifest(manifest, "phase1-e2e-example") def test_manifest_rejects_every_resource_booking_target(self): manifest = example_manifest() manifest["cases"][9]["steps"][0]["target"] = "resource_booking" with self.assertRaisesRegex(MODULE.HarnessError, "forbidden non-Timetabler"): MODULE.validate_execution_manifest(manifest, "phase1-e2e-example") def test_conflict_case_requires_one_timetabler_step_and_epoch(self): manifest = example_manifest() conflict = manifest["cases"][9] del conflict["execute_at_epoch"] with self.assertRaisesRegex(MODULE.HarnessError, "execute_at_epoch"): MODULE.validate_execution_manifest(manifest, "phase1-e2e-example") def test_manifest_rejects_unresolved_variables_before_mutation(self): manifest = example_manifest() manifest["cases"][0]["steps"][0]["payload"]["name"] = "{{missing_name}}" with self.assertRaisesRegex(MODULE.HarnessError, "unresolved pre-mutation"): MODULE.validate_execution_manifest(manifest, "phase1-e2e-example") def test_manifest_rejects_credential_like_variables(self): manifest = example_manifest() manifest["variables"]["resource_booking_token"] = "forbidden" with self.assertRaisesRegex(MODULE.HarnessError, "credential-like"): MODULE.validate_execution_manifest(manifest, "phase1-e2e-example") def test_bounded_multi_row_selector_is_allowed_for_bulk_fixture(self): manifest = example_manifest() selector = manifest["reference_selectors"]["bulk_activity_ids"] MODULE.validate_selector( "bulk_activity_ids", selector, run_id="phase1-e2e-example", require_run_tag=True, ) self.assertEqual(selector["expected_count"], 2) def test_evidence_hash_covers_canonical_payload(self): payload = {"run_id": "phase1-e2e-example", "final_watermark": 79} sealed = MODULE.seal_evidence(payload) expected = hashlib.sha256( json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() self.assertEqual(sealed["evidence_sha256"], expected) self.assertNotEqual( sealed["evidence_sha256"], MODULE.seal_evidence({**payload, "final_watermark": 80})["evidence_sha256"], ) def test_selector_must_be_dedicated_and_bounded(self): selector = { "model": "activity", "filters": {"code": "unrelated"}, "expected_count": 1, "dedicated_e2e": False, } with self.assertRaisesRegex(MODULE.HarnessError, "not marked dedicated"): MODULE.validate_selector( "unsafe", selector, run_id="phase1-e2e-example", require_run_tag=True ) def test_renderer_preserves_id_and_array_types(self): variables = {"activity_id": 17, "activity_ids": [17, 18]} self.assertEqual(MODULE._render("{{activity_id}}", variables), 17) self.assertEqual(MODULE._render("{{activity_ids}}", variables), [17, 18]) def test_harness_contains_no_direct_orm_write(self): source = SCRIPT.read_text() for forbidden in ( ".objects.create(", ".objects.update(", ".objects.delete(", ".save(", "raw(", ): self.assertNotIn(forbidden, source) def test_cleanup_uses_three_supported_api_transactions_and_exact_fence(self): source = (ROOT / "deploy/phase1_e2e_provision.py").read_text() self.assertEqual(source.count('"name": "delete-fixture-'), 3) for path in ( "/api/admin/activity/delete", "/api/admin/staff/delete", "/api/admin/location/delete", ): self.assertIn(f'"path": "{path}"', source) self.assertIn("cleanup_final = conflict_final + 3", source) def test_cleanup_attests_fixture_deletion_durability_and_publisher_catchup(self): source = SCRIPT.read_text() for required in ( "run-owned fixture rows remain after supported cleanup", "cleanup changed pre-existing durable outbox evidence", "timed out waiting for publisher to reach cleanup watermark", '"durable_state_deleted": False', '"direct_database_mutation": False', '"publisher_readiness"', ): self.assertIn(required, source) def test_operational_driver_never_sources_dotenv_or_deletes_durable_rows(self): source = (ROOT / "deploy/phase1_e2e_operations.sh").read_text() self.assertNotIn('source "$APP_DIR/.env"', source) self.assertNotIn("manage.py flush", source) self.assertNotIn("TRUNCATE", source) self.assertNotIn("DELETE FROM", source) self.assertIn("--disable-only", source) self.assertNotIn("RB_PHASE1_E2E_CONSUMER", source) self.assertNotIn("RB_CONSUMER", source) def test_workflow_passes_discovered_runtime_to_operational_driver(self): workflow = (ROOT / ".github/workflows/deploy.yml").read_text() branch = workflow.split(" restart|rollback)\n", 1)[1].split( " *)\n", 1 )[0] self.assertIn('TT_PHASE1_APP_DIR="$app_dir"', branch) self.assertIn('TT_PHASE1_PYTHON_BIN="$python_bin"', branch) self.assertIn("bash deploy/phase1_e2e_operations.sh", branch) def test_timetabler_harness_contains_no_rb_credentials_or_mutations(self): source = SCRIPT.read_text() workflow = ( (ROOT / ".github/workflows/deploy.yml") .read_text() .split(" phase1-e2e:\n", 1)[1] ) for forbidden in ( "ResourceBookingClient", "RB_PHASE1_E2E_ADMIN_TOKEN", "RB_PHASE1_E2E_SIGNATURE_SECRET", 'target == "resource_booking"', 'target": "resource_booking"', ): self.assertNotIn(forbidden, source) self.assertNotIn(forbidden, workflow) def test_final_stage_cannot_claim_cross_service_completion(self): source = SCRIPT.read_text() self.assertIn('"e2e13_completion_claim": False', source) self.assertNotIn("rb_unfiltered_reconciliation_run_id", source) def test_final_runner_does_not_resolve_deleted_fixtures(self): config = type("Config", (), {"run_id": "phase1-e2e-example"})() manifest = {"variables": {}, "reference_selectors": {"deleted": {}}} with mock.patch.object(MODULE, "resolve_selector") as resolve: runner = MODULE.ScenarioRunner( config=config, manifest=manifest, session=object(), resolve_fixtures=False, ) resolve.assert_not_called() self.assertEqual(runner.fixture_ids, {}) def test_final_stage_attests_absence_durable_outbox_and_readiness(self): source = SCRIPT.read_text() for required in ( 'resolve_fixtures=stage != "final"', "fixture_absence = gate.final_fixture_absence", '"durable_state_deleted": False', '"resource_booking_process_touched": False', 'payload["final_assertions"]', ): self.assertIn(required, source)