tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
27bca33b0921e7c3d0620239ac1826912ce1be3f
cyclone/backend/tests/test_serialize_837.py
T
Tyler 9bca4b608a feat(release): v0.2.0 — batch 837 export, ClaimCard, theme tokens 
Backend:

- New POST /api/batches/{id}/export-837: regenerate X12 837 files
  for a list of claim_ids into a ZIP using HCPF file naming standards,
  with a unique interchange/group control number per export. Wire
  the clearhouse Loop 1000A (NM1*41 + PER) and per-payer receiver
  (NM1*40) blocks so the serializer no longer falls back to
  CYCLONE / RECEIVER placeholders.
- /api/parse-837 and /api/parse-835 now surface the server-side
  batch_id in both JSON and NDJSON response shapes so the frontend
  can hit batch-scoped endpoints without an extra listBatches
  round-trip.
- Filename helpers and the 837 serializer updated to match the new
  HCPF envelope; tests cover batch export, parse batch_id, and the
  serializer's control-number uniqueness guarantee.

Frontend:
- New shared components: ClaimCard, ClaimCard837, DominantKpiCard,
  EditorialNote, ExportBar, TickerTape, and a charts/ set
  (BarChart, HBarChart, SegmentedBar, AgingBars).
- New useBatchExport hook driving ExportBar's download flow against
  the new endpoint.
- ClaimDrawer, Lane, and Layout migrated from raw CSS-variable
  colors to Tailwind theme tokens (bg-card, text-foreground,
  border/60, etc.) for consistency with the rest of the instrument
  chrome; the active tab indicator gains a subtle accent glow.
- Upload, Inbox, Batches, BatchDiff, Reconciliation, and Acks pages
  reworked to compose the new shared components and consume the new
  batch-scoped API surface (notably ExportBar wired into Batches).

Tooling / Docs:
- Add audit-uiux.mjs and a docs/goodclaim.x12 sample fixture.
- Update ClaimDrawer testids and add coverage for the new
  components and the useBatchExport hook.

Rolls up into the v0.2.0 release tag.
2026-06-22 11:01:58 -06:00

596 lines
19 KiB
Python
Raw Blame History

"""Tests for the outbound 837P serializer (full rebuild approach)."""
from datetime import date
from decimal import Decimal
import pytest
from cyclone.parsers.parse_837 import parse
from cyclone.parsers.payer import PayerConfig
from cyclone.parsers.serialize_837 import (
SerializeError,
_build_bht,
_build_clm,
_build_dmg,
_build_dtp_472,
_build_gs,
_build_hl,
_build_isa,
_build_lx,
_build_n3,
_build_n4,
_build_nm1,
_build_per,
_build_ref,
_build_ref_g1,
_build_se,
_build_sbr,
_build_st,
_build_sv1,
serialize_837,
serialize_837_for_resubmit,
)
_CFG = PayerConfig(name="CO_MEDICAID")
def _load_claim(path: str = "tests/fixtures/co_medicaid_837p.txt"):
text = open(path).read()
result = parse(text, _CFG)
assert result.claims, f"no claims parsed from {path}"
return result.claims[0]
# ---------------------------------------------------------------------------
# Envelope helpers
# ---------------------------------------------------------------------------
def test_build_isa_is_106_chars_with_standard_delimiters():
isa = _build_isa("SENDER", "RECEIVER", "000000001")
assert isa.startswith("ISA*")
# 106 chars total: 105 content + 1 segment terminator.
assert len(isa) == 106
# ISA16 (component separator, ':') sits right before the segment terminator.
assert isa.endswith(":~")
def test_build_gs_emits_gs_segment_with_hc_functional_id():
gs = _build_gs("SENDER", "RECEIVER", "1")
parts = gs.rstrip("~").split("*")
assert parts[0] == "GS"
assert parts[1] == "HC"
assert parts[2] == "SENDER"
assert parts[3] == "RECEIVER"
assert parts[6] == "1"
def test_build_gs_uses_gs04_yyyymmdd_8_digits():
"""GS-04 must be CCYYMMDD (8 digits) per X12; ISA uses YYMMDD (6).
A 6-digit value like '260622' is rejected by EDI validators with
'Element GS-04 must use CCYYMMDD date format'.
"""
gs = _build_gs("SENDER", "RECEIVER", "1")
parts = gs.rstrip("~").split("*")
# parts[4] = GS-04 date
assert len(parts[4]) == 8, f"expected 8-digit CCYYMMDD, got {parts[4]!r}"
# Must parse as a CCYYMMDD date
from datetime import datetime
datetime.strptime(parts[4], "%Y%m%d")
def test_build_st_emits_837_segment():
st = _build_st("0001")
assert st.startswith("ST*837*0001*005010X222A1~")
def test_build_se_emits_se_segment():
se = _build_se(10, "0001")
assert se == "SE*10*0001~"
# ---------------------------------------------------------------------------
# Hierarchy helpers
# ---------------------------------------------------------------------------
def test_build_nm1_non_person_entity_uses_nm103_for_name():
nm1 = _build_nm1("85", "85", "CLINIC A", "XX", "1234567890")
parts = nm1.rstrip("~").split("*")
assert parts[0] == "NM1"
assert parts[1] == "85"
assert parts[2] == "2" # non-person
assert parts[3] == "CLINIC A"
assert parts[8] == "XX"
assert parts[9] == "1234567890"
def test_build_nm1_person_entity_splits_first_last():
nm1 = _build_nm1("IL", "IL", "Doe Jane", "MI", "M12345")
parts = nm1.rstrip("~").split("*")
assert parts[1] == "IL"
assert parts[2] == "1"
assert parts[3] == "Doe"
assert parts[4] == "Jane"
def test_build_per_emits_per01_even_with_no_contact():
"""PER is required by X12 Loop 1000A — at least PER01 must be present."""
per = _build_per(None, None)
assert per == "PER*IC~"
per = _build_per("", "")
assert per == "PER*IC~"
def test_build_per_emits_segment_with_phone_contact():
per = _build_per("Jane Doe", "5551234567")
parts = per.rstrip("~").split("*")
assert parts[0] == "PER"
assert parts[1] == "IC"
assert parts[2] == "Jane Doe"
assert parts[3] == "TE"
assert parts[4] == "5551234567"
def test_build_per_emits_segment_with_email_contact():
"""When email is given, it wins over phone (HCPF expects email)."""
per = _build_per("Tyler Martinez", None, contact_email="tyler@dzinesco.com")
parts = per.rstrip("~").split("*")
assert parts[0] == "PER"
assert parts[1] == "IC"
assert parts[2] == "Tyler Martinez"
assert parts[3] == "EM"
assert parts[4] == "tyler@dzinesco.com"
def test_build_per_email_takes_precedence_over_phone():
"""If both phone and email are given, email is emitted (PER04)."""
per = _build_per("Tyler", "555-1234", contact_email="t@example.com")
parts = per.rstrip("~").split("*")
assert parts[3] == "EM"
assert parts[4] == "t@example.com"
def test_build_n3_returns_empty_when_no_address():
assert _build_n3(None, None) == ""
assert _build_n3("", "") == ""
def test_build_n4_returns_empty_when_no_address():
assert _build_n4(None, None, None) == ""
def test_build_dmg_returns_empty_when_no_demographics():
assert _build_dmg(None, None) == ""
def test_build_hl_emits_segment():
hl = _build_hl("1", "", "20", "1")
assert hl == "HL*1**20*1~"
def test_build_sbr_emits_segment_with_correct_slots():
"""SBR01=Payer Responsibility Seq Code (default 'P'),
SBR02=Individual Relationship Code (e.g. '18' for self),
SBR09=Claim Filing Indicator Code (e.g. 'MC' for Medicaid)."""
sbr = _build_sbr("18", "MC")
parts = sbr.rstrip("~").split("*")
assert parts[0] == "SBR"
# SBR01 — primary
assert parts[1] == "P"
# SBR02 — individual relationship (self = 18)
assert parts[2] == "18"
# SBR09 — claim filing indicator
assert parts[9] == "MC"
# Member ID and payer name do NOT belong in SBR — they live in
# NM109 and NM1*PR.NM103 respectively.
assert "M123" not in sbr
assert "PAYER" not in sbr
def test_build_sbr_defaults_relationship_to_self_and_filing_to_empty():
"""When called with all-None, SBR01/02 fall back to safe defaults
and SBR09 is left empty (the validator's R202 rule will then skip)."""
sbr = _build_sbr(None, None)
parts = sbr.rstrip("~").split("*")
assert parts[1] == "P"
assert parts[2] == "18"
assert parts[9] == ""
def test_build_ref_returns_empty_when_no_value():
assert _build_ref("EI", None) == ""
assert _build_ref("EI", "") == ""
def test_build_ref_emits_segment_with_value():
assert _build_ref("EI", "123456789") == "REF*EI*123456789~"
# ---------------------------------------------------------------------------
# Claim-level segment builders
# ---------------------------------------------------------------------------
def _stub_claim_header():
from cyclone.parsers.models import ClaimHeader
return ClaimHeader(
claim_id="CLM-1",
total_charge=Decimal("100.00"),
place_of_service="11",
frequency_code="1",
)
def test_build_bht_uses_default_transaction_type_ch_when_none():
bht = _build_bht(
transaction_type_code=None,
reference_id="REF1",
transaction_date=date(2026, 6, 20),
transaction_time="1200",
)
parts = bht.rstrip("~").split("*")
assert parts[0] == "BHT"
assert parts[6] == "CH"
def test_build_bht_propagates_transaction_type_code():
bht = _build_bht(
transaction_type_code="RP",
reference_id="REF1",
transaction_date=date(2026, 6, 20),
transaction_time="1200",
)
parts = bht.rstrip("~").split("*")
assert parts[6] == "RP"
def test_build_clm_emits_clm01_to_clm05():
claim = _stub_claim_header()
clm = _build_clm(claim)
parts = clm.rstrip("~").split("*")
assert parts[0] == "CLM"
assert parts[1] == "CLM-1"
assert parts[2] == "100.00"
# CLM05 is a composite "POS:facility_qualifier:frequency_code" — the stub
# has POS="11" and frequency_code="1", no qualifier set.
assert parts[5] == "11:1"
def test_build_clm_emits_clm08_defaulting_to_y():
"""CLM-08 (Benefits Assignment Certification) is required by X12
837P when CLM-07 = 'Y'. Default to 'Y' when the source didn't
capture one (matches what 99% of HCPF files look like).
"""
claim = _stub_claim_header()
clm = _build_clm(claim)
parts = clm.rstrip("~").split("*")
# parts[8] = CLM-08
assert parts[8] == "Y", f"CLM-08 should default to 'Y', got {parts[8]!r}"
def test_build_clm_propagates_captured_clm08():
from cyclone.parsers.models import ClaimHeader
claim = ClaimHeader(
claim_id="CLM-1",
total_charge=Decimal("100.00"),
place_of_service="11",
frequency_code="1",
assignment="Y",
benefits_assignment_certification="N",
release_of_info="Y",
)
clm = _build_clm(claim)
parts = clm.rstrip("~").split("*")
assert parts[8] == "N"
def test_build_ref_g1_returns_empty_when_no_prior_auth():
assert _build_ref_g1(None) == ""
assert _build_ref_g1("") == ""
def test_build_ref_g1_emits_segment_when_prior_auth_set():
assert _build_ref_g1("AUTH123") == "REF*G1*AUTH123~"
# ---------------------------------------------------------------------------
# Service-line segment builders
# ---------------------------------------------------------------------------
def _stub_service_line(charge="100.00", units=Decimal("1"), date_value=date(2026, 6, 15)):
from cyclone.parsers.models import Procedure, ServiceLine
return ServiceLine(
line_number=1,
procedure=Procedure(qualifier="HC", code="99213", modifiers=["25"]),
charge=Decimal(charge),
units=units,
unit_type="UN",
service_date=date_value,
)
def test_build_lx_emits_lx_segment():
assert _build_lx(1) == "LX*1~"
def test_build_sv1_emits_procedure_modifiers_charge_units():
line = _stub_service_line()
sv1 = _build_sv1(line)
parts = sv1.rstrip("~").split("*")
assert parts[0] == "SV1"
assert parts[1].startswith("HC:99213")
assert ":25" in parts[1]
assert parts[2] == "100.00"
assert parts[3] == "UN"
assert parts[4] == "1"
def test_build_sv1_emits_sv1_06_and_sv1_07_when_dx_pointer_given():
"""SV1-07 (Diagnosis Code Pointer) is required by X12/HCPF when the
parent claim has an HI segment (i.e. has at least one diagnosis).
Per 005010X222A1, SV1-06 is "Not Used" by the guide and MUST be
empty. Unit basis (UN/MJ/...) goes only in SV1-03.
"""
line = _stub_service_line()
sv1 = _build_sv1(line, dx_pointer="1")
parts = sv1.rstrip("~").split("*")
# parts layout: SV1, comp(SV1-01), charge(02), unit_basis(03),
# units(04), pos(05), ""(06 NOT USED), sv1_07(07)
assert len(parts) == 8, f"expected 8 elements, got {parts}"
# SV1-03 = unit basis
assert parts[3] == "UN"
# SV1-06 = "" (Not Used by 837P guide)
assert parts[6] == "", f"SV1-06 must be empty (Not Used by 837P), got {parts[6]!r}"
# SV1-07 = pointer
assert parts[7] == "1"
def test_build_sv1_omits_sv1_07_when_no_dx_pointer():
"""When the claim has no HI segment, SV1-07 should be empty."""
line = _stub_service_line()
sv1 = _build_sv1(line) # no dx_pointer kwarg
parts = sv1.rstrip("~").split("*")
assert len(parts) == 8
assert parts[6] == "" # SV1-06 still empty
assert parts[7] == "" # SV1-07 empty
def test_build_sv1_matches_goodclaim_layout():
"""Layout must match the known-good reference at docs/goodclaim.x12:
SV1*HC:T1019:U1:KX*125.40*UN*19.00***1~
i.e. 8 fields total: comp, charge, UN, units, '', '', '1'.
"""
from cyclone.parsers.models import Procedure, ServiceLine
line = ServiceLine(
line_number=1,
procedure=Procedure(qualifier="HC", code="T1019", modifiers=["U1", "KX"]),
charge=Decimal("125.40"),
units=Decimal("19.00"),
unit_type="UN",
place_of_service=None,
)
sv1 = _build_sv1(line, dx_pointer="1")
assert sv1 == "SV1*HC:T1019:U1:KX*125.40*UN*19.00***1~"
def test_build_dtp_472_emits_service_date():
assert _build_dtp_472(date(2026, 6, 15)) == "DTP*472*D8*20260615~"
def test_build_dtp_472_returns_empty_when_no_date():
assert _build_dtp_472(None) == ""
# ---------------------------------------------------------------------------
# serialize_837 — envelope + body composition
# ---------------------------------------------------------------------------
def test_serialize_837_envelope_has_106_char_isa():
claim = _load_claim()
text = serialize_837(claim)
# The ISA segment is 106 chars including the segment terminator.
isa_segment = next(part + "~" for part in text.split("~") if part.startswith("ISA"))
assert len(isa_segment) == 106
def test_serialize_837_emits_envelope_in_order():
claim = _load_claim()
text = serialize_837(claim)
seg_ids = [seg.split("*")[0] for seg in text.split("~") if seg]
# Envelope order
assert seg_ids[0] == "ISA"
assert seg_ids[1] == "GS"
assert seg_ids[2] == "ST"
assert seg_ids[3] == "BHT"
# Termination
assert seg_ids[-2] == "GE"
assert seg_ids[-1] == "IEA"
# Body order (after BHT, before SE)
body = seg_ids[3:-2]
assert "CLM" in body
assert "HI" in body
assert "NM1" in body
assert "HL" in body
assert "SV1" in body
def test_serialize_837_round_trips_canonical_fields():
claim = _load_claim()
text = serialize_837(claim)
reparsed = parse(text, _CFG).claims[0]
# Canonical header fields
assert reparsed.claim.claim_id == claim.claim.claim_id
assert Decimal(reparsed.claim.total_charge) == claim.claim.total_charge
assert reparsed.claim.frequency_code == claim.claim.frequency_code
assert reparsed.claim.place_of_service == claim.claim.place_of_service
# Diagnoses
assert [d.code for d in reparsed.diagnoses] == [d.code for d in claim.diagnoses]
# Service lines (charge, units)
assert len(reparsed.service_lines) == len(claim.service_lines)
for src, out in zip(claim.service_lines, reparsed.service_lines):
assert out.procedure.code == src.procedure.code
assert Decimal(out.charge) == src.charge
def test_serialize_837_edited_charge_propagates():
claim = _load_claim()
claim.claim.total_charge = Decimal("999.99")
text = serialize_837(claim)
assert "CLM*" in text and "*999.99*" in text
reparsed = parse(text, _CFG).claims[0]
assert Decimal(reparsed.claim.total_charge) == Decimal("999.99")
def test_serialize_837_edited_frequency_propagates():
claim = _load_claim()
claim.claim.frequency_code = "7"
text = serialize_837(claim)
reparsed = parse(text, _CFG).claims[0]
assert reparsed.claim.frequency_code == "7"
def test_serialize_837_edited_prior_auth_propagates():
claim = _load_claim()
claim.claim.prior_auth = "AUTH999"
text = serialize_837(claim)
assert "REF*G1*AUTH999" in text
reparsed = parse(text, _CFG).claims[0]
assert reparsed.claim.prior_auth == "AUTH999"
def test_serialize_837_added_diagnosis_propagates():
claim = _load_claim()
from cyclone.parsers.models import Diagnosis
claim.diagnoses.append(Diagnosis(code="Z00.00", qualifier="BF"))
text = serialize_837(claim)
assert "BF:Z00.00" in text
reparsed = parse(text, _CFG).claims[0]
codes = [d.code for d in reparsed.diagnoses]
assert "Z00.00" in codes
def test_serialize_837_edited_service_line_charge_propagates():
claim = _load_claim()
claim.service_lines[0].charge = Decimal("555.55")
text = serialize_837(claim)
assert "555.55" in text
reparsed = parse(text, _CFG).claims[0]
assert any(Decimal(sl.charge) == Decimal("555.55") for sl in reparsed.service_lines)
def test_serialize_837_edited_service_date_propagates():
claim = _load_claim()
claim.service_lines[0].service_date = date(2027, 1, 15)
text = serialize_837(claim)
assert "20270115" in text
def test_serialize_837_uses_custom_sender_receiver_ids():
claim = _load_claim()
text = serialize_837(
claim,
sender_id="AXISCARE",
receiver_id="CO_MED",
submitter_name="AXISCARE BILLING",
)
assert "AXISCARE" in text
assert "CO_MED" in text
# Sanity: file still parses
parse(text, _CFG)
def test_serialize_837_emits_per_segment_in_submitter_block():
"""X12 Loop 1000A (Submitter Name) requires a PER segment after
NM1*41. The serializer must emit one even with no contact info
(PER01='IC' is the only required element)."""
claim = _load_claim()
text = serialize_837(claim)
# The first NM1*41 should be followed immediately by a PER segment.
seg_ids = [seg.split("*")[0] for seg in text.split("~") if seg]
nm1_41_idx = seg_ids.index("NM1") # first NM1 is the submitter
assert nm1_41_idx >= 0
# The very next segment must be PER (PER01='IC' is required by spec).
assert seg_ids[nm1_41_idx + 1] == "PER"
per_line = next(seg for seg in text.split("~") if seg.startswith("PER*IC"))
assert per_line.startswith("PER*IC")
def test_serialize_837_per_segment_includes_email_from_kwargs():
"""Passing submitter_contact_email should emit PER*IC*<name>*EM*<email>."""
claim = _load_claim()
text = serialize_837(
claim,
sender_id="DZINESCO",
submitter_name="Dzinesco",
submitter_contact_name="Tyler Martinez",
submitter_contact_email="tyler@dzinesco.com",
)
assert "PER*IC*Tyler Martinez*EM*tyler@dzinesco.com" in text
# And the ISA sender id should be the clearhouse TPID, not "CYCLONE".
assert "ZZ*DZINESCO" in text
assert "ZZ*CYCLONE" not in text
def test_serialize_837_sbr09_uses_claim_filing_indicator_code_kwarg():
"""SBR09 must be the claim filing indicator (e.g. 'MC' for Medicaid),
not the member id. The serializer takes it from the kwarg."""
claim = _load_claim()
text = serialize_837(claim, claim_filing_indicator_code="MC")
sbr_line = next(seg for seg in text.split("~") if seg.startswith("SBR*"))
parts = sbr_line.rstrip("~").split("*")
# SBR01 = P (primary), SBR02 = 18 (self), SBR09 = MC
assert parts[1] == "P"
assert parts[2] == "18"
assert parts[9] == "MC"
# And the member id should NOT be in SBR.
assert claim.subscriber.member_id not in sbr_line
def test_serialize_837_for_resubmit_forwards_kwargs_to_serialize_837():
"""serialize_837_for_resubmit is a thin wrapper — it must forward
clearhouse + payer kwargs so the export endpoint can use it."""
claim = _load_claim()
text = serialize_837_for_resubmit(
claim,
interchange_index=7,
sender_id="DZINESCO",
submitter_name="Dzinesco",
submitter_contact_name="Tyler Martinez",
submitter_contact_email="tyler@dzinesco.com",
receiver_id="COMEDASSISTPROG",
receiver_name="COLORADO MEDICAL ASSISTANCE PROGRAM",
claim_filing_indicator_code="MC",
)
assert "ZZ*DZINESCO" in text
assert "ZZ*COMEDASSISTPROG" in text
assert "PER*IC*Tyler Martinez*EM*tyler@dzinesco.com" in text
# Control numbers reflect the resubmit index.
isa = next(seg for seg in text.split("~") if seg.startswith("ISA*"))
assert "000000007" in isa
def test_serialize_error_is_an_exception():
assert issubclass(SerializeError, Exception)
# ---------------------------------------------------------------------------
# serialize_837_for_resubmit
# ---------------------------------------------------------------------------
def test_serialize_837_for_resubmit_assigns_unique_control_numbers():
claim = _load_claim()
text_a = serialize_837_for_resubmit(claim, interchange_index=42)
text_b = serialize_837_for_resubmit(claim, interchange_index=43)
isa_a = next(line for line in text_a.split("~") if line.startswith("ISA"))
isa_b = next(line for line in text_b.split("~") if line.startswith("ISA"))
assert isa_a != isa_b
assert "000000042" in isa_a
assert "000000043" in isa_b
Reference in New Issue View Git Blame Copy Permalink