8f99eeb154
- 5519 unit tests covering all modules (aco, bcda, bls, cms, pfs, rex, bib) - ruff lint + format enforcement across entire codebase (377 files reformatted) - pre-commit hook: ruff check, ruff format, pytest - Woodpecker CI split into ci.yml (quality gate) and deploy.yml (package + images) - ci.yml: lint → test → validate-compose, runs on every push/PR - deploy.yml: build + publish Python package to Gitea PyPI registry, then container image builds, Trivy scans, and registry push (main branch only) - Gitea branch protection on main: requires CI status checks to pass - .gitignore updated for .coverage, dist/, *.egg-info/ - grafana config moved to dev/grafana/ Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
487 lines
18 KiB
Python
487 lines
18 KiB
Python
"""Tests for rex.sieve — Zone, Sieve, classify(), sift(), FieldMap."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Iterator
|
|
|
|
import pytest
|
|
|
|
from rex.sieve import FieldMap, Sieve, Zone, classify, sift
|
|
|
|
# ── Zone ──────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestZone:
|
|
def test_zone_values_are_strings(self) -> None:
|
|
assert Zone.HEAD == "head"
|
|
assert Zone.RULE == "rule"
|
|
assert Zone.DATA == "data"
|
|
assert Zone.SKIP == "skip"
|
|
|
|
def test_zone_is_str_enum(self) -> None:
|
|
assert isinstance(Zone.HEAD, str)
|
|
assert isinstance(Zone.DATA, str)
|
|
|
|
def test_zone_members(self) -> None:
|
|
members = set(Zone)
|
|
assert members == {Zone.HEAD, Zone.RULE, Zone.DATA, Zone.SKIP}
|
|
|
|
|
|
# ── Sieve construction ────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestSieveConstruction:
|
|
def test_minimal_sieve_data_only(self) -> None:
|
|
s = Sieve(name="minimal", data=re.compile(r"^\d"))
|
|
assert s.name == "minimal"
|
|
assert s.head is None
|
|
assert s.rule is None
|
|
assert s.skip is None
|
|
assert s.fields is None
|
|
assert s.encoding == "utf-8"
|
|
|
|
def test_full_sieve(self) -> None:
|
|
s = Sieve(
|
|
name="full",
|
|
head=re.compile(r"^HEAD"),
|
|
rule=re.compile(r"^---"),
|
|
data=re.compile(r"^\d"),
|
|
skip=re.compile(r"^\s*$"),
|
|
)
|
|
assert s.head is not None
|
|
assert s.rule is not None
|
|
assert s.data is not None
|
|
assert s.skip is not None
|
|
|
|
def test_custom_encoding(self) -> None:
|
|
s = Sieve(name="ebcdic", data=re.compile(r"."), encoding="cp037")
|
|
assert s.encoding == "cp037"
|
|
|
|
def test_sieve_with_fields(self, fixed_width_field_map: FieldMap) -> None:
|
|
s = Sieve(
|
|
name="with_fields", data=re.compile(r"."), fields=fixed_width_field_map
|
|
)
|
|
assert s.fields is not None
|
|
assert "claim_id" in s.fields.positions
|
|
|
|
def test_sieve_name_preserved(self) -> None:
|
|
s = Sieve(name="my_format", data=re.compile(r"^X"))
|
|
assert s.name == "my_format"
|
|
|
|
|
|
# ── classify() ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestClassify:
|
|
@pytest.fixture
|
|
def full_sieve(self) -> Sieve:
|
|
return Sieve(
|
|
name="test",
|
|
head=re.compile(r"^The SAS System", re.IGNORECASE),
|
|
rule=re.compile(r"^\s+Obs\s+"),
|
|
data=re.compile(r"^\s+\d+\s+\w"),
|
|
skip=re.compile(r"^\s*(Total|NOTE:|$)"),
|
|
)
|
|
|
|
def test_head_line_classified_as_head(self, full_sieve: Sieve) -> None:
|
|
assert classify("The SAS System 10:30 AM", full_sieve) == Zone.HEAD
|
|
|
|
def test_rule_line_classified_as_rule(self, full_sieve: Sieve) -> None:
|
|
assert classify(" Obs claim_id paid_amount", full_sieve) == Zone.RULE
|
|
|
|
def test_data_line_classified_as_data(self, full_sieve: Sieve) -> None:
|
|
assert classify(" 1 CLM001 75.00", full_sieve) == Zone.DATA
|
|
|
|
def test_skip_line_classified_as_skip(self, full_sieve: Sieve) -> None:
|
|
assert classify(" Total: 450.00", full_sieve) == Zone.SKIP
|
|
|
|
def test_blank_line_classified_as_skip(self, full_sieve: Sieve) -> None:
|
|
assert classify("", full_sieve) == Zone.SKIP
|
|
|
|
def test_unmatched_line_defaults_to_skip(self, full_sieve: Sieve) -> None:
|
|
# A line that matches no pattern should default to SKIP
|
|
assert classify("??GARBAGE??LINE", full_sieve) == Zone.SKIP
|
|
|
|
def test_head_takes_priority_over_data(self) -> None:
|
|
"""head pattern tested before data — head wins when both match."""
|
|
s = Sieve(
|
|
name="priority",
|
|
head=re.compile(r"^X"),
|
|
data=re.compile(r"^X"), # same pattern
|
|
)
|
|
assert classify("XDATA", s) == Zone.HEAD
|
|
|
|
def test_skip_takes_priority_over_data(self) -> None:
|
|
"""skip pattern tested before data — skip wins when both match."""
|
|
s = Sieve(
|
|
name="priority_skip",
|
|
skip=re.compile(r"^\d"),
|
|
data=re.compile(r"^\d"),
|
|
)
|
|
assert classify("123 data", s) == Zone.SKIP
|
|
|
|
def test_data_only_sieve(self) -> None:
|
|
s = Sieve(name="data_only", data=re.compile(r"^\d"))
|
|
assert classify("123 abc", s) == Zone.DATA
|
|
assert classify("abc 123", s) == Zone.SKIP # no match → skip
|
|
|
|
def test_none_patterns_do_not_raise(self) -> None:
|
|
s = Sieve(name="sparse", data=re.compile(r"^\d"))
|
|
# Should not raise even if head/rule/skip are None
|
|
assert classify("not a digit line", s) == Zone.SKIP
|
|
|
|
|
|
# ── sift() ───────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestSift:
|
|
@pytest.fixture
|
|
def sas_sieve(self) -> Sieve:
|
|
return Sieve(
|
|
name="sas",
|
|
head=re.compile(r"^\s*(The SAS System|\x0c)"),
|
|
rule=re.compile(r"^\s+Obs\s+"),
|
|
data=re.compile(r"^\s+\d+\s+\w"),
|
|
skip=re.compile(r"^\s*(Total|NOTE:|$)"),
|
|
)
|
|
|
|
def _lines(self, text: str) -> Iterator[str]:
|
|
for line in text.splitlines():
|
|
yield line
|
|
|
|
def test_sift_yields_correct_zones(self, sas_sieve: Sieve) -> None:
|
|
text = (
|
|
"The SAS System 10:30 AM\n"
|
|
"\n"
|
|
" Obs claim_id paid_amount\n"
|
|
" 1 CLM001 75.00\n"
|
|
" 2 CLM002 150.00\n"
|
|
" Total: 225.00\n"
|
|
)
|
|
results = list(sift(self._lines(text), sas_sieve))
|
|
zones = [z for z, _ in results]
|
|
|
|
assert Zone.HEAD in zones
|
|
assert Zone.RULE in zones
|
|
assert Zone.DATA in zones
|
|
assert Zone.SKIP in zones
|
|
|
|
def test_sift_data_lines_count(self, sas_sieve: Sieve) -> None:
|
|
text = (
|
|
"The SAS System\n"
|
|
" Obs claim_id\n"
|
|
" 1 CLM001\n"
|
|
" 2 CLM002\n"
|
|
" 3 CLM003\n"
|
|
" Total:\n"
|
|
)
|
|
data_lines = [
|
|
line
|
|
for zone, line in sift(self._lines(text), sas_sieve)
|
|
if zone == Zone.DATA
|
|
]
|
|
assert len(data_lines) == 3
|
|
|
|
def test_sift_empty_input_yields_nothing(self, sas_sieve: Sieve) -> None:
|
|
results = list(sift(iter([]), sas_sieve))
|
|
assert results == []
|
|
|
|
def test_sift_all_skip_file(self, sas_sieve: Sieve) -> None:
|
|
text = "\n\n\nNOTE: something\nNOTE: else\n"
|
|
results = list(sift(self._lines(text), sas_sieve))
|
|
for zone, _ in results:
|
|
assert zone == Zone.SKIP
|
|
|
|
def test_sift_preserves_data_content(self, sas_sieve: Sieve) -> None:
|
|
text = "The SAS System\n Obs id\n 1 HELLO\n"
|
|
data_lines = [
|
|
line
|
|
for zone, line in sift(self._lines(text), sas_sieve)
|
|
if zone == Zone.DATA
|
|
]
|
|
assert len(data_lines) == 1
|
|
assert "HELLO" in data_lines[0]
|
|
|
|
def test_sift_yields_tuples_of_zone_and_str(self, sas_sieve: Sieve) -> None:
|
|
text = " 1 CLM001\n"
|
|
results = list(sift(self._lines(text), sas_sieve))
|
|
for item in results:
|
|
assert len(item) == 2
|
|
zone, line = item
|
|
assert isinstance(zone, Zone)
|
|
assert isinstance(line, str)
|
|
|
|
def test_sift_builds_field_map_from_ruler_line(self) -> None:
|
|
"""When sieve.fields is None, sift should auto-build FieldMap from first rule line."""
|
|
s = Sieve(
|
|
name="automap",
|
|
rule=re.compile(r"^\s+Obs\s+"),
|
|
data=re.compile(r"^\s+\d+\s+\w"),
|
|
)
|
|
text = " Obs claim_id paid_amount\n 1 CLM001 75.00\n"
|
|
# Should not raise — FieldMap detection is part of sift's contract
|
|
results = list(sift(self._lines(text), s))
|
|
data_lines = [l for z, l in results if z == Zone.DATA]
|
|
assert len(data_lines) == 1
|
|
|
|
def test_sift_uses_explicit_field_map_when_provided(
|
|
self, fixed_width_field_map: FieldMap
|
|
) -> None:
|
|
s = Sieve(
|
|
name="explicit",
|
|
data=re.compile(r"^CLM"),
|
|
fields=fixed_width_field_map,
|
|
)
|
|
text = "CLM001 P001 20240115000007500\n"
|
|
results = list(sift(iter([text.strip()]), s))
|
|
assert len(results) == 1
|
|
assert results[0][0] == Zone.DATA
|
|
# fields should still be the provided map
|
|
assert s.fields is fixed_width_field_map
|
|
|
|
|
|
# ── FieldMap ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestFieldMap:
|
|
# ── construction ──────────────────────────────────────────────────────────
|
|
|
|
def test_empty_field_map(self) -> None:
|
|
fm = FieldMap()
|
|
assert fm.positions == {}
|
|
assert fm.delimiter == ""
|
|
assert fm.indices == {}
|
|
|
|
def test_fixed_width_field_map_construction(
|
|
self, fixed_width_field_map: FieldMap
|
|
) -> None:
|
|
assert "claim_id" in fixed_width_field_map.positions
|
|
assert "person_id" in fixed_width_field_map.positions
|
|
assert fixed_width_field_map.positions["claim_id"] == (0, 10)
|
|
|
|
def test_delimited_field_map_construction(
|
|
self, delimited_field_map: FieldMap
|
|
) -> None:
|
|
assert delimited_field_map.delimiter == "|"
|
|
assert delimited_field_map.indices["claim_id"] == 0
|
|
assert delimited_field_map.indices["paid_amount"] == 3
|
|
|
|
# ── extract() — fixed-width mode ──────────────────────────────────────────
|
|
|
|
def test_extract_fixed_width_basic(self, fixed_width_field_map: FieldMap) -> None:
|
|
# "CLM001 P001 20240115000007500"
|
|
# positions: claim_id=(0,10), person_id=(10,20), service_date=(20,28), paid_amount=(28,37)
|
|
line = "CLM001 P001 20240115000007500"
|
|
result = fixed_width_field_map.extract(line)
|
|
assert "claim_id" in result
|
|
assert result["claim_id"].strip() == "CLM001"
|
|
|
|
def test_extract_fixed_width_person_id(
|
|
self, fixed_width_field_map: FieldMap
|
|
) -> None:
|
|
line = "CLM001 P001 20240115000007500"
|
|
result = fixed_width_field_map.extract(line)
|
|
assert result["person_id"].strip() == "P001"
|
|
|
|
def test_extract_fixed_width_service_date(
|
|
self, fixed_width_field_map: FieldMap
|
|
) -> None:
|
|
line = "CLM001 P001 20240115000007500"
|
|
result = fixed_width_field_map.extract(line)
|
|
assert result["service_date"] == "20240115"
|
|
|
|
def test_extract_fixed_width_returns_dict(
|
|
self, fixed_width_field_map: FieldMap
|
|
) -> None:
|
|
line = "CLM001 P001 20240115000007500"
|
|
result = fixed_width_field_map.extract(line)
|
|
assert isinstance(result, dict)
|
|
assert set(result.keys()) == {
|
|
"claim_id",
|
|
"person_id",
|
|
"service_date",
|
|
"paid_amount",
|
|
}
|
|
|
|
def test_extract_fixed_width_short_line(
|
|
self, fixed_width_field_map: FieldMap
|
|
) -> None:
|
|
"""Extraction on a short line should not raise — fields beyond end are empty."""
|
|
line = "CLM001"
|
|
result = fixed_width_field_map.extract(line)
|
|
assert "claim_id" in result
|
|
# Fields that extend beyond the line length should be empty string
|
|
assert (
|
|
result.get("service_date", "") == ""
|
|
or result.get("service_date") is not None
|
|
)
|
|
|
|
def test_extract_fixed_width_multiple_records(
|
|
self, fixed_width_field_map: FieldMap, fixed_width_text: str
|
|
) -> None:
|
|
lines = fixed_width_text.strip().splitlines()
|
|
results = [fixed_width_field_map.extract(line) for line in lines]
|
|
assert len(results) == 3
|
|
claim_ids = [r["claim_id"].strip() for r in results]
|
|
assert claim_ids == ["CLM001", "CLM002", "CLM003"]
|
|
|
|
# ── extract() — delimited mode ────────────────────────────────────────────
|
|
|
|
def test_extract_delimited_basic(self, delimited_field_map: FieldMap) -> None:
|
|
line = "CLM001|P001|20240115|75.00"
|
|
result = delimited_field_map.extract(line)
|
|
assert result["claim_id"] == "CLM001"
|
|
assert result["person_id"] == "P001"
|
|
assert result["paid_amount"] == "75.00"
|
|
|
|
def test_extract_delimited_multiple_rows(
|
|
self, delimited_field_map: FieldMap, pipe_delimited_text: str
|
|
) -> None:
|
|
data_lines = pipe_delimited_text.strip().splitlines()[1:] # skip header
|
|
results = [delimited_field_map.extract(line) for line in data_lines]
|
|
assert len(results) == 2
|
|
assert results[0]["claim_id"] == "CLM001"
|
|
assert results[1]["claim_id"] == "CLM002"
|
|
|
|
def test_extract_delimited_returns_dict(
|
|
self, delimited_field_map: FieldMap
|
|
) -> None:
|
|
line = "CLM001|P001|20240115|75.00"
|
|
result = delimited_field_map.extract(line)
|
|
assert isinstance(result, dict)
|
|
|
|
# ── from_ruler() ──────────────────────────────────────────────────────────
|
|
|
|
def test_from_ruler_basic(self) -> None:
|
|
ruler = " Obs claim_id paid_amount diag_code"
|
|
fm = FieldMap.from_ruler(ruler)
|
|
assert isinstance(fm, FieldMap)
|
|
assert "claim_id" in fm.positions or len(fm.positions) > 0
|
|
|
|
def test_from_ruler_returns_field_map(self) -> None:
|
|
ruler = " Obs id amount"
|
|
fm = FieldMap.from_ruler(ruler)
|
|
assert isinstance(fm, FieldMap)
|
|
|
|
def test_from_ruler_positions_are_tuples(self) -> None:
|
|
ruler = " Obs claim_id paid_amount"
|
|
fm = FieldMap.from_ruler(ruler)
|
|
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 should have 2 elements"
|
|
start, end = pos
|
|
assert start >= 0
|
|
assert end > start
|
|
|
|
def test_from_ruler_non_overlapping_positions(self) -> None:
|
|
ruler = " Obs claim_id paid_amount diag_code"
|
|
fm = FieldMap.from_ruler(ruler)
|
|
positions = sorted(fm.positions.values(), key=lambda p: p[0])
|
|
for i in range(len(positions) - 1):
|
|
_, end_a = positions[i]
|
|
start_b, _ = positions[i + 1]
|
|
assert end_a <= start_b, (
|
|
f"Overlapping positions detected: {positions[i]} and {positions[i + 1]}"
|
|
)
|
|
|
|
def test_from_ruler_column_names_stripped(self) -> None:
|
|
ruler = " Obs claim_id paid_amount"
|
|
fm = FieldMap.from_ruler(ruler)
|
|
for name in fm.positions:
|
|
assert name == name.strip(), f"Column name {name!r} has whitespace"
|
|
|
|
def test_from_ruler_with_sieve_integration(self) -> None:
|
|
"""FieldMap.from_ruler result can be used in a Sieve."""
|
|
ruler = " Obs claim_id paid_amount"
|
|
fm = FieldMap.from_ruler(ruler)
|
|
s = Sieve(name="dynamic", data=re.compile(r"^\s+\d"), fields=fm)
|
|
assert s.fields is fm
|
|
|
|
|
|
# ── Integration: classify + sift + extract ────────────────────────────────────
|
|
|
|
|
|
class TestSieveIntegration:
|
|
def test_full_pipeline_sas_listing(self, sas_listing_text: str) -> None:
|
|
"""Classify, sift, and extract a complete SAS listing sample."""
|
|
sieve = Sieve(
|
|
name="sas",
|
|
head=re.compile(r"^\s*(The SAS System|\x0c)"),
|
|
rule=re.compile(r"^\s+Obs\s+"),
|
|
data=re.compile(r"^\s+\d+\s+\w"),
|
|
skip=re.compile(r"^\s*(Total|NOTE:|$)"),
|
|
)
|
|
field_map = FieldMap(
|
|
positions={
|
|
"obs": (0, 5),
|
|
"claim_id": (5, 15),
|
|
"paid_amount": (15, 30),
|
|
"diag_code": (30, 40),
|
|
}
|
|
)
|
|
sieve = Sieve(
|
|
name="sas",
|
|
head=re.compile(r"^\s*(The SAS System|\x0c)"),
|
|
rule=re.compile(r"^\s+Obs\s+"),
|
|
data=re.compile(r"^\s+\d+\s+\w"),
|
|
skip=re.compile(r"^\s*(Total|NOTE:|$)"),
|
|
fields=field_map,
|
|
)
|
|
|
|
lines = sas_listing_text.splitlines()
|
|
tagged = list(sift(iter(lines), sieve))
|
|
|
|
data_records = [field_map.extract(line) for z, line in tagged if z == Zone.DATA]
|
|
assert len(data_records) == 3
|
|
|
|
def test_pipe_delimited_full_pipeline(self, pipe_delimited_text: str) -> None:
|
|
"""Classify and extract from a pipe-delimited file."""
|
|
sieve = Sieve(
|
|
name="pipe",
|
|
rule=re.compile(r"^claim_id\|"),
|
|
data=re.compile(r"^CLM"),
|
|
)
|
|
fm = FieldMap(
|
|
delimiter="|", indices={"claim_id": 0, "person_id": 1, "paid_amount": 3}
|
|
)
|
|
|
|
lines = pipe_delimited_text.strip().splitlines()
|
|
results = []
|
|
for zone, line in sift(iter(lines), sieve):
|
|
if zone == Zone.DATA:
|
|
results.append(fm.extract(line))
|
|
|
|
assert len(results) == 2
|
|
assert results[0]["claim_id"] == "CLM001"
|
|
assert results[1]["claim_id"] == "CLM002"
|
|
|
|
def test_classify_priority_order(self) -> None:
|
|
"""Priority: head > skip > rule > data."""
|
|
# A line matching all four — head should win
|
|
s = Sieve(
|
|
name="priority",
|
|
head=re.compile(r"^A"),
|
|
rule=re.compile(r"^A"),
|
|
data=re.compile(r"^A"),
|
|
skip=re.compile(r"^A"),
|
|
)
|
|
assert classify("AAAA", s) == Zone.HEAD
|
|
|
|
# Without head — skip should win
|
|
s2 = Sieve(
|
|
name="priority2",
|
|
rule=re.compile(r"^A"),
|
|
data=re.compile(r"^A"),
|
|
skip=re.compile(r"^A"),
|
|
)
|
|
assert classify("AAAA", s2) == Zone.SKIP
|
|
|
|
# Without skip — rule should win
|
|
s3 = Sieve(
|
|
name="priority3",
|
|
rule=re.compile(r"^A"),
|
|
data=re.compile(r"^A"),
|
|
)
|
|
assert classify("AAAA", s3) == Zone.RULE
|