ManikandanandClaude Opus 5.5
ba15a574ab
Snapshot of the working project as of 2026-10-23 (local 2d4d4fc), one commit with no earlier history: the tracked files were de-identified first (synthetic claim ids and patient data only; see Context.md section 7). Credentials, run logs, screenshots, claim reports and client workbooks are gitignored and not included. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
459 lines
17 KiB
Python
459 lines
17 KiB
Python
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())
|