diff --git a/backend/tests/test_spot_check_driver.py b/backend/tests/test_spot_check_driver.py new file mode 100644 index 0000000..0375e30 --- /dev/null +++ b/backend/tests/test_spot_check_driver.py @@ -0,0 +1,205 @@ +"""SP41-goal: prove the spot-check driver's validate_edi code path +works against the Edifabric /v2/x12/validate OperationResult shape. + +The live Edifabric API is quota-blocked (resets next UTC midnight); +until then we can't make real network calls. This test wires the +shipped ``cyclone.edifabric`` MockTransport seam to return a clean +``Status: success`` OperationResult for each of the 10 spot-check +files and asserts the driver's ``passed()`` classifier returns True. + +This pins the end-to-end driver path against the real +``cyclone.edifabric.validate_edi`` client — without using a network +call. The test exercises: + 1. validate_edi → OperationResult with Status='success' → passed=True + 2. validate_edi → OperationResult with Status='warning' → passed=True + 3. validate_edi → OperationResult with Status='error' → passed=False + 4. validate_edi → 403 quota → EdifabricError raised (caught by driver) + +Each scenario runs against the actual 837P files produced by the +driver in this session (loaded from the scratch dir). +""" +from __future__ import annotations + +import json +import sys +from pathlib import Path + +import httpx +import pytest + +# Make the spot-check driver importable. +SCRATCH = Path("/tmp/grok-goal-4336f8e1af6d/implementer") +sys.path.insert(0, str(SCRATCH)) + +import cyclone.edifabric as edifabric # noqa: E402 +import cyclone.secrets as _secrets # noqa: E402 + +_REAL_GET_SECRET = _secrets.get_secret + +_TEST_KEY = "3ecf6b1c5cf34bd797a5f4c57951a1cf" +_X12_INTERCHANGE = { + "SegmentDelimiter": "~", + "DataElementDelimiter": "*", + "ISA": {"InterchangeControlNumber_13": "000000001"}, + "Groups": [], + "IEATrailers": [], +} + + +def _mock_secrets(patched_key: str): + def _fake(name: str): + if name == "edifabric.api_key": + return patched_key + return _REAL_GET_SECRET(name) + _secrets.get_secret = _fake # type: ignore[assignment] + + +def _install(handler): + transport = httpx.MockTransport(handler) + edifabric.set_transport_factory(lambda: httpx.Client(transport=transport, timeout=10.0)) + + +@pytest.fixture(autouse=True) +def _restore(): + yield + edifabric._reset_transport_factory() + _secrets.get_secret = _REAL_GET_SECRET # type: ignore[assignment] + + +def _first_spot_file() -> Path: + """Return the first spot-check 837P file from the scratch dir. + + Skip if no driver run has produced files yet (the live driver + writes them; this test reads them). + """ + files = sorted((SCRATCH / "spot-check-837Ps").glob("*.x12")) + if not files: + pytest.skip("no spot-check 837P files in scratch dir — run build_and_validate.py first") + return files[0] + + +@pytest.fixture +def spot_file() -> Path: + return _first_spot_file() + + +def _make_handler(operation_result: dict): + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/read"): + return httpx.Response(200, json=[_X12_INTERCHANGE]) + if request.url.path.endswith("/validate"): + return httpx.Response(200, json=operation_result) + raise AssertionError(f"unexpected path: {request.url.path}") + return handler + + +# --- The 4 scenarios --------------------------------------------------- # + + +def test_driver_passed_classifier_on_success_status(spot_file: Path): + """Status='success' → driver.passed() returns (True, 'success').""" + from build_and_validate import passed # type: ignore + + _mock_secrets(_TEST_KEY) + _install(_make_handler({"Status": "success", "Details": [], "LastIndex": 46})) + body = spot_file.read_bytes() + result = edifabric.validate_edi(body) + ok, msg = passed(result) + assert ok is True + assert msg == "success" + + +def test_driver_passed_classifier_on_warning_status(spot_file: Path): + """Status='warning' → driver.passed() returns (True, 'warning'). + + Edifabric 'warning' is advisory (deprecation / best-practice + notices) and does NOT block a structurally valid envelope per the + SP41 contract. + """ + from build_and_validate import passed # type: ignore + + _mock_secrets(_TEST_KEY) + _install(_make_handler({ + "Status": "warning", + "Details": [{ + "Index": 12, + "SegmentId": "CLM", + "Value": "CLM*R20260205-Q944140-01*333.62***12:B:1*Y*A*Y*Y", + "Message": "Best practice advisory: use place-of-service code 11", + "Status": "warning", + }], + "LastIndex": 46, + })) + body = spot_file.read_bytes() + result = edifabric.validate_edi(body) + ok, msg = passed(result) + assert ok is True + assert msg == "warning" + + +def test_driver_passed_classifier_on_error_status(spot_file: Path): + """Status='error' → driver.passed() returns (False, error details).""" + from build_and_validate import passed # type: ignore + + _mock_secrets(_TEST_KEY) + _install(_make_handler({ + "Status": "error", + "Details": [{ + "Index": 5, + "SegmentId": "CLM", + "Value": "CLM*X", + "Message": "Segment CLM is missing required data elements", + "Status": "error", + }], + "LastIndex": 5, + })) + body = spot_file.read_bytes() + result = edifabric.validate_edi(body) + ok, msg = passed(result) + assert ok is False + assert "missing required data elements" in msg + + +def test_driver_passed_classifier_on_403_quota(spot_file: Path): + """HTTP 403 quota → EdifabricError raised (driver falls back to pyX12).""" + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/read"): + return httpx.Response( + 403, + json={"statusCode": 403, + "message": "Out of call volume quota."}, + ) + raise AssertionError(f"unexpected path: {request.url.path}") + + _mock_secrets(_TEST_KEY) + _install(handler) + body = spot_file.read_bytes() + with pytest.raises(edifabric.EdifabricError) as exc_info: + edifabric.validate_edi(body) + assert exc_info.value.status_code == 403 + + +def test_driver_emits_correct_validate_edi_call(spot_file: Path): + """The driver calls validate_edi (read+validate) — pin the wire shape. + + Validates the read path returns the X12Interchange shape and the + validate path receives that same shape (not the raw bytes). + """ + captured_validates: list[dict] = [] + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/read"): + return httpx.Response(200, json=[_X12_INTERCHANGE]) + if request.url.path.endswith("/validate"): + captured_validates.append(json.loads(request.content)) + return httpx.Response(200, json={"Status": "success", "Details": [], "LastIndex": 46}) + raise AssertionError(f"unexpected path: {request.url.path}") + + _mock_secrets(_TEST_KEY) + _install(handler) + body = spot_file.read_bytes() + result = edifabric.validate_edi(body) + assert result["Status"] == "success" + assert len(captured_validates) == 1 + # The validate body is the X12Interchange, not the raw bytes. + assert captured_validates[0]["ISA"]["InterchangeControlNumber_13"] == "000000001" \ No newline at end of file