Files
stack/tests/bcda/test_ui_views.py
kert 2cc13008a9
Some checks failed
CI / lint (push) Successful in 54s
CI / notebooks-smoke (push) Successful in 1m43s
CI / test (push) Failing after 2m56s
Infra CI / zotero (push) Successful in 15s
Infra CI / notebooks (push) Successful in 1m1s
Infra CI / api (push) Successful in 1m7s
Infra CI / docs (push) Successful in 1m50s
Infra CI / llm (push) Successful in 49s
Infra CI / mc (push) Failing after 35s
Infra CI / bcda (push) Successful in 1m42s
CI / lint (pull_request) Successful in 45s
CI / notebooks-smoke (pull_request) Successful in 1m38s
CI / test (pull_request) Failing after 2m53s
Infra CI / mc (pull_request) Failing after 30s
Infra CI / zotero (pull_request) Successful in 16s
Infra CI / notebooks (pull_request) Successful in 58s
Infra CI / docs (pull_request) Successful in 16s
Infra CI / api (pull_request) Successful in 15s
Infra CI / bcda (pull_request) Successful in 21s
Infra CI / llm (pull_request) Successful in 17s
feat(bcda): close the upstream-review gaps — partial exports, claim identity across systems, attribution checks, filter grammar, vendored behaviour
Closes #1293 #1294 #1295 #1296 #1297 #1298 #1299 #1300 (upstream review
docs/superpowers/specs/2026-10-25-bcda-upstream-review.md).

Partial exports (#1293). The manifest's `error` list is downloaded as
OperationOutcome[.N].ndjson with resource_type=OperationOutcome; every
URL is attempted and failures are recorded per file, then raised once as
DownloadError. A job with error files is `partial`: it still reconciles
(its rows are real) but never becomes the watermark, check_since warns
`partial_job`, the Store page lists the issues and `stack bcda errors`
prints them.

Claim identity (#1294), per bcda.cms.gov "Matching multiple EOBs to the
same claim": three new header columns — clm_cntl_num_match (VMS leading 1
stripped on SharedSystem DME), clm_natural_key (Part D events without a
control number: bene · line from-date · provider qualifier · service
provider · RX number · fill number) and superseded_by — plus
prvdr_srvc_id_qlfyr_cd. bcda.sync keys claims on match → control number
→ natural key → id, marks every claim another current claim names in
`related` as superseded (chain walked), counts `superseded`, and a new
invariant forbids a live claim being named predecessor by another live
claim. The connector dedupes on the same identity and drops superseded
rows. Partitions and current tables flattened before these columns
existed get them as NULLs.

Attribution (#1295): `attribution_expired` (block) from
expiration_dates, `attribution_floor` (warn) from the earliest
last_attribution_update recorded on job rows (new column), and the
server's 404/400/429 texts explained by explain_server_error.

Adoption (#1296): completed_jobs() drops jobs older than the 24 h file
retention; downloads no longer stop at the first failed file.

_typeFilter (#1297): build_type_filter follows bcda/fhir/param.go —
outcome lists, service-date earliest and latest bounds with eq/gt/ge/lt/le
on YYYY[-MM[-DD]], no comma lists, no conflicting prefixes; CLI and
Exports page expose them.

Runout (#1298): `runout_window` warn after July; population column on
the Sync page ledger.

Vendoring (#1299): the manifest now pins bcda-app's param.go,
bluebutton.go, service.go, constants, models and prod.yml, the docs
source (CMSgov/bcda-static-site: filter/access/migrate/PAC/FAQ pages and
announcements) and the PAC and unmapped-fields dictionaries.

Docs (#1300): the reconciliation design record states the server's
window `(since, transactionTime]`, the lookback bound, the silent
exclusions, re-loads, partial exports and the identity rules.

refs #1139
2026-10-25 17:10:24 -04:00

593 lines
22 KiB
Python

"""bcda.ui.views — every page through Streamlit's AppTest with the
service layer stubbed, so no network, no replica, no real store."""
from __future__ import annotations
import logging
import threading
import time
from pathlib import Path
import duckdb
import pytest
from streamlit.testing.v1 import AppTest
from bcda.store import Store
from bcda.ui import service as svc
from bcda.ui import views
from bcda.ui.runner import Runner
SMALL = svc.PRESETS["Small Model Entity"]
def _page(
name: str, *, conn: svc.Connection | None = SMALL, timeout: float = 60
) -> AppTest:
at = AppTest.from_string(
f"from bcda.ui import views\nviews.{name}()\n", default_timeout=timeout
)
if conn is not None:
at.session_state["conn"] = conn
return at
def _button(at: AppTest, label: str): # noqa: ANN202
return next(b for b in at.button if b.label == label)
@pytest.fixture(autouse=True)
def fresh_runner(monkeypatch):
"""Each test gets its own Runner so background runs never leak across tests."""
r = Runner()
monkeypatch.setattr(views, "runner", r)
return r
@pytest.fixture
def fake_client(monkeypatch):
"""A client stand-in; service functions are stubbed to accept it."""
class FakeClient:
base_url = svc.SANDBOX
version = "v3"
def close(self) -> None:
pass
client = FakeClient()
monkeypatch.setattr(svc, "make_client", lambda conn, output_dir=None: client)
return client
@pytest.fixture
def tmp_store(monkeypatch, tmp_path):
monkeypatch.setattr(svc, "store_path", lambda: tmp_path)
return Store(path=str(tmp_path))
# ── Connect ──────────────────────────────────────────────────
class TestConnect:
def test_preset_then_preflight_and_auth(self, monkeypatch, fake_client) -> None:
monkeypatch.setattr(
svc,
"preflight",
lambda c: {
"ok": True,
"fhir_version": "4.0.1",
"software": "r302",
"release_date": "2026",
"resources": ["Patient"],
"health": {"database": "ok"},
"server_version": {"version": "r302"},
},
)
monkeypatch.setattr(
svc,
"auth_check",
lambda c: {"token_chars": 40, "last_attribution_update": "x"},
)
at = _page("connect", conn=None).run()
assert at.session_state.get("conn") is None
_button(at, "Use this connection").click().run()
assert at.session_state["conn"].label == "Large Model Entity"
_button(at, "Preflight (no token)").click().run()
assert any(m.value == "4.0.1" for m in at.metric)
_button(at, "Authenticate + attribution").click().run()
assert any("Token issued" in s.value for s in at.success)
def test_preflight_failure_and_auth_error(self, monkeypatch, fake_client) -> None:
monkeypatch.setattr(
svc, "preflight", lambda c: {"ok": False, "error": "unreachable"}
)
def bad_auth(c): # noqa: ANN001, ANN202
raise RuntimeError("401 invalid client")
monkeypatch.setattr(svc, "auth_check", bad_auth)
at = _page("connect").run()
_button(at, "Preflight (no token)").click().run()
assert any("unreachable" in e.value for e in at.error)
_button(at, "Authenticate + attribution").click().run()
assert any("invalid client" in e.value for e in at.error)
def test_environment_and_custom_sources(self, monkeypatch) -> None:
monkeypatch.setenv("BCDA_CLIENT_ID", "env-id")
monkeypatch.setenv("BCDA_CLIENT_SECRET", "env-secret-long-enough")
at = _page("connect", conn=None).run()
radio = at.radio[0]
assert radio.options[0] == "Environment credentials"
_button(at, "Use this connection").click().run()
assert at.session_state["conn"].client_id == "env-id"
radio.set_value("Custom").run()
assert not [
b for b in at.button if b.label == "Use this connection"
] # needs id + secret first
at.text_input(key="custom_id").input("abc").run()
at.text_input(key="custom_secret").input("def").run()
at.selectbox(key="custom_version").set_value("v2").run()
_button(at, "Use this connection").click().run()
conn = at.session_state["conn"]
assert (conn.client_id, conn.client_secret, conn.version, conn.label) == (
"abc",
"def",
"v2",
"custom",
)
# ── Exports ──────────────────────────────────────────────────
class TestExports:
def test_requires_connection(self) -> None:
at = _page("exports", conn=None).run()
assert any("Connect" in i.value for i in at.info)
def test_start_export_per_type_and_one_job(
self, monkeypatch, fake_client, tmp_store, fresh_runner
) -> None:
calls: list[dict] = []
monkeypatch.setattr(
svc,
"export_per_type",
lambda store, client, **kw: (
calls.append({"per_type": True, **kw}) or {"Patient": "K1"}
),
)
monkeypatch.setattr(
svc,
"start_export",
lambda store, client, **kw: calls.append({"per_type": False, **kw}) or "K2",
)
at = _page("exports").run()
at.multiselect(key="exp_types").set_value(["Patient"])
at.multiselect(key="exp_system_types").set_value(["DDPS"])
at.multiselect(key="exp_outcome").set_value(["complete"])
at.radio(key="exp_since_mode").set_value("date")
# an empty store has no baseline: the checks block and nothing starts
_button(at, "Start export").click().run()
assert fresh_runner.all() == []
assert "no_baseline" in at.dataframe[0].value["code"].tolist()
assert any("blocks this export" in e.value for e in at.error)
# forced through
at.checkbox(key="exp_force").check()
_button(at, "Start export").click().run()
assert any("despite" in w.value for w in at.warning)
run = fresh_runner.wait(fresh_runner.all()[0].id)
assert run.state == "done" and run.result == {"Patient": "K1"}
assert calls[0]["per_type"] and calls[0]["types"] == ["Patient"]
assert calls[0]["type_filter"].endswith("|DDPS&outcome=complete") and calls[0][
"since"
].endswith("T00:00:00+00:00")
assert calls[0]["force"] is True
assert any("_typeFilter=" in c.value for c in at.caption)
# a filter the builder rejects stops before anything starts
monkeypatch.setattr(
svc,
"build_type_filter",
lambda **kw: (_ for _ in ()).throw(ValueError("conflicting prefix")),
)
n_runs = len(fresh_runner.all())
_button(at, "Start export").click().run()
assert any("_typeFilter: conflicting prefix" in e.value for e in at.error)
assert len(fresh_runner.all()) == n_runs
monkeypatch.undo()
monkeypatch.setattr(
svc,
"export_per_type",
lambda store, client, **kw: (
calls.append({"per_type": True, **kw}) or {"Patient": "K1"}
),
)
monkeypatch.setattr(
svc,
"start_export",
lambda store, client, **kw: calls.append({"per_type": False, **kw}) or "K2",
)
# watermark mode hands the service the sentinel
at.radio(key="exp_since_mode").set_value("watermark")
_button(at, "Start export").click().run()
fresh_runner.wait(fresh_runner.all()[0].id)
assert calls[-1]["since"] == svc.SINCE_AUTO
at.radio(key="exp_since_mode").set_value("off")
at.checkbox(key="exp_per_type").uncheck()
at.checkbox(key="exp_use_svc_date").check()
_button(at, "Start export").click().run()
fresh_runner.wait(fresh_runner.all()[0].id)
assert (
calls[-1]["per_type"] is False
and "service-date=gt" in calls[-1]["type_filter"]
)
def test_jobs_adopt_and_cancel(
self, monkeypatch, fake_client, tmp_store, fresh_runner
) -> None:
jobs = [
{
"job_id": "11",
"status": "completed",
"types": "Patient,Coverage",
"endpoint": "Group/all",
"started": "s",
"ended": "e",
"reusable": True,
"job_url": "u11",
},
{
"job_id": "12",
"status": "in-progress",
"types": "(all)",
"endpoint": "Group/all",
"started": "s",
"ended": "",
"reusable": False,
"job_url": "u12",
},
]
monkeypatch.setattr(svc, "server_jobs", lambda c: jobs)
adopted: list = []
monkeypatch.setattr(
svc,
"adopt_job",
lambda store, client, url, types=None: adopted.append((url, types)) or "KA",
)
cancelled: list = []
monkeypatch.setattr(svc, "cancel_job", lambda c, url: cancelled.append(url))
at = _page("exports").run()
_button(at, "Refresh jobs").click().run()
assert len(at.dataframe) == 1
_button(at, "Adopt").click().run()
run = fresh_runner.wait(fresh_runner.all()[0].id)
assert run.result == "KA" and adopted == [("u11", ["Patient", "Coverage"])]
_button(at, "Cancel").click().run()
assert cancelled == ["u12"] and "server_jobs" not in at.session_state
def test_jobs_errors(self, monkeypatch, fake_client, tmp_store) -> None:
def boom(*a, **k): # noqa: ANN002, ANN003, ANN202
raise RuntimeError("nope")
monkeypatch.setattr(svc, "server_jobs", boom)
at = _page("exports").run()
_button(at, "Refresh jobs").click().run()
assert any("Could not list jobs" in e.value for e in at.error)
jobs = [
{
"job_id": "12",
"status": "in-progress",
"types": "(all)",
"endpoint": "Group/all",
"started": "s",
"ended": "",
"reusable": False,
"job_url": "u12",
}
]
monkeypatch.setattr(svc, "server_jobs", lambda c: jobs)
monkeypatch.setattr(svc, "cancel_job", boom)
_button(at, "Refresh jobs").click().run()
_button(at, "Cancel").click().run()
assert any("Cancel failed" in e.value for e in at.error)
# ── Store / Load ─────────────────────────────────────────────
class TestStoreAndLoad:
def test_partial_job_warning_and_diagnostics(self, tmp_store, tmp_path) -> None:
d = Path(tmp_store.root) / "exports" / "P1"
d.mkdir(parents=True)
oo = d / "OperationOutcome.ndjson"
oo.write_text(
'{"resourceType":"OperationOutcome","issue":[{"severity":"error","code":"exception","diagnostics":"BFD 502"}]}\n'
)
tmp_store._append_file(
job_key="P1",
resource_type="OperationOutcome",
url="u",
storage_path=str(oo),
file_size=1,
status="complete",
)
tmp_store._append_job(
key="P1", endpoint="Group/all", types="Patient", status="complete"
)
at = _page("store").run()
assert not at.exception
assert any(
"Partial export" in w.value and "P1 (1)" in w.value for w in at.warning
)
assert any("OperationOutcome issue" in e.label for e in at.expander)
def test_store_empty_and_populated(self, tmp_store, tmp_path) -> None:
at = _page("store").run()
assert any("No store jobs" in i.value for i in at.info)
tmp_store._append_job(
key="K1",
status="complete",
endpoint="Group/all",
types="Patient",
job_url="u",
)
tmp_store._append_file(
job_key="K1",
resource_type="Patient",
url="u1",
storage_path="p",
file_size=2048,
status="complete",
)
d = tmp_path / "exports" / "K1"
d.mkdir(parents=True)
(d / "Patient.ndjson").write_text("{}\n")
at = _page("store").run()
assert at.selectbox(key="store_key").value == "K1"
assert len(at.dataframe) == 3 # jobs, files, dirs
assert (
views._fmt_bytes(2048) == "2.0 KB"
and views._fmt_bytes(5) == "5 B"
and views._fmt_bytes(3 * 1024**4) == "3.0 TB"
)
def test_load_runs_in_background(self, monkeypatch, fresh_runner) -> None:
seen = {}
monkeypatch.setattr(
svc, "load_and_publish", lambda **kw: seen.update(kw) or {"bcda.patient": 3}
)
at = _page("load").run()
at.checkbox(key="load_skip").check()
_button(at, "Load now").click().run()
run = fresh_runner.wait(fresh_runner.all()[0].id)
assert run.result == {"bcda.patient": 3} and seen == {
"run_pipeline": True,
"skip_flatten": True,
}
at.run() # the runs panel now shows the finished run
assert any("flatten + load" in e.label for e in at.expander)
_button(at, "Clear finished runs").click().run()
assert fresh_runner.all() == []
# ── Data ─────────────────────────────────────────────────────
class TestData:
def test_no_replica(self, monkeypatch) -> None:
def boom(): # noqa: ANN202
raise OSError("no replica")
monkeypatch.setattr(svc, "replica_connect", boom)
at = _page("data").run()
assert any("Replica not available" in w.value for w in at.warning)
def test_empty_replica(self, monkeypatch) -> None:
con = duckdb.connect()
monkeypatch.setattr(svc, "replica_connect", lambda: con)
at = _page("data").run()
assert any("No `bcda.*`" in i.value for i in at.info)
def test_tables_summaries_and_sql(self, monkeypatch, tmp_path) -> None:
db = tmp_path / "r.duckdb"
con = duckdb.connect(str(db))
con.execute("create schema bcda")
con.execute(
"create table bcda.explanation_of_benefit as select 'DDPS' system_type, 'PDE' eob_type, 'FinalAction' final_action, 1.0 clm_pmt_amt"
)
con.close()
ro = duckdb.connect(str(db), read_only=True)
monkeypatch.setattr(svc, "replica_connect", lambda: ro)
at = _page("data").run()
assert (
at.metric[0].label == "bcda.explanation_of_benefit"
and at.metric[0].value == "1"
)
assert len(at.dataframe) == 1 # the default summary
at.selectbox(key="summary_pick").set_value("Coverage rows by plan").run()
assert any(
"coverage" in w.value.lower() for w in at.warning
) # table absent → warning
_button(at, "Run").click().run()
assert any("1 rows" in c.value for c in at.caption)
at.text_area(key="sql").input("select * from nope").run()
_button(at, "Run").click().run()
assert at.error
# ── Dictionary / Upstream ────────────────────────────────────
class TestDictionaryAndUpstream:
def test_dictionary_search(self, monkeypatch) -> None:
monkeypatch.setattr(svc, "replica_connect", lambda: duckdb.connect())
at = _page("dictionary").run()
assert len(at.dataframe) == 3
at.text_input(key="dict_search").input("CLM_CNTL_NUM").run()
assert (
"Dictionary (" in at.tabs[0].label and at.tabs[0].label != "Dictionary (0)"
)
at.checkbox(key="xw_mapped").uncheck().run()
assert len(at.dataframe) == 3
def test_dictionary_missing(self, monkeypatch) -> None:
def boom(): # noqa: ANN202
raise OSError("no replica")
monkeypatch.setattr(svc, "replica_connect", boom)
monkeypatch.setattr(svc, "dictionary_frames", lambda con: {})
at = _page("dictionary").run()
assert any("Crosswalk seeds not found" in i.value for i in at.info)
def test_upstream(self, monkeypatch, fresh_runner) -> None:
monkeypatch.setattr(svc, "vendor_status", lambda: {"available": False})
at = _page("upstream").run()
assert any("not available" in i.value for i in at.info)
status = {
"available": True,
"sources": [
{
"name": "bcda-app",
"kind": "github",
"origin": "CMSgov/bcda-app",
"commit": "abc",
"files": 1,
}
],
"files": [{"path": "bcda-app/LICENSE", "sha256": "0" * 12}],
"problems": ["missing bcda-app/LICENSE"],
}
monkeypatch.setattr(svc, "vendor_status", lambda: status)
monkeypatch.setattr(svc, "vendor_check", lambda: [])
at = _page("upstream").run()
assert any("does not match" in e.value for e in at.error)
_button(at, "Check upstream for drift (network)").click().run()
assert fresh_runner.wait(fresh_runner.all()[0].id).result == []
status["problems"] = []
at = _page("upstream").run()
assert any("matches the manifest" in s.value for s in at.success)
# ── runs panel states ────────────────────────────────────────
class TestRunsPanel:
def test_failed_and_active_runs_render(self, fresh_runner) -> None:
gate = threading.Event()
logging.getLogger("bcda.ui.test")
def fail() -> None:
raise RuntimeError("kaput")
fresh_runner.wait(fresh_runner.submit("will fail", fail).id)
fresh_runner.submit("hangs", gate.wait, 10)
time.sleep(0.05)
at = _page("load").run()
labels = [e.label for e in at.expander]
assert any("will fail" in lbl and "failed" in lbl for lbl in labels)
assert any("hangs" in lbl and "running" in lbl for lbl in labels)
assert any("kaput" in e.value for e in at.error)
gate.set()
fresh_runner.wait(fresh_runner.all()[0].id)
# ── Sync ─────────────────────────────────────────────────────
class TestSync:
def test_empty_store(self, tmp_store) -> None:
at = _page("sync").run()
assert not at.exception
assert any("No complete jobs" in i.value for i in at.info)
assert at.title[0].value == "Sync"
def test_with_jobs_and_reconcile(
self, monkeypatch, tmp_store, fresh_runner
) -> None:
from datetime import datetime, timezone
from bcda import sync
from .test_sync import _rows, add_job
t0 = datetime(2026, 10, 1, tzinfo=timezone.utc)
add_job(tmp_store, "B1", {"Patient": _rows("Patient")[:5]}, transaction_time=t0)
from .test_sync import _manifest_with_errors
add_job(
tmp_store,
"P2",
{"Patient": _rows("Patient")[5:6]},
since="2026-10-01T00:00:00+00:00",
transaction_time=t0,
)
_manifest_with_errors(tmp_store, "P2", t0)
at = _page("sync").run()
assert not at.exception
assert any(
"Partial (error files" in w.value and "P2" in w.value for w in at.warning
)
assert any("2 pending" in b.label for b in at.button)
assert any("nothing applied" in c.value for c in at.caption)
calls: list = []
monkeypatch.setattr(
svc,
"load_and_publish",
lambda **kw: calls.append(kw) or {"bcda.patient": 5},
)
_button(at, "Reconcile now (2 pending)").click().run()
run = fresh_runner.wait(fresh_runner.all()[0].id)
assert run.state == "done" and run.result["applied"] == ["B1", "P2"]
assert calls == [{"run_pipeline": True, "skip_flatten": True}]
assert [r["job_key"] for r in sync.read_ledger(tmp_store)] == ["B1", "P2"]
# the ledger renders, and the inflated page still has no pending work
at = _page("sync").run()
assert not at.exception
assert any("Reconcile now" == b.label for b in at.button)
assert any("patient" in d.value.to_string() for d in at.dataframe)
# a failed ledger row surfaces as an error
sync._append_ledger(
tmp_store,
{
"applied_seq": 3,
"job_key": "BAD",
"kind": "delta",
"endpoint": "Group/all",
"type_filter": "",
"types": "Patient",
"since": "x",
"transaction_time": "",
"transaction_source": "job_row",
"applied_at": "",
"ok": False,
"counts": "{}",
"notes": '["scope differs"]',
"error": "SyncInvariantError: boom",
},
)
at = _page("sync").run()
assert any("boom" in e.value for e in at.error)
assert any("note(s)" in e.label for e in at.expander)
def test_app_shell_renders() -> None:
at = AppTest.from_file(
str(Path(views.__file__).with_name("app.py")), default_timeout=60
).run()
assert not at.exception
assert any("not connected" in c.value for c in at.sidebar.caption)
# masthead first, then one page link per page, in order
assert any("fw-mast" in h.value for h in at.sidebar.html)
assert [p.label for p in at.sidebar.page_link] == [
"Connect",
"Exports",
"Store",
"Flatten + load",
"Sync",
"Data",
"Dictionary + crosswalk",
"Upstream",
]
at.session_state["conn"] = SMALL
at.run()
assert any("Small Model Entity" in c.value for c in at.sidebar.caption)