7e53d9360a
dev/vendor/MANIFEST.toml pins bcda-app@7decdcc (v3 handler + tests, request validation middleware, OperationOutcome writer, v3 Postman collection, gzip fixtures, sandbox config, LICENSE) and the bcda.cms.gov v3 data dictionary, v2 dictionary and official v3 sample ndjson. vendor_bcda_app.py --verify | --check | --sync (stdlib only); the bcda-vendor workflow runs --check weekly and files an issue on drift — nothing auto-commits, a payload change upstream is something a human reads before the flattener is re-pointed. gen_bcda_v3_crosswalk.py writes dev/seeds/bcda_v3_data_dictionary.csv (fhirPath, IDR sourceColumn, ccwMapping, cclfMapping per v3 field), bcda_v3_cclf_crosswalk.csv (293 CCLF elements -> v3 field) and bcda_v3_cclf_absent.csv (52 CCLF fields v3 does not carry); the CCLF sheet has label/name swapped on data rows and is normalised. Tests assert the committed seeds match the vendored workbook. refs #1139
101 lines
3.1 KiB
Python
101 lines
3.1 KiB
Python
"""dev/scripts/gen_bcda_v3_crosswalk.py — dictionary → seed CSVs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SCRIPT = ROOT / "dev" / "scripts" / "gen_bcda_v3_crosswalk.py"
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def gen():
|
|
spec = importlib.util.spec_from_file_location("gen_bcda_v3_crosswalk", SCRIPT)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
sys.modules["gen_bcda_v3_crosswalk"] = mod
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
|
|
def test_label_name_normalisation(gen) -> None:
|
|
assert gen._label_name("CLM_CNTL_NUM", "Claim Control Number") == (
|
|
"Claim Control Number",
|
|
"CLM_CNTL_NUM",
|
|
)
|
|
assert gen._label_name("File Number Label", "Title") == (
|
|
"File Number Label",
|
|
"Title",
|
|
)
|
|
assert gen._label_name(None, "#N/A") == ("", "")
|
|
|
|
|
|
def test_helpers(gen) -> None:
|
|
assert (
|
|
gen._clean(None) == "" and gen._clean("#N/A") == "" and gen._clean("None") == ""
|
|
)
|
|
assert gen._clean(" a \n b ") == "a b"
|
|
assert gen._list("['A', 'B']") == "A;B"
|
|
assert gen._list("[]") == ""
|
|
assert gen._list("plain") == "plain"
|
|
|
|
|
|
def test_generates_seeds_from_vendored_dictionary(gen, tmp_path) -> None:
|
|
counts = gen.main(seeds=tmp_path)
|
|
assert counts["bcda_v3_data_dictionary"] > 300
|
|
assert counts["bcda_v3_cclf_crosswalk"] > 250
|
|
assert counts["bcda_v3_cclf_absent"] > 40
|
|
|
|
with (tmp_path / "bcda_v3_data_dictionary.csv").open() as f:
|
|
rows = list(csv.DictReader(f))
|
|
assert {
|
|
"fhir_resource",
|
|
"field_name",
|
|
"fhir_path",
|
|
"source_column",
|
|
"ccw_mapping",
|
|
"cclf_mapping",
|
|
} <= set(rows[0])
|
|
resources = {r["fhir_resource"] for r in rows}
|
|
assert resources == {"Patient", "Coverage", "ExplanationOfBenefit"}
|
|
# the crosswalk the flattener follows: CCW names on v3 fields
|
|
assert any(
|
|
r["source_column"] == "CLM_NRLN_RIC_CD"
|
|
and "NCH_NEAR_LINE_REC_IDENT_CD" in r["ccw_mapping"]
|
|
for r in rows
|
|
)
|
|
assert any(r["cclf_mapping"].startswith("CCLF7.") for r in rows)
|
|
|
|
with (tmp_path / "bcda_v3_cclf_crosswalk.csv").open() as f:
|
|
xw = list(csv.DictReader(f))
|
|
mapped = [r for r in xw if r["bcda_v3_field"]]
|
|
assert len(mapped) > 200
|
|
assert all(r["fhir_path"] for r in mapped)
|
|
assert any(
|
|
r["cclf_file"] == "CCLF1"
|
|
and r["field_name"] == "CLM_CNTL_NUM"
|
|
and r["field_label"] == "Claim Control Number"
|
|
for r in mapped
|
|
)
|
|
|
|
with (tmp_path / "bcda_v3_cclf_absent.csv").open() as f:
|
|
absent = list(csv.DictReader(f))
|
|
assert any(r["field_name"] == "HCPCS_BETOS_CD" for r in absent)
|
|
|
|
|
|
def test_committed_seeds_are_current(gen, tmp_path) -> None:
|
|
"""The seeds in dev/seeds are what the vendored dictionary generates."""
|
|
gen.main(seeds=tmp_path)
|
|
for name in (
|
|
"bcda_v3_data_dictionary",
|
|
"bcda_v3_cclf_crosswalk",
|
|
"bcda_v3_cclf_absent",
|
|
):
|
|
assert (tmp_path / f"{name}.csv").read_text() == (
|
|
ROOT / "dev" / "seeds" / f"{name}.csv"
|
|
).read_text(), name
|