import copy import hashlib import datetime import importlib.util import json import os from pathlib import Path import re import stat import sys import threading import time from concurrent.futures import Future, ThreadPoolExecutor from tempfile import TemporaryDirectory from types import SimpleNamespace from unittest import TestCase, mock ROOT = Path(__file__).resolve().parents[2] SCRIPT = ROOT / "deploy" / "phase1_load_harness.py" SPEC = importlib.util.spec_from_file_location("phase1_load_harness", SCRIPT) MODULE = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = MODULE SPEC.loader.exec_module(MODULE) PROVISION_SCRIPT = ROOT / "deploy" / "phase1_load_provision.py" PROVISION_SPEC = importlib.util.spec_from_file_location( "phase1_load_provision_for_tests", PROVISION_SCRIPT ) PROVISION_MODULE = importlib.util.module_from_spec(PROVISION_SPEC) sys.modules[PROVISION_SPEC.name] = PROVISION_MODULE PROVISION_SPEC.loader.exec_module(PROVISION_MODULE) EXAMPLE = ROOT / "tests/contracts/resource_booking_phase1_load/manifest.example.json" RUN_ID = "phase1-load-20260808t010203z-a1b2c3d4e5f6" class _FakeRelation: def __init__(self, values=()): self.values = tuple(values) self.ordering = None def count(self): return len(self.values) def exists(self): return bool(self.values) def order_by(self, *fields): self.ordering = fields return sorted(self.values, key=lambda value: (value.week, value.id)) def values_list(self, field, *, flat=False): if not flat: return [(getattr(value, field),) for value in self.values] return [getattr(value, field) for value in self.values] def _template_with_weeks( count, *, academic_term_id=26, has_staff=False, module_term_id=26, module_week_count=None, module_availability_pattern="0" * 280, ): week_relation = _FakeRelation( SimpleNamespace(id=1000 + index, week=index) for index in reversed(range(1, count + 1)) ) module_week_count = count if module_week_count is None else module_week_count module_week_relation = _FakeRelation( SimpleNamespace(id=1000 + index, week=index) for index in reversed(range(1, module_week_count + 1)) ) return ( SimpleNamespace( id=7, academic_term_id=academic_term_id, academic_term=SimpleNamespace( week=week_relation, default_availability="0" * 280, ), module=SimpleNamespace( status=1, academic_term_id=module_term_id, week_pattern_id=None, week=module_week_relation, availability_id=None, availability_pattern=module_availability_pattern, ), staff=_FakeRelation((object(),) if has_staff else ()), location=_FakeRelation(), staff_suitability=_FakeRelation(), location_suitability=_FakeRelation(), ), week_relation, ) def manifest(): return json.loads(EXAMPLE.read_text()) def with_current_recovery_control( value, *, original_sha="a" * 40, recovery_sha="c" * 40, settled_watermark=4000, ): value["commits"] = { "timetabler": original_sha, "resource_booking": "b" * 40, } original_manifest_sha256 = MODULE._sha256(value) value["recovery_control"] = { "failed_run_id": RUN_ID, "failed_github_run_id": "31550768598", "original_manifest_sha256": original_manifest_sha256, "original_timetabler_sha": original_sha, "recovery_timetabler_sha": recovery_sha, "abort_evidence_sha256": "d" * 64, "settled_source_watermark": settled_watermark, "approved_actions": sorted(MODULE.CURRENT_LOAD02_RECOVERY_ACTIONS), "record_ref": "codex-thread:test-recovery", "approved_at": value["approved_window"]["starts_at"], } return value class Phase1LoadHarnessTests(TestCase): def test_example_is_complete_but_not_approved_or_armed(self): result = MODULE.validate_manifest( manifest(), run_id=RUN_ID, environment="staging", source_scope="default", require_approval=False, ) self.assertEqual( set(result["expected_stage_start_watermarks"]), set(MODULE.STAGES) ) self.assertEqual( set(result["workloads"]), set(MODULE.EXECUTION_WORKLOAD_STAGES) ) self.assertFalse(result["approved"]) self.assertTrue( result["workloads"]["load-07"]["restart_fence_policy"][ "armed_by_separate_artifact" ] ) self.assertTrue(result["workloads"]["load-05"]["maximum_recurrence_defined"]) self.assertEqual( result["workloads"]["load-05"]["expected_occurrence_count"], 384 ) self.assertEqual( list(result["stage_predecessors"]), list(MODULE.EXPECTED_STAGE_PREDECESSORS), ) def test_execution_rejects_unapproved_example(self): with self.assertRaisesRegex(MODULE.LoadHarnessError, "not approved"): MODULE.validate_manifest( manifest(), run_id=RUN_ID, environment="staging", source_scope="default", require_approval=True, ) def test_rate_mix_is_only_term_26_schedule_and_unschedule_transactions(self): value = manifest() validated = MODULE.validate_manifest( value, run_id=RUN_ID, environment="staging", source_scope="default", require_approval=False, ) self.assertEqual( validated["load_acceptance_scope"], MODULE.ACADEMIC_TERM_LOAD_SCOPE ) self.assertEqual( [step["path"] for step in validated["traffic_mix_steps"]], [ "/api/admin/schedule-request", "/api/admin/unschedule", "/api/admin/schedule-request", "/api/admin/unschedule", ], ) self.assertEqual( { step["workload_classification"] for step in validated["traffic_mix_steps"] }, {"schedule", "unschedule"}, ) def test_current_baseline_requires_exact_authoritative_term_26_binding(self): value = manifest() value["accepted_baseline"]["source_watermark"] = 2653 with self.assertRaisesRegex(MODULE.LoadHarnessError, "term 26"): MODULE.validate_manifest( value, run_id=RUN_ID, require_approval=False ) value["accepted_baseline"]["authoritative_term"] = { "academic_term_id": 26, "start_date": "2026-01-05", "end_date": "2026-06-26", "activity_count": 45, "first_activity_id": 672, "last_activity_id": 716, "activity_ids_sha256": "1b599dcd8709fcdb29b292b110ef1e5cb011929ba371011ef1ae055c9e408453", "source_transaction_id": "fc63fa1f-09e6-5af6-b4a9-699cebfd6693", "source_watermark": 2640, } value["load01_receiver_attestation"]["source_watermark"] = 2653 value["expected_stage_start_watermarks"]["entry"] = 2653 value["expected_stage_final_watermarks"]["entry"] = 2653 value["expected_stage_start_watermarks"]["load-01"] = 2653 value["expected_stage_final_watermarks"]["load-01"] = 2653 value["expected_stage_start_watermarks"]["provision"] = 2653 value["stage_predecessors"]["entry"]["external_root_sha256"] = value[ "e2e_final_evidence_sha256" ] # Shift every downstream fence by the same accepted-baseline delta. delta = 2653 - 117 for stage in MODULE.STAGES[3:]: value["expected_stage_start_watermarks"][stage] += delta value["expected_stage_final_watermarks"][stage] += delta value["expected_stage_final_watermarks"]["provision"] += delta MODULE.validate_manifest(value, run_id=RUN_ID, require_approval=False) def test_abandoned_load02_cleanup_requires_exact_run_fence_and_mutation_confirmation(self): environment = { "TT_PHASE1_LOAD_ACTION": MODULE.ABANDONED_LOAD02_CLEANUP_ACTION, "TT_PHASE1_LOAD_RUN_ID": MODULE.LOAD02_CHECKPOINT_RUN_ID, "TT_PHASE1_LOAD_CONFIRM_MUTATION": "False", "TT_PHASE1_LOAD_CONFIRM_OPERATIONAL": "False", } with mock.patch.dict(os.environ, environment, clear=False): with self.assertRaisesRegex(MODULE.LoadHarnessError, "mutation confirmation"): MODULE.cleanup_abandoned_load02_from_environment() def test_provision_boundary_cleanup_requires_zero_durable_load02_work_and_unallocated_fixtures(self): value = manifest() states = {} for name in ( "traffic_activity_ids", "traffic_recurrence_activity_ids", "maximum_recurrence_activity_ids", "bulk_activity_ids", "boundary_activity_ids", ): expected = value["fixture_selectors"][name]["expected_count"] states[name] = { "activity_count": expected, "present_activity_count": expected, "absent_activity_count": 0, "scheduled_count": 0, "staff_relation_count": 0, "location_relation_count": 0, } audit = MODULE._provision_boundary_cleanup_audit( manifest=value, expected_watermark=value["expected_stage_start_watermarks"]["load-02"], correlated_delivery_count=0, correlated_outbox_count=0, activity_states=states, ) self.assertTrue(audit["containment_passed"]) self.assertTrue(audit["provision_boundary_recovery"]) self.assertIsNone( MODULE._provision_boundary_cleanup_audit( manifest=value, expected_watermark=value["expected_stage_start_watermarks"]["load-02"] + 1, correlated_delivery_count=0, correlated_outbox_count=0, activity_states=states, ) ) with self.assertRaisesRegex(MODULE.LoadHarnessError, "durable LOAD-02"): MODULE._provision_boundary_cleanup_audit( manifest=value, expected_watermark=value["expected_stage_start_watermarks"]["load-02"], correlated_delivery_count=1, correlated_outbox_count=0, activity_states=states, ) allocated = copy.deepcopy(states) allocated["traffic_activity_ids"]["scheduled_count"] = 1 with self.assertRaisesRegex(MODULE.LoadHarnessError, "safely unallocated"): MODULE._provision_boundary_cleanup_audit( manifest=value, expected_watermark=value["expected_stage_start_watermarks"]["load-02"], correlated_delivery_count=0, correlated_outbox_count=0, activity_states=allocated, ) def test_provision_slot_attestation_exposes_sealed_mismatch(self): value = manifest() fixture = { "slot_a": 19, "slot_b": 20, } evidence = { "profile": "PROVISION", "service": "timetabler", "run_id": RUN_ID, "fixture_manifest": fixture, "stage_assertions": {"fixture_manifest": copy.deepcopy(fixture)}, } attestation = MODULE._provision_slot_attestation(evidence, value, RUN_ID) self.assertEqual(attestation["actual"], {"slot_a": 19, "slot_b": 20}) self.assertEqual(attestation["declared"], {"slot_a": 1, "slot_b": 2}) self.assertFalse(attestation["matches_manifest"]) evidence["stage_assertions"]["fixture_manifest"]["slot_b"] = 21 with self.assertRaisesRegex(MODULE.LoadHarnessError, "contradictory"): MODULE._provision_slot_attestation(evidence, value, RUN_ID) def test_completed_engine_future_failure_is_raised_during_submission(self): success = Future() success.set_result({"request_id": "ok"}) failure = Future() failure.set_exception(MODULE.LoadHarnessError("engine failed")) consumed = set() self.assertEqual( MODULE._consume_completed_futures([success], consumed), [{"request_id": "ok"}], ) self.assertEqual(MODULE._consume_completed_futures([success], consumed), []) with self.assertRaisesRegex(MODULE.LoadHarnessError, "engine failed"): MODULE._consume_completed_futures([failure], consumed) def test_provision_rejects_manifest_slot_mismatch_before_writes(self): value = manifest() facts = {"slot_a": 19, "slot_b": 20} with self.assertRaisesRegex( PROVISION_MODULE.ProvisionError, "manifest slots differ" ): PROVISION_MODULE._assert_manifest_slots_match_reference(value, facts) value["variables"].update(facts) self.assertEqual( PROVISION_MODULE._assert_manifest_slots_match_reference(value, facts), facts, ) def test_abandoned_cleanup_retry_accepts_only_the_exact_contiguous_prefix(self): rows = [ ( f"{MODULE.LOAD02_CHECKPOINT_RUN_ID}-cleanup-{index:06d}", MODULE.ABANDONED_LOAD02_CLEANUP_INITIAL_WATERMARK + index, ) for index in range(1, 12) ] state = MODULE._abandoned_cleanup_retry_state( run_id=MODULE.LOAD02_CHECKPOINT_RUN_ID, initial_watermark=MODULE.ABANDONED_LOAD02_CLEANUP_INITIAL_WATERMARK, current_watermark=2651, expected_current_watermark=2651, cleanup_transaction_count=13, completed_rows=rows, ) self.assertEqual(state["initial_watermark"], 2640) self.assertEqual(state["completed_transaction_count"], 11) self.assertEqual(state["remaining_transaction_count"], 2) self.assertEqual(state["final_watermark"], 2653) with self.assertRaisesRegex(MODULE.LoadHarnessError, "contiguous prefix"): MODULE._abandoned_cleanup_retry_state( run_id=MODULE.LOAD02_CHECKPOINT_RUN_ID, initial_watermark=MODULE.ABANDONED_LOAD02_CLEANUP_INITIAL_WATERMARK, current_watermark=2642, expected_current_watermark=2642, cleanup_transaction_count=13, completed_rows=[rows[0], rows[2]], ) with self.assertRaisesRegex(MODULE.LoadHarnessError, "outside its fence"): MODULE._abandoned_cleanup_retry_state( run_id=MODULE.LOAD02_CHECKPOINT_RUN_ID, initial_watermark=MODULE.ABANDONED_LOAD02_CLEANUP_INITIAL_WATERMARK, current_watermark=2641, expected_current_watermark=2641, cleanup_transaction_count=13, completed_rows=[(rows[0][0], 9999)], ) def test_run_id_is_collision_resistant_and_traversal_safe(self): MODULE.validate_run_id(RUN_ID) for invalid in ( "phase1-load-example", "phase1-load-20260808t010203z-short", "phase1-load-20260808t010203z-../../etc", "phase1-load-20260808T010203Z-a1b2c3d4e5f6", ): with self.assertRaises(MODULE.LoadHarnessError): MODULE.validate_run_id(invalid) def test_default_rates_and_durations_are_exact(self): value = manifest() self.assertEqual(value["load"], MODULE.DEFAULT_LOAD) self.assertEqual(value["profile_timing"], MODULE.DEFAULT_PROFILE_TIMING) value["load"]["p_tps"] = 3 value["load"]["burst_tps"] = 15 with self.assertRaisesRegex(MODULE.LoadHarnessError, "approved telemetry"): MODULE.validate_manifest( value, run_id=RUN_ID, require_approval=False, ) def test_non_timetabler_target_and_direct_path_are_rejected(self): value = manifest() value["traffic_mix_steps"][0]["target"] = "resource_booking" with self.assertRaisesRegex(MODULE.LoadHarnessError, "non-Timetabler"): MODULE.validate_manifest(value, run_id=RUN_ID, require_approval=False) def test_canonical_manifest_requires_all_named_approvals_and_window(self): value = manifest() self.assertEqual(set(value["operators"]), set(MODULE.REQUIRED_APPROVAL_ROLES)) del value["operators"]["resource_booking_owner"] with self.assertRaisesRegex(MODULE.LoadHarnessError, "role inventory"): MODULE.validate_manifest(value, run_id=RUN_ID, require_approval=False) value = manifest() del value["operators"]["product_owner"]["approved_at"] with self.assertRaisesRegex(MODULE.LoadHarnessError, "product_owner"): MODULE.validate_manifest(value, run_id=RUN_ID, require_approval=False) def test_execution_window_must_contain_full_profile_observation(self): value = manifest() now = MODULE._parse_iso8601("2026-08-09T00:00:00Z", "test") MODULE.assert_approved_window(value, "LOAD-02", now=now) too_late = MODULE._parse_iso8601("2026-08-09T07:40:00Z", "test") with self.assertRaisesRegex(MODULE.LoadHarnessError, "complete profile"): MODULE.assert_approved_window(value, "LOAD-02", now=too_late) outside = MODULE._parse_iso8601("2026-08-08T23:59:59Z", "test") with self.assertRaisesRegex(MODULE.LoadHarnessError, "outside"): MODULE.assert_approved_window(value, "LOAD-02", now=outside) value = manifest() value["traffic_mix_steps"][0]["path"] = "/api/admin/database/reset" with self.assertRaisesRegex(MODULE.LoadHarnessError, "unsupported Timetabler"): MODULE.validate_manifest(value, run_id=RUN_ID, require_approval=False) def test_manifest_rejects_credentials_and_reverse_phase2_or_dr(self): for key, value, message in ( ("reverse_delivery_enabled", True, "reverse delivery"), ("phase2_enabled", True, "Phase 2"), ("dr_executed", True, "DR"), ): candidate = manifest() candidate[key] = value with self.assertRaisesRegex(MODULE.LoadHarnessError, message): MODULE.validate_manifest( candidate, run_id=RUN_ID, require_approval=False ) candidate = manifest() candidate["variables"]["rb_password"] = "forbidden" with self.assertRaisesRegex(MODULE.LoadHarnessError, "credential-like"): MODULE.validate_manifest(candidate, run_id=RUN_ID, require_approval=False) candidate = manifest() candidate["evidence_exchange"]["resource_booking_reader_uid"] = 1073 with self.assertRaisesRegex(MODULE.LoadHarnessError, "evidence exchange"): MODULE.validate_manifest(candidate, run_id=RUN_ID, require_approval=False) def test_manifest_provision_correction_declaration_is_exact(self): value = manifest() declaration = value["evidence_corrections"]["provision"] self.assertEqual(set(declaration), MODULE.PROVISION_CORRECTION_KEYS) self.assertEqual( declaration["artifact_filename"], "provision-corrected-v7.json" ) self.assertEqual(declaration["source_timezone"], "Asia/Kuala_Lumpur") self.assertEqual(declaration["reason"], "missing-timezone-offset") MODULE.validate_manifest(value, run_id=RUN_ID, require_approval=False) for compatible_filename in ( "provision-corrected.json", "provision-corrected-v2.json", "provision-corrected-v3.json", "provision-corrected-v4.json", "provision-corrected-v5.json", "provision-corrected-v6.json", ): compatible = manifest() compatible["evidence_corrections"]["provision"]["artifact_filename"] = ( compatible_filename ) MODULE.validate_manifest(compatible, run_id=RUN_ID, require_approval=False) for key, invalid in ( ("artifact_filename", "../provision.json"), ("artifact_filename", "provision-corrected-v8.json"), ("source_timezone", "UTC"), ("reason", "different-reason"), ): candidate = manifest() candidate["evidence_corrections"]["provision"][key] = invalid with self.assertRaisesRegex( MODULE.LoadHarnessError, "correction declaration" ): MODULE.validate_manifest( candidate, run_id=RUN_ID, require_approval=False ) def test_engine_aware_load02_plan_rejects_partial_source_progress(self): value = manifest() with self.assertRaisesRegex(MODULE.LoadHarnessError, "not resumable"): MODULE._rate_execution_plan( manifest=value, run_id=RUN_ID, stage="load-02", rate=2, duration_seconds=1800, current_watermark=1707, ) def test_load02_resume_fails_closed_on_fence_or_prefix_mismatch(self): value = manifest() start = value["expected_stage_start_watermarks"]["load-02"] final = value["expected_stage_final_watermarks"]["load-02"] for watermark in (start - 1, final, final + 1): with self.assertRaisesRegex(MODULE.LoadHarnessError, "resume fence"): MODULE._load02_resume_prefix_evidence( manifest=value, run_id=RUN_ID, current_watermark=watermark, ) mismatched = [ { "source_sequence": start + 1, "request_id": f"{RUN_ID}-load-02-unrelated", } ] with ( mock.patch.object(MODULE, "_transaction_evidence", return_value=mismatched), self.assertRaisesRegex(MODULE.LoadHarnessError, "deterministic contiguous"), ): MODULE._load02_resume_prefix_evidence( manifest=value, run_id=RUN_ID, current_watermark=start + 1, ) def test_resume_requires_finalized_complete_transaction_members(self): valid = [ SimpleNamespace( transaction_finalized=True, transaction_count=2, transaction_index=index, ) for index in (1, 2) ] MODULE._assert_complete_transaction_members(valid) incomplete_sets = ( [ SimpleNamespace( transaction_finalized=False, transaction_count=1, transaction_index=1, ) ], [ SimpleNamespace( transaction_finalized=True, transaction_count=2, transaction_index=1, ) ], [ SimpleNamespace( transaction_finalized=True, transaction_count=2, transaction_index=2, ), SimpleNamespace( transaction_finalized=True, transaction_count=2, transaction_index=1, ), ], ) for members in incomplete_sets: with self.assertRaisesRegex(MODULE.LoadHarnessError, "not finalized"): MODULE._assert_complete_transaction_members(members) def test_load02_entry_requires_exact_engine_aware_fence(self): value = manifest() config = mock.Mock(action="load-02", run_id=RUN_ID) source = {"transport_watermark": 3433} with ( mock.patch.object( MODULE, "_assert_source_safety", return_value=source ) as source_safety, mock.patch.object(MODULE, "_process_counts", return_value={}), ): entry = MODULE._entry_evidence(config, value, {"profile": "PROVISION"}) source_safety.assert_called_once_with(expected_watermark=3433) self.assertEqual( entry["load02_resume"], { "approved_entry_watermark": 3433, "resume_watermark": 3433, "resumed_prefix_transaction_count": 0, "prefix_verified": False, "engine_aware_non_resumable": True, }, ) def test_run_rate_does_not_submit_after_partial_engine_aware_progress(self): value = manifest() fixtures = { name: list(range(selector["expected_count"])) for name, selector in value["fixture_selectors"].items() } session = mock.Mock() config = SimpleNamespace(run_id=RUN_ID, action="load-02") with mock.patch.object(MODULE, "_resolve_selectors", return_value=fixtures): runner = MODULE.LoadRunner( config=config, manifest=value, session=session, ) with mock.patch.object(MODULE, "_current_watermark", return_value=1610): with self.assertRaisesRegex(MODULE.LoadHarnessError, "not resumable"): runner.run_rate("load-02", rate=2, duration_seconds=1800) session.post.assert_not_called() def test_source_transaction_timestamps_localize_only_naive_values(self): naive = datetime.datetime(2026, 8, 8, 12, 30, 45) aware = datetime.datetime( 2026, 8, 8, 12, 30, 45, tzinfo=datetime.timezone(datetime.timedelta(hours=-4)), ) self.assertEqual( MODULE._timezone_qualified_iso(naive, source_timezone="Asia/Kuala_Lumpur"), "2026-08-08T12:30:45+08:00", ) self.assertEqual( MODULE._timezone_qualified_iso(aware, source_timezone="Asia/Kuala_Lumpur"), "2026-08-08T12:30:45-04:00", ) def test_corrected_provision_evidence_changes_only_bounded_fields(self): value = manifest() value["commits"] = { "timetabler": "a" * 40, "resource_booking": "b" * 40, } original = MODULE.seal_evidence( { "schema_version": 1, "profile": "PROVISION", "git_sha": value["evidence_corrections"]["provision"][ "original_timetabler_sha" ], "resource_booking_git_sha_attestation": value["evidence_corrections"][ "provision" ]["original_resource_booking_sha"], "github_run_id": "31263932393", "source_transactions": [ { "committed_at": "2026-08-08T20:00:00", "published_at": "2026-08-08T20:00:01+03:00", } ], } ) corrected = MODULE._build_corrected_provision_evidence( original=original, manifest=value, github_run_id="31270000000", inventory_counts={"all_fixture_activity_ids": 1364}, source_progress_attestation={ "historical_provision_final_watermark": 1609, "current_source_watermark": 2639, "load02_prefix_transaction_count": 1030, "load02_prefix_source_transactions_sha256": "d" * 64, "load02_prefix_verified": True, }, ) self.assertEqual( corrected["evidence_correction"], value["evidence_corrections"]["provision"], ) self.assertEqual(corrected["git_sha"], "a" * 40) self.assertEqual(corrected["resource_booking_git_sha_attestation"], "b" * 40) self.assertEqual(corrected["github_run_id"], "31270000000") transaction = corrected["source_transactions"][0] self.assertEqual(transaction["committed_at"], "2026-08-08T20:00:00+08:00") self.assertEqual(transaction["published_at"], "2026-08-08T20:00:01+03:00") self.assertFalse(corrected["correction_action_assertions"]["database_write"]) self.assertEqual( corrected["post_provision_source_progress"]["current_source_watermark"], 2639, ) self.assertEqual(len(corrected["source_transactions"]), 1) self.assertEqual( corrected["evidence_sha256"], MODULE._sha256( { key: item for key, item in corrected.items() if key != "evidence_sha256" } ), ) def test_correction_progress_attests_prefix_without_mixing_transactions(self): value = manifest() value["expected_stage_start_watermarks"]["load-02"] = 1609 value["expected_stage_final_watermarks"]["load-02"] = 5209 value["expected_stage_final_watermarks"]["provision"] = 1609 value["provision_plan"]["expected_source_transactions"] = 1492 prefix = [ { "source_sequence": 1610 + index, "request_id": MODULE._rate_request_id( MODULE.LOAD02_CHECKPOINT_RUN_ID, "load-02", index + 1 ), } for index in range(1030) ] with mock.patch.object( MODULE, "_load02_resume_prefix_evidence", return_value=prefix ) as verify: progress = MODULE._load02_source_progress_attestation( manifest=value, run_id=MODULE.LOAD02_CHECKPOINT_RUN_ID, current_watermark=2639, ) verify.assert_called_once_with( manifest=value, run_id=MODULE.LOAD02_CHECKPOINT_RUN_ID, current_watermark=2639, ) self.assertEqual(progress["historical_provision_final_watermark"], 1609) self.assertEqual(progress["current_source_watermark"], 2639) self.assertEqual(progress["load02_prefix_transaction_count"], 1030) self.assertEqual( progress["load02_prefix_source_transactions_sha256"], MODULE._sha256(prefix), ) self.assertEqual( progress["source_transactions_in_corrected_provision_evidence"], 1492 ) self.assertEqual( progress["prefix_source_transactions_in_corrected_provision_evidence"], 0, ) def test_correction_action_accepts_only_a_reverified_partial_load02_prefix(self): value = manifest() value["commits"] = { "timetabler": "a" * 40, "resource_booking": "b" * 40, } value["approved_window"] = { "starts_at": "2026-01-01T00:00:00Z", "ends_at": "2027-01-01T00:00:00Z", } original_sha = value["evidence_corrections"]["provision"][ "original_evidence_sha256" ] original = {"evidence_sha256": original_sha} progress = { "historical_provision_final_watermark": 1609, "load02_approved_final_watermark": 5209, "current_source_watermark": 2639, "load02_prefix_transaction_count": 1030, "load02_prefix_source_transactions_sha256": "d" * 64, "load02_prefix_verified": True, "source_transactions_in_corrected_provision_evidence": 1492, "prefix_source_transactions_in_corrected_provision_evidence": 0, } metrics = { "transport_watermark": 2639, "publisher_last_sequence": 2639, } corrected = MODULE.seal_evidence( { "schema_version": 1, "profile": "PROVISION", "service": "timetabler", "run_id": RUN_ID, "source_transactions": [], } ) inventory = { key: list(range(selector["expected_count"])) for key, selector in value["fixture_selectors"].items() } environment = { "TT_PHASE1_LOAD_ACTION": "correct-provision-evidence", "TT_PHASE1_LOAD_RUN_ID": RUN_ID, "TT_PHASE1_LOAD_PRIOR_EVIDENCE_SHA256": original_sha, "TT_PHASE1_LOAD_GITHUB_RUN_ID": "31273000002", "TT_PHASE1_LOAD_MANIFEST_JSON": "{}", "TT_PHASE1_LOAD_ENVIRONMENT": "staging", "TT_PHASE1_LOAD_SOURCE_SCOPE": "default", "TT_PHASE1_LOAD_EXPECTED_SHA": "a" * 40, "TT_PHASE1_LOAD_CONFIRM_MUTATION": "False", "TT_PHASE1_LOAD_CONFIRM_OPERATIONAL": "False", } with TemporaryDirectory() as directory: root = Path(directory) / "exchange" run_dir = root / RUN_ID original_path = run_dir / "provision.json" root.mkdir(mode=0o755) run_dir.mkdir(mode=0o755) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(MODULE, "load_manifest", return_value=value), mock.patch.object(MODULE, "assert_evidence_exchange_runtime"), mock.patch.object( MODULE, "_assert_source_safety", side_effect=(metrics, metrics), ) as source_safety, mock.patch.object( MODULE, "_load02_source_progress_attestation", side_effect=(progress, progress), ) as source_progress, mock.patch.object( MODULE, "_evidence_path", return_value=(root, run_dir, original_path), ), mock.patch.object( MODULE, "_read_immutable_exchange_json", side_effect=((original, mock.Mock()), (original, mock.Mock())), ), mock.patch.object(MODULE, "_validate_original_provision_evidence"), mock.patch.object(MODULE, "_resolve_selectors", return_value=inventory), mock.patch.object( MODULE, "_build_corrected_provision_evidence", return_value=corrected, ) as build, mock.patch.object(MODULE, "TIMETABLER_EVIDENCE_OWNER_UID", os.getuid()), ): result = MODULE.correct_provision_evidence_from_environment() self.assertEqual(result["final_watermark"], 2639) self.assertEqual(result["load02_prefix_transaction_count"], 1030) self.assertFalse(result["mutation_executed"]) self.assertFalse(result["operational_action_executed"]) self.assertFalse(result["load_executed"]) self.assertEqual( [call.kwargs for call in source_safety.call_args_list], [{"expected_watermark": None}, {"expected_watermark": 2639}], ) self.assertEqual( [ call.kwargs["current_watermark"] for call in source_progress.call_args_list ], [2639, 2639], ) self.assertEqual( build.call_args.kwargs["source_progress_attestation"], progress ) invalid_environment = {**environment, "TT_PHASE1_LOAD_ACTION": "load-02"} with ( mock.patch.dict(os.environ, invalid_environment, clear=False), self.assertRaisesRegex(MODULE.LoadHarnessError, "CLI action differs"), ): MODULE.correct_provision_evidence_from_environment() def test_checkpoint_load02_persists_exact_immutable_read_only_prefix(self): value = manifest() value["expected_stage_start_watermarks"]["load-02"] = 1609 value["expected_stage_final_watermarks"]["load-02"] = 5209 value["commits"] = { "timetabler": "a" * 40, "resource_booking": "b" * 40, } prefix = [ { "source_sequence": 1610 + index, "request_id": MODULE._rate_request_id( MODULE.LOAD02_CHECKPOINT_RUN_ID, "load-02", index + 1 ), "workload_classification": "staff_lifecycle", } for index in range(1030) ] metrics = { "transport_watermark": 2639, "publisher_last_sequence": 2639, "publisher_live": True, "reverse_delivery_enabled": False, } environment = { "TT_PHASE1_LOAD_ACTION": "checkpoint-load-02", "TT_PHASE1_LOAD_RUN_ID": MODULE.LOAD02_CHECKPOINT_RUN_ID, "TT_PHASE1_LOAD_GITHUB_RUN_ID": "31273000000", "TT_PHASE1_LOAD_PRIOR_EVIDENCE_SHA256": "c" * 64, "TT_PHASE1_LOAD_MANIFEST_JSON": "{}", "TT_PHASE1_LOAD_ENVIRONMENT": "staging", "TT_PHASE1_LOAD_SOURCE_SCOPE": "default", "TT_PHASE1_LOAD_EXPECTED_SHA": "a" * 40, "TT_PHASE1_LOAD_CONFIRM_MUTATION": "False", "TT_PHASE1_LOAD_CONFIRM_OPERATIONAL": "False", } with TemporaryDirectory() as directory: root = Path(directory) / "exchange" run_dir = root / MODULE.LOAD02_CHECKPOINT_RUN_ID root.mkdir(mode=0o755) run_dir.mkdir(mode=0o755) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(MODULE, "load_manifest", return_value=value), mock.patch.object(MODULE, "assert_evidence_exchange_runtime"), mock.patch.object( MODULE, "verify_predecessor_evidence", return_value={ "profile": "PROVISION", "evidence_sha256": "c" * 64, }, ), mock.patch.object(MODULE, "assert_approved_window"), mock.patch.object( MODULE, "_assert_source_safety", return_value=metrics ), mock.patch.object( MODULE, "_load02_resume_prefix_evidence", return_value=prefix ) as verify_prefix, mock.patch.object( MODULE, "_ensure_evidence_exchange_directories", return_value=(root, run_dir), ), mock.patch.object(MODULE, "TIMETABLER_EVIDENCE_OWNER_UID", os.getuid()), ): first = MODULE.checkpoint_load02_from_environment() second = MODULE.checkpoint_load02_from_environment() os.environ["TT_PHASE1_LOAD_GITHUB_RUN_ID"] = "31273000001" with self.assertRaisesRegex( MODULE.LoadHarnessError, "different or unsafe" ): MODULE.checkpoint_load02_from_environment() target = run_dir / "load-02-checkpoint.json" artifact = json.loads(target.read_bytes()) self.assertEqual(first["evidence_sha256"], second["evidence_sha256"]) self.assertEqual(stat.S_IMODE(target.stat().st_mode), 0o444) self.assertEqual(artifact["entry_watermark"], 1609) self.assertEqual(artifact["current_watermark"], 2639) self.assertEqual(artifact["prefix_transaction_count"], 1030) self.assertEqual(artifact["source_transactions"], prefix) self.assertEqual( artifact["source_transactions_sha256"], MODULE._sha256(prefix) ) self.assertFalse(artifact["load_executed"]) self.assertFalse(artifact["mutation_executed"]) self.assertFalse(artifact["database_write_executed"]) self.assertFalse(artifact["resource_booking_process_touched"]) self.assertFalse(artifact["phase2_enabled"]) self.assertFalse(artifact["reverse_delivery_enabled"]) self.assertFalse(artifact["dr_executed"]) self.assertEqual(verify_prefix.call_count, 6) def test_checkpoint_load02_rejects_wrong_run_or_any_confirmation(self): base = { "TT_PHASE1_LOAD_ACTION": "checkpoint-load-02", "TT_PHASE1_LOAD_RUN_ID": RUN_ID, "TT_PHASE1_LOAD_CONFIRM_MUTATION": "False", "TT_PHASE1_LOAD_CONFIRM_OPERATIONAL": "False", } with ( mock.patch.dict(os.environ, base, clear=False), self.assertRaisesRegex(MODULE.LoadHarnessError, "fixed recovery"), ): MODULE.checkpoint_load02_from_environment() for confirmation in ( "TT_PHASE1_LOAD_CONFIRM_MUTATION", "TT_PHASE1_LOAD_CONFIRM_OPERATIONAL", ): candidate = { **base, "TT_PHASE1_LOAD_RUN_ID": MODULE.LOAD02_CHECKPOINT_RUN_ID, confirmation: "True", } with ( mock.patch.dict(os.environ, candidate, clear=False), self.assertRaisesRegex(MODULE.LoadHarnessError, "false mutation"), ): MODULE.checkpoint_load02_from_environment() def test_load02_diagnostic_reports_only_bounded_first_unexpected_attribution(self): value = manifest() start = value["expected_stage_start_watermarks"]["load-02"] run_id = MODULE.LOAD02_CHECKPOINT_RUN_ID rows = [ SimpleNamespace( transport_sequence=start + 1, transaction_finalized=True, transaction_count=1, transaction_index=1, status="published", request_id=MODULE._rate_request_id(run_id, "load-02", 1), event_type="timetabler.staff.updated", ), SimpleNamespace( transport_sequence=start + 2, transaction_finalized=True, transaction_count=1, transaction_index=1, status="published", request_id=f"{run_id}-load-02-retry-000002", event_type="timetabler.location.updated", ), SimpleNamespace( transport_sequence=start + 3, transaction_finalized=True, transaction_count=1, transaction_index=1, status="published", request_id="unrelated-private-attribution", event_type="timetabler.activity.updated", ), ] result = MODULE._diagnose_load02_prefix_rows( rows=rows, manifest=value, run_id=run_id, current_watermark=start + 3, ) self.assertFalse(result["strict_prefix_attributable"]) self.assertFalse(result["resume_or_correction_permitted"]) self.assertEqual(result["unexpected_transaction_count"], 2) self.assertEqual(result["first_unexpected"]["source_sequence"], start + 2) self.assertEqual( result["first_unexpected"]["request_attributions"][0], { "classification": "same_run_load02_variant", "request_id_present": True, "request_id_length": len(f"{run_id}-load-02-retry-000002"), "request_id_sha256": hashlib.sha256( f"{run_id}-load-02-retry-000002".encode() ).hexdigest(), "safe_same_run_suffix": "load-02-retry-000002", }, ) foreign = MODULE._bounded_request_attribution( actual_request_id="unrelated-private-attribution", expected_request_id=MODULE._rate_request_id(run_id, "load-02", 3), run_id=run_id, ) self.assertEqual( foreign["classification"], "foreign_or_unclassified_request_id" ) self.assertIsNone(foreign["safe_same_run_suffix"]) self.assertNotIn("unrelated-private-attribution", json.dumps(foreign)) def test_load02_diagnostic_proves_order_independent_exact_iteration_bijection(self): value = manifest() start = value["expected_stage_start_watermarks"]["load-02"] run_id = MODULE.LOAD02_CHECKPOINT_RUN_ID rows = [ SimpleNamespace( transport_sequence=start + offset, transaction_finalized=True, transaction_count=1, transaction_index=1, status="published", request_id=MODULE._rate_request_id(run_id, "load-02", iteration), event_type="timetabler.staff.updated", ) for offset, iteration in enumerate((2, 3, 1), start=1) ] result = MODULE._diagnose_load02_prefix_rows( rows=rows, manifest=value, run_id=run_id, current_watermark=start + 3, ) inventory = result["request_iteration_inventory"] self.assertEqual(result["unexpected_transaction_count"], 3) self.assertFalse(result["strict_prefix_attributable"]) self.assertTrue(inventory["exact_current_prefix_bijection"]) self.assertEqual( inventory["present_iteration_ranges"], [{"start": 1, "end": 3}] ) self.assertEqual(inventory["missing_current_prefix_iteration_count"], 0) self.assertEqual(inventory["duplicate_iteration_count"], 0) self.assertEqual(inventory["out_of_range_iteration_count"], 0) self.assertEqual( inventory["missing_approved_iteration_ranges"], [{"start": 4, "end": 3600}], ) self.assertFalse(result["resume_or_correction_permitted"]) def test_load02_diagnostic_action_is_read_only_and_writes_no_artifact(self): value = manifest() value["commits"] = { "timetabler": "a" * 40, "resource_booking": "b" * 40, } metrics = { "transport_watermark": 4000, "publisher_last_sequence": 4000, "publisher_live": True, "reverse_delivery_enabled": False, } diagnosis = { "strict_prefix_attributable": False, "first_unexpected": {"source_sequence": 1708}, "resume_or_correction_permitted": False, } environment = { "TT_PHASE1_LOAD_ACTION": "diagnose-load-02-prefix", "TT_PHASE1_LOAD_RUN_ID": MODULE.LOAD02_CHECKPOINT_RUN_ID, "TT_PHASE1_LOAD_GITHUB_RUN_ID": "31274000000", "TT_PHASE1_LOAD_MANIFEST_JSON": "{}", "TT_PHASE1_LOAD_ENVIRONMENT": "staging", "TT_PHASE1_LOAD_SOURCE_SCOPE": "default", "TT_PHASE1_LOAD_EXPECTED_SHA": "a" * 40, "TT_PHASE1_LOAD_CONFIRM_MUTATION": "False", "TT_PHASE1_LOAD_CONFIRM_OPERATIONAL": "False", } rows = [mock.Mock()] with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(MODULE, "load_manifest", return_value=value), mock.patch.object( MODULE, "_assert_source_safety", side_effect=(metrics, metrics) ) as source_safety, mock.patch.object( MODULE, "_diagnose_load02_prefix_rows", return_value=diagnosis ) as diagnose, mock.patch.object(MODULE, "_read_load02_prefix_rows", return_value=rows), mock.patch.object(MODULE, "_persist_immutable_exchange_file") as persist, ): result = MODULE.diagnose_load02_prefix_from_environment() self.assertEqual(result["diagnosis"], diagnosis) self.assertTrue(result["two_read_inventory_stable"]) self.assertFalse(result["evidence_written"]) self.assertFalse(result["api_authentication_performed"]) self.assertFalse(result["mutation_executed"]) self.assertFalse(result["load_executed"]) self.assertFalse(result["restart_executed"]) self.assertFalse(result["database_write_executed"]) self.assertFalse(result["phase2_enabled"]) self.assertFalse(result["reverse_delivery_enabled"]) self.assertFalse(result["dr_executed"]) persist.assert_not_called() self.assertEqual( [call.kwargs for call in source_safety.call_args_list], [{"expected_watermark": None}, {"expected_watermark": 4000}], ) self.assertEqual(diagnose.call_count, 2) for call in diagnose.call_args_list: self.assertEqual( call.kwargs, { "rows": rows, "manifest": value, "run_id": MODULE.LOAD02_CHECKPOINT_RUN_ID, "current_watermark": 4000, }, ) def test_failed_load02_engine_audit_correlates_terminal_outcomes(self): value = manifest() value["workloads"]["load-02"]["expected_class_counts"] = { name: 1 for name in MODULE.LEGACY_TRAFFIC_CLASSIFICATIONS } operations = [ { "iteration": index, "request_id": f"request-{index}", "workload_classification": name, "target_id": 100 + index, } for index, name in enumerate(MODULE.LEGACY_TRAFFIC_CLASSIFICATIONS, 1) ] engine_operations = { operation["workload_classification"]: operation for operation in operations if operation["workload_classification"] in {"schedule", "reschedule", "recurrence_schedule"} } deliveries = [] kafka_logs = [] receipts = [] outbox_transactions = [ { "source_sequence": 1610, "change_set_id": "direct-staff", "request_ids": ["request-1"], "complete": True, "published": True, }, { "source_sequence": 1611, "change_set_id": "direct-location", "request_ids": ["request-2"], "complete": True, "published": True, }, ] outcomes = { "schedule": ("success", True, "applied-change"), "reschedule": ("failed", False, "failed-change"), "recurrence_schedule": ("success", True, "noop-change"), } for classification, operation in engine_operations.items(): delivery_id = f"delivery-{classification}" status, successful, change_set = outcomes[classification] deliveries.append( { "delivery_id": delivery_id, "correlation_id": operation["request_id"], "channel": "kafka", "method": "schedule", "status": "published", "payload_activity_ids": [operation["target_id"]], } ) kafka_logs.append( { "request_id": delivery_id, "request_present": True, "request_payload_valid": True, "response_at": "2026-08-08T10:00:00+00:00", "response": { "present": True, "protocol_valid": True, "status": status, "successful_assignment": successful, "canonical_response_hash": f"hash-{classification}", }, } ) receipts.append( { "request_id": delivery_id, "response_hash": f"hash-{classification}", "change_set_id": change_set, } ) outbox_transactions.append( { "source_sequence": 1612, "change_set_id": "applied-change", "request_ids": ["delivery-schedule"], "complete": True, "published": True, } ) snapshot = { "operations": operations, "api_logs": [ { "id": index, "workload_classification": name, "scode": 200, } for index, name in enumerate(MODULE.LEGACY_TRAFFIC_CLASSIFICATIONS, 1) ], "deliveries": deliveries, "kafka_logs": kafka_logs, "receipts": receipts, "quarantines": [], "outbox_transactions": outbox_transactions, "activity_states": {"traffic_activity_ids": {"scheduled_count": 1}}, } result = MODULE._classify_failed_load02_engine_audit( snapshot=snapshot, manifest=value ) self.assertTrue(result["all_engine_work_terminal"]) self.assertFalse(result["late_mutation_possible"]) self.assertTrue(result["containment_passed"]) self.assertFalse(result["load02_acceptance_passed"]) self.assertFalse(result["cleanup_permitted"]) self.assertEqual(result["workload_class_counts"]["schedule"]["applied"], 1) self.assertEqual(result["workload_class_counts"]["reschedule"]["failed"], 1) self.assertEqual( result["workload_class_counts"]["recurrence_schedule"]["no_op"], 1, ) self.assertEqual(result["workload_class_counts"]["terminal"]["no_op"], 1) def test_failed_load02_engine_audit_inventory_is_exact_and_run_owned(self): value = manifest() fixtures = { name: list(range(1000, 1000 + selector["expected_count"])) for name, selector in value["fixture_selectors"].items() } operations = MODULE._failed_load02_expected_operations( manifest=value, run_id=MODULE.LOAD02_CHECKPOINT_RUN_ID, fixtures=fixtures, ) counts = { name: sum( operation["workload_classification"] == name for operation in operations ) for name in MODULE.REQUIRED_TRAFFIC_CLASSIFICATIONS } self.assertEqual(len(operations), 3600) self.assertEqual(counts, value["workloads"]["load-02"]["expected_class_counts"]) self.assertEqual( operations[0]["request_id"], f"{MODULE.LOAD02_CHECKPOINT_RUN_ID}-load-02-000001", ) self.assertEqual( operations[0]["target_id"], fixtures["traffic_activity_ids"][0] ) self.assertEqual( operations[4]["target_id"], fixtures["traffic_activity_ids"][1] ) def test_failed_load02_engine_response_parser_requires_exact_delivery_and_fixture( self, ): response = { "request_id": "engine-delivery-id", "status": "success", "schedule": json.dumps( [ { "activity": "42", "teaching_staff": "[7]", "location": "[8]", "start_slot": "21", } ] ), } parsed = MODULE._parse_engine_response_for_audit( json.dumps(response), delivery_id="engine-delivery-id", expected_activity_id=42, ) self.assertTrue(parsed["protocol_valid"]) self.assertTrue(parsed["successful_assignment"]) self.assertRegex(parsed["canonical_response_hash"], r"^[0-9a-f]{64}$") mismatched = MODULE._parse_engine_response_for_audit( json.dumps(response), delivery_id="different-delivery-id", expected_activity_id=42, ) self.assertFalse(mismatched["protocol_valid"]) def test_failed_load02_engine_audit_reports_unresolved_response_as_late(self): value = manifest() value["workloads"]["load-02"]["expected_class_counts"] = { name: 0 for name in MODULE.LEGACY_TRAFFIC_CLASSIFICATIONS } value["workloads"]["load-02"]["expected_class_counts"]["schedule"] = 1 operation = { "iteration": 3, "request_id": "request-3", "workload_classification": "schedule", "target_id": 103, } snapshot = { "operations": [operation], "api_logs": [ { "id": 9, "workload_classification": "schedule", "scode": 200, } ], "deliveries": [ { "delivery_id": "delivery-schedule", "correlation_id": "request-3", "channel": "kafka", "method": "schedule", "status": "published", "payload_activity_ids": [103], } ], "kafka_logs": [ { "request_id": "delivery-schedule", "request_present": True, "request_payload_valid": True, "response_at": None, "response": {"present": False}, } ], "receipts": [], "quarantines": [], "outbox_transactions": [], "activity_states": {}, } result = MODULE._classify_failed_load02_engine_audit( snapshot=snapshot, manifest=value ) self.assertFalse(result["all_engine_work_terminal"]) self.assertTrue(result["automatic_late_mutation_possible"]) self.assertTrue(result["late_mutation_possible"]) self.assertFalse(result["containment_passed"]) self.assertEqual(result["ambiguous_operation_count"], 1) def test_current_failed_load02_audit_accepts_only_exact_terminal_prefix(self): value = manifest() operations = [ { "iteration": 1, "request_id": "run-load-02-000001", "workload_classification": "staff_lifecycle", "target_id": 101, }, { "iteration": 2, "request_id": "run-load-02-000002", "workload_classification": "location_lifecycle", "target_id": 102, }, { "iteration": 3, "request_id": "run-load-02-000003", "workload_classification": "schedule", "target_id": 103, }, { "iteration": 4, "request_id": "run-load-02-000004", "workload_classification": "reschedule", "target_id": 104, }, ] snapshot = { "operations": operations, "api_logs": [ { "id": index, "workload_classification": operation[ "workload_classification" ], "scode": 200, } for index, operation in enumerate(operations[:3], 1) ], "deliveries": [ { "delivery_id": "delivery-schedule", "correlation_id": "run-load-02-000003", "channel": "kafka", "method": "schedule", "status": "published", "payload_activity_ids": [103], }, { "delivery_id": "delivery-downstream-projection", "correlation_id": "run-load-02-000003", "channel": "kafka", "method": "schedule", "status": "published", "payload_activity_ids": None, }, ], "kafka_logs": [ { "request_id": "delivery-schedule", "request_present": True, "request_payload_valid": True, "response_at": "2026-08-09T11:00:00+00:00", "response": { "present": True, "protocol_valid": True, "status": "success", "successful_assignment": True, "canonical_response_hash": "response-hash", }, } ], "receipts": [ { "request_id": "delivery-schedule", "response_hash": "response-hash", "change_set_id": "schedule-change", } ], "quarantines": [], "outbox_transactions": [ { "source_sequence": 1, "change_set_id": "staff-change", "request_ids": ["run-load-02-000001"], "complete": True, "published": True, }, { "source_sequence": 2, "change_set_id": "location-change", "request_ids": ["run-load-02-000002"], "complete": True, "published": True, }, { "source_sequence": 3, "change_set_id": "schedule-change", "request_ids": ["delivery-schedule"], "complete": True, "published": True, }, ], "activity_states": {}, } result = MODULE._classify_failed_load02_engine_audit( snapshot=snapshot, manifest=value, allow_partial_prefix=True, ) self.assertTrue(result["partial_prefix_attributable"]) self.assertEqual(result["submitted_operation_count"], 3) self.assertTrue(result["all_engine_work_terminal"]) self.assertFalse(result["late_mutation_possible"]) self.assertTrue(result["containment_passed"]) self.assertEqual( result["post_commit_delivery"]["auxiliary_delivery_count"], 1 ) snapshot["api_logs"] = [ snapshot["api_logs"][2], snapshot["api_logs"][0], snapshot["api_logs"][1], ] reordered = MODULE._classify_failed_load02_engine_audit( snapshot=snapshot, manifest=value, allow_partial_prefix=True, ) self.assertTrue(reordered["partial_prefix_attributable"]) self.assertTrue(reordered["containment_passed"]) snapshot["api_logs"][1]["workload_classification"] = "schedule" with self.assertRaisesRegex( MODULE.LoadHarnessError, "exact durably attributed operation set" ) as caught: MODULE._classify_failed_load02_engine_audit( snapshot=snapshot, manifest=value, allow_partial_prefix=True, ) self.assertIn("submitted=3", str(caught.exception)) self.assertIn( 'class_count_deltas={"schedule":1,"staff_lifecycle":-1}', str(caught.exception), ) def test_current_failed_load02_audit_accepts_terminal_out_of_order_abort_set(self): value = manifest() operations = [ { "iteration": iteration, "request_id": f"run-load-02-{iteration:06d}", "workload_classification": classification, "target_id": 100 + iteration, } for iteration, classification in ( (1, "staff_lifecycle"), (2, "location_lifecycle"), (3, "schedule"), (4, "recurrence_terminal"), (5, "schedule"), ) ] snapshot = { "operations": operations, "api_logs": [ {"id": 1, "workload_classification": "staff_lifecycle", "scode": 200}, {"id": 2, "workload_classification": "location_lifecycle", "scode": 200}, {"id": 3, "workload_classification": "schedule", "scode": 200}, {"id": 4, "workload_classification": "schedule", "scode": 200}, ], "deliveries": [ { "delivery_id": "delivery-3", "correlation_id": "run-load-02-000003", "channel": "kafka", "method": "schedule", "status": "published", "payload_activity_ids": [103], }, { "delivery_id": "delivery-5", "correlation_id": "run-load-02-000005", "channel": "kafka", "method": "schedule", "status": "published", "payload_activity_ids": [105], }, ], "kafka_logs": [ { "request_id": "delivery-3", "request_present": True, "request_payload_valid": True, "response_at": "2026-08-12T00:00:00+00:00", "response": { "protocol_valid": True, "status": "success", "successful_assignment": True, "canonical_response_hash": "response-3", }, }, { "request_id": "delivery-5", "request_present": True, "request_payload_valid": True, "response_at": "2026-08-12T00:00:01+00:00", "response": { "protocol_valid": True, "status": "success", "successful_assignment": False, "canonical_response_hash": "response-5", }, }, ], "receipts": [ { "request_id": "delivery-3", "response_hash": "response-3", "change_set_id": "schedule-change", }, { "request_id": "delivery-5", "response_hash": "response-5", "change_set_id": "failed-change", }, ], "quarantines": [], "outbox_transactions": [ { "source_sequence": 1, "change_set_id": "staff-change", "request_ids": ["run-load-02-000001"], "complete": True, "published": True, }, { "source_sequence": 2, "change_set_id": "location-change", "request_ids": ["run-load-02-000002"], "complete": True, "published": True, }, { "source_sequence": 3, "change_set_id": "schedule-change", "request_ids": ["delivery-3"], "complete": True, "published": True, }, ], "activity_states": {}, } result = MODULE._classify_failed_load02_engine_audit( snapshot=snapshot, manifest=value, allow_partial_prefix=True, ) self.assertFalse(result["partial_prefix_attributable"]) self.assertTrue(result["partial_submission_set_attributable"]) self.assertTrue(result["all_engine_work_terminal"]) self.assertFalse(result["late_mutation_possible"]) self.assertTrue(result["containment_passed"]) self.assertEqual( result["engine_delivery"]["terminal_status_counts"], {"applied": 1, "engine_failed_or_no_slot": 1}, ) def test_failed_load02_engine_audit_action_is_two_read_and_read_only(self): value = manifest() value["commits"] = { "timetabler": "a" * 40, "resource_booking": "b" * 40, } snapshot = {"bounded": "inventory"} metrics = { "transport_watermark": 4000, "publisher_last_sequence": 4000, "publisher_live": False, "outbox_depth": 0, "outbox_dead_letter_count": 0, "post_commit_dead_letter_count": 0, "engine_response_quarantine_count": 0, "reverse_delivery_enabled": False, } classified = { "containment_passed": False, "all_engine_work_terminal": False, "late_mutation_possible": True, } environment = { "TT_PHASE1_LOAD_ACTION": MODULE.LOAD02_ENGINE_AUDIT_ACTION, "TT_PHASE1_LOAD_RUN_ID": MODULE.LOAD02_CHECKPOINT_RUN_ID, "TT_PHASE1_LOAD_GITHUB_RUN_ID": "31275000000", "TT_PHASE1_LOAD_MANIFEST_JSON": "{}", "TT_PHASE1_LOAD_ENVIRONMENT": "staging", "TT_PHASE1_LOAD_SOURCE_SCOPE": "default", "TT_PHASE1_LOAD_EXPECTED_SHA": "a" * 40, "TT_PHASE1_LOAD_CONFIRM_MUTATION": "False", "TT_PHASE1_LOAD_CONFIRM_OPERATIONAL": "False", "RB_INTEGRATION_PHASE": "phase1", } with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(MODULE, "load_manifest", return_value=value), mock.patch.object( MODULE, "_source_metrics", return_value=metrics ) as source, mock.patch.object( MODULE, "_read_failed_load02_engine_audit_snapshot", return_value=snapshot, ) as read_snapshot, mock.patch.object( MODULE, "_classify_failed_load02_engine_audit", return_value=classified, ), mock.patch.object(MODULE, "_persist_immutable_exchange_file") as persist, ): result = MODULE.audit_failed_load02_engine_work_from_environment() self.assertEqual(read_snapshot.call_count, 2) self.assertEqual(source.call_count, 2) persist.assert_not_called() self.assertFalse(result["evidence_written"]) self.assertFalse(result["api_authentication_performed"]) self.assertFalse(result["database_write_executed"]) self.assertFalse(result["kafka_publish_executed"]) self.assertFalse(result["cleanup_executed"]) self.assertFalse(result["load_executed"]) self.assertFalse(result["restart_executed"]) self.assertFalse(result["phase2_enabled"]) self.assertFalse(result["reverse_delivery_enabled"]) self.assertFalse(result["dr_executed"]) def test_current_load02_engine_audit_accepts_only_manifest_bound_run(self): value = with_current_recovery_control(manifest()) metrics = { "transport_watermark": 4000, "publisher_last_sequence": 4000, "publisher_live": True, "outbox_depth": 0, "outbox_dead_letter_count": 0, "post_commit_dead_letter_count": 0, "engine_response_quarantine_count": 0, "reverse_delivery_enabled": False, } environment = { "TT_PHASE1_LOAD_ACTION": MODULE.CURRENT_LOAD02_ENGINE_AUDIT_ACTION, "TT_PHASE1_LOAD_RUN_ID": RUN_ID, "TT_PHASE1_LOAD_GITHUB_RUN_ID": "31275000001", "TT_PHASE1_LOAD_MANIFEST_JSON": "{}", "TT_PHASE1_LOAD_ENVIRONMENT": "staging", "TT_PHASE1_LOAD_SOURCE_SCOPE": "default", "TT_PHASE1_LOAD_EXPECTED_SHA": "c" * 40, "TT_PHASE1_LOAD_CONFIRM_MUTATION": "False", "TT_PHASE1_LOAD_CONFIRM_OPERATIONAL": "False", "RB_INTEGRATION_PHASE": "phase1", } with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(MODULE, "load_manifest", return_value=value) as load, mock.patch.object(MODULE, "_source_metrics", return_value=metrics), mock.patch.object( MODULE, "_read_failed_load02_engine_audit_snapshot", return_value={"bounded": "inventory"}, ), mock.patch.object( MODULE, "_classify_failed_load02_engine_audit", return_value={ "containment_passed": True, "all_engine_work_terminal": True, "late_mutation_possible": False, }, ), ): result = MODULE.audit_failed_load02_engine_work_from_environment() self.assertEqual(result["run_id"], RUN_ID) self.assertTrue(result["audit"]["containment_passed"]) wrong_deployment = copy.deepcopy(value) wrong_deployment["recovery_control"]["recovery_timetabler_sha"] = "e" * 40 with self.assertRaisesRegex( MODULE.LoadHarnessError, "differs from its sealed control" ): MODULE._assert_current_load02_recovery_control( manifest=wrong_deployment, run_id=RUN_ID, action=MODULE.CURRENT_LOAD02_ENGINE_AUDIT_ACTION, exact_sha="c" * 40, ) load.assert_called_once_with( "{}", run_id=RUN_ID, environment="staging", source_scope="default", require_approval=True, ) def test_size_boundary_requires_under_accept_and_over_reject(self): candidate = manifest() candidate["workloads"]["load-06"]["steps"][1][ "expected_source_transactions" ] = 1 with self.assertRaisesRegex(MODULE.LoadHarnessError, "boundary contract"): MODULE.validate_manifest(candidate, run_id=RUN_ID, require_approval=False) def test_evidence_hash_covers_all_fields_except_itself(self): payload = {"profile": "LOAD-02", "run_id": RUN_ID, "source_transactions": []} sealed = MODULE.seal_evidence(payload) self.assertEqual(sealed["evidence_sha256"], MODULE._sha256(payload)) self.assertNotEqual( sealed["evidence_sha256"], MODULE.seal_evidence({**payload, "profile": "LOAD-03"})["evidence_sha256"], ) def test_evidence_persistence_is_owner_only_and_immutable(self): with TemporaryDirectory() as directory: root = Path(directory) / "exchange" run_dir = root / RUN_ID def exchange_path(_run_id, profile): return root, run_dir, run_dir / f"{profile.lower()}.json" owner = os.getuid() patches = ( mock.patch.object(MODULE, "_evidence_path", side_effect=exchange_path), mock.patch.object(MODULE, "TIMETABLER_EVIDENCE_OWNER_UID", owner), ) for patcher in patches: patcher.start() self.addCleanup(patcher.stop) payload = MODULE.seal_evidence( { "schema_version": 1, "profile": "LOAD-02", "run_id": RUN_ID, "evidence_exchange": MODULE.evidence_exchange_attestation(), } ) first = MODULE.persist_evidence(payload) path = Path(first["evidence_path"]) self.assertEqual(stat.S_IMODE(root.stat().st_mode), 0o755) self.assertEqual(stat.S_IMODE(run_dir.stat().st_mode), 0o755) self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o444) self.assertEqual(path.stat().st_uid, os.getuid()) self.assertEqual( MODULE.persist_evidence(payload)["evidence_sha256"], first["evidence_sha256"], ) changed = MODULE.seal_evidence( { **{ key: value for key, value in payload.items() if key != "evidence_sha256" }, "changed": True, } ) with self.assertRaisesRegex(MODULE.LoadHarnessError, "different or unsafe"): MODULE.persist_evidence(changed) def test_evidence_exchange_rejects_credentials_before_persistence(self): payload = { "schema_version": 1, "profile": "LOAD-02", "run_id": RUN_ID, "evidence_exchange": MODULE.evidence_exchange_attestation(), "api_password": "must-not-cross-users", } with self.assertRaisesRegex(MODULE.LoadHarnessError, "credential-like"): MODULE._assert_exchange_payload_has_no_credentials(payload) def test_evidence_exchange_rejects_symlinked_root_before_write(self): with TemporaryDirectory() as directory: base = Path(directory) actual = base / "actual" actual.mkdir(mode=0o700) root = base / "exchange" root.symlink_to(actual, target_is_directory=True) run_dir = root / RUN_ID def exchange_path(_run_id, profile): return root, run_dir, run_dir / f"{profile.lower()}.json" with ( mock.patch.object(MODULE, "_evidence_path", side_effect=exchange_path), mock.patch.object(MODULE, "TIMETABLER_EVIDENCE_OWNER_UID", os.getuid()), ): with self.assertRaisesRegex(MODULE.LoadHarnessError, "unsafe"): MODULE._ensure_evidence_exchange_directories(RUN_ID) self.assertEqual(stat.S_IMODE(actual.stat().st_mode), 0o700) def test_proc_host_sample_math_is_bounded(self): meminfo = "MemTotal: 1000 kB\nMemAvailable: 250 kB\n" statline = "cpu 10 20 30 40 5 6 7 8\n" with mock.patch( "builtins.open", side_effect=[ mock.mock_open(read_data=meminfo).return_value, mock.mock_open(read_data=statline).return_value, ], ): sample = MODULE._read_proc_host_sample() self.assertEqual(sample["memory_percent"], 75.0) self.assertGreater(sample["cpu_total_ticks"], sample["cpu_idle_ticks"]) def test_contract_schema_and_example_are_valid_json(self): schema = json.loads( ( ROOT / "tests/contracts/resource_booking_phase1_load/timetabler-load-evidence.schema.json" ).read_text() ) self.assertEqual( schema["properties"]["reverse_delivery_enabled"]["const"], False ) self.assertIn("LOAD-08", schema["properties"]["profile"]["enum"]) self.assertIn("PROVISION", schema["properties"]["profile"]["enum"]) self.assertEqual( schema["properties"]["evidence_exchange"]["properties"]["file_mode"][ "const" ], "0444", ) self.assertEqual( manifest()["evidence_exchange"], MODULE.evidence_exchange_attestation() ) self.assertEqual( MODULE.evidence_exchange_attestation()["path_sha256"], hashlib.sha256(b"/var/tmp/mayvins-timetabler-phase1-load").hexdigest(), ) correction_schema = schema["properties"]["evidence_correction"] self.assertFalse(correction_schema["additionalProperties"]) self.assertEqual( set(correction_schema["properties"]["artifact_filename"]["enum"]), MODULE.PROVISION_CORRECTION_ARTIFACT_FILENAMES, ) for timestamp in ("committed_at", "published_at"): self.assertIn( "[+-]", schema["properties"]["source_transactions"]["items"]["properties"][ timestamp ]["pattern"], ) fence_schema = json.loads( ( ROOT / "tests/contracts/resource_booking_phase1_load/load-07-fence.schema.json" ).read_text() ) self.assertEqual(fence_schema["additionalProperties"], False) self.assertEqual( set(fence_schema["required"]), { "schema_version", "run_id", "timetabler_sha", "resource_booking_sha", "execute_at_epoch", "nonce", "approval_ref", "created_at", "expires_at", "coordination_sha256", }, ) def test_representative_common_evidence_for_all_12_profiles_matches_schema(self): schema = json.loads( ( ROOT / "tests/contracts/resource_booking_phase1_load/timetabler-load-evidence.schema.json" ).read_text() ) profiles = [stage.upper() for stage in MODULE.STAGES] self.assertEqual(set(profiles), set(schema["properties"]["profile"]["enum"])) for profile in profiles: representative = { "schema_version": 1, "profile": profile, "service": "timetabler", "repository": "Mayvins/timetabler-be", "run_id": RUN_ID, "environment": "staging", "production": False, "git_sha": "1" * 40, "resource_booking_git_sha_attestation": "2" * 40, "github_run_id": "123", "source_scope": "default", "entry_watermark": 117, "final_watermark": 117, "prior_evidence_sha256": "3" * 64, "predecessor_evidence": { "profile": "ENTRY", "evidence_sha256": "3" * 64, "external_accepted_root": False, }, "profile_timing": MODULE.DEFAULT_PROFILE_TIMING[profile], "approved_window": manifest()["approved_window"], "configuration_snapshot_sha256": "4" * 64, "evidence_exchange": MODULE.evidence_exchange_attestation(), "workload_parameters": manifest()["load"], "thresholds": MODULE.DEFAULT_THRESHOLDS, "stage_assertions": {}, "source_transactions": [], "source_metric_samples": [], "host_metric_samples": [], "final_source_metrics": {}, "load_executed": profile in {stage.upper() for stage in MODULE.LOAD_EXECUTION_STAGES}, "mutation_executed": profile in {stage.upper() for stage in MODULE.MUTATING_STAGES}, "restart_executed": profile == "LOAD-07", "reverse_delivery_enabled": False, "phase2_enabled": False, "dr_executed": False, "resource_booking_database_touched": False, "resource_booking_process_touched": False, "direct_database_mutation": False, "manual_outbox_edit": False, "evidence_sha256": "5" * 64, } self.assertFalse(set(schema["required"]) - set(representative)) for key, definition in schema["properties"].items(): if "const" in definition and key in representative: self.assertEqual(representative[key], definition["const"]) if "pattern" in definition and key in representative: self.assertRegex( str(representative[key]), re.compile(definition["pattern"]) ) def test_harness_has_no_direct_domain_write_or_sql(self): for source in ( SCRIPT.read_text(), (ROOT / "deploy/phase1_load_provision.py").read_text(), ): for forbidden in ( ".objects.create(", ".objects.update(", ".objects.delete(", ".save(", ".raw(", "cursor.execute(", "TRUNCATE ", "DELETE FROM ", ): self.assertNotIn(forbidden, source) def test_workflow_is_default_disabled_and_does_not_deploy_or_control_rb(self): workflow = (ROOT / ".github/workflows/deploy.yml").read_text() job = workflow.split(" phase1-load:\n", 1)[1] self.assertIn("inputs.phase1_load_action != 'none'", job) self.assertIn("TT_PHASE1_LOAD_MANIFEST_JSON", job) self.assertIn("TT_PHASE1_LOAD_PRIOR_EVIDENCE_SHA256", job) self.assertIn("command_timeout: 120m", job) self.assertIn("phase1_load_provision.py", job) self.assertIn('TT_PHASE1_LOAD_ACTION" = "attest"', job) self.assertIn("phase1-publisher-deployment-gate.sh", job) self.assertIn("/var/tmp/mayvins-timetabler-phase1-load", job) self.assertIn("TT_PHASE1_LOAD_EVIDENCE_OWNER_UID: '1073'", job) self.assertIn("TT_PHASE1_LOAD_EVIDENCE_READER_UID: '1069'", job) self.assertIn("TT_PHASE1_LOAD_EVIDENCE_FILE_MODE: '0444'", job) self.assertIn("- arm-fence", workflow) self.assertIn("- correct-provision-evidence", workflow) self.assertIn("- diagnose-load-02-prefix", workflow) self.assertIn("- audit-failed-load-02-engine-work", workflow) self.assertIn("- audit-current-load-02-engine-work", workflow) self.assertIn("- cleanup-current-abandoned-load-02", workflow) self.assertIn("- checkpoint-load-02", workflow) self.assertIn("deploy/phase1_load_harness.py correct-provision-evidence", job) self.assertIn("deploy/phase1_load_harness.py diagnose-load-02-prefix", job) self.assertIn( "deploy/phase1_load_harness.py audit-failed-load-02-engine-work", job ) self.assertIn( "deploy/phase1_load_harness.py audit-current-load-02-engine-work", job ) self.assertIn( "deploy/phase1_load_harness.py cleanup-current-abandoned-load-02", job ) self.assertIn("deploy/phase1_load_harness.py checkpoint-load-02", job) correction_invocation = job.split( 'TT_PHASE1_LOAD_ACTION" = "correct-provision-evidence"', 1 )[1].split('TT_PHASE1_LOAD_ACTION" = "diagnose-load-02-prefix"', 1)[0] self.assertIn("TT_PHASE1_LOAD_EMAIL=''", correction_invocation) self.assertIn("TT_PHASE1_LOAD_PASSWORD=''", correction_invocation) diagnostic_invocation = job.split( 'TT_PHASE1_LOAD_ACTION" = "diagnose-load-02-prefix"', 1 )[1].split('TT_PHASE1_LOAD_ACTION" = "audit-failed-load-02-engine-work"', 1)[0] self.assertIn("TT_PHASE1_LOAD_EMAIL=''", diagnostic_invocation) self.assertIn("TT_PHASE1_LOAD_PASSWORD=''", diagnostic_invocation) audit_invocation = job.split( 'TT_PHASE1_LOAD_ACTION" = "audit-failed-load-02-engine-work"', 1 )[1].split('TT_PHASE1_LOAD_ACTION" = "checkpoint-load-02"', 1)[0] self.assertIn("TT_PHASE1_LOAD_EMAIL=''", audit_invocation) self.assertIn("TT_PHASE1_LOAD_PASSWORD=''", audit_invocation) checkpoint_invocation = job.split( 'TT_PHASE1_LOAD_ACTION" = "checkpoint-load-02"', 1 )[1].split('TT_PHASE1_LOAD_ACTION" = "arm-fence"', 1)[0] self.assertIn("TT_PHASE1_LOAD_EMAIL=''", checkpoint_invocation) self.assertIn("TT_PHASE1_LOAD_PASSWORD=''", checkpoint_invocation) self.assertIn( "attest|diagnose-engine-consumer|entry|load-01|correct-provision-evidence|diagnose-load-02-prefix|audit-failed-load-02-engine-work|audit-current-load-02-engine-work|checkpoint-load-02|arm-fence|final)", job, ) self.assertIn('TT_PHASE1_LOAD_ACTION" = "arm-fence"', job) self.assertIn("deploy/phase1_load_harness.py arm-fence", job) self.assertNotIn("deploy-staging.sh", job) self.assertNotIn("RB_PHASE1_E2E_EMAIL", job) self.assertNotIn("RB_PHASE1_E2E_PASSWORD", job) self.assertNotIn("RB_PHASE1_E2E_ADMIN_TOKEN", job) self.assertNotIn("resource-booking-be", job) def test_workflow_uses_masked_e2e_credential_fallback_without_logging(self): workflow = (ROOT / ".github/workflows/deploy.yml").read_text() job = workflow.split(" phase1-load:\n", 1)[1] self.assertIn( "TT_PHASE1_LOAD_EMAIL: ${{ secrets.TT_PHASE1_LOAD_EMAIL || secrets.TT_PHASE1_E2E_EMAIL }}", job, ) self.assertIn( "TT_PHASE1_LOAD_PASSWORD: ${{ secrets.TT_PHASE1_LOAD_PASSWORD || secrets.TT_PHASE1_E2E_PASSWORD }}", job, ) self.assertNotRegex( job, r"(?:echo|printf|printenv).*TT_PHASE1_LOAD_(?:EMAIL|PASSWORD)", ) self.assertNotIn("set -x", job) def test_load07_driver_controls_only_exact_tt_publisher(self): source = (ROOT / "deploy/phase1_load_operations.sh").read_text() self.assertIn('"$PM2_BIN" restart "$PUBLISHER_NAME"', source) self.assertIn("resource_booking_process_touched", source) self.assertNotIn("RB_CONSUMER", source) self.assertNotIn("resource-booking-be", source) self.assertNotIn('source "$APP_DIR/.env"', source) self.assertNotIn("TRUNCATE", source) self.assertNotIn("DELETE FROM", source) self.assertIn("0 <= skew <= 5", source) self.assertIn('"publisher_restart_count":1', source) def test_complete_transaction_evidence_fields_are_present(self): source = SCRIPT.read_text() for field in ( '"source_transaction_id"', '"source_sequence"', '"committed_at"', '"published_at"', '"canonical_bytes"', '"member_count"', '"workload_classification"', ): self.assertIn(field, source) def test_evidence_file_hash_uses_canonical_payload(self): value = {"a": 1, "b": [2, 3]} self.assertEqual( MODULE._sha256(value), hashlib.sha256(b'{"a":1,"b":[2,3]}').hexdigest(), ) def test_predecessor_file_is_required_owner_only_canonical_and_exact(self): value = manifest() with TemporaryDirectory() as directory: root = Path(directory) / "exchange" run_dir = root / RUN_ID def exchange_path(_run_id, profile): return root, run_dir, run_dir / f"{profile.lower()}.json" owner = os.getuid() patches = ( mock.patch.object(MODULE, "_evidence_path", side_effect=exchange_path), mock.patch.object(MODULE, "TIMETABLER_EVIDENCE_OWNER_UID", owner), ) for patcher in patches: patcher.start() self.addCleanup(patcher.stop) payload = MODULE.seal_evidence( { "schema_version": 1, "profile": "ENTRY", "service": "timetabler", "run_id": RUN_ID, "reverse_delivery_enabled": False, "phase2_enabled": False, "dr_executed": False, "evidence_exchange": MODULE.evidence_exchange_attestation(), } ) with self.assertRaisesRegex(MODULE.LoadHarnessError, "missing"): MODULE.verify_predecessor_evidence( run_id=RUN_ID, stage="load-01", manifest=value, supplied_sha256=payload["evidence_sha256"], ) persisted = MODULE.persist_evidence(payload) verified = MODULE.verify_predecessor_evidence( run_id=RUN_ID, stage="load-01", manifest=value, supplied_sha256=payload["evidence_sha256"], ) self.assertEqual(verified["profile"], "ENTRY") with self.assertRaisesRegex(MODULE.LoadHarnessError, "contradictory"): MODULE.verify_predecessor_evidence( run_id=RUN_ID, stage="load-01", manifest=value, supplied_sha256="2" * 64, ) Path(persisted["evidence_path"]).chmod(0o644) with self.assertRaisesRegex(MODULE.LoadHarnessError, "unsafe"): MODULE.verify_predecessor_evidence( run_id=RUN_ID, stage="load-01", manifest=value, supplied_sha256=payload["evidence_sha256"], ) def test_load02_uses_declared_corrected_provision_predecessor(self): value = manifest() with TemporaryDirectory() as directory: root = Path(directory) / "exchange" run_dir = root / RUN_ID def exchange_path(_run_id, profile): return root, run_dir, run_dir / f"{profile.lower()}.json" owner = os.getuid() with ( mock.patch.object(MODULE, "_evidence_path", side_effect=exchange_path), mock.patch.object(MODULE, "TIMETABLER_EVIDENCE_OWNER_UID", owner), ): MODULE._ensure_evidence_exchange_directories(RUN_ID) fixture_manifest = { "academic_term_id": 26, "academic_term_start_date": "2026-01-05", "academic_term_end_date": "2026-06-26", "authoritative_term_scope_enforced": True, "term_template_donor_only": True, "template_candidate_defaults_authoritative": False, "candidate_defaults_replaced_before_scheduling": True, "run_owned_activity_final_allocations_empty_at_provision": True, "advisory_requirement_source_watermark_unchanged": True, "activity_template_id": 7, "donor_activity_template_id": 7, } payload = MODULE.seal_evidence( { "schema_version": 1, "profile": "PROVISION", "service": "timetabler", "run_id": RUN_ID, "git_sha": value["commits"]["timetabler"], "resource_booking_git_sha_attestation": value["commits"][ "resource_booking" ], "evidence_correction": value["evidence_corrections"][ "provision" ], "reverse_delivery_enabled": False, "phase2_enabled": False, "dr_executed": False, "evidence_exchange": MODULE.evidence_exchange_attestation(), "fixture_manifest": fixture_manifest, "stage_assertions": { "fixture_manifest": fixture_manifest }, } ) target = ( run_dir / value["evidence_corrections"]["provision"]["artifact_filename"] ) MODULE._persist_immutable_exchange_file( target=target, encoded=MODULE._canonical_bytes(payload) + b"\n", label="test corrected evidence", ) verified = MODULE.verify_predecessor_evidence( run_id=RUN_ID, stage="load-02", manifest=value, supplied_sha256=payload["evidence_sha256"], ) self.assertEqual(verified["profile"], "PROVISION") self.assertTrue( verified["evidence_path"].endswith("provision-corrected-v7.json") ) def test_entry_uses_external_e2e_root_without_future_manifest_hashes(self): value = manifest() root = value["e2e_final_evidence_sha256"] verified = MODULE.verify_predecessor_evidence( run_id=RUN_ID, stage="entry", manifest=value, supplied_sha256=root, ) self.assertTrue(verified["external_accepted_root"]) self.assertNotIn("prior_evidence_sha256", value) def test_sustained_memory_abort_requires_full_300_seconds_and_resets(self): sampler = MODULE.HostSampler(interval_seconds=5, hard_memory_percent=90) self.assertFalse(sampler._memory_breach_is_sustained(91, now_monotonic=10)) self.assertFalse(sampler._memory_breach_is_sustained(99, now_monotonic=309.9)) self.assertTrue(sampler._memory_breach_is_sustained(91, now_monotonic=310)) self.assertFalse(sampler._memory_breach_is_sustained(90, now_monotonic=311)) self.assertFalse(sampler._memory_breach_is_sustained(99, now_monotonic=700)) def test_abort_shutdown_cancels_queued_submissions(self): gate = threading.Event() ran_queued = threading.Event() executor = ThreadPoolExecutor(max_workers=1) running = executor.submit(gate.wait) queued = executor.submit(ran_queued.set) shutdown = threading.Thread( target=MODULE._shutdown_executors, args=([executor],), kwargs={"abort_pending": True}, ) shutdown.start() time.sleep(0.05) gate.set() shutdown.join(timeout=2) self.assertTrue(running.done()) self.assertTrue(queued.cancelled()) self.assertFalse(ran_queued.is_set()) def test_rate_fidelity_fails_closed_on_queue_delay_or_excess_skew(self): policy = manifest()["rate_fidelity"] healthy = [ {"submitted_at_epoch": index / 10, "submission_scheduling_skew_ms": 20} for index in range(100) ] result = MODULE._rate_fidelity_metrics(healthy, rate=10, policy=policy) self.assertGreaterEqual(result["achieved_request_start_tps"], 9.5) delayed = [ {"submitted_at_epoch": index / 5, "submission_scheduling_skew_ms": 1100} for index in range(100) ] with self.assertRaisesRegex(MODULE.LoadHarnessError, "approved rate"): MODULE._rate_fidelity_metrics(delayed, rate=10, policy=policy) def test_cleanup_batches_are_bounded_and_precalibrated(self): value = manifest() cleanup = value["workloads"]["cleanup"] self.assertEqual(cleanup["expected_activity_batch_count"], 16) self.assertEqual(cleanup["expected_source_transactions"], 18) self.assertLessEqual(cleanup["activity_batch_max_members"], 128) self.assertGreaterEqual(cleanup["transport_margin_bytes"], 65536) source = SCRIPT.read_text() self.assertIn("_calibrate_activity_deletion", source) self.assertIn("retry_safe_stable_request_ids", source) def test_cleanup_resource_step_submits_the_full_bounded_selector(self): value = manifest() fixtures = { name: list(range(1000, 1000 + selector["expected_count"])) for name, selector in value["fixture_selectors"].items() } session = mock.Mock() config = SimpleNamespace(run_id=RUN_ID, action="cleanup") with mock.patch.object( MODULE, "_resolve_cleanup_selectors", return_value=fixtures ): runner = MODULE.LoadRunner( config=config, manifest=value, session=session, ) step = value["workloads"]["cleanup"]["steps"][1] measurement = runner._submit_step("cleanup", 12, step) selector_name = step["fixture_selector"] session.post.assert_called_once_with( step["path"], {"id": fixtures[selector_name]}, request_id=f"{RUN_ID}-cleanup-000012", ) self.assertEqual(measurement["fixture_shard"], None) def test_schedule_step_waits_for_atomic_engine_transaction_attribution(self): value = manifest() fixtures = { name: list(range(1000, 1000 + selector["expected_count"])) for name, selector in value["fixture_selectors"].items() } session = mock.Mock() config = SimpleNamespace(run_id=RUN_ID, action="load-02") with mock.patch.object(MODULE, "_resolve_selectors", return_value=fixtures): runner = MODULE.LoadRunner( config=config, manifest=value, session=session, ) step = value["traffic_mix_steps"][0] engine = { "engine_delivery_id": "12345678-1234-4234-8234-123456789abc", "engine_response_sha256": "a" * 64, "engine_change_set_id": "12345678-1234-4234-8234-123456789abd", "engine_source_sequence": 1610, "source_request_id": "12345678-1234-4234-8234-123456789abc", "source_transaction": {"source_sequence": 1610}, } assignment = { **value["load_acceptance_scope"], "recurrence_occurrence_count": 2, "staff_assignment_count": 1, "location_assignment_count": 1, "staff_preset_count": 1, "location_preset_count": 1, "assignment_transition": "assigned", } with ( mock.patch.object( MODULE, "_wait_for_engine_applied_transaction", return_value=engine ) as wait, mock.patch.object( MODULE, "_recurring_assignment_state", return_value=assignment ), ): measurement = runner._submit_step( "load-02", 3, step, shard_index=0, cycle=0 ) request_id = MODULE._rate_request_id(RUN_ID, "load-02", 3) session.post.assert_called_once_with( step["path"], { "activity_ids": [fixtures["traffic_activity_ids"][0]], "slot": value["variables"]["slot_a"], }, request_id=request_id, ) wait.assert_called_once_with( correlation_id=request_id, expected_activity_id=fixtures["traffic_activity_ids"][0], timeout_seconds=120, ) self.assertTrue(measurement["engine_aware"]) self.assertEqual( measurement["source_request_id"], engine["source_request_id"] ) self.assertEqual( measurement["engine_source_sequence"], engine["engine_source_sequence"] ) def test_http_failure_retains_safe_operation_attribution(self): value = manifest() fixtures = { name: list(range(1000, 1000 + selector["expected_count"])) for name, selector in value["fixture_selectors"].items() } request_error = MODULE.identity_gate.EntryGateError( "endpoint returned HTTP 500", safe_details={ "method": "POST", "path": "/api/admin/schedule-request", "http_status": 500, "response_body_sha256": "a" * 64, }, ) session = mock.Mock() session.post.side_effect = request_error config = SimpleNamespace(run_id=RUN_ID, action="load-02") with mock.patch.object(MODULE, "_resolve_selectors", return_value=fixtures): runner = MODULE.LoadRunner( config=config, manifest=value, session=session, ) step = value["traffic_mix_steps"][0] with self.assertRaises(MODULE.identity_gate.EntryGateError) as raised: runner._submit_step( "load-02", 590, step, shard_index=0, cycle=147 ) request_id = MODULE._rate_request_id(RUN_ID, "load-02", 590) expected = { "method": "POST", "path": "/api/admin/schedule-request", "http_status": 500, "response_body_sha256": "a" * 64, "stage": "load-02", "iteration": 590, "request_id": request_id, "workload_classification": "schedule", "fixture_selector": "traffic_activity_ids", } self.assertEqual(raised.exception.safe_details, expected) self.assertEqual(runner.api_failures, [expected]) def test_abort_evidence_includes_safely_attributed_api_failures(self): source = SCRIPT.read_text() self.assertIn('"api_failures_observed": sorted(', source) def test_engine_request_delivery_ignores_same_correlation_downstream_projection(self): engine_request = { "channel": "kafka", "method": "schedule", "payload_activity_ids": [103], } downstream_projection = { "channel": "kafka", "method": "schedule", "payload_activity_ids": None, } self.assertTrue( MODULE._is_engine_request_delivery( engine_request, expected_activity_id=103 ) ) self.assertFalse( MODULE._is_engine_request_delivery( downstream_projection, expected_activity_id=103 ) ) def test_engine_transaction_wait_retries_while_publication_is_in_flight(self): source = SCRIPT.read_text() self.assertIn( 'if str(error) != "source transaction is not completely published"', source, ) self.assertIn("time.sleep(0.2)\n continue", source) def test_provision_is_separate_reachable_and_exactly_3316_api_transactions(self): value = manifest() self.assertEqual(sum(value["provision_plan"]["counts"].values()), 3316) self.assertEqual(value["provision_plan"]["expected_source_transactions"], 3316) config = mock.Mock(action="provision") expected = {"profile": "PROVISION", "source_transactions": [object()] * 3316} with mock.patch.object(MODULE, "run_provision", return_value=expected) as run: self.assertIs(MODULE.execute(config), expected) run.assert_called_once_with(config) workflow = (ROOT / ".github/workflows/deploy.yml").read_text() self.assertIn("- provision", workflow) self.assertIn("deploy/phase1_load_provision.py", workflow) def test_provision_separates_ordinary_and_rate_recurrence_resource_pools(self): value = manifest() provisioner = object.__new__(PROVISION_MODULE.Provisioner) provisioner.manifest = value provisioner.facts = { "short_week_ids": [1, 2], "all_week_ids": list(range(1, 13)), "slot_a": 1, "slot_b": 2, } provisioner.engine_feasibility_preflight = {} provisioner.ensure_staff = mock.Mock( side_effect=lambda _category, index, _key: SimpleNamespace(id=index) ) provisioner.ensure_location = mock.Mock( side_effect=lambda _category, index, _key: SimpleNamespace(id=index) ) activity_calls = [] def ensure_activity(**kwargs): activity_calls.append(kwargs) return SimpleNamespace(id=len(activity_calls)) provisioner.ensure_activity = mock.Mock(side_effect=ensure_activity) provisioner.assert_engine_feasibility = mock.Mock( return_value={"verified": True} ) provisioner.run() ordinary = [item for item in activity_calls if item["category"] == "traffic"] rate_recurrence = [ item for item in activity_calls if item["category"] == "rate-recurrence" ] maximum_recurrence = [ item for item in activity_calls if item["category"] == "recurrence" ] self.assertEqual({item["staff_id"] for item in ordinary}, set(range(1, 321))) self.assertEqual( {item["staff_id"] for item in rate_recurrence}, set(range(321, 641)) ) self.assertEqual( {item["location_id"] for item in ordinary}, set(range(1, 321)) ) self.assertEqual( {item["location_id"] for item in rate_recurrence}, set(range(321, 641)) ) self.assertEqual({tuple(item["week_ids"]) for item in ordinary}, {(1, 2)}) self.assertEqual( {tuple(item["week_ids"]) for item in rate_recurrence}, {(1, 2)} ) self.assertEqual( {tuple(item["week_ids"]) for item in maximum_recurrence}, {tuple(range(1, 13))}, ) self.assertEqual( {item["count_key"] for item in rate_recurrence}, {"rate_recurrence_activity"}, ) def test_restart_fence_has_one_non_circular_seal_field(self): source = SCRIPT.read_text() self.assertIn('if key != "coordination_sha256"', source) self.assertNotIn("evidence_seal", source) def test_load04_and_load06_require_real_maximum_and_near_boundary_approval(self): value = manifest() self.assertEqual( value["workloads"]["load-04"]["approved_real_maximum_activity_count"], value["workloads"]["load-04"]["approved_maximum_member_count"], ) self.assertLessEqual( value["workloads"]["load-06"]["below_boundary_tolerance_bytes"], 4096 ) value["workloads"]["load-04"]["approved_real_maximum_activity_count"] = 99 with self.assertRaisesRegex(MODULE.LoadHarnessError, "maximum"): MODULE.validate_manifest(value, run_id=RUN_ID, require_approval=False) def test_load05_canonical_occurrence_count_is_32_times_12(self): value = manifest() self.assertEqual(value["provision_plan"]["counts"]["recurrence_activity"], 32) self.assertEqual(value["provision_plan"]["maximum_recurrence_week_count"], 12) self.assertEqual( value["workloads"]["load-05"]["expected_occurrence_count"], 384 ) value["workloads"]["load-05"]["expected_occurrence_count"] = 383 with self.assertRaisesRegex(MODULE.LoadHarnessError, "recurrence count"): MODULE.validate_manifest(value, run_id=RUN_ID, require_approval=False) def test_reference_template_with_longer_term_binds_exact_first_12_weeks(self): template, week_relation = _template_with_weeks(15) selected, weeks = PROVISION_MODULE._select_reference_template( [template], 12, approved_academic_term_id=26 ) self.assertIs(selected, template) self.assertEqual(week_relation.ordering, ("week", "id")) self.assertEqual([week.week for week in weeks], list(range(1, 13))) self.assertEqual(len(weeks), 12) def test_reference_template_rejects_short_or_resource_bearing_term(self): short_template, _ = _template_with_weeks(11) resource_template, _ = _template_with_weeks(15, has_staff=True) for candidate in (short_template, resource_template): with self.assertRaisesRegex( PROVISION_MODULE.ProvisionError, "no engine-feasible maximum-recurrence template", ): PROVISION_MODULE._select_reference_template( [candidate], 12, approved_academic_term_id=26 ) selected, _ = PROVISION_MODULE._select_reference_template( [resource_template], 12, approved_academic_term_id=26, require_resource_free=False, ) self.assertIs(selected, resource_template) def test_reference_template_rejects_module_without_all_recurrence_weeks(self): candidate, _ = _template_with_weeks(15, module_week_count=11) with self.assertRaisesRegex( PROVISION_MODULE.ProvisionError, "no engine-feasible maximum-recurrence template", ): PROVISION_MODULE._select_reference_template( [candidate], 12, approved_academic_term_id=26 ) def test_reference_template_rejects_module_without_matching_term(self): candidate, _ = _template_with_weeks(15, module_term_id=None) with self.assertRaisesRegex( PROVISION_MODULE.ProvisionError, "no engine-feasible maximum-recurrence template", ): PROVISION_MODULE._select_reference_template( [candidate], 12, approved_academic_term_id=26 ) def test_reference_template_rejects_module_without_two_common_slots(self): candidate, _ = _template_with_weeks( 15, module_availability_pattern="0" + ("1" * 279), ) with self.assertRaisesRegex( PROVISION_MODULE.ProvisionError, "no engine-feasible maximum-recurrence template", ): PROVISION_MODULE._select_reference_template( [candidate], 12, approved_academic_term_id=26, slot_per_day=40, ) def test_reference_template_rejects_a_different_active_term(self): foreign, _ = _template_with_weeks( 15, academic_term_id=27, module_term_id=27 ) with self.assertRaisesRegex( PROVISION_MODULE.ProvisionError, "approved academic term", ): PROVISION_MODULE._select_reference_template( [foreign], 12, approved_academic_term_id=26 ) def test_provision_requires_authoritative_term_scope(self): value = manifest() with self.assertRaisesRegex( PROVISION_MODULE.ProvisionError, "lacks the authoritative academic-term scope", ): PROVISION_MODULE._approved_term_scope(value) value["accepted_baseline"]["authoritative_term"] = { "academic_term_id": 26, "start_date": "2026-01-05", "end_date": "2026-06-26", } self.assertEqual( PROVISION_MODULE._approved_term_scope(value), { "academic_term_id": 26, "start_date": "2026-01-05", "end_date": "2026-06-26", }, ) def test_load02_rejects_unscoped_provision_evidence(self): value = manifest() value["accepted_baseline"]["authoritative_term"] = { "academic_term_id": 26, "start_date": "2026-01-05", "end_date": "2026-06-26", } evidence = { "fixture_manifest": {}, "stage_assertions": {"fixture_manifest": {}}, } with self.assertRaisesRegex( MODULE.LoadHarnessError, "not bound to the authoritative academic-term scope", ): MODULE._assert_provision_term_scope(evidence, value) scoped = { "academic_term_id": 26, "academic_term_start_date": "2026-01-05", "academic_term_end_date": "2026-06-26", "authoritative_term_scope_enforced": True, "activity_template_id": 81, "donor_activity_template_id": 81, "term_template_donor_only": True, "template_candidate_defaults_authoritative": False, "candidate_defaults_replaced_before_scheduling": True, "run_owned_activity_final_allocations_empty_at_provision": True, "advisory_requirement_source_watermark_unchanged": True, } evidence = { "run_id": RUN_ID, "fixture_manifest": scoped, "stage_assertions": {"fixture_manifest": dict(scoped)}, } self.assertIsNone(MODULE._assert_provision_term_scope(evidence, value)) def test_preschedule_canary_requires_both_exact_load_slots(self): delivery_id = "4ad9aa31-37ef-4acd-a652-f9ed8a9c0d15" response = { "request_id": delivery_id, "status": "success", "schedule": json.dumps( [{"activity": "91", "preference": [0, 9, 7, 0]}] ), } accepted = PROVISION_MODULE._parse_preschedule_feasibility_response( response, delivery_id=delivery_id, activity_id=91, required_slots=(1, 2), ) rejected = PROVISION_MODULE._parse_preschedule_feasibility_response( response, delivery_id=delivery_id, activity_id=91, required_slots=(0, 1), ) self.assertTrue(accepted["protocol_valid"]) self.assertTrue(accepted["required_slots_available"]) self.assertEqual(accepted["required_preferences"], [9, 7]) self.assertTrue(rejected["protocol_valid"]) self.assertFalse(rejected["required_slots_available"]) def test_load07_short_lived_fence_is_separate_owner_only_and_exact(self): value = manifest() now = datetime.datetime(2026, 8, 9, 1, 0, tzinfo=datetime.UTC) unsigned = { "schema_version": 1, "run_id": RUN_ID, "timetabler_sha": value["commits"]["timetabler"], "resource_booking_sha": value["commits"]["resource_booking"], "execute_at_epoch": int(now.timestamp()) + 120, "nonce": "a" * 64, "approval_ref": "joint-load07-example", "created_at": "2026-08-09T00:59:30Z", "expires_at": "2026-08-09T01:03:00Z", } artifact = { **unsigned, "coordination_sha256": MODULE._sha256(unsigned), } with TemporaryDirectory() as directory: root = Path(directory) / "exchange" run_dir = root / RUN_ID def exchange_path(_run_id, profile): return root, run_dir, run_dir / f"{profile.lower()}.json" owner = os.getuid() patches = ( mock.patch.object(MODULE, "_evidence_path", side_effect=exchange_path), mock.patch.object(MODULE, "TIMETABLER_EVIDENCE_OWNER_UID", owner), ) for patcher in patches: patcher.start() self.addCleanup(patcher.stop) root.mkdir(mode=0o755) run_dir.mkdir(mode=0o755) path = run_dir / "load-07-fence.json" path.write_bytes(MODULE._canonical_bytes(artifact) + b"\n") path.chmod(0o444) verified = MODULE.verify_restart_fence( run_id=RUN_ID, manifest=value, supplied_sha256=artifact["coordination_sha256"], now=now, ) self.assertEqual(verified["execute_at_epoch"], int(now.timestamp()) + 120) with self.assertRaisesRegex(MODULE.LoadHarnessError, "hash"): MODULE.verify_restart_fence( run_id=RUN_ID, manifest=value, supplied_sha256="f" * 64, now=now, ) def test_arm_fence_creates_canonical_read_only_artifact_and_reuses_only_exact(self): value = manifest() now = datetime.datetime(2026, 8, 9, 1, 0, tzinfo=datetime.UTC) epoch = int(now.timestamp()) + 120 with TemporaryDirectory() as directory: root = Path(directory) / "exchange" run_dir = root / RUN_ID def exchange_path(_run_id, profile): return root, run_dir, run_dir / f"{profile.lower()}.json" owner = os.getuid() with ( mock.patch.object(MODULE, "_evidence_path", side_effect=exchange_path), mock.patch.object(MODULE, "TIMETABLER_EVIDENCE_OWNER_UID", owner), mock.patch.object( MODULE, "verify_predecessor_evidence", return_value={"profile": "LOAD-06"}, ), mock.patch.object( MODULE, "_assert_source_safety", return_value={"transport_watermark": 8212}, ), mock.patch.object(MODULE.secrets, "token_hex", return_value="a" * 64), ): value["evidence_exchange"] = MODULE.evidence_exchange_attestation() first = MODULE.arm_restart_fence( run_id=RUN_ID, manifest=value, supplied_predecessor_sha256="b" * 64, execute_at_epoch=epoch, approval_ref="approval/load-07/2026-08-09", now=now, ) target = run_dir / "load-07-fence.json" artifact = json.loads(target.read_bytes()) self.assertEqual(stat.S_IMODE(root.stat().st_mode), 0o755) self.assertEqual(stat.S_IMODE(run_dir.stat().st_mode), 0o755) self.assertEqual(stat.S_IMODE(target.stat().st_mode), 0o444) self.assertEqual( set(artifact), { "schema_version", "run_id", "timetabler_sha", "resource_booking_sha", "execute_at_epoch", "nonce", "approval_ref", "created_at", "expires_at", "coordination_sha256", }, ) self.assertEqual( first["coordination_sha256"], artifact["coordination_sha256"] ) self.assertFalse(first["load_executed"]) self.assertFalse(first["restart_executed"]) self.assertFalse(first["database_mutation_executed"]) self.assertFalse(first["domain_mutation_executed"]) second = MODULE.arm_restart_fence( run_id=RUN_ID, manifest=value, supplied_predecessor_sha256="b" * 64, execute_at_epoch=epoch, approval_ref="approval/load-07/2026-08-09", now=now, ) self.assertEqual( second["coordination_sha256"], first["coordination_sha256"] ) with self.assertRaisesRegex(MODULE.LoadHarnessError, "differs"): MODULE.arm_restart_fence( run_id=RUN_ID, manifest=value, supplied_predecessor_sha256="b" * 64, execute_at_epoch=epoch + 60, approval_ref="approval/load-07/2026-08-09", now=now, ) def test_attest_is_manifest_free_and_emits_shared_host_identity_hashes(self): source = SCRIPT.read_text() self.assertIn('args.action == "attest"', source) self.assertIn('"runtime_uid"', source) self.assertIn('"evidence_exchange"', source) self.assertNotIn('"runtime_home_sha256"', source) self.assertIn('"timetabler_database_sha256"', source) self.assertIn('"configuration_sha256_log_parts"', source) self.assertIn('"timetabler_database_sha256_log_parts"', source) def test_attestation_sha256_log_parts_round_trip_non_secret_hash(self): digest = "a" * 64 parts = MODULE._sha256_log_parts(digest) self.assertEqual(parts, ["a" * 32, "a" * 32]) self.assertEqual("".join(parts), digest) with self.assertRaisesRegex( MODULE.LoadHarnessError, "attestation SHA-256 is invalid" ): MODULE._sha256_log_parts("not-a-sha")