"""Spec <-> implementation drift gate. lambdas/api/openapi.json is the published contract; lambdas/api/router.py is what the Lambda actually serves. This test makes them the SAME set: an endpoint added/removed/renamed on one side without the other fails CI here, so the docs page can never silently lie. Also pins the phase-2 x-planned markers, the four outbound webhook events, and the enum values the spec promises against the WO extraction contract (prompts.py). """ import json from pathlib import Path from tests.support import REPO_ROOT, load_lambda_module _SPEC_PATH = Path(REPO_ROOT) / "lambdas" / "api" / "openapi.json" _HTTP_METHODS = {"get", "put", "post", "patch", "delete", "head", "options"} def _spec(): return json.loads(_SPEC_PATH.read_text(encoding="utf-8")) def _spec_routes(spec): implemented, planned = set(), set() for path, ops in spec["paths"].items(): for method, op in ops.items(): if method not in _HTTP_METHODS: continue key = (method.upper(), path) if op.get("x-planned"): planned.add(key) else: implemented.add(key) return implemented, planned def test_spec_is_openapi_31(): assert _spec()["openapi"] == "3.1.0" def test_implemented_routes_match_router_exactly(): router = load_lambda_module("api", "router") implemented, planned = _spec_routes(_spec()) assert implemented == set(router.DATA_ROUTES) | set(router.DOCS_ROUTES) assert planned == set(router.PLANNED_ROUTES) def test_every_data_route_has_a_repo_function(): router = load_lambda_module("api", "router") handler = load_lambda_module("api", "handler") assert set(router.DATA_ROUTES.values()) == set(handler._REPO_FUNCS) def test_webhooks_section_documents_all_four_events(): spec = _spec() assert set(spec["webhooks"]) == { "work_order.created", "work_order.updated", "work_order.cancelled", "work_order.comment_added", } for ops in spec["webhooks"].values(): assert "post" in ops def test_spec_enums_match_wo_extraction_contract(): # The WO pipeline's prompts.py is the enum source of truth (the webhook # contract pins it too). If the pipeline ever widens wo_status or # record_type, the published spec must move in the same PR. prompts = load_lambda_module("wo", "email_processor/prompts") prompt_text = prompts.EXTRACTION_PROMPT spec = _spec() schemas = spec["components"]["schemas"] wo_status_enum = { value for value in schemas["WorkOrder"]["properties"]["wo_status"]["enum"] if value is not None } record_type_enum = { value for value in schemas["WorkOrder"]["properties"]["record_type"]["enum"] if value is not None } for value in wo_status_enum - {"unknown"}: assert value in prompt_text, f"wo_status {value!r} not in extraction prompt" for value in record_type_enum: assert value in prompt_text, f"record_type {value!r} not in extraction prompt" event_data = schemas["WorkOrderEventData"]["properties"] assert set(event_data["wo_status"]["enum"]) == set( schemas["WorkOrder"]["properties"]["wo_status"]["enum"] ) def test_spec_has_no_script_breakout_sequence(): # The spec is inlined into a