test(driver): prove validate_edi path works against Edifabric response shape

The live /v2/x12/validate endpoint is quota-blocked; until it resets at next UTC midnight, the driver can't make real network calls. This test wires the shipped cyclone.edifabric MockTransport seam to return canned OperationResult payloads and asserts the driver's passed() classifier returns the expected verdict. Covers 4 scenarios: 1. Status='success' → passed=True 2. Status='warning' → passed=True (advisory notices don't block) 3. Status='error' → passed=False (with error details in message) 4. HTTP 403 quota → EdifabricError raised (driver falls back to pyX12) Plus 1 wire-shape test that pins the validate call body is the X12Interchange JSON (not raw bytes) per the /v2/x12/validate contract.
2026-07-08 03:34:00 -06:00
parent e9e0b68746
commit 61e90b64b3
1 changed files with 205 additions and 0 deletions
@@ -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"