eb1b574c04
Some checks failed
CI / lint (push) Successful in 48s
CI / notebooks-smoke (push) Successful in 1m46s
Infra CI / zotero (push) Successful in 15s
Infra CI / llm (push) Successful in 1m4s
Infra CI / notebooks (pull_request) Successful in 55s
Infra CI / docs (pull_request) Successful in 17s
Infra CI / api (pull_request) Successful in 14s
Infra CI / llm (pull_request) Successful in 20s
CI / test (push) Failing after 2m26s
Infra CI / notebooks (push) Successful in 52s
Infra CI / api (push) Successful in 1m14s
Infra CI / docs (push) Successful in 1m45s
Infra CI / mc (push) Failing after 42s
CI / lint (pull_request) Successful in 27s
CI / notebooks-smoke (pull_request) Successful in 1m41s
CI / test (pull_request) Failing after 2m20s
Infra CI / zotero (pull_request) Successful in 21s
Infra CI / mc (pull_request) Failing after 32s
Package Supply Chain / pkg-supply-chain (push) Successful in 2m3s
bcda.ui mirrors the surface of the CMS server (CMSgov/bcda-app, vendored under dev/vendor): /_version, /_health, metadata, /auth/token, attribution_status, $export on Group/all · Group/runout · Patient with _type / _since / _typeFilter, job history, status, cancel and /data download — plus the local side: the Store's job and file history, flatten + load into bcda.*, the connector into input_layer.*, the replica for browsing, the v3 data dictionary and CCLF crosswalk, and the vendored-upstream manifest with an on-demand drift check. bcda.ui.service has no Streamlit in it: every operation takes a Connection / Client / Store, so `stack bcda preflight|auth|jobs|export| adopt|cancel|store|load|vendor` reuse it verbatim and tests hand in httpx.MockTransport clients and temp stores. Connections are the four public sandbox entities, BCDA_CLIENT_ID/SECRET from the environment, or custom (sandbox or production, v3 or v2). bcda.ui.runner runs exports, adoption, loads and drift checks on daemon threads in a process-level registry that survives Streamlit reruns, captures the bcda.* log lines of each run's own thread, and flips state last so readers never see a finished run with an incomplete log; a st.fragment panel shows runs live. Exports default to one job per type with cancel retries (the only shape that survives the shared sandbox's sweeps); the _typeFilter builder composes System-Type, outcome and service-date; completed jobs under 24 h are offered for adopt, in-progress ones for cancel. Data pages read the read-only replica only. `stack bcda ui` launches streamlit on src/bcda/ui/app.py; streamlit is added to the bcda extra. CLI reference pages regenerated (the generator renumbers sidebar positions, hence the touched pages). Tests: service, runner, every page through AppTest (buttons, forms, error paths), the app shell, and the CLI — new modules at 100% line coverage. refs #1139
92 lines
2.5 KiB
Python
92 lines
2.5 KiB
Python
"""bcda.ui.runner — background runs with per-thread log capture."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
import time
|
|
|
|
from bcda.ui.runner import Runner
|
|
|
|
|
|
def test_run_success_captures_bcda_logs() -> None:
|
|
r = Runner()
|
|
log = logging.getLogger("bcda.test")
|
|
|
|
def work(x: int) -> int:
|
|
log.info("working on %d", x)
|
|
return x * 2
|
|
|
|
run = r.submit("double", work, 21)
|
|
done = r.wait(run.id)
|
|
assert done.state == "done" and done.result == 42
|
|
assert (
|
|
"working on 21" in done.log
|
|
and done.log[0] == "started: double"
|
|
and done.log[-1] == "finished"
|
|
)
|
|
assert not done.active and done.elapsed >= 0 and done.finished is not None
|
|
assert r.get(run.id) is done and r.get("nope") is None
|
|
assert r.all() == [done] and r.active() == []
|
|
assert r.clear_finished() == 1 and r.all() == []
|
|
|
|
|
|
def test_run_failure_records_error_and_traceback() -> None:
|
|
r = Runner()
|
|
|
|
def boom() -> None:
|
|
raise ValueError("bad input")
|
|
|
|
run = r.wait(r.submit("boom", boom).id)
|
|
assert run.state == "failed" and run.error == "ValueError: bad input"
|
|
assert any("Traceback" in line for line in run.log)
|
|
|
|
|
|
def test_logs_are_routed_by_thread() -> None:
|
|
r = Runner()
|
|
log = logging.getLogger("bcda.routing")
|
|
gate = threading.Event()
|
|
|
|
def slow() -> str:
|
|
log.info("from slow")
|
|
gate.wait(5)
|
|
return "ok"
|
|
|
|
def quick() -> str:
|
|
log.info("from quick")
|
|
return "ok"
|
|
|
|
slow_run = r.submit("slow", slow)
|
|
quick_run = r.wait(r.submit("quick", quick).id)
|
|
log.info("from main thread") # no run owns the main thread → dropped
|
|
gate.set()
|
|
r.wait(slow_run.id)
|
|
assert "from quick" in quick_run.log and "from slow" not in quick_run.log
|
|
assert "from slow" in slow_run.log and "from quick" not in slow_run.log
|
|
assert "from main thread" not in slow_run.log + quick_run.log
|
|
assert r.active() == []
|
|
|
|
|
|
def test_log_is_bounded() -> None:
|
|
r = Runner()
|
|
log = logging.getLogger("bcda.flood")
|
|
|
|
def flood() -> None:
|
|
for i in range(500):
|
|
log.info("line %d", i)
|
|
|
|
run = r.wait(r.submit("flood", flood).id)
|
|
assert len(run.log) == 400 and run.log[-1] == "finished"
|
|
|
|
|
|
def test_wait_times_out_on_active_run() -> None:
|
|
r = Runner()
|
|
gate = threading.Event()
|
|
run = r.submit("hang", gate.wait, 5)
|
|
t0 = time.time()
|
|
waited = r.wait(run.id, timeout=0.1)
|
|
assert waited.active and time.time() - t0 < 2
|
|
assert r.active() == [waited] and r.clear_finished() == 0
|
|
gate.set()
|
|
r.wait(run.id)
|