Nora
5fe61fddd3
EdifabricError now carries an optional retry_after_seconds kwarg parsed from the upstream Retry-After header (API Management quota policy always sends this on 429/403 quota responses). The spot-check driver threads the value into each transport_error entry of spot-check.json so the operator sees the exact moment quota replenishes rather than parsing it out of the response body. Adds a MockTransport test confirming retry_after_seconds is set when the upstream sends Retry-After. Existing callers are unaffected (retry_after_seconds defaults to None and is keyword-only).
210 lines
7.4 KiB
Python
210 lines
7.4 KiB
Python
"""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) |