Files
stack/tests/bcda/test_pipe_cclf.py
kert 52ebad29cd
Some checks failed
CI / lint (push) Successful in 31s
CI / test (push) Failing after 2m16s
Infra CI / notebooks (push) Successful in 53s
Infra CI / docs (push) Successful in 1m30s
Infra CI / api (push) Successful in 1m0s
Infra CI / mc (push) Failing after 32s
Infra CI / llm (push) Successful in 47s
CI / notebooks-smoke (push) Successful in 1m40s
Infra CI / zotero (push) Successful in 16s
CI / lint (pull_request) Successful in 47s
Infra CI / zotero (pull_request) Successful in 14s
Infra CI / docs (pull_request) Successful in 19s
Infra CI / notebooks (pull_request) Successful in 56s
Infra CI / api (pull_request) Successful in 16s
CI / notebooks-smoke (pull_request) Successful in 1m33s
CI / test (pull_request) Failing after 2m35s
Infra CI / mc (pull_request) Failing after 32s
Infra CI / llm (pull_request) Successful in 14s
feat(bcda): rewrite the input_layer connector for v3 semantics
The connector (bcda.express.cclf / bcda.pipe.cclf) still routed on the v2
near-line RIC code, read v2-only amount columns, and had never run end to
end: its contracts named CCLF columns the EOB table does not have, so the
runner's exact-match check would have refused every stage.

One header stage now carries the v3 semantics. _stg_claim_header drops
NotFinalAction claims (9% of the sandbox), derives claim_family from
EOB-TYPE → CLM-TYPE-CD → pde_id → RIC (v3 emits the RIC on only some
claims), collapses a claim seen in two sources on its claim control number
(NationalClaimsHistory/DDPS over SharedSystem, then newest last_updated;
rows without a control number are never collapsed) and resolves the MBI.
The four family stages are filters on it.

One mapping produces input_layer.medical_claim for institutional, carrier
and DME lines: admission/discharge from the C4BB admission period, NUBC
type of bill (new clm_bill_type_cd on the EOB table; rebuilt as "0" + the
three-character v3 frequency code when the coding is missing — the v2
fac+clsfctn+freq concatenation yields "88787J" on v3), MS-DRG, revenue
centre, HCPCS + modifiers, attending→operating→rendering / rendering /
rendering→ordering NPIs, billing Organization NPI and tax number,
paid/allowed/charge/coinsurance/deductible from the consolidated v3
Adjudication amounts, paid date from clm_pd_dt → NCH weekly → CMS process.
Part D uses the PDE id, prescribing care-team NPI, contained pharmacy NPI,
NDC quantity, total drug cost as charge and paid-by-patient as copayment.
Eligibility reads the Part A row (Part B fallback) for status, entitlement
and enrollment, the Dual row for dual status, and accepts both CCW sex
codes and the FHIR words the v2 flattener passed through.

Contracts inherit: header stages subclass ExplanationOfBenefit, outputs
subclass aco.table.input_layer.InputLayer*, so downstream pipelines run on
BCDA output exactly as on CCLF output. load_bcda(run_pipeline=True),
stack load bcda --run-pipeline and stack run bcda land input_layer.*.

Large-entity sandbox pull: 221,021 medical lines on 28,282 claims (equal to
the FinalAction non-PDE EOBs by type), 11,968 pharmacy rows, 9,582
eligibility rows; all 16 reconciliation checks in the notebook pass. Tests
run the whole connector on CMS's official v3 samples through DuckDB.

refs #1139
2026-10-25 13:42:24 -04:00

155 lines
5.7 KiB
Python

"""Tests for bcda.pipe.cclf — lazy pipeline module."""
from __future__ import annotations
import sys
import pytest
class TestBuildPipeline:
def test_build_returns_pipeline(self) -> None:
"""_build_pipeline() returns a Pipeline with 13 exprs."""
from bcda.pipe.cclf import _build_pipeline
pipeline = _build_pipeline()
assert hasattr(pipeline, "exprs")
assert len(pipeline.exprs) == 13
assert hasattr(pipeline, "run")
def test_expr_names(self) -> None:
"""All 13 expected expression names are present."""
from bcda.pipe.cclf import _build_pipeline
pipeline = _build_pipeline()
names = [e.name for e in pipeline.exprs]
assert "bcda._stg_beneficiary_xref" in names
assert "bcda._stg_claim_header" in names
assert "bcda._stg_part_a_header" in names
assert "bcda._int_institutional_medical_claim" in names
assert "bcda._stg_part_b_physician_header" in names
assert "bcda._int_physician_medical_claim" in names
assert "bcda._stg_part_b_dme_header" in names
assert "bcda._int_dme_medical_claim" in names
assert "bcda.medical_claim" in names
assert "bcda._stg_part_d_header" in names
assert "bcda.pharmacy_claim" in names
assert "bcda._stg_beneficiary_demographics" in names
assert "bcda.eligibility" in names
class TestLazyGetattr:
def test_pipeline_attr(self) -> None:
"""Accessing .pipeline triggers lazy init."""
# Force re-import to test __getattr__.
mod_name = "bcda.pipe.cclf"
if mod_name in sys.modules:
mod = sys.modules[mod_name]
# Clear cached globals to force lazy init.
mod.__dict__.pop("pipeline", None)
mod.__dict__.pop("run", None)
import bcda.pipe.cclf as cclf_mod
pipeline = cclf_mod.pipeline
assert hasattr(pipeline, "exprs")
assert len(pipeline.exprs) == 13
def test_run_attr(self) -> None:
"""Accessing .run triggers lazy init and returns callable."""
mod_name = "bcda.pipe.cclf"
if mod_name in sys.modules:
mod = sys.modules[mod_name]
mod.__dict__.pop("pipeline", None)
mod.__dict__.pop("run", None)
import bcda.pipe.cclf as cclf_mod
run = cclf_mod.run
assert callable(run)
def test_unknown_attr_raises(self) -> None:
"""Accessing unknown attribute raises AttributeError."""
import bcda.pipe.cclf as cclf_mod
with pytest.raises(
AttributeError,
match="has no attribute 'nonexistent'",
):
_ = cclf_mod.nonexistent
class TestRunOnOfficialSamples:
"""The whole connector against CMS's v3 sample files through DuckDB.
Every expression's output is checked against its contract by the
runner, so this is the proof that the v3-flattened tables carry
what the connector reads and that the outputs are input_layer-shaped.
"""
def test_end_to_end(self, tmp_path) -> None:
import shutil
from pathlib import Path
import duckdb
from aco.lake.context import DuckDBContext
from aco.load.bcda import load_bcda
from bcda.pipe.cclf import pipeline
samples = Path(__file__).resolve().parents[2] / "dev" / "vendor" / "bcda-site"
src = tmp_path / "ndjson"
src.mkdir()
for rtype in ("Patient", "Coverage", "ExplanationOfBenefit"):
shutil.copy(samples / f"v3_{rtype}.ndjson", src / f"{rtype}.ndjson")
db = tmp_path / "aco.duckdb"
from unittest.mock import patch
with patch("conf.path", side_effect=lambda k: tmp_path / "store"):
load_bcda(ndjson_dir=src, database=str(db))
ctx = DuckDBContext(database=str(db), read_only=True)
cache = pipeline.run(ctx.load)
assert set(cache) >= {
"bcda._stg_claim_header",
"bcda.medical_claim",
"bcda.pharmacy_claim",
"bcda.eligibility",
}
header = cache["bcda._stg_claim_header"]
# 36 NotFinalAction sample claims are dropped; the rest are routed
assert len(header) == 233
assert set(header["claim_family"].to_list()) == {
"institutional",
"professional",
"dme",
"pharmacy",
}
medical = cache["bcda.medical_claim"]
assert medical["claim_id"].n_unique() == 131
assert set(medical["claim_type"].to_list()) == {"institutional", "professional"}
inst = medical.filter(medical["claim_type"] == "institutional")
assert inst["bill_type_code"].null_count() == 0
assert (
inst["bill_type_code"].str.len_chars().max() == 4
) # NUBC, not fac+clsfctn+freq
assert inst["admission_date"].null_count() == 0
assert inst["paid_amount"].null_count() == 0
assert inst["deductible_amount"].null_count() == 0
prof = medical.filter(medical["claim_type"] == "professional")
assert prof["place_of_service_code"].null_count() < len(prof) * 0.05
assert prof["rendering_npi"].null_count() == 0
assert len(cache["bcda.pharmacy_claim"]) == 102
rx = cache["bcda.pharmacy_claim"]
assert rx["ndc_code"].null_count() == 0 and rx["days_supply"].null_count() == 0
assert rx["charge_amount"].null_count() == 0 # total drug cost
elig = cache["bcda.eligibility"]
assert len(elig) == 94
assert set(elig["gender"].to_list()) == {"male", "female"}
assert elig["medicare_status_code"].null_count() < len(elig)
assert set(elig["enrollment_status"].drop_nulls().to_list()) <= {
"active",
"inactive",
}
duckdb.connect(str(db), read_only=True).close()