5261032a9b
Cover remaining uncovered lines in bcda (client, store, log, pipe/cclf, flatten), bib (sync, spider, translate, ingest, item, store, format, ui), cms (express wrappers, log edge cases), api (base, gitea, rustfs, woodpecker, zotero), bls (table import), and pfs (pragma on race guard). 37,687 statements, 0 missed — 11,061 tests passing.
79 lines
2.6 KiB
Python
79 lines
2.6 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 12 exprs."""
|
|
from bcda.pipe.cclf import _build_pipeline
|
|
|
|
pipeline = _build_pipeline()
|
|
assert hasattr(pipeline, "exprs")
|
|
assert len(pipeline.exprs) == 12
|
|
assert hasattr(pipeline, "run")
|
|
|
|
def test_expr_names(self) -> None:
|
|
"""All 12 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_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) == 12
|
|
|
|
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
|