"""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) def test_read_interchange_raises_with_retry_after_when_quota_blocked(): """On HTTP 403 with a ``Retry-After`` header (API Management quota policy), the raised EdifabricError exposes ``retry_after_seconds`` so callers can sleep exactly until quota replenishes.""" def handler(request: httpx.Request) -> httpx.Response: return httpx.Response( 403, headers={"Retry-After": "50492"}, json={"statusCode": 403, "message": "Out of call volume quota."}, ) _install_factory(handler) with pytest.raises(edifabric.EdifabricError) as exc_info: edifabric.read_interchange(b"ISA*...", api_key=_TEST_KEY) err = exc_info.value assert err.status_code == 403 assert err.retry_after_seconds == 50492 assert "Out of call volume quota" in str(err.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)