feat(sp40): Edifabric HTTP client (read + validate + composed)

Two-step wrapper around https://api.edination.com/v2/x12/read and /x12/validate. Used by the validate-837 CLI and the pre-upload gate inside resubmit-rejected-claims. cyclone.edifabric public surface: - read_interchange(edi_bytes) -> X12Interchange - validate_interchange(x12_json) -> OperationResult - validate_edi(edi_bytes) -> OperationResult - EdifabricError(status, body) for non-2xx responses API key resolved lazily via cyclone.secrets.get_secret('edifabric.api_key') which maps to CYCLONE_EDIFABRIC_API_KEY env var or keychain. The validate_edi / read_interchange / validate_interchange functions all accept an api_key= kwarg so tests can skip the secrets lookup. Test seam: set_transport_factory(client_factory) lets tests inject an httpx.Client backed by httpx.MockTransport — no live HTTP. The secrets module is never read in tests; the autouse _reset_transport fixture restores the default after each test. 7 mocked tests cover: read returns first of array, empty array 502s, validate success, validate Status=error doesn't raise (data, not exception), 5xx raises EdifabricError, validate_edi composes read + validate, missing API key surfaces clear error.
2026-07-07 18:07:53 -06:00
parent c5604fdfa0
commit b237b9cfc4
2 changed files with 404 additions and 0 deletions
@@ -0,0 +1,188 @@
+"""SP40: tests for the cyclone.edifabric HTTP client.
+
+All tests use httpx.MockTransport — no live HTTP hits the network.
+The API key is supplied directly via the ``api_key=`` kwarg so the
+secrets module is never read during tests.
+"""
+from __future__ import annotations
+
+import json
+
+import httpx
+import pytest
+
+from cyclone import edifabric
+
+
+_TEST_KEY = "test-edifabric-key-0123456789abcdef"
+
+
+def _make_client(handler):
+    """Build an httpx.Client whose transport is the given handler."""
+    transport = httpx.MockTransport(handler)
+    return httpx.Client(transport=transport, timeout=10.0)
+
+
+def _install_factory(handler):
+    """Swap in the mocked httpx.Client for the duration of a test."""
+    return edifabric.set_transport_factory(lambda: _make_client(handler))
+
+
+@pytest.fixture(autouse=True)
+def _reset_transport():
+    """Restore the default transport after each test (so a failing test
+    can't poison the next)."""
+    yield
+    edifabric._reset_transport_factory()
+
+
+# --- /x12/read ---------------------------------------------------------
+
+
+def test_read_interchange_returns_first_x12_from_array():
+    """The /x12/read endpoint returns a list (multi-interchange file).
+    Cyclone calls return the first element."""
+    x12 = {
+        "SegmentDelimiter": "~",
+        "DataElementDelimiter": "*",
+        "ISA": {"InterchangeControlNumber_13": "000000001"},
+        "Groups": [],
+        "IEATrailers": [],
+    }
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.headers["Ocp-Apim-Subscription-Key"] == _TEST_KEY
+        assert request.headers["Content-Type"] == "application/octet-stream"
+        return httpx.Response(200, json=[x12])
+
+    _install_factory(handler)
+    result = edifabric.read_interchange(b"ISA*...~IEA*0*000000001~", api_key=_TEST_KEY)
+    assert result["ISA"]["InterchangeControlNumber_13"] == "000000001"
+
+
+def test_read_interchange_rejects_empty_response():
+    """If /x12/read returns an empty list, surface a 502-style error."""
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json=[])
+
+    _install_factory(handler)
+    with pytest.raises(edifabric.EdifabricError) as exc_info:
+        edifabric.read_interchange(b"ISA*...~IEA*0*000000001~", api_key=_TEST_KEY)
+    assert exc_info.value.status_code == 502
+
+
+# --- /x12/validate -----------------------------------------------------
+
+
+def test_validate_interchange_returns_operation_result():
+    """A 200 response is returned verbatim — Status + Details."""
+    operation_result = {
+        "Status": "success",
+        "Details": [],
+        "LastIndex": 46,
+    }
+    x12 = {
+        "SegmentDelimiter": "~",
+        "DataElementDelimiter": "*",
+        "ISA": {"InterchangeControlNumber_13": "000000001"},
+        "Groups": [],
+        "IEATrailers": [],
+    }
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content)
+        assert body["ISA"]["InterchangeControlNumber_13"] == "000000001"
+        return httpx.Response(200, json=operation_result)
+
+    _install_factory(handler)
+    result = edifabric.validate_interchange(x12, api_key=_TEST_KEY)
+    assert result["Status"] == "success"
+    assert result["Details"] == []
+
+
+def test_validate_interchange_does_not_raise_on_status_error():
+    """OperationResult.Status='error' is data, not an exception — the
+    caller (the gate / CLI) decides whether to fail-closed."""
+    operation_result = {
+        "Status": "error",
+        "Details": [
+            {"Index": 5, "SegmentId": "PER", "Message": "PER-04 is required", "Status": "error"},
+        ],
+        "LastIndex": 5,
+    }
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json=operation_result)
+
+    _install_factory(handler)
+    result = edifabric.validate_interchange({"ISA": {}}, api_key=_TEST_KEY)
+    assert result["Status"] == "error"
+    assert result["Details"][0]["Message"] == "PER-04 is required"
+
+
+# --- 4xx / 5xx ---------------------------------------------------------
+
+
+def test_validate_interchange_raises_on_5xx():
+    """Non-2xx responses raise EdifabricError; the body is preserved."""
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(503, text="upstream overloaded")
+
+    _install_factory(handler)
+    with pytest.raises(edifabric.EdifabricError) as exc_info:
+        edifabric.validate_interchange({"ISA": {}}, api_key=_TEST_KEY)
+    assert exc_info.value.status_code == 503
+    assert "upstream overloaded" in str(exc_info.value.body)
+
+
+# --- validate_edi (composed) ------------------------------------------
+
+
+def test_validate_edi_composes_read_then_validate():
+    """validate_edi should call read first, then validate with the
+    X12Interchange JSON from read's response."""
+    x12 = {
+        "SegmentDelimiter": "~",
+        "DataElementDelimiter": "*",
+        "ISA": {"InterchangeControlNumber_13": "000000099"},
+        "Groups": [],
+        "IEATrailers": [],
+    }
+    operation_result = {"Status": "success", "Details": [], "LastIndex": 10}
+    seen_calls: list[str] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path.endswith("/read"):
+            seen_calls.append("read")
+            return httpx.Response(200, json=[x12])
+        if request.url.path.endswith("/validate"):
+            seen_calls.append("validate")
+            # Verify the validate body is the X12Interchange JSON
+            body = json.loads(request.content)
+            assert body["ISA"]["InterchangeControlNumber_13"] == "000000099"
+            return httpx.Response(200, json=operation_result)
+        raise AssertionError(f"unexpected path: {request.url.path}")
+
+    _install_factory(handler)
+    result = edifabric.validate_edi(b"ISA*...~IEA*0*000000099~", api_key=_TEST_KEY)
+    assert seen_calls == ["read", "validate"]
+    assert result["Status"] == "success"
+
+
+# --- API key handling --------------------------------------------------
+
+
+def test_validate_edi_raises_when_api_key_missing(monkeypatch):
+    """With no key configured anywhere, validate_edi surfaces a clear
+    error to the operator (not a generic 500)."""
+    monkeypatch.delenv("CYCLONE_EDIFABRIC_API_KEY", raising=False)
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        raise AssertionError("transport should not be called when key is missing")
+
+    _install_factory(handler)
+    with pytest.raises(edifabric.EdifabricError) as exc_info:
+        edifabric.validate_edi(b"ISA*...~IEA*0*000000001~")
+    assert exc_info.value.status_code == 0
+    assert "API key not configured" in str(exc_info.value.body)