Tyler
76278ec9f6
Adds full TA1 (Interchange Acknowledgment, X12 envelope-level) support
to mirror the existing 999 transaction-set ACK pipeline.
Parser & models
- parsers/models_ta1.py: Pydantic Ta1Ack + ParseResultTa1 models
(AckCode = Literal['A','E','R'], date serializer mirroring 999)
- parsers/parse_ta1.py: TA1 segment parser
- Tolerant YYMMDD (6-digit) + CCYYMMDD (8-digit) date handling
for CO Medicaid interchange_date / ack_generated_date
- 106-char ISA validation (15-char sender/receiver IDs)
- source_batch_id = 'TA1-<ISA13>'
Persistence
- migrations/0005_create_ta1_acks.sql: ta1_acks table + indexes
on source_batch_id and ack_code
- db.py: Ta1Ack ORM model (flat columns + raw_json, mirrors Ack)
- store.py: add_ta1_ack, list_ta1_acks (returns all rows),
get_ta1_ack
- db version bumped 4 -> 5 (test_acks.py updated)
API (api.py)
- POST /api/parse-ta1: text/file ingest, persists ta1_ack,
returns detail-ready payload
- GET /api/ta1-acks: list with limit + total (mirrors 999 pattern)
- GET /api/ta1-acks/{ack_id}: detail
- _ta1_to_ui / _serialize_ta1 / _serialize_ta1_from_row helpers
Tests (17 new, 508 total passing)
- tests/test_parse_ta1.py (8): CCYYMMDD + YYMMDD acceptance,
E/R codes, source_batch_id, missing ISA/TA1, short defaults
- tests/test_api_ta1.py (9): happy path, rejected persists,
empty/malformed 400, missing file 422, detail regenerates
segment, 404, empty list, newest-first
- tests/test_prodfiles_smoke.py: extended to smoke-test 352
production TA1 files (A=0, R=352, E=0)
- tests/test_api_parse_persists.py: cross-pipeline reconciliation
test asserting invariants across 837P + 835 prod files
Real-data finding: all 352 production TA1s are R (rejected);
operator follow-up warranted.
128 lines
4.9 KiB
Python
128 lines
4.9 KiB
Python
"""Tests for the TA1 (Interchange Acknowledgment) parser orchestrator."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
|
|
import pytest
|
|
|
|
from cyclone.parsers.exceptions import CycloneParseError
|
|
from cyclone.parsers.parse_ta1 import parse_ta1_text
|
|
|
|
|
|
# A minimal accepted TA1 in canonical CCYYMMDD form (per X12 spec).
|
|
# ISA must be exactly 106 chars (15-char sender/receiver IDs).
|
|
ACCEPTED_CCYYMMDD = (
|
|
"ISA*00* *00* *ZZ*SENDER *ZZ*RECEIVER "
|
|
"*260520*1750*^*00501*000000001*0*P*:~"
|
|
"TA1*000000001*20260520*1750*A*000*20260520~"
|
|
"IEA*1*000000001~"
|
|
)
|
|
|
|
# A rejected TA1 with Colorado's YYMMDD date format (6 digits instead of 8).
|
|
# This is what FromHPE/tp11525703-837P_M019044596-…_TA1.x12 looks like in
|
|
# production — same shape but YYMMDD and ack_code=R, note_code=006.
|
|
REJECTED_YYMMDD = (
|
|
"ISA*00* *00* *ZZ*COMEDASSISTPROG*ZZ*TP11525703 "
|
|
"*260520*1750*^*00501*000000001*0*P*:~"
|
|
"TA1*320293557*260520*2338*R*006~"
|
|
"IEA*0*000000001~"
|
|
)
|
|
|
|
# Accepted variant with errors (E) — uncommon but spec-legal.
|
|
ACCEPTED_WITH_ERRORS = (
|
|
"ISA*00* *00* *ZZ*SENDER *ZZ*RECEIVER "
|
|
"*260520*1750*^*00501*000000001*0*P*:~"
|
|
"TA1*000000001*20260520*1750*E*007*20260520~"
|
|
"IEA*1*000000001~"
|
|
)
|
|
|
|
|
|
def test_parse_accepted_ccyymmdd_returns_a():
|
|
"""Spec-compliant TA1 with CCYYMMDD dates surfaces ack_code='A'."""
|
|
result = parse_ta1_text(ACCEPTED_CCYYMMDD, input_file="accepted.ta1")
|
|
assert result.ta1.ack_code == "A"
|
|
assert result.ta1.control_number == "000000001"
|
|
assert result.ta1.interchange_date == date(2026, 5, 20)
|
|
assert result.ta1.interchange_time == "1750"
|
|
assert result.ta1.note_code == "000"
|
|
assert result.ta1.ack_generated_date == date(2026, 5, 20)
|
|
# Envelope mirrors the ISA
|
|
assert result.envelope.sender_id == "SENDER"
|
|
assert result.envelope.receiver_id == "RECEIVER"
|
|
assert result.envelope.control_number == "000000001"
|
|
# Summary reflects the single ack
|
|
assert result.summary.total_claims == 1
|
|
assert result.summary.passed == 1
|
|
assert result.summary.failed == 0
|
|
|
|
|
|
def test_parse_rejected_yymmdd_from_production():
|
|
"""Colorado's YYMMDD date format (6 digits) must still parse."""
|
|
result = parse_ta1_text(REJECTED_YYMMDD, input_file="rejected.ta1")
|
|
assert result.ta1.ack_code == "R"
|
|
assert result.ta1.note_code == "006"
|
|
assert result.ta1.interchange_date == date(2026, 5, 20), (
|
|
"YYMMDD must be inferred as 20YY-MM-DD"
|
|
)
|
|
assert result.ta1.interchange_time == "2338"
|
|
assert result.summary.passed == 0
|
|
assert result.summary.failed == 1
|
|
|
|
|
|
def test_parse_accepted_with_errors_returns_e():
|
|
"""TA1 ack_code='E' (Accepted with errors) is preserved as the literal."""
|
|
result = parse_ta1_text(ACCEPTED_WITH_ERRORS, input_file="with_errors.ta1")
|
|
assert result.ta1.ack_code == "E"
|
|
assert result.ta1.note_code == "007"
|
|
|
|
|
|
def test_source_batch_id_format():
|
|
"""source_batch_id must be 'TA1-<ISA13>' so the FK target is stable."""
|
|
result = parse_ta1_text(ACCEPTED_CCYYMMDD, input_file="x.ta1")
|
|
assert result.source_batch_id == "TA1-000000001"
|
|
|
|
|
|
def test_missing_isa_raises():
|
|
"""No ISA envelope at file start → CycloneParseError (caught by tokenize)."""
|
|
text = "TA1*000000001*20260520*1750*A*000*20260520~"
|
|
with pytest.raises(CycloneParseError, match="ISA"):
|
|
parse_ta1_text(text, input_file="bad.ta1")
|
|
|
|
|
|
def test_missing_ta1_raises():
|
|
"""ISA but no TA1 segment → CycloneParseError."""
|
|
text = (
|
|
"ISA*00* *00* *ZZ*SENDER *ZZ*RECEIVER "
|
|
"*260520*1750*^*00501*000000001*0*P*:~"
|
|
"IEA*0*000000001~"
|
|
)
|
|
with pytest.raises(CycloneParseError, match="No TA1 segment"):
|
|
parse_ta1_text(text, input_file="no_ta1.ta1")
|
|
|
|
|
|
def test_short_ta1_segment_uses_defaults():
|
|
"""A TA1 with fewer than 6 elements still parses with sensible defaults."""
|
|
text = (
|
|
"ISA*00* *00* *ZZ*SENDER *ZZ*RECEIVER "
|
|
"*260520*1750*^*00501*000000001*0*P*:~"
|
|
"TA1*000000001~"
|
|
"IEA*0*000000001~"
|
|
)
|
|
result = parse_ta1_text(text, input_file="short.ta1")
|
|
assert result.ta1.ack_code == "R" # default to rejected when missing
|
|
assert result.ta1.note_code is None
|
|
assert result.ta1.interchange_date is None
|
|
assert result.ta1.interchange_time is None
|
|
|
|
|
|
def test_garbage_date_returns_none_not_raises():
|
|
"""A garbage date in TA102 doesn't raise — it surfaces as None."""
|
|
text = (
|
|
"ISA*00* *00* *ZZ*SENDER *ZZ*RECEIVER "
|
|
"*260520*1750*^*00501*000000001*0*P*:~"
|
|
"TA1*000000001*not-a-date*1750*A*000~"
|
|
"IEA*0*000000001~"
|
|
)
|
|
result = parse_ta1_text(text, input_file="bad_date.ta1")
|
|
assert result.ta1.interchange_date is None
|
|
assert result.ta1.ack_code == "A" # the rest still parses |