Files
stack/tests/rex/test_copybook.py

583 lines
24 KiB
Python

"""Tests for rex.copybook — cobol_to_snake, pic_byte_length, parse_copybook."""
from __future__ import annotations
import math
import pytest
from rex.copybook import CopybookField, cobol_to_snake, parse_copybook, pic_byte_length
from rex.sieve import FieldMap
# ── cobol_to_snake ────────────────────────────────────────────────────────────
class TestCobolToSnake:
"""cobol_to_snake converts COBOL-CASE hyphenated names to snake_case."""
# ── basic conversions ─────────────────────────────────────────────────────
def test_single_hyphen(self) -> None:
assert cobol_to_snake("CLAIM-ID") == "claim_id"
def test_two_hyphens(self) -> None:
assert cobol_to_snake("PAID-AMOUNT-USD") == "paid_amount_usd"
def test_three_hyphens(self) -> None:
assert cobol_to_snake("WS-PROVIDER-NPI") == "ws_provider_npi"
def test_no_hyphens_single_word(self) -> None:
assert cobol_to_snake("FILLER") == "filler"
def test_no_hyphens_already_lower(self) -> None:
assert cobol_to_snake("filler") == "filler"
def test_mixed_case_input(self) -> None:
# Input may not be all-caps in some copybooks
assert cobol_to_snake("Claim-Id") == "claim_id"
# ── whitespace handling ───────────────────────────────────────────────────
def test_leading_whitespace_stripped(self) -> None:
result = cobol_to_snake(" CLAIM-ID")
assert result == "claim_id"
def test_trailing_whitespace_stripped(self) -> None:
result = cobol_to_snake("CLAIM-ID ")
assert result == "claim_id"
# ── docstring examples ────────────────────────────────────────────────────
def test_docstring_example_claim_id(self) -> None:
assert cobol_to_snake("CLAIM-ID") == "claim_id"
def test_docstring_example_paid_amount(self) -> None:
assert cobol_to_snake("PAID-AMOUNT") == "paid_amount"
def test_docstring_example_ws_provider_npi(self) -> None:
assert cobol_to_snake("WS-PROVIDER-NPI") == "ws_provider_npi"
# ── output invariants ─────────────────────────────────────────────────────
def test_output_is_lowercase(self) -> None:
result = cobol_to_snake("CLAIM-ID")
assert result == result.lower()
def test_output_has_no_hyphens(self) -> None:
result = cobol_to_snake("CLAIM-PROVIDER-NPI-CODE")
assert "-" not in result
def test_output_uses_underscores(self) -> None:
result = cobol_to_snake("CLAIM-ID")
assert "_" in result
def test_idempotent_snake_input(self) -> None:
result = cobol_to_snake("claim_id")
assert result == "claim_id"
# ── healthcare field name examples ────────────────────────────────────────
def test_bene_hic_num(self) -> None:
assert cobol_to_snake("BENE-HIC-NUM") == "bene_hic_num"
def test_clm_from_dt(self) -> None:
assert cobol_to_snake("CLM-FROM-DT") == "clm_from_dt"
def test_icd_dgns_cd(self) -> None:
assert cobol_to_snake("ICD-DGNS-CD-1") == "icd_dgns_cd_1"
def test_npi_num(self) -> None:
assert cobol_to_snake("NPI-NUM") == "npi_num"
def test_hcpcs_cd(self) -> None:
assert cobol_to_snake("HCPCS-CD") == "hcpcs_cd"
# ── pic_byte_length ───────────────────────────────────────────────────────────
class TestPicByteLengthDisplay:
"""DISPLAY mode (default usage=""): one byte per character position."""
# ── PIC X(n) ─────────────────────────────────────────────────────────────
def test_x_15(self) -> None:
assert pic_byte_length("X(15)") == 15
def test_x_1(self) -> None:
assert pic_byte_length("X(1)") == 1
def test_x_10(self) -> None:
assert pic_byte_length("X(10)") == 10
def test_x_80(self) -> None:
assert pic_byte_length("X(80)") == 80
def test_x_bare(self) -> None:
# PIC X without parens = 1 byte
assert pic_byte_length("X") == 1
# ── PIC 9(n) ─────────────────────────────────────────────────────────────
def test_9_8(self) -> None:
assert pic_byte_length("9(8)") == 8
def test_9_3(self) -> None:
assert pic_byte_length("9(3)") == 3
def test_9_10(self) -> None:
assert pic_byte_length("9(10)") == 10
def test_9_bare(self) -> None:
assert pic_byte_length("9") == 1
# ── PIC S9(n) (signed, DISPLAY) ──────────────────────────────────────────
def test_s9_7(self) -> None:
# Signed display: sign in last byte — same digit count
assert pic_byte_length("S9(7)") == 7
def test_s9_5(self) -> None:
assert pic_byte_length("S9(5)") == 5
# ── PIC 9(n)V9(m) (implied decimal) ─────────────────────────────────────
def test_9_7_v_99(self) -> None:
# PIC S9(7)V99 = 7 + 2 = 9 bytes in DISPLAY
assert pic_byte_length("S9(7)V99") == 9
def test_9_5_v_9_2(self) -> None:
# PIC 9(5)V9(2) = 5 + 2 = 7 bytes
assert pic_byte_length("9(5)V9(2)") == 7
def test_9_5_v_9_4(self) -> None:
assert pic_byte_length("9(5)V9(4)") == 9
def test_9_3_v_99(self) -> None:
# 9(3)V99 = 3 + 2 = 5 bytes
assert pic_byte_length("9(3)V99") == 5
def test_v_is_not_counted_as_byte(self) -> None:
# V is an implied decimal point — it takes no space
without_decimal = pic_byte_length("9(5)")
with_decimal = pic_byte_length("9(5)V9(0)")
assert with_decimal == without_decimal
# ── explicit display usage ────────────────────────────────────────────────
def test_explicit_display_usage(self) -> None:
assert pic_byte_length("X(10)", usage="DISPLAY") == 10
def test_empty_usage_same_as_display(self) -> None:
assert pic_byte_length("9(8)", usage="") == pic_byte_length("9(8)")
class TestPicByteLengthComp3:
"""COMP-3 (packed decimal): ceil((digits + 1) / 2) bytes."""
def test_s9_7_v99_comp3(self) -> None:
# 7 + 2 = 9 digits → ceil((9 + 1) / 2) = ceil(10/2) = 5 bytes
assert pic_byte_length("S9(7)V99", usage="COMP-3") == 5
def test_s9_5_comp3(self) -> None:
# 5 digits → ceil((5 + 1) / 2) = 3 bytes
assert pic_byte_length("S9(5)", usage="COMP-3") == 3
def test_s9_3_comp3(self) -> None:
# 3 digits → ceil((3 + 1) / 2) = 2 bytes
assert pic_byte_length("S9(3)", usage="COMP-3") == 2
def test_s9_1_comp3(self) -> None:
# 1 digit → ceil((1 + 1) / 2) = 1 byte
assert pic_byte_length("S9(1)", usage="COMP-3") == 1
def test_9_9_comp3(self) -> None:
# 9 digits → ceil((9 + 1) / 2) = 5 bytes
assert pic_byte_length("9(9)", usage="COMP-3") == 5
def test_9_10_comp3(self) -> None:
# 10 digits → ceil((10 + 1) / 2) = ceil(5.5) = 6 bytes
assert pic_byte_length("9(10)", usage="COMP-3") == 6
def test_s9_15_v99_comp3(self) -> None:
# 15 + 2 = 17 digits → ceil((17 + 1) / 2) = ceil(9) = 9 bytes
assert pic_byte_length("S9(15)V99", usage="COMP-3") == 9
def test_comp3_always_at_least_one_byte(self) -> None:
result = pic_byte_length("9(1)", usage="COMP-3")
assert result >= 1
def test_formula_matches_math(self) -> None:
"""Verify the formula matches ceil((n+1)/2) for several digit counts."""
cases = [
("S9(1)", 1),
("S9(3)", 2),
("S9(5)", 3),
("S9(7)", 4),
("S9(9)", 5),
("S9(11)", 6),
]
for pic, expected in cases:
n = int(pic[3:-1])
assert math.ceil((n + 1) / 2) == expected, f"Math check for {pic}"
assert pic_byte_length(pic, usage="COMP-3") == expected, (
f"pic_byte_length({pic!r}, 'COMP-3') should be {expected}"
)
class TestPicByteLengthComp:
"""COMP / COMP-4 (binary): size depends on digit count."""
def test_1_digit_comp(self) -> None:
# 1-4 digits → 2 bytes
assert pic_byte_length("9(1)", usage="COMP") == 2
def test_4_digit_comp(self) -> None:
assert pic_byte_length("9(4)", usage="COMP") == 2
def test_5_digit_comp(self) -> None:
# 5-9 digits → 4 bytes
assert pic_byte_length("9(5)", usage="COMP") == 4
def test_9_digit_comp(self) -> None:
assert pic_byte_length("9(9)", usage="COMP") == 4
def test_10_digit_comp(self) -> None:
# 10-18 digits → 8 bytes
assert pic_byte_length("9(10)", usage="COMP") == 8
def test_18_digit_comp(self) -> None:
assert pic_byte_length("9(18)", usage="COMP") == 8
def test_comp4_same_as_comp(self) -> None:
assert pic_byte_length("9(4)", usage="COMP-4") == pic_byte_length(
"9(4)", usage="COMP"
)
def test_comp4_5_digit(self) -> None:
assert pic_byte_length("9(5)", usage="COMP-4") == 4
def test_signed_comp(self) -> None:
assert pic_byte_length("S9(4)", usage="COMP") == 2
def test_signed_comp_5_digit(self) -> None:
assert pic_byte_length("S9(5)", usage="COMP") == 4
# ── CopybookField model ───────────────────────────────────────────────────────
class TestCopybookField:
"""CopybookField is a Pydantic intermediate representation."""
def test_basic_construction(self) -> None:
field = CopybookField(
level=5,
name="claim_id",
pic="X(15)",
)
assert field.level == 5
assert field.name == "claim_id"
assert field.pic == "X(15)"
assert field.usage == ""
assert field.redefines == ""
assert field.byte_length == 0
assert field.offset == 0
def test_with_usage(self) -> None:
field = CopybookField(
level=5,
name="paid_amount",
pic="S9(7)V99",
usage="COMP-3",
)
assert field.usage == "COMP-3"
def test_with_redefines(self) -> None:
field = CopybookField(
level=5,
name="alt_view",
pic="X(5)",
redefines="original_field",
)
assert field.redefines == "original_field"
def test_with_computed_offset(self) -> None:
field = CopybookField(
level=5, name="npi", pic="X(10)", byte_length=10, offset=15
)
assert field.byte_length == 10
assert field.offset == 15
def test_level_01_group(self) -> None:
field = CopybookField(level=1, name="claim_record", pic="")
assert field.level == 1
assert field.pic == ""
# ── parse_copybook ────────────────────────────────────────────────────────────
class TestParseCopybook:
"""parse_copybook turns copybook text into a FieldMap with byte positions."""
@pytest.fixture
def simple_copybook(self) -> str:
return """
01 CLAIM-RECORD.
05 CLAIM-ID PIC X(10).
05 PERSON-ID PIC X(10).
05 SERVICE-DATE PIC 9(8).
"""
@pytest.fixture
def comp3_copybook(self) -> str:
return """
01 PAYMENT-RECORD.
05 CLAIM-ID PIC X(15).
05 PROVIDER-NPI PIC X(10).
05 SERVICE-DATE PIC 9(8).
05 PAID-AMOUNT PIC S9(7)V99 COMP-3.
05 FILLER PIC X(5).
05 DX-CODE PIC X(7).
"""
@pytest.fixture
def filler_copybook(self) -> str:
return """
01 MMR-RECORD.
05 BENE-HIC-NUM PIC X(12).
05 FILLER PIC X(3).
05 CLM-FROM-DT PIC 9(8).
05 FILLER PIC X(2).
05 PAID-AMT PIC S9(9)V99 COMP-3.
"""
@pytest.fixture
def redefines_copybook(self) -> str:
return """
01 DATE-RECORD.
05 DATE-PACKED PIC S9(7) COMP-3.
05 DATE-DISPLAY REDEFINES DATE-PACKED PIC X(4).
05 EXTRA-FIELD PIC X(10).
"""
# ── return type ───────────────────────────────────────────────────────────
def test_returns_field_map(self, simple_copybook: str) -> None:
result = parse_copybook(simple_copybook)
assert isinstance(result, FieldMap)
def test_positions_is_dict(self, simple_copybook: str) -> None:
fm = parse_copybook(simple_copybook)
assert isinstance(fm.positions, dict)
# ── field names ───────────────────────────────────────────────────────────
def test_field_names_are_snake_case(self, simple_copybook: str) -> None:
fm = parse_copybook(simple_copybook)
for name in fm.positions:
assert "-" not in name, f"Field name {name!r} contains a hyphen"
assert name == name.lower(), f"Field name {name!r} is not lowercase"
def test_claim_id_in_positions(self, simple_copybook: str) -> None:
fm = parse_copybook(simple_copybook)
assert "claim_id" in fm.positions
def test_person_id_in_positions(self, simple_copybook: str) -> None:
fm = parse_copybook(simple_copybook)
assert "person_id" in fm.positions
def test_service_date_in_positions(self, simple_copybook: str) -> None:
fm = parse_copybook(simple_copybook)
assert "service_date" in fm.positions
def test_group_item_not_in_positions(self, simple_copybook: str) -> None:
# The 01-level group "CLAIM-RECORD" has no PIC, so it should not appear
fm = parse_copybook(simple_copybook)
assert "claim_record" not in fm.positions
# ── byte positions — simple layout ────────────────────────────────────────
def test_claim_id_starts_at_zero(self, simple_copybook: str) -> None:
fm = parse_copybook(simple_copybook)
start, _ = fm.positions["claim_id"]
assert start == 0
def test_claim_id_ends_at_10(self, simple_copybook: str) -> None:
fm = parse_copybook(simple_copybook)
_, end = fm.positions["claim_id"]
assert end == 10
def test_person_id_starts_after_claim_id(self, simple_copybook: str) -> None:
fm = parse_copybook(simple_copybook)
_, claim_end = fm.positions["claim_id"]
person_start, _ = fm.positions["person_id"]
assert person_start == claim_end
def test_service_date_starts_at_20(self, simple_copybook: str) -> None:
fm = parse_copybook(simple_copybook)
start, _ = fm.positions["service_date"]
assert start == 20
def test_service_date_ends_at_28(self, simple_copybook: str) -> None:
fm = parse_copybook(simple_copybook)
_, end = fm.positions["service_date"]
assert end == 28
def test_positions_are_non_overlapping(self, simple_copybook: str) -> None:
fm = parse_copybook(simple_copybook)
sorted_fields = sorted(fm.positions.values(), key=lambda p: p[0])
for i in range(len(sorted_fields) - 1):
_, end_a = sorted_fields[i]
start_b, _ = sorted_fields[i + 1]
assert end_a <= start_b, (
f"Overlapping positions: {sorted_fields[i]} and {sorted_fields[i + 1]}"
)
def test_positions_are_valid_tuples(self, simple_copybook: str) -> None:
fm = parse_copybook(simple_copybook)
for name, pos in fm.positions.items():
assert isinstance(pos, tuple), f"{name}: position is not a tuple"
assert len(pos) == 2, f"{name}: position tuple does not have 2 elements"
start, end = pos
assert isinstance(start, int), f"{name}: start is not int"
assert isinstance(end, int), f"{name}: end is not int"
assert start >= 0, f"{name}: start < 0"
assert end > start, f"{name}: end <= start"
# ── FILLER handling ───────────────────────────────────────────────────────
def test_filler_not_in_positions(self, filler_copybook: str) -> None:
fm = parse_copybook(filler_copybook)
assert "filler" not in fm.positions
def test_filler_contributes_to_offset(self, filler_copybook: str) -> None:
"""CLM-FROM-DT starts at offset 15 (12 bene_hic_num + 3 filler)."""
fm = parse_copybook(filler_copybook)
start, _ = fm.positions["clm_from_dt"]
assert start == 15
def test_multiple_fillers_accumulate(self, filler_copybook: str) -> None:
"""PAID-AMT starts at 12 + 3 + 8 + 2 = 25."""
fm = parse_copybook(filler_copybook)
start, _ = fm.positions["paid_amt"]
assert start == 25
# ── COMP-3 byte sizes ─────────────────────────────────────────────────────
def test_comp3_field_correct_byte_size(self, comp3_copybook: str) -> None:
"""PAID-AMOUNT S9(7)V99 COMP-3 → 5 bytes."""
fm = parse_copybook(comp3_copybook)
start, end = fm.positions["paid_amount"]
assert (end - start) == 5
def test_comp3_offsets_are_sequential(self, comp3_copybook: str) -> None:
"""Fields after COMP-3 use the packed size, not the display size."""
fm = parse_copybook(comp3_copybook)
# CLAIM-ID(0-15) + PROVIDER-NPI(15-25) + SERVICE-DATE(25-33) + PAID-AMOUNT(33-38)
paid_start, paid_end = fm.positions["paid_amount"]
assert paid_start == 33
assert paid_end == 38
def test_dx_code_starts_after_filler(self, comp3_copybook: str) -> None:
"""DX-CODE starts at offset 43 (15+10+8+5+5 filler)."""
fm = parse_copybook(comp3_copybook)
start, end = fm.positions["dx_code"]
assert start == 43
assert end == 50
# ── REDEFINES handling ────────────────────────────────────────────────────
def test_redefines_field_included_in_positions(
self, redefines_copybook: str
) -> None:
"""REDEFINES fields should appear in positions alongside the original."""
fm = parse_copybook(redefines_copybook)
assert "date_packed" in fm.positions or "date_display" in fm.positions
def test_redefines_same_offset_as_original(self, redefines_copybook: str) -> None:
"""REDEFINES field must share the same start offset as the field it redefines."""
fm = parse_copybook(redefines_copybook)
if "date_packed" in fm.positions and "date_display" in fm.positions:
packed_start, _ = fm.positions["date_packed"]
display_start, _ = fm.positions["date_display"]
assert packed_start == display_start
def test_field_after_redefines_correct_offset(
self, redefines_copybook: str
) -> None:
"""EXTRA-FIELD must start after the REDEFINES group, not be shifted by it."""
fm = parse_copybook(redefines_copybook)
# date_packed S9(7) COMP-3 → ceil((7+1)/2) = 4 bytes
# extra_field starts at offset 4
if "extra_field" in fm.positions:
start, _ = fm.positions["extra_field"]
assert start == 4
# ── general contract ──────────────────────────────────────────────────────
def test_empty_copybook_returns_empty_field_map(self) -> None:
fm = parse_copybook("")
assert isinstance(fm, FieldMap)
assert fm.positions == {}
def test_whitespace_only_copybook_returns_empty_field_map(self) -> None:
fm = parse_copybook(" \n\n\t \n")
assert isinstance(fm, FieldMap)
assert fm.positions == {}
def test_copybook_with_only_group_item_returns_empty_positions(self) -> None:
"""A copybook with only a 01-level group and no PIC fields has no positions."""
text = "01 EMPTY-RECORD.\n"
fm = parse_copybook(text)
assert "empty_record" not in fm.positions
def test_parse_copybook_from_fixture(self, copybook_text: str) -> None:
"""parse_copybook successfully parses the conftest.copybook_text fixture."""
fm = parse_copybook(copybook_text)
assert isinstance(fm, FieldMap)
assert len(fm.positions) > 0
def test_fixture_copybook_field_names(self, copybook_text: str) -> None:
"""conftest copybook has claim_id, person_id, service_date, dx_code."""
fm = parse_copybook(copybook_text)
for name in ("claim_id", "person_id", "service_date"):
assert name in fm.positions, f"Expected {name!r} in positions"
def test_fixture_copybook_filler_absent(self, copybook_text: str) -> None:
fm = parse_copybook(copybook_text)
assert "filler" not in fm.positions
def test_all_field_ends_positive(self, simple_copybook: str) -> None:
fm = parse_copybook(simple_copybook)
for name, (start, end) in fm.positions.items():
assert end > 0, f"{name}: end <= 0"
def test_sequential_monotone_offsets(self, simple_copybook: str) -> None:
"""Fields must be in non-decreasing start-offset order."""
fm = parse_copybook(simple_copybook)
starts = sorted(start for start, _ in fm.positions.values())
assert starts == sorted(starts)
def test_field_map_delimiter_empty_for_copybook(self, simple_copybook: str) -> None:
"""parse_copybook always returns fixed-width mode — delimiter must be empty."""
fm = parse_copybook(simple_copybook)
assert fm.delimiter == ""
def test_field_map_indices_empty_for_copybook(self, simple_copybook: str) -> None:
"""parse_copybook always returns fixed-width mode — indices must be empty."""
fm = parse_copybook(simple_copybook)
assert fm.indices == {}
def test_skips_non_matching_lines(self) -> None:
"""Lines that don't start with level+field (comments, directives) are skipped."""
text = (
"* THIS IS A COMMENT\n"
" COPY SOME-OTHER-BOOK.\n"
"01 MY-RECORD.\n"
" 05 MY-FIELD PIC X(10).\n"
)
fm = parse_copybook(text)
assert "my_field" in fm.positions