Files
2026-10-30 10:58:35 +05:30

607 lines
24 KiB
Python

r"""
export_medcor_ready - what the client sees: one box per claim, in English,
with no "?" pages (review item D9, and P6's ungated Dental box).
python tests\test_claim_boxes.py
Before D9 (2026-10-05), with the operator's settings (SAVE_CORRECTIONS=0,
REVIEW_BEFORE_CLOSE=1, SHOW_EACH_CLAIM=1):
* a corrected UB/HCFA claim showed THREE boxes - "N condition(s) matched
- about to apply", "review before close", then the per-claim box - three
OK clicks, up to 180 s each, in a client demo;
* a Dental claim showed its own Dental box (even with every flag off),
then a per-claim box of "Patient : ? ... Charges : ?";
* the boxes said "innum N" and "Chart locked: Aamaa/Illa", and
"Conditions : none matched" right after a box said one was about to
apply.
Every case drives the real `_process_claims` / `_read_claim_form` with each
page step stubbed and `msgbox.info_box` recording (title, body) into the
same event list as `close_claim`, so the ORDER is checked, not assumed.
save_claim and the save box are forbidden throughout: nothing here may
write. No browser, no eCW.
"""
import io
import re
import sys
import types
import pathlib
import contextlib
ROOT = pathlib.Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "src"))
from ecw_bot import settings as cfg # noqa: E402
from ecw_bot.core import msgbox # noqa: E402
from ecw_bot.pages import claim_form, progress_note # noqa: E402
from ecw_bot.conditions.base import ConditionResult # noqa: E402
from ecw_bot.flows import export_medcor_ready as flow # noqa: E402
CASES = []
def case(fn):
CASES.append(fn)
return fn
def _forbidden(*a, **k):
raise AssertionError("must not be called on this path")
# Every attribute any case replaces, restored after each case. The real .env
# is loaded at import, so every setting a case depends on is set here.
_PATCHED = [
(cfg, "SAVE_CORRECTIONS"), (cfg, "REVIEW_BEFORE_CLOSE"),
(cfg, "SHOW_EACH_CLAIM"), (cfg, "SHOW_CLAIM_IDS"),
(cfg, "CLAIM_PROCESS_LIMIT"), (cfg, "OPEN_CLAIM_FORM"),
(cfg, "OPEN_PROGRESS_NOTES"), (cfg, "APPLY_CORRECTIONS"),
(cfg, "TEST_CLAIM_IDS_FILE"), (cfg, "CHECK_HIGH_RISK"),
(cfg, "CHECK_SLIDING_FEE"), (cfg, "MSGBOX_TIMEOUT_SECONDS"),
(msgbox, "info_box"), (msgbox, "ok_cancel"), (msgbox, "info"),
(claim_form, "open_claim"), (claim_form, "wait_claim_loaded"),
(claim_form, "is_dental_claim_form"), (claim_form, "open_cpt_hcpcs_tab"),
(claim_form, "extract"), (claim_form, "save"), (claim_form, "close_claim"),
(claim_form, "save_claim"), (claim_form, "open_progress_notes"),
(flow.conditions, "CONDITIONS"), (flow.conditions, "BY_NAME"),
(flow.claim_results, "search_claim"),
(flow.claim_filters, "clear_claim_no"), (flow, "_write_charge_summary"),
# Never the real results database, whatever this machine's .env says.
(cfg, "RESULTS_DB"),
(flow.dental_notes, "check"), (flow.sliding_fee, "check"),
]
_ORIG = [(obj, name, getattr(obj, name)) for obj, name in _PATCHED]
def _restore():
for obj, name, value in _ORIG:
setattr(obj, name, value)
# A UB/HCFA read that leaves nothing for the Charge Summary to send to
# Review: one CPT row, one ICD row (an empty ICD list next to CPT rows is a
# review reason since the final review, 2026-10-05) and a progress note
# that was actually read.
_UB = {"claim_no": "9103000",
"patient": {"name": "Doe, Jane", "age_sex": "34 F",
"account_no": "A1"},
"header": {"claim_date": "09/01/2026",
"claim_status_label": "Medcor Ready",
"claim_status_code": "MR"},
"totals": {"charges": "224.00", "balance": "224.00"},
"cpt": [{"code": "99213", "sdos": "09/01/2026", "units": "1"}],
"icd": [{"#": "1", "code": "R69"}], "revenue": []}
_NOTE = {"opened": "modal", "sections": {"Assessments": "x"}}
_RESULT = ConditionResult(
condition_name="depression_dx_pointer", changed=True,
changes=[{"row": 0, "code": "G8510",
"before": {"icd": ["1", "", "", ""]},
"after": {"icd": ["2", "", "", ""]}}])
def _fake_condition(fires, detect_raises=False):
mod = types.ModuleType("fake_cond")
mod.NAME = "fake_cond"
if detect_raises:
def detect(claim_id, data):
raise KeyError("rows")
mod.detect = detect
else:
mod.detect = lambda claim_id, data: fires
mod.apply = lambda page, claim_id, data: _RESULT
return mod
class _DescribeRaises:
"""Stands in for depression_dx_pointer in BY_NAME: describe() raises."""
NAME = "depression_dx_pointer"
@staticmethod
def describe(result):
raise ValueError("unexpected change shape")
def _search_raises(page, cid):
raise RuntimeError("Lookup button never became visible")
def _run(*, fires=False, review=True, show_each=True, dental=False,
open_claim=None, test_file="", high_risk=True, save=False,
detect_raises=False, describe_raises=False, search=None):
"""
One claim through the real `_process_claims`. Returns (events, printed
text, outcomes). events: ("box", title, body) and ("close",), in order.
"""
events, written = [], []
cfg.SAVE_CORRECTIONS = save
cfg.REVIEW_BEFORE_CLOSE = review
cfg.SHOW_EACH_CLAIM = show_each
cfg.SHOW_CLAIM_IDS = False
cfg.CLAIM_PROCESS_LIMIT = 0
cfg.OPEN_CLAIM_FORM = True
cfg.OPEN_PROGRESS_NOTES = True
cfg.APPLY_CORRECTIONS = True
cfg.TEST_CLAIM_IDS_FILE = test_file
cfg.CHECK_HIGH_RISK = True
cfg.CHECK_SLIDING_FEE = True
cfg.MSGBOX_TIMEOUT_SECONDS = 7
if test_file:
# The [TEST] prefix is added inside msgbox itself (titled()), so
# here the layer under info_box records instead - the title the
# user would actually see.
msgbox.info = lambda t, m, s: events.append(("box", t, m)) or \
msgbox.OK
else:
msgbox.info_box = lambda t, m: events.append(("box", t, m))
# SAVE_CORRECTIONS=1 only: the save decision box, recorded and declined
# (so save_claim, still forbidden, is never reached).
msgbox.ok_cancel = (
(lambda t, m, default_cancel=False:
events.append(("decision", t, m)) or False) if save
else _forbidden)
claim_form.open_claim = open_claim or (lambda page, cid: True)
claim_form.wait_claim_loaded = lambda page, cid: None
claim_form.is_dental_claim_form = lambda page: dental
claim_form.open_cpt_hcpcs_tab = lambda page: None
claim_form.extract = lambda page: {k: (dict(v) if isinstance(v, dict)
else list(v) if isinstance(v, list)
else v)
for k, v in _UB.items()}
claim_form.save = lambda data: pathlib.Path("unused.json")
claim_form.close_claim = lambda page: events.append(("close",))
claim_form.save_claim = _forbidden
claim_form.open_progress_notes = lambda page: dict(_NOTE)
flow.conditions.CONDITIONS = [_fake_condition(fires, detect_raises)]
if describe_raises:
flow.conditions.BY_NAME = dict(flow.conditions.BY_NAME,
depression_dx_pointer=_DescribeRaises)
flow.claim_results.search_claim = search or (lambda page, cid: {
"id": cid, "count": 1, "found": True, "seconds": 3.0})
flow.claim_filters.clear_claim_no = lambda page: None
cfg.RESULTS_DB = False
flow._write_charge_summary = lambda info, outcomes: written.append(
outcomes)
flow.dental_notes.check = lambda page, cid: (
{"high_risk": True, "match": "HIGH RISK"} if high_risk
else {"high_risk": False, "match": ""})
flow.sliding_fee.check = lambda page, cid, data: {
"verdict": "active", "reason": "DOS inside the range",
"dos": "09/01/2026", "from_date": "01/01/2026",
"to_date": "12/31/2026"}
out = io.StringIO()
try:
with contextlib.redirect_stdout(out):
# No page to recover on: flows/recovery.py's steps are
# its own tests' business (test_recovery.py).
real_recover = flow.recovery.recover
flow.recovery.recover = lambda *a: "cleared"
try:
flow._process_claims(None, {"ids": ["9103000"], "file": None})
finally:
flow.recovery.recover = real_recover
finally:
_restore()
return events, out.getvalue(), written[0]
def _boxes(events):
return [e for e in events if e[0] == "box"]
# ── corrected UB/HCFA claim ─────────────────────────────────────────────────
@case
def corrected_claim_with_review_on_gets_exactly_one_box_before_close():
events, out, outcomes = _run(fires=True)
boxes = _boxes(events)
assert len(boxes) == 1, events
# Shown while the form is still open: before close_claim, not after.
assert events[0][0] == "box" and events[1] == ("close",), events
_, title, body = boxes[0]
assert title == "eCW Bot — Claim 9103000: review before close", title
# The per-claim header _process_claims builds ...
for want in ("Claim 1/1", "Claim No : 9103000", "Found : Yes",
"Total Counts : 1", "Time taken : 3s"):
assert want in body, f"{want!r} missing:\n{body}"
# ... what the condition changed (its own describe()) ...
assert "G8510" in body, body
# ... the claim's Summary line, as the sheet will say it ...
oc = outcomes["9103000"]
assert oc["Category"] == "Fail", oc
assert (f"Summary : Fail — {oc['Current Status']}") in body, body
# ... and the footer: nothing saved, OK closes with Cancel, auto-close.
assert "Nothing is saved" in body and "close the claim with Cancel" \
in body and "after 7s" in body, body
# The before-apply box is gone; its log line stays.
assert "1 condition(s) matched — about to apply: fake_cond" in out, out
@case
def corrected_claim_with_review_off_gets_one_box_after_close():
events, _, _ = _run(fires=True, review=False)
boxes = _boxes(events)
assert len(boxes) == 1, events
assert events[0] == ("close",) and events[1][0] == "box", events
_, title, body = boxes[0]
assert title == "eCW Bot — Claim 9103000", title
assert "Changed : depression_dx_pointer" in body, body
assert "Summary : Fail — Corrections pending (not saved)" in body, \
body
@case
def save_on_keeps_the_decision_box_then_the_per_claim_box():
"""
SAVE_CORRECTIONS=1 (brief item 4): the save decision box is unchanged
and the per-claim info box still follows after close. No review box:
the decision box is already the pause before closing.
"""
events, _, _ = _run(fires=True, save=True)
kinds = [e[0] for e in events]
assert kinds == ["decision", "close", "box"], events
assert events[0][1] == "eCW Bot — Claim 9103000: 1 condition(s) fired", \
events
assert events[2][1] == "eCW Bot — Claim 9103000", events
@case
def corrected_claim_with_every_box_off_shows_nothing():
events, _, _ = _run(fires=True, review=False, show_each=False)
assert _boxes(events) == [], events
@case
def clean_claim_says_nothing_needed_changing_and_pass():
events, _, outcomes = _run(fires=False)
boxes = _boxes(events)
assert len(boxes) == 1, events
assert events[0] == ("close",), events
_, title, body = boxes[0]
assert title == "eCW Bot — Claim 9103000", title
assert "Changed : nothing needed changing" in body, body
assert outcomes["9103000"]["Category"] == "Pass", outcomes
assert "Summary : Pass — Ready to Submit" in body, body
# The old wording that read as "nothing fired" is gone.
assert "none matched" not in body, body
# The form detail is there, with no "?" lines.
assert "Patient : Doe, Jane (34 F)" in body, body
assert not re.search(r":\s*\?\s*$", body, re.M), body
@case
def a_review_box_that_raises_still_closes_the_claim():
"""
Unit F review, finding 3: the review box is built in
_read_claim_form's `finally`, before close_claim. A condition whose
describe() raises must not skip close_claim (the form would stay open
with the edits on it) nor end the run. The claim gets its per-claim box
after close instead, and the Charge Summary is still written.
describe() raises first inside the read (_confirm_and_save logs it), so
the claim is an error; the `finally` then calls describe() again
for the review box, and before the fix that second raise escaped
_read_claim_form and _process_claims both.
Updated in the final review (2026-10-05): the error came AFTER the
conditions ran, and their account is now kept on the error dict — so
the box and the sheet say "failed after the conditions ran", not
"Could not read this claim", and the condition that changed the form
is still named (this case used to assert the "Could not read" line).
"""
events, out, outcomes = _run(fires=True, describe_raises=True)
assert events[0] == ("close",), events
boxes = _boxes(events)
assert len(boxes) == 1, events
_, title, body = boxes[0]
assert title == "eCW Bot — Claim 9103000", title
assert ("The bot failed after the conditions ran: unexpected change "
"shape") in body, body
assert "Could not read this claim" not in body, body
assert "Changed : depression_dx_pointer" in body, body
c = outcomes["9103000"]["Comments"]
assert ("Bot failed after the conditions ran (unexpected change shape)"
in c and c.startswith("ICD pointer issue")), c
assert "[COND 9103000] review box failed: unexpected change shape" \
in out, out
# The run went on to the Charge Summary, with the claim on it.
assert outcomes["9103000"]["Category"] == "Review", outcomes
@case
def a_failed_condition_with_nothing_changed_has_no_changed_line():
"""
Unit F review, finding 2: the brief allows two forms of the line,
"<names>" and "nothing needed changing". When nothing fired but a
condition raised, the second would call an unchecked claim clean, so
there is no Changed line; the Needs review line says what failed.
"""
events, _, outcomes = _run(detect_raises=True)
_, _, body = _boxes(events)[0]
assert "Changed" not in body, body
assert "Needs review : fake_cond failed:" in body, body
assert outcomes["9103000"]["Category"] == "Review", outcomes
assert "Summary : Review — Needs manual review" in body, body
def _box_for(**extra):
"""The per-claim box body for a _UB read with `extra` merged in, with
the conditions run and nothing changed."""
data = {k: (dict(v) if isinstance(v, dict) else
list(v) if isinstance(v, list) else v)
for k, v in _UB.items()}
data["progress_notes"] = dict(_NOTE)
data["_conditions"] = {"fired": [], "saved": False, "failed": [],
"review": [], "comments": {}}
data.update(extra)
outcome = flow.charge_summary.claim_outcome(True, data)
return flow._claim_box("Claim 1/1", "9103000", data, outcome)[1]
@case
def nothing_needed_changing_is_never_said_of_a_claim_that_was_not_checked():
"""
Final review, 2026-10-05: the box said "Changed : nothing needed
changing" right above "Needs review : Progress note not read" — any
review reason at all means the claim was not fully checked.
"""
for extra, reason in (
({"progress_notes": {"opened": "nothing"}},
"Needs review : Progress note not read (nothing)"),
({"cpt": []}, "Needs review : No CPT rows were read")):
body = _box_for(**extra)
assert "Changed" not in body, body
assert reason in body, body
assert "Summary : Review — Needs manual review" in body, body
# The control: a fully read claim still says it.
body = _box_for()
assert "Changed : nothing needed changing" in body, body
assert "Summary : Pass" in body, body
@case
def a_not_loaded_reason_is_shown_once_not_twice():
"""
The box leads with NOT_LOADED_NOTE and the reason under it, so
_findings_lines leaves `_not_evaluated` out of its Needs review lines;
without that filter the client saw the same reason twice.
"""
why = "HCFA form did not finish loading in 20s (0 CPT rows)"
data = {"claim_no": "9103000", "cpt": [], "icd": [], "revenue": [],
"_not_evaluated": why,
"progress_notes": {"opened": "skipped"}}
outcome = flow.charge_summary.claim_outcome(True, data)
body = flow._claim_box("Claim 1/1", "9103000", data, outcome)[1]
assert body.count(why) == 1, body
assert claim_form.NOT_LOADED_NOTE in body, body
@case
def a_search_that_raised_shows_no_question_mark_count():
"""
Unit F review, finding 4: _process_claims records count "?" when
search_claim raises; the box leaves the line out.
Final review, 2026-10-05: and it no longer says "Found : No" — a search
that raised established nothing about the queue. The box says the
search failed and why (updated: this case used to assert "Found : No").
"""
events, _, _ = _run(search=_search_raises)
boxes = _boxes(events)
assert len(boxes) == 1, events
_, _, body = boxes[0]
assert "Found : Unknown — the search failed" in body, body
assert "Found : No" not in body, body
assert "Could not search this claim:" in body, body
# Said once, by that line — not again as a "Needs review" line.
assert "Needs review : Bot could not search" not in body, body
assert "Summary : Review — Needs manual review" in body, body
assert "Total Counts" not in body and "?" not in body, body
@case
def a_zero_count_is_still_shown():
"""Control: 0 is a fact (searched, nothing came back), not a blank."""
head = flow._claim_head(1, 1, "9103000",
{"count": 0, "found": False, "seconds": 2.0})
assert "Total Counts : 0" in head, head
# ── Dental ──────────────────────────────────────────────────────────────────
@case
def dental_claim_gets_one_box_with_its_results_and_no_question_marks():
events, _, _ = _run(dental=True)
boxes = _boxes(events)
assert len(boxes) == 1, events
# After close, like every per-claim box: _dental_checks has no box now.
assert events[0] == ("close",), events
_, title, body = boxes[0]
assert title == "eCW Bot — Dental claim 9103000: HIGH RISK, " \
"sliding ACTIVE", title
assert "High risk : YES" in body, body
assert "Sliding fee : ACTIVE" in body, body
assert "Read-only — nothing was changed in eCW." in body, body
# claim_form.summary() is not used for a Dental read: no form fields.
for gone in ("Patient", "Charges", "Acct No", "ICD 0"):
assert gone not in body, f"{gone!r} in:\n{body}"
assert "?" not in body, body
assert "Summary : Fail" in body, body
@case
def dental_claim_shows_no_box_at_all_with_show_each_claim_off():
"""P6: the Dental box used to appear whatever was set."""
events, _, outcomes = _run(dental=True, show_each=False)
assert _boxes(events) == [], events
# Still evaluated and recorded - only the box is off.
assert outcomes["9103000"]["Category"] == "Fail", outcomes
# ── a claim that could not be read ──────────────────────────────────────────
def _cannot_open(page, cid):
raise RuntimeError("Timeout 30000ms exceeded.\nCall log:\n - waiting")
@case
def read_error_gets_one_box_with_the_error_line():
events, _, _ = _run(open_claim=_cannot_open)
boxes = _boxes(events)
assert len(boxes) == 1, events
_, title, body = boxes[0]
assert "Could not read this claim: Timeout 30000ms exceeded." in body, \
body
assert "Summary : Review — Needs manual review" in body, body
assert "Patient" not in body and "?" not in body, body
# ── test mode ───────────────────────────────────────────────────────────────
@case
def test_mode_prefixes_every_box_title():
events, _, _ = _run(fires=False, test_file="C:/x/Standard Claims.xlsx")
boxes = _boxes(events)
assert len(boxes) == 1, events
assert boxes[0][1] == "[TEST] eCW Bot — Claim 9103000", boxes
@case
def test_mode_prefixes_the_review_before_close_box_too():
events, _, _ = _run(fires=True, test_file="C:/x/Standard Claims.xlsx")
boxes = _boxes(events)
assert len(boxes) == 1, events
assert boxes[0][1] == ("[TEST] eCW Bot — Claim 9103000: review before "
"close"), boxes
@case
def a_live_run_has_no_test_prefix():
events, _, _ = _run(fires=False)
assert not _boxes(events)[0][1].startswith("[TEST]"), events
# ── the strings themselves ──────────────────────────────────────────────────
@case
def summary_leaves_out_lines_that_came_back_empty_or_question_mark():
text = claim_form.summary({
"claim_no": "9109000",
"patient": {"name": "?", "age_sex": "", "account_no": ""},
"header": {"claim_date": "?", "facility": None},
"totals": {},
"cpt": [], "icd": [], "revenue": []})
assert "Claim No : 9109000" in text, text
for gone in ("Patient", "Acct No", "Claim Date", "Status", "Facility",
"Charges", "Balance"):
assert gone not in text, f"{gone!r} in:\n{text}"
assert "?" not in text, text
# The counts line stays: 0 is a fact, not a blank.
assert "ICD 0 CPT 0 Revenue 0" in text, text
@case
def summary_of_an_empty_dict_has_no_question_marks():
"""What a Dental dict used to get: a page of '?'."""
text = claim_form.summary({})
assert "?" not in text, text
@case
def summary_keeps_a_status_with_only_its_label():
text = claim_form.summary({"header": {"claim_status_label": "Medcor Ready",
"claim_status_code": ""}})
assert "Status : Medcor Ready" in text, text
assert "()" not in text, text
@case
def summary_says_and_n_more_in_english():
rows = [{"code": f"9921{i}", "sdos": "09/01/2026", "units": "1"}
for i in range(7)]
text = claim_form.summary({"cpt": rows})
assert " ... and 2 more" in text, text
@case
def progress_note_summary_says_yes_or_no_for_chart_locked():
assert "Chart locked: Yes" in progress_note.summary(
{"chart_locked": True})
assert "Chart locked: No" in progress_note.summary(
{"chart_locked": False})
@case
def progress_note_summary_skips_rows_with_no_code():
text = progress_note.summary({
"assessments": [{"code": ""}, {"code": "R69", "primary": True}],
"procedure_codes": [{"code": None}, {"code": "H0023",
"modifiers": ["GT"]}]})
assert "Assessment : R69 (Primary)" in text, text
assert "Procedure : H0023 mod GT" in text, text
assert "?" not in text, text
assert text.count("Assessment") == 1 and text.count("Procedure") == 1, \
text
@case
def no_tamil_word_is_left_anywhere_in_src():
"""
The words the review found in message-box strings, and the others the
brief names. Scanned over all of src/, not just the boxes found so far,
so a new box cannot bring one back.
"""
words = re.compile(r"\b(innum|aamaa|illa|aagum|pannu)\b", re.I)
hits = []
for path in (ROOT / "src").rglob("*.py"):
for n, line in enumerate(path.read_text(encoding="utf-8")
.splitlines(), start=1):
if words.search(line):
hits.append(f"{path.relative_to(ROOT)}:{n}: {line.strip()}")
assert not hits, "\n".join(hits)
def main() -> int:
failures = 0
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:
_restore()
print(f"\n{len(CASES) - failures}/{len(CASES)} passed")
return 1 if failures else 0
if __name__ == "__main__":
sys.exit(main())