r""" The Condition Framework's runner — exception isolation, not condition logic. python tests\test_conditions_runner.py See docs\superpowers\specs\2026-09-16-condition-framework-design.md. This file tests `conditions.base.run_all` in isolation, against small fake condition modules built for each case — never the real telehealth logic (that stays in test_conditions_telehealth.py and test_telehealth_fix.py). The one thing this file exists to prove: **one condition's bug can never block or corrupt another's result on the same claim.** No Playwright needed for most cases — `apply()` on a fake condition never touches `page`, so `page` is passed through untouched (`None` is fine). """ import io import sys import pathlib import types import contextlib ROOT = pathlib.Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT / "src")) from ecw_bot.conditions.base import ConditionResult, run_all # noqa: E402 import ecw_bot.conditions as conditions_pkg # noqa: E402 CASES = [] def case(fn): CASES.append(fn) return fn def _fake_condition(name, *, detects=True, changed=True, raises_in=None): """ Builds a minimal condition module (matching the real detect/apply/ describe contract) as a throwaway `types.ModuleType`, so these tests never need real files on disk for scenarios that only exist to prove the runner's exception handling. """ mod = types.ModuleType(name) mod.NAME = name def detect(claim_id, data): if raises_in == "detect": raise RuntimeError(f"{name} detect() blew up") return detects def apply(page, claim_id, data): if raises_in == "apply": raise RuntimeError(f"{name} apply() blew up") return ConditionResult( condition_name=name, changed=changed, changes=[{"row": 0, "code": "X", "before": {}, "after": {}}] if changed else [], ) def describe(result): return f"{name}: {len(result.changes)} change(s)" mod.detect, mod.apply, mod.describe = detect, apply, describe return mod @case def a_condition_that_does_not_detect_is_never_applied(): calls = [] def detect(claim_id, data): return False def apply(page, claim_id, data): calls.append("apply") return ConditionResult.none("never") mod = types.ModuleType("never") mod.NAME = "never" mod.detect, mod.apply, mod.describe = detect, apply, lambda r: "" results = run_all(None, "C1", {}, conditions=[mod]) assert results == [], results assert calls == [], "apply() must not run when detect() returned False" @case def results_is_empty_when_every_condition_detects_false(): a = _fake_condition("a", detects=False) b = _fake_condition("b", detects=False) results = run_all(None, "C1", {}, conditions=[a, b]) assert results == [], results @case def a_raising_condition_is_skipped_and_does_not_block_the_next_one(): broken = _fake_condition("broken", raises_in="apply") healthy = _fake_condition("healthy", changed=True) results = run_all(None, "C1", {}, conditions=[broken, healthy]) assert len(results) == 1, results assert results[0].condition_name == "healthy", results @case def a_raising_detect_is_also_skipped_without_blocking_others(): broken = _fake_condition("broken_detect", raises_in="detect") healthy = _fake_condition("healthy2", changed=True) results = run_all(None, "C1", {}, conditions=[broken, healthy]) assert len(results) == 1, results assert results[0].condition_name == "healthy2", results @case def a_condition_that_detects_but_changes_nothing_contributes_no_result(): mod = _fake_condition("no_op", detects=True, changed=False) results = run_all(None, "C1", {}, conditions=[mod]) assert results == [], results @case def the_package_exposes_run_all_the_way_export_medcor_ready_calls_it(): """ Regression test for a real bug (run_20260916_162654.log, claim 9130000): `conditions/__init__.py` never re-exported `run_all` from `base.py`, so `conditions.run_all(...)` — exactly how export_medcor_ready.py calls it (`from .. import conditions` then `conditions.run_all(...)`) — raised AttributeError on every single claim. Every other test here imported `run_all` straight from `conditions.base`, so none of them caught it. This one imports the package itself, the same way the real caller does, so it cannot pass by accident the way the others did. """ assert hasattr(conditions_pkg, "run_all"), ( "conditions.run_all is missing — export_medcor_ready.py's " "conditions.run_all(...) call would raise AttributeError on " "every claim") assert conditions_pkg.run_all is run_all @case def multiple_fired_conditions_all_appear_in_order(): a = _fake_condition("a", changed=True) b = _fake_condition("b", changed=True) results = run_all(None, "C1", {}, conditions=[a, b]) assert [r.condition_name for r in results] == ["a", "b"], results @case def before_apply_receives_only_fired_conditions_and_runs_before_any_apply(): """ The before/after message-box pair the operator sees per claim needs a checkpoint between "everything has detect()ed" and "the first apply() touches the DOM" — this proves `before_apply` lands exactly there: called once, with only the conditions that actually detected True (never the one that didn't), and before either fired condition's apply() has run. """ calls = [] def tracking_apply(name): def apply(page, claim_id, data): calls.append(f"apply:{name}") return ConditionResult(condition_name=name, changed=True, changes=[{"row": 0}]) return apply a = _fake_condition("a", detects=True) a.apply = tracking_apply("a") b = _fake_condition("b", detects=False) b.apply = tracking_apply("b") c = _fake_condition("c", detects=True) c.apply = tracking_apply("c") def before_apply(fired): calls.append("before_apply:" + ",".join(m.NAME for m in fired)) results = run_all(None, "C1", {}, conditions=[a, b, c], before_apply=before_apply) assert calls == ["before_apply:a,c", "apply:a", "apply:c"], calls assert [r.condition_name for r in results] == ["a", "c"], results @case def before_apply_is_not_called_when_nothing_fires(): calls = [] a = _fake_condition("a", detects=False) run_all(None, "C1", {}, conditions=[a], before_apply=lambda fired: calls.append(fired)) assert calls == [], calls # ============================================================================= # report= — what really happened, not only what changed (review items D2/P2) # ============================================================================= # Before 2026-10-05 run_all returned the changed results and nothing else, so # a condition that raised, or matched and declined to act, left no trace and # the Charge Summary called the claim "Clean / Pass". def _quiet(fn, *args, **kwargs): with contextlib.redirect_stdout(io.StringIO()): return fn(*args, **kwargs) def _hooked(name, *, detects, needs_review=None, summary_comment=None, result=None): """A fake condition with the optional hooks, and a fixed apply() result.""" mod = _fake_condition(name, detects=detects) if result is not None: mod.apply = lambda page, claim_id, data: result if needs_review is not None: mod.needs_review = needs_review if summary_comment is not None: mod.summary_comment = summary_comment return mod @case def report_records_a_detect_that_raised_as_failed(): report = {} broken = _fake_condition("broken_detect", raises_in="detect") healthy = _fake_condition("healthy") results = _quiet(run_all, None, "C1", {}, conditions=[broken, healthy], report=report) assert [r.condition_name for r in results] == ["healthy"], results assert report["failed"] == [{"name": "broken_detect", "stage": "detect", "error": "broken_detect detect() blew up"}], \ report assert report["fired"] == ["healthy"], report assert report["changed"] == ["healthy"], report @case def report_records_an_apply_that_raised_as_failed_but_still_fired(): report = {} broken = _fake_condition("broken", raises_in="apply") healthy = _fake_condition("healthy") results = _quiet(run_all, None, "C1", {}, conditions=[broken, healthy], report=report) assert [r.condition_name for r in results] == ["healthy"], results assert report["fired"] == ["broken", "healthy"], report assert report["changed"] == ["healthy"], report assert report["failed"] == [{"name": "broken", "stage": "apply", "error": "broken apply() blew up"}], report @case def report_keeps_only_the_first_line_of_a_long_error(): report = {} mod = _fake_condition("long") def apply(page, claim_id, data): raise RuntimeError("Timeout 30000ms exceeded.\nCall log:\n - waiting") mod.apply = apply _quiet(run_all, None, "C1", {}, conditions=[mod], report=report) assert report["failed"][0]["error"] == "Timeout 30000ms exceeded.", report @case def a_declined_result_is_review_not_changed(): report = {} mod = _hooked("decliner", detects=True, result=ConditionResult.declined("decliner", "tab gone")) results = run_all(None, "C1", {}, conditions=[mod], report=report) assert results == [], results assert report["fired"] == ["decliner"] and report["changed"] == [], report assert report["review"] == [{"name": "decliner", "reason": "tab gone"}], \ report assert report["failed"] == [], report @case def declined_is_not_changed_and_none_is_unchanged(): d = ConditionResult.declined("x", "why") assert (d.changed, d.changes, d.review) == (False, [], "why"), d n = ConditionResult.none("x") assert (n.changed, n.changes, n.review) == (False, [], ""), n # The new field defaults, so every pre-existing construction still works. assert ConditionResult("x", True, [1]).review == "" @case def a_changed_result_with_a_review_reason_is_both_changed_and_review(): report = {} res = ConditionResult(condition_name="half", changed=True, changes=[{"row": 0}], review="check by hand") mod = _hooked("half", detects=True, result=res) results = run_all(None, "C1", {}, conditions=[mod], report=report) assert results == [res], results assert report["changed"] == ["half"], report assert report["review"] == [{"name": "half", "reason": "check by hand"}], report @case def needs_review_hook_is_asked_only_when_detect_said_no(): asked = [] def hook(claim_id, data): asked.append(claim_id) return "payer not covered" report = {} quiet = _hooked("quiet", detects=False, needs_review=hook) loud = _hooked("loud", detects=True, needs_review=hook) run_all(None, "C1", {}, conditions=[quiet, loud], report=report) assert asked == ["C1"], "asked once — for the one that did not fire" assert report["review"] == [{"name": "quiet", "reason": "payer not covered"}], report @case def needs_review_returning_nothing_adds_no_entry(): report = {} mod = _hooked("fine", detects=False, needs_review=lambda c, d: None) run_all(None, "C1", {}, conditions=[mod], report=report) assert report["review"] == [] and report["failed"] == [], report @case def a_raising_needs_review_hook_is_failed_and_blocks_nothing_else(): def hook(claim_id, data): raise ValueError("bad mapping") report = {} broken = _hooked("hooked", detects=False, needs_review=hook) healthy = _fake_condition("healthy") results = _quiet(run_all, None, "C1", {}, conditions=[broken, healthy], report=report) assert [r.condition_name for r in results] == ["healthy"], results assert report["failed"] == [{"name": "hooked", "stage": "needs_review", "error": "bad mapping"}], report @case def summary_comment_hook_words_the_comment_of_a_changed_result(): report = {} mod = _hooked("worded", detects=True, summary_comment=lambda r: f"{len(r.changes)} line(s) fixed") plain = _fake_condition("plain") run_all(None, "C1", {}, conditions=[mod, plain], report=report) assert report["comments"] == {"worded": "1 line(s) fixed"}, report @case def summary_comment_is_not_asked_for_an_unchanged_result(): asked = [] mod = _hooked("same", detects=True, result=ConditionResult.none("same"), summary_comment=lambda r: asked.append(r) or "x") report = {} run_all(None, "C1", {}, conditions=[mod], report=report) assert asked == [] and report["comments"] == {}, report @case def a_raising_summary_comment_falls_back_without_tainting_the_claim(): def hook(result): raise KeyError("missing") report = {} mod = _hooked("worded", detects=True, summary_comment=hook) results = _quiet(run_all, None, "C1", {}, conditions=[mod], report=report) assert [r.condition_name for r in results] == ["worded"], results assert report["comments"] == {} and report["failed"] == [], report assert report["review"] == [], report @case def return_value_is_unchanged_without_a_report(): """Old callers (no report=) see exactly the list they always did.""" a = _fake_condition("a") b = _hooked("b", detects=True, result=ConditionResult.declined("b", "why")) c = _fake_condition("c", raises_in="apply") with_report = _quiet(run_all, None, "C1", {}, conditions=[a, b, c], report={}) without = _quiet(run_all, None, "C1", {}, conditions=[a, b, c]) assert [r.condition_name for r in without] == ["a"], without assert without == with_report, (without, with_report) @case def a_report_passed_twice_is_reset_not_appended_to(): report = {} broken = _fake_condition("broken", raises_in="apply") _quiet(run_all, None, "C1", {}, conditions=[broken], report=report) _quiet(run_all, None, "C2", {}, conditions=[broken], report=report) assert len(report["failed"]) == 1, report # ============================================================================= # Every real condition: a CPT/HCPCS tab that will not open is declined # ============================================================================= @case def every_registered_condition_declines_when_the_cpt_tab_cannot_open(): """ Every registered `apply()` re-clicks CPT/HCPCS first and used to return `none()` when that failed — "nothing to do", reported Clean, on a claim that was never checked (review item D2). The branch runs before any page access, so `page=None` is enough. """ from ecw_bot.conditions import CONDITIONS from ecw_bot.conditions.base import CPT_TAB_NOT_CONFIRMED from ecw_bot.pages import claim_form def tab_that_raises(page): raise RuntimeError("No CPT/HCPCS tab matched (simulated)") real = claim_form.open_cpt_hcpcs_tab claim_form.open_cpt_hcpcs_tab = tab_that_raises try: for cond in CONDITIONS: result = _quiet(cond.apply, None, "C1", {}) assert result == ConditionResult.declined( cond.NAME, CPT_TAB_NOT_CONFIRMED), (cond.NAME, result) finally: claim_form.open_cpt_hcpcs_tab = real # The whole registry: the common rule's two halves bracket it, and no # condition is listed twice. names = [c.NAME for c in CONDITIONS] assert len(set(names)) == len(names), names assert names[0] == "zero_fee_uncheck", names assert names[-1] == "priced_line_check", names def main() -> int: failures = 0 # run_all screenshots a condition that raised (base.shot), and shot() # creates the operator's dated Screenshots\ folder before it fails on a # page of None. A no-op here keeps the suite out of that folder (final # review, 2026-10-05), as test_save_corrections_gate.py already does. from ecw_bot.conditions import base orig_shot = base.shot base.shot = lambda page, name: None try: for fn in CASES: name = fn.__name__.replace("_", " ") try: fn() print(f" ok {name}") except AssertionError as exc: print(f" FAIL {name}: {exc}") failures += 1 except Exception as exc: print(f" ERROR {name}: {type(exc).__name__}: {exc}") failures += 1 finally: base.shot = orig_shot print(f"\n{len(CASES) - failures}/{len(CASES)} passed") return 1 if failures else 0 if __name__ == "__main__": sys.exit(main())