Tyler
133 lines
4.5 KiB
Python
133 lines
4.5 KiB
Python
"""Tests for the 999 ACK Pydantic models.
|
|
|
|
SP3 Phase 3 (T10): minimal typed shape needed by the parser (T11),
|
|
serializer (T12), and auto-ACK builder (T15). The spec defines a richer
|
|
``FunctionalGroupResponse`` model with nested ``SetHeader`` /
|
|
``ElementError`` types, but the actual code paths only need the
|
|
``AK1``/``AK2`` headers + ``AK5`` accept/reject + ``AK9`` counts. We keep
|
|
the leaner models in this file; richer shapes can be added later without
|
|
breaking the existing public API.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import date
|
|
|
|
from cyclone.parsers.models import BatchSummary, Envelope
|
|
from cyclone.parsers.models_999 import (
|
|
AcknowledgmentHeader,
|
|
FunctionalGroupAck,
|
|
ParseResult999,
|
|
SegmentContext,
|
|
SegmentError,
|
|
SetAcceptReject,
|
|
SetFunctionalGroupResponse,
|
|
)
|
|
|
|
|
|
def _build_result() -> ParseResult999:
|
|
return ParseResult999(
|
|
envelope=Envelope(
|
|
sender_id="RECEIVERID",
|
|
receiver_id="SUBMITTERID",
|
|
control_number="000000001",
|
|
transaction_date=date(2024, 1, 1),
|
|
implementation_guide="005010X231A1",
|
|
),
|
|
functional_group_acks=[
|
|
FunctionalGroupAck(
|
|
ak1=AcknowledgmentHeader(
|
|
functional_id_code="HC",
|
|
group_control_number="0001",
|
|
),
|
|
received_count=1,
|
|
accepted_count=1,
|
|
rejected_count=0,
|
|
ack_code="A",
|
|
),
|
|
],
|
|
set_responses=[
|
|
SetFunctionalGroupResponse(
|
|
ak2=AcknowledgmentHeader(
|
|
functional_id_code="837",
|
|
group_control_number="0001",
|
|
),
|
|
set_control_number="0001",
|
|
transaction_set_identifier="837",
|
|
segment_errors=[],
|
|
set_accept_reject=SetAcceptReject(code="A"),
|
|
),
|
|
],
|
|
summary=BatchSummary(
|
|
input_file="minimal_999.txt",
|
|
control_number="000000001",
|
|
transaction_date=date(2024, 1, 1),
|
|
total_claims=1,
|
|
passed=1,
|
|
failed=0,
|
|
),
|
|
)
|
|
|
|
|
|
def test_parse_result_999_minimal_round_trip():
|
|
"""Build a minimal ParseResult999 and round-trip via JSON."""
|
|
r = _build_result()
|
|
blob = json.loads(r.model_dump_json())
|
|
assert blob["envelope"]["control_number"] == "000000001"
|
|
assert blob["functional_group_acks"][0]["ack_code"] == "A"
|
|
assert blob["set_responses"][0]["set_accept_reject"]["code"] == "A"
|
|
# Round-trip back to the model
|
|
r2 = ParseResult999.model_validate(blob)
|
|
assert r2.functional_group_acks[0].received_count == 1
|
|
assert r2.set_responses[0].ak2.functional_id_code == "837"
|
|
|
|
|
|
def test_set_accept_reject_codes():
|
|
"""A / R / E / W / X all parse cleanly on SetAcceptReject."""
|
|
for code in ("A", "R", "E", "W", "X"):
|
|
sar = SetAcceptReject(code=code)
|
|
assert sar.code == code
|
|
# round-trip via model_dump
|
|
assert SetAcceptReject.model_validate(sar.model_dump()).code == code
|
|
|
|
|
|
def test_functional_group_ack_counts_serialize():
|
|
"""accepted/rejected counts on FunctionalGroupAck round-trip."""
|
|
fg = FunctionalGroupAck(
|
|
ak1=AcknowledgmentHeader(functional_id_code="HC", group_control_number="1"),
|
|
received_count=5,
|
|
accepted_count=3,
|
|
rejected_count=2,
|
|
ack_code="P", # "partial" — surfaced in AK909
|
|
)
|
|
blob = json.loads(fg.model_dump_json())
|
|
assert blob["received_count"] == 5
|
|
assert blob["accepted_count"] == 3
|
|
assert blob["rejected_count"] == 2
|
|
assert blob["ack_code"] == "P"
|
|
# default values populated
|
|
rebuilt = FunctionalGroupAck.model_validate(blob)
|
|
assert rebuilt.ak1.functional_id_code == "HC"
|
|
assert rebuilt.received_count == 5
|
|
|
|
|
|
def test_segment_error_carries_optional_element_ref():
|
|
"""SegmentError.element_position is optional; defaults to None when absent."""
|
|
seg_err = SegmentError(
|
|
context=SegmentContext(
|
|
segment_id="CLM",
|
|
segment_position=12,
|
|
loop_id="2300",
|
|
implementation_convention_ref=None,
|
|
),
|
|
error_code="8", # "Segment Exceeds Maximum Use"
|
|
)
|
|
assert seg_err.element_position is None
|
|
assert seg_err.element_reference is None
|
|
# Round-trip
|
|
blob = json.loads(seg_err.model_dump_json())
|
|
assert blob["error_code"] == "8"
|
|
assert blob["element_position"] is None
|
|
rebuilt = SegmentError.model_validate(blob)
|
|
assert rebuilt.context.segment_id == "CLM"
|