#!/usr/bin/env python3 """Provision isolated Phase 1 E2E fixtures through Timetabler APIs only. ORM access in this module is read-only. Every domain mutation is an authenticated request to an existing Timetabler admin API and is followed by exact outbox transaction evidence. The command is idempotent for one run id and aborts on a foreign or incompatible collision. """ from __future__ import annotations import base64 from dataclasses import dataclass from datetime import timedelta import hashlib import json import os from pathlib import Path import re import sys import time from typing import Any SCRIPT_DIR = Path(__file__).resolve().parent if str(SCRIPT_DIR) not in sys.path: sys.path.insert(0, str(SCRIPT_DIR)) import phase1_e2e_entry_gate as gate # noqa: E402 import phase1_e2e_harness as harness # noqa: E402 PAIR_NAMES = ( "bulk", "drag-drop", "booking", "reschedule", "variant", "constraint-break", "conflict", ) ACTIVITY_NAMES = ( "bulk-01", "bulk-02", "bulk-unschedule", "drag-drop", "reschedule", "variant", "constraint-break", "conflict", ) ACTIVITY_PAIR = { "bulk-01": "bulk", "bulk-02": "bulk", "bulk-unschedule": "bulk", "drag-drop": "drag-drop", "reschedule": "reschedule", "variant": "variant", "constraint-break": "constraint-break", "conflict": "conflict", } REQUIRED_PERMISSIONS = frozenset( { "staff;create", "staff;update", "staff;delete", "location;create", "location;update", "location;delete", "activity;create", "activity;delete", "resources;update_requirement", "schedule;request", "variant;create", "booking;create", "booking;schedule", "booking;delete", } ) CONFLICT_SENTINEL_EPOCH = 4_102_444_800 class ProvisionError(harness.HarnessError): pass @dataclass(frozen=True) class ProvisionConfig: run_id: str environment: str source_scope: str timetabler_base_url: str timetabler_email: str timetabler_password: str expected_sha: str github_run_id: str @classmethod def from_environment(cls) -> "ProvisionConfig": mapping = { "run_id": "TT_PHASE1_E2E_RUN_ID", "environment": "TT_PHASE1_E2E_ENVIRONMENT", "source_scope": "TT_PHASE1_E2E_SOURCE_SCOPE", "timetabler_base_url": "TT_PHASE1_E2E_BASE_URL", "timetabler_email": "TT_PHASE1_E2E_EMAIL", "timetabler_password": "TT_PHASE1_E2E_PASSWORD", "expected_sha": "TT_PHASE1_E2E_EXPECTED_SHA", "github_run_id": "TT_PHASE1_E2E_GITHUB_RUN_ID", } values = { field: ( os.environ.get(name, "") if field == "timetabler_password" else os.environ.get(name, "").strip() ) for field, name in mapping.items() } missing = sorted(name for field, name in mapping.items() if not values[field]) if missing: raise ProvisionError( "fixture provisioning configuration is incomplete: " + ", ".join(missing) ) config = cls(**values) gate.validate_run_id(config.run_id) if config.environment != "staging": raise ProvisionError("fixture provisioning is restricted to staging") if not re.fullmatch(r"[0-9a-f]{40}", config.expected_sha): raise ProvisionError("deployed git SHA attestation is invalid") if not config.github_run_id.isdigit(): raise ProvisionError("GitHub workflow run id is invalid") return config @dataclass(frozen=True) class ReferenceFacts: academic_term_id: int activity_template_id: int booking_type_id: int week_ids: tuple[int, int] booking_from_date: str booking_to_date: str minute_per_slot: int slots: tuple[int, ...] class Provisioner: def __init__( self, *, config: ProvisionConfig, session: gate.TimetablerAdminSession, actor_id: int, ): self.config = config self.session = session self.actor_id = actor_id self.mutations: list[dict[str, Any]] = [] self.created_counts = {"staff": 0, "location": 0, "activity": 0} self.reused_counts = {"staff": 0, "location": 0, "activity": 0} def _api_mutation( self, *, path: str, payload: dict[str, Any], label: str, expected_source_transactions: int = 1, ) -> None: request_id = f"{self.config.run_id}-provision-{label}" before = harness._assert_local_safety() self.session.post(path, payload, request_id=request_id) try: after, transactions = harness._wait_for_transaction_delta( start=before, expected=expected_source_transactions, request_id=request_id, timeout_seconds=90, ) except harness.HarnessError as error: raise ProvisionError(f"{label}: {error}") from error self.mutations.append( { "label": label, "before_watermark": before, "after_watermark": after, "transactions": transactions, } ) def _owned_resource(self, model, code: str): rows = list(model.objects.filter(code=code).order_by("id")[:2]) if len(rows) > 1: raise ProvisionError(f"duplicate fixture code collision: {code}") if not rows: return None row = rows[0] if row.created_by != self.actor_id or row.status != 1: raise ProvisionError( f"fixture code is not owned by the E2E identity: {code}" ) return row def ensure_staff(self, pair: str): from api.models import TtStaff code = f"{self.config.run_id}-fixture-staff-{pair}" staff = self._owned_resource(TtStaff, code) if staff is None: self._api_mutation( path="/api/admin/staff/create", payload={ "code": code, "name": f"{self.config.run_id} fixture staff {pair}", "maximum_period": 1, "contract_period": 1, }, label=f"staff-{pair}", ) staff = self._owned_resource(TtStaff, code) if staff is None: raise ProvisionError(f"staff fixture was not created: {code}") self.created_counts["staff"] += 1 else: self.reused_counts["staff"] += 1 return staff def ensure_location(self, pair: str): from api.models import TtLocation code = f"{self.config.run_id}-fixture-location-{pair}" location = self._owned_resource(TtLocation, code) if location is None: self._api_mutation( path="/api/admin/location/create", payload={ "code": code, "name": f"{self.config.run_id} fixture location {pair}", "maximum_period": 1, "contract_period": 1, "capacity": 1, }, label=f"location-{pair}", ) location = self._owned_resource(TtLocation, code) if location is None: raise ProvisionError(f"location fixture was not created: {code}") self.created_counts["location"] += 1 else: self.reused_counts["location"] += 1 return location def ensure_activity( self, *, name: str, pair_staff_id: int, pair_location_id: int, facts: ReferenceFacts, ): from api.models import TtActivity code = f"{self.config.run_id}-{name}" activity = self._owned_resource(TtActivity, code) if activity is None: self._api_mutation( path="/api/admin/activity/create", payload={ "code": code, "name": f"{self.config.run_id} {name}", "duration": facts.minute_per_slot, "planned_size": 1, "academic_term_id": facts.academic_term_id, "activity_template_id": facts.activity_template_id, "week_pattern": list(facts.week_ids), }, label=f"activity-{name}", ) activity = self._owned_resource(TtActivity, code) if activity is None: raise ProvisionError(f"activity fixture was not created: {code}") self.created_counts["activity"] += 1 else: self.reused_counts["activity"] += 1 if ( activity.academic_term_id != facts.academic_term_id or activity.activity_template_id != facts.activity_template_id or activity.scheduled or tuple(activity.week.order_by("id").values_list("id", flat=True)) != tuple(sorted(facts.week_ids)) ): raise ProvisionError(f"activity fixture has incompatible state: {code}") staff_ids = set(activity.staff_preset.values_list("id", flat=True)) location_ids = set(activity.location_preset.values_list("id", flat=True)) if staff_ids != {pair_staff_id} or location_ids != {pair_location_id}: if staff_ids - {pair_staff_id} or location_ids - {pair_location_id}: raise ProvisionError( f"activity fixture references a non-dedicated resource: {code}" ) self._api_mutation( path="/api/admin/resources/update-requirement", payload={ "activity_id": activity.id, "staff_requirement": 1, "staff_preset": [pair_staff_id], "staff_suitability": [], "location_requirement": 1, "location_preset": [pair_location_id], "location_suitability": [], }, label=f"requirements-{name}", expected_source_transactions=0, ) activity = self._owned_resource(TtActivity, code) if ( activity is None or set(activity.staff_preset.values_list("id", flat=True)) != {pair_staff_id} or set(activity.location_preset.values_list("id", flat=True)) != {pair_location_id} ): raise ProvisionError( f"activity fixture requirements are incomplete: {code}" ) return activity def discover_reference_facts() -> ReferenceFacts: from api.models import TtActivityTemplate, TtActivityType, TtSetting templates = ( TtActivityTemplate.objects.filter( status=1, academic_term__status=1, activity_type__booking=False, module__isnull=False, ) .select_related("academic_term", "activity_type", "module") .prefetch_related( "academic_term__week", "staff", "location", "staff_suitability", "location_suitability", ) .order_by("id") ) template = next( ( candidate for candidate in templates if candidate.academic_term.week.count() >= 2 and not candidate.staff.exists() and not candidate.location.exists() and not candidate.staff_suitability.exists() and not candidate.location_suitability.exists() ), None, ) if template is None: raise ProvisionError( "no active read-only activity template without resource references is available" ) booking_type = ( TtActivityType.objects.filter(status=1, booking=True).order_by("id").first() ) if booking_type is None: raise ProvisionError("no active native booking activity type is available") weeks = list(template.academic_term.week.order_by("week", "id")[:2]) if len(weeks) != 2: raise ProvisionError("reference academic term does not contain two weeks") settings = TtSetting.get_multiple_setting( {"minute_per_slot", "slot_per_day", "slot_per_week"} ) minute_per_slot = int(settings["minute_per_slot"]) slot_per_day = int(settings["slot_per_day"]) pattern = template.academic_term.default_availability or "" candidates = [ index for index, value in enumerate(pattern) if value == "0" and index % slot_per_day < slot_per_day - 1 ] if len(candidates) < 6: raise ProvisionError("reference academic term has insufficient available slots") selected_slots = tuple(candidates[index] for index in range(6)) booking_start = weeks[0].start_date return ReferenceFacts( academic_term_id=template.academic_term_id, activity_template_id=template.id, booking_type_id=booking_type.id, week_ids=(weeks[0].id, weeks[1].id), booking_from_date=booking_start.isoformat(), booking_to_date=(booking_start + timedelta(days=6)).isoformat(), minute_per_slot=minute_per_slot, slots=selected_slots, ) def wait_for_publisher(timeout_seconds: int = 180) -> dict[str, Any]: from api.services.integration.readiness import phase1_readiness deadline = time.monotonic() + timeout_seconds while True: readiness = phase1_readiness(require_enabled=True) if readiness["reverse_delivery_enabled"]: raise ProvisionError("reverse delivery became enabled during provisioning") if readiness["errors"] or readiness["dead_letter_count"]: raise ProvisionError("Timetabler publisher readiness failed") if ( readiness["publisher_live"] and readiness["publisher_last_sequence"] == readiness["transport_watermark"] ): return readiness if time.monotonic() >= deadline: raise ProvisionError("publisher did not reach the provisioned watermark") time.sleep(1) def selector(model: str, filters: dict[str, Any], expected_count: int = 1): return { "model": model, "filters": filters, "expected_count": expected_count, "dedicated_e2e": True, } def build_manifest( *, config: ProvisionConfig, facts: ReferenceFacts, watermark: int ) -> dict[str, Any]: prefix = config.run_id functional_final = watermark + 19 conflict_final = functional_final + 1 cleanup_final = conflict_final + 3 selectors = { "all_fixture_staff_ids": selector( "staff", {"code__startswith": f"{prefix}-fixture-staff-"}, 7 ), "all_fixture_location_ids": selector( "location", {"code__startswith": f"{prefix}-fixture-location-"}, 7 ), "all_fixture_activity_ids": selector( "activity", {"code__startswith": f"{prefix}-"}, 8 ), "bulk_activity_ids": selector( "activity", {"code__startswith": f"{prefix}-bulk-"}, 3 ), "unschedule_activity_id": selector( "activity", {"code": f"{prefix}-bulk-unschedule"} ), "drag_drop_activity_id": selector("activity", {"code": f"{prefix}-drag-drop"}), "reschedule_activity_id": selector( "activity", {"code": f"{prefix}-reschedule"} ), "variant_parent_id": selector("activity", {"code": f"{prefix}-variant"}), "constraint_break_activity_id": selector( "activity", {"code": f"{prefix}-constraint-break"} ), "conflict_activity_id": selector("activity", {"code": f"{prefix}-conflict"}), "booking_staff_id": selector( "staff", {"code": f"{prefix}-fixture-staff-booking"} ), "booking_location_id": selector( "location", {"code": f"{prefix}-fixture-location-booking"} ), "conflict_staff_id": selector( "staff", {"code": f"{prefix}-fixture-staff-conflict"} ), "conflict_location_id": selector( "location", {"code": f"{prefix}-fixture-location-conflict"} ), } cases = [ { "id": "E2E-01", "stage": "functional", "driver": "timetabler_admin_api", "fixture_refs": ["lifecycle_staff", "lifecycle_location"], "reverse_delivery_expected": False, "steps": [ { "name": "staff-create", "kind": "api", "path": "/api/admin/staff/create", "payload": { "code": "{{run_id}}-staff", "name": "{{run_id}} Staff", "maximum_period": 1, "contract_period": 1, }, "expected_source_transactions": 1, "capture": { "staff_id": selector("staff", {"code": f"{prefix}-staff"}) }, }, { "name": "staff-update-archive", "kind": "api", "path": "/api/admin/staff/update", "payload": {"id": ["{{staff_id}}"], "status": 2}, "expected_source_transactions": 1, }, { "name": "staff-delete", "kind": "api", "path": "/api/admin/staff/delete", "payload": {"id": ["{{staff_id}}"]}, "expected_source_transactions": 1, }, { "name": "location-create", "kind": "api", "path": "/api/admin/location/create", "payload": { "code": "{{run_id}}-location", "name": "{{run_id}} Location", "maximum_period": 1, "contract_period": 1, }, "expected_source_transactions": 1, "capture": { "location_id": selector( "location", {"code": f"{prefix}-location"} ) }, }, { "name": "location-update-archive", "kind": "api", "path": "/api/admin/location/update", "payload": {"id": ["{{location_id}}"], "status": 2}, "expected_source_transactions": 1, }, { "name": "location-delete", "kind": "api", "path": "/api/admin/location/delete", "payload": {"id": ["{{location_id}}"]}, "expected_source_transactions": 1, }, ], }, { "id": "E2E-02", "stage": "functional", "driver": "scheduling_engine_final_response", "fixture_refs": ["bulk_activity_ids"], "reverse_delivery_expected": False, "steps": [ { "name": "bulk-final", "kind": "api", "path": "/api/admin/schedule-request", "payload": {"activity_ids": "{{bulk_activity_ids}}"}, "expected_source_transactions": 1, "timeout_seconds": 120, } ], }, { "id": "E2E-03", "stage": "functional", "driver": "preschedule_advice", "fixture_refs": ["drag_drop_activity_id"], "reverse_delivery_expected": False, "steps": [ { "name": "advice-only", "kind": "api", "path": "/api/admin/preschedule-request", "payload": {"activity_ids": ["{{drag_drop_activity_id}}"]}, "expected_source_transactions": 0, } ], }, { "id": "E2E-04", "stage": "functional", "driver": "accepted_drag_drop", "fixture_refs": ["drag_drop_activity_id"], "reverse_delivery_expected": False, "steps": [ { "name": "accepted-drop", "kind": "api", "path": "/api/admin/schedule-request", "payload": { "activity_ids": ["{{drag_drop_activity_id}}"], "slot": "{{approved_slot}}", }, "expected_source_transactions": 1, "timeout_seconds": 120, } ], }, { "id": "E2E-05", "stage": "functional", "driver": "timetabler_booking_api", "fixture_refs": ["booking_staff_id", "booking_location_id"], "reverse_delivery_expected": False, "steps": [ { "name": "booking-create", "kind": "api", "path": "/api/admin/booking/create", "payload": { "code": "{{run_id}}-booking", "name": "{{run_id}} Booking", "slot_required": 1, "planned_size": 1, "activity_type_id": "{{booking_type_id}}", "from_date": "{{booking_from_date}}", "to_date": "{{booking_to_date}}", }, "expected_source_transactions": 1, "capture": { "booking_activity_id": selector( "activity", {"code": f"{prefix}-booking"} ) }, }, { "name": "booking-schedule-staff", "kind": "api", "path": "/api/admin/booking/schedule", "payload": { "activity_id": "{{booking_activity_id}}", "type": "staff", "resource_ids": ["{{booking_staff_id}}"], "slot": "{{booking_slot}}", }, "expected_source_transactions": 1, }, { "name": "booking-schedule-location", "kind": "api", "path": "/api/admin/booking/schedule", "payload": { "activity_id": "{{booking_activity_id}}", "type": "location", "resource_ids": ["{{booking_location_id}}"], "slot": "{{booking_slot}}", }, "expected_source_transactions": 1, }, ], }, { "id": "E2E-06", "stage": "functional", "driver": "timetabler_reschedule_api", "fixture_refs": ["reschedule_activity_id"], "reverse_delivery_expected": False, "steps": [ { "name": "initial-schedule", "kind": "api", "path": "/api/admin/schedule-request", "payload": { "activity_ids": ["{{reschedule_activity_id}}"], "slot": "{{initial_slot}}", }, "expected_source_transactions": 1, "timeout_seconds": 120, }, { "name": "reschedule", "kind": "api", "path": "/api/admin/schedule-request", "payload": { "activity_ids": ["{{reschedule_activity_id}}"], "slot": "{{replacement_slot}}", }, "expected_source_transactions": 1, "timeout_seconds": 120, }, ], }, { "id": "E2E-07", "stage": "functional", "driver": "timetabler_lifecycle_api", "fixture_refs": ["unschedule_activity_id", "variant_parent_id"], "reverse_delivery_expected": False, "steps": [ { "name": "unschedule", "kind": "api", "path": "/api/admin/unschedule", "payload": {"activity_ids": ["{{unschedule_activity_id}}"]}, "expected_source_transactions": 1, }, { "name": "booking-unschedule", "kind": "api", "path": "/api/admin/booking/unschedule", "payload": {"activity_ids": ["{{booking_activity_id}}"]}, "expected_source_transactions": 1, }, { "name": "booking-delete", "kind": "api", "path": "/api/admin/booking/delete", "payload": {"id": ["{{booking_activity_id}}"]}, "expected_source_transactions": 1, }, { "name": "variant-create", "kind": "api", "path": "/api/admin/variant-create", "payload": { "activity_id": "{{variant_parent_id}}", "week_ids": "{{variant_week_ids}}", }, "expected_source_transactions": 1, "capture": { "variant_activity_id": selector( "activity", {"code": f"{prefix}-variant 002"} ) }, }, { "name": "variant-delete-merge", "kind": "api", "path": "/api/admin/activity/delete", "payload": {"id": ["{{variant_activity_id}}"]}, "expected_source_transactions": 1, }, ], }, { "id": "E2E-08", "stage": "functional", "driver": "manual_constraint_break", "fixture_refs": ["constraint_break_activity_id"], "reverse_delivery_expected": False, "steps": [ { "name": "manual-final", "kind": "api", "path": "/api/admin/schedule-request", "payload": { "activity_ids": ["{{constraint_break_activity_id}}"], "slot": "{{constraint_break_slot}}", "constraint_breaking_mode": 1, }, "expected_source_transactions": 1, } ], }, { "id": "E2E-09", "stage": "redelivery", "driver": "immutable_transport_redelivery", "fixture_refs": ["run-owned-published-transaction"], "reverse_delivery_expected": False, "correlation_prefix": prefix, "delay_seconds": 5, }, { "id": "E2E-10", "stage": "conflict", "driver": "cross_service_mirror_only_race", "fixture_refs": [ "conflict_activity_id", "conflict_staff_id", "conflict_location_id", ], "reverse_delivery_expected": False, "execute_at_epoch": CONFLICT_SENTINEL_EPOCH, "steps": [ { "name": "tt-contender", "kind": "api", "target": "timetabler", "path": "/api/admin/schedule-request", "payload": { "activity_ids": ["{{conflict_activity_id}}"], "slot": "{{conflict_slot}}", }, "expected_source_transactions": 1, "timeout_seconds": 120, } ], }, { "id": "E2E-11", "stage": "restart", "driver": "coordinated_process_restart", "fixture_refs": ["attested_tt_publisher"], "reverse_delivery_expected": False, }, { "id": "E2E-12", "stage": "rollback", "driver": "safe_release_config_rollback", "fixture_refs": ["approved_release_and_config"], "reverse_delivery_expected": False, }, { "id": "E2E-13", "stage": "final", "driver": "unfiltered_reconciliation", "fixture_refs": ["immutable_rb_reconciliation_evidence_owned_by_rb"], "reverse_delivery_expected": False, "cleanup_steps": [ { "name": "delete-fixture-activities", "kind": "api", "path": "/api/admin/activity/delete", "payload": {"id": "{{all_fixture_activity_ids}}"}, "expected_source_transactions": 1, }, { "name": "delete-fixture-staff", "kind": "api", "path": "/api/admin/staff/delete", "payload": {"id": "{{all_fixture_staff_ids}}"}, "expected_source_transactions": 1, }, { "name": "delete-fixture-locations", "kind": "api", "path": "/api/admin/location/delete", "payload": {"id": "{{all_fixture_location_ids}}"}, "expected_source_transactions": 1, }, ], }, ] manifest = { "schema_version": 1, "environment": "staging", "source_scope": config.source_scope, "run_id": prefix, "dedicated_fixture_prefix": prefix, "expected_stage_start_watermarks": { "entry": watermark, "functional": watermark, "redelivery": functional_final, "conflict": functional_final, "restart": conflict_final, "rollback": conflict_final, "cleanup": conflict_final, "final": cleanup_final, }, "approved": True, "approval_ref": f"github-actions:{config.github_run_id}:provision", "conflict_armed": False, "load_enabled": False, "dr_enabled": False, "variables": { "approved_slot": facts.slots[0], "initial_slot": facts.slots[1], "replacement_slot": facts.slots[2], "constraint_break_slot": facts.slots[3], "conflict_slot": facts.slots[4], "booking_slot": facts.slots[5], "booking_from_date": facts.booking_from_date, "booking_to_date": facts.booking_to_date, "booking_type_id": facts.booking_type_id, "variant_week_ids": [facts.week_ids[1]], }, "reference_selectors": selectors, "cases": cases, } gate.load_and_validate_manifest( json.dumps(manifest), environment="staging", source_scope=config.source_scope, require_approval=True, ) harness.validate_execution_manifest(manifest, prefix) return manifest def execute() -> dict[str, Any]: config = ProvisionConfig.from_environment() gate.bootstrap_django_runtime() before = harness._assert_local_safety() facts = discover_reference_facts() session = gate.TimetablerAdminSession( base_url=gate.validate_base_url( config.timetabler_base_url, "Timetabler base URL" ), email=config.timetabler_email, password=config.timetabler_password, run_id=config.run_id, ) try: session.login_and_attest() if not REQUIRED_PERMISSIONS <= session.permissions: raise ProvisionError( "dedicated Timetabler identity lacks required E2E API permissions" ) if session.user_id is None: raise ProvisionError("dedicated Timetabler user id is unavailable") provisioner = Provisioner( config=config, session=session, actor_id=session.user_id ) staff_by_pair = {pair: provisioner.ensure_staff(pair) for pair in PAIR_NAMES} location_by_pair = { pair: provisioner.ensure_location(pair) for pair in PAIR_NAMES } for activity_name in ACTIVITY_NAMES: pair = ACTIVITY_PAIR[activity_name] provisioner.ensure_activity( name=activity_name, pair_staff_id=staff_by_pair[pair].id, pair_location_id=location_by_pair[pair].id, facts=facts, ) readiness = wait_for_publisher() watermark = int(readiness["transport_watermark"]) manifest = build_manifest(config=config, facts=facts, watermark=watermark) finally: session.logout() canonical = json.dumps( manifest, sort_keys=True, separators=(",", ":"), ensure_ascii=True ) manifest_sha256 = hashlib.sha256(canonical.encode()).hexdigest() transaction_ids = [ transaction["source_transaction_id"] for mutation in provisioner.mutations for transaction in mutation["transactions"] ] summary = { "fixture_provisioning": "passed", "run_id": config.run_id, "git_sha": config.expected_sha, "github_run_id": config.github_run_id, "before_watermark": before, "after_watermark": watermark, "created_counts": provisioner.created_counts, "reused_counts": provisioner.reused_counts, "source_transaction_count": len(transaction_ids), "source_transaction_ids": transaction_ids, "manifest_sha256": manifest_sha256, "manifest_case_count": len(manifest["cases"]), "conflict_armed": False, "reverse_delivery_enabled": False, "load_executed": False, "dr_executed": False, } print(json.dumps(summary, sort_keys=True)) print( "TT_PHASE1_E2E_MANIFEST_BASE64=" + base64.b64encode(canonical.encode()).decode() ) return summary def main() -> int: try: execute() return 0 except (ProvisionError, gate.EntryGateError) as error: print( json.dumps( {"fixture_provisioning": "blocked", "reason": str(error)}, sort_keys=True, ), file=sys.stderr, ) return 3 if __name__ == "__main__": raise SystemExit(main())