ManikandanandClaude Opus 5.5
d500f750da
Snapshot of D:\Ecw Process a11c9ff (2026-10-30). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
418 lines
15 KiB
Python
418 lines
15 KiB
Python
r"""
|
|
Every claim starts from a clean search page (operator, 2026-10-12): what
|
|
the last claim left open is closed, Claim No and Lookup are checked to be
|
|
on screen with nothing over them, and only then is the claim searched. A
|
|
page that will not come clean makes that ONE claim Review — the next claim
|
|
clears and checks again. Live run that day: claim 1's viewer stayed open
|
|
and claims 2 and 3 were never searched.
|
|
|
|
python tests\test_clean_between_claims.py
|
|
"""
|
|
|
|
import io
|
|
import sys
|
|
import pathlib
|
|
import inspect
|
|
import contextlib
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT / "src"))
|
|
|
|
FIXTURES = pathlib.Path(__file__).resolve().parent / "fixtures"
|
|
HCFA = FIXTURES / "hcfa_claim.html"
|
|
|
|
from ecw_bot import settings as cfg # noqa: E402
|
|
from ecw_bot.flows import export_medcor_ready as flow # noqa: E402
|
|
from ecw_bot.pages import claim_form, claim_results # noqa: E402
|
|
from ecw_bot.selectors import claim_lookup as lsel # noqa: E402
|
|
|
|
CASES = []
|
|
|
|
|
|
def case(fn):
|
|
CASES.append(fn)
|
|
return fn
|
|
|
|
|
|
def _quiet(fn, *a, **k):
|
|
with contextlib.redirect_stdout(io.StringIO()) as out:
|
|
result = fn(*a, **k)
|
|
return result, out.getvalue()
|
|
|
|
|
|
# The two controls the search needs, as eCW's Claim Lookup has them.
|
|
_LOOKUP = """
|
|
<div id="filters" style="padding:20px">
|
|
<input id="claimLookupIpt10" ng-model="_InvId" style="width:200px">
|
|
<button id="btnclaimlookup">Lookup</button>
|
|
</div>
|
|
"""
|
|
|
|
# A viewer-like dialog over the whole page, with a patient's name in it.
|
|
_COVER = """
|
|
<div class="modal in" id="pnViewer" style="display:block;position:fixed;
|
|
left:0;top:0;width:100%;height:100%;z-index:1050;background:#fff">
|
|
PATIENT DOE, JANE</div>
|
|
"""
|
|
|
|
|
|
# ── the check before the search ──────────────────────────────────────────────
|
|
|
|
@case
|
|
def a_clean_search_page_is_ready(page):
|
|
page.set_content(_LOOKUP)
|
|
assert claim_results.search_blocker(page) is None
|
|
|
|
|
|
@case
|
|
def a_dialog_over_lookup_is_named_without_its_text(page):
|
|
page.set_content(_LOOKUP + _COVER)
|
|
why = claim_results.search_blocker(page)
|
|
assert why and "covered by" in why and "pnViewer" in why, why
|
|
assert "DOE" not in why and "JANE" not in why, why
|
|
|
|
|
|
@case
|
|
def a_missing_lookup_button_is_named(page):
|
|
page.set_content('<input id="claimLookupIpt10">')
|
|
why = claim_results.search_blocker(page)
|
|
assert why == "Lookup is not on screen", why
|
|
|
|
|
|
@case
|
|
def a_disabled_lookup_button_is_named(page):
|
|
page.set_content(_LOOKUP.replace("<button id", "<button disabled id"))
|
|
assert claim_results.search_blocker(page) == "Lookup is disabled"
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def _ready_wait(ms):
|
|
"""The search-ready wait shortened for a test that has to sit it out."""
|
|
orig = claim_results._SEARCH_READY_MS
|
|
claim_results._SEARCH_READY_MS = ms
|
|
try:
|
|
yield
|
|
finally:
|
|
claim_results._SEARCH_READY_MS = orig
|
|
|
|
|
|
@case
|
|
def search_claim_types_nothing_when_the_page_is_not_ready(page):
|
|
page.set_content(_LOOKUP + _COVER)
|
|
with _ready_wait(1_000):
|
|
try:
|
|
_quiet(claim_results.search_claim, page, "9103000")
|
|
except RuntimeError as exc:
|
|
assert "not ready after 1s" in str(exc), exc
|
|
assert "covered by" in str(exc), exc
|
|
else:
|
|
raise AssertionError("search_claim went ahead over a dialog")
|
|
assert page.locator(lsel.SEL_CLAIM_NO).input_value() == ""
|
|
|
|
|
|
@case
|
|
def a_page_taking_no_clicks_is_still_loading_not_covered(page):
|
|
"""Live run 2026-10-13, claim 9126000: eCW loading, the click fell
|
|
through to <html> — logged as "covered by html.jelly-page"."""
|
|
page.set_content(_LOOKUP)
|
|
page.evaluate("() => { document.body.style.pointerEvents = 'none'; }")
|
|
why = claim_results.search_blocker(page)
|
|
assert why == "Claim No is not taking clicks yet (eCW still loading)", why
|
|
|
|
|
|
@case
|
|
def the_route_spinner_on_screen_means_still_loading(page):
|
|
page.set_content(_LOOKUP + '<div id="routeLoader" '
|
|
'style="width:40px;height:40px">...</div>')
|
|
why = claim_results.search_blocker(page)
|
|
assert why == "eCW is still loading (its spinner is on screen)", why
|
|
|
|
|
|
@case
|
|
def a_page_busy_for_a_moment_is_ready_once_it_comes_free(page):
|
|
"""The claim is searched, not sent to Review, when the page clears
|
|
within the wait."""
|
|
page.set_content(_LOOKUP + _COVER)
|
|
page.evaluate("() => setTimeout(() => "
|
|
"document.getElementById('pnViewer').remove(), 1200)")
|
|
why, out = _quiet(claim_results.wait_search_ready, page, "9103000")
|
|
assert why is None, (why, out)
|
|
assert "not ready yet" in out and "came free after" in out, out
|
|
assert "DOE" not in out and "JANE" not in out, out
|
|
|
|
|
|
@case
|
|
def a_page_that_stays_busy_gives_its_last_reason_after_the_wait(page):
|
|
page.set_content(_LOOKUP + _COVER)
|
|
with _ready_wait(1_000):
|
|
why, _ = _quiet(claim_results.wait_search_ready, page, "9103000")
|
|
assert why and "covered by" in why and "pnViewer" in why, why
|
|
|
|
|
|
# ── clearing the screen ──────────────────────────────────────────────────────
|
|
|
|
@case
|
|
def a_clean_page_needs_nothing_and_says_nothing(page):
|
|
page.set_content(_LOOKUP)
|
|
done, out = _quiet(claim_form.clear_screen, page, "9103000",
|
|
page.context.pages)
|
|
assert done == [] and out == "", (done, out)
|
|
|
|
|
|
@case
|
|
def a_tab_opened_during_a_claim_is_closed_and_the_bot_tabs_kept(page):
|
|
page.set_content(_LOOKUP)
|
|
keep = list(page.context.pages)
|
|
reader = page.context.new_page() # e.g. a note reader tab
|
|
done, out = _quiet(claim_form.clear_screen, page, "9103000", keep)
|
|
assert reader.is_closed(), out
|
|
assert not page.is_closed()
|
|
assert "an extra browser tab" in done, (done, out)
|
|
|
|
|
|
@case
|
|
def a_claim_form_left_open_is_cancelled(page):
|
|
page.goto(HCFA.as_uri())
|
|
assert claim_form.any_claim_form_open(page)
|
|
done, out = _quiet(claim_form.clear_screen, page, "9103000",
|
|
page.context.pages)
|
|
assert "a claim form" in done, (done, out)
|
|
assert not claim_form.any_claim_form_open(page), out
|
|
|
|
|
|
@case
|
|
def a_leftover_form_is_described_in_the_log_before_it_is_cancelled(page):
|
|
page.goto(HCFA.as_uri())
|
|
_, out = _quiet(claim_form.clear_screen, page, "9103000",
|
|
page.context.pages)
|
|
assert "Open form found:" in out, out
|
|
assert "a click there hits" in out, out
|
|
|
|
|
|
@case
|
|
def a_ghost_form_says_how_it_is_hidden_and_what_takes_its_click(page):
|
|
"""The evidence for a form the bot sees open that a person does not."""
|
|
page.set_content("""
|
|
<div class="modal fade claimModalBackdrop" id="ghost"
|
|
style="display:block;opacity:0;position:fixed;left:0;top:0;
|
|
width:400px;height:200px">
|
|
<button ng-click="closeClaimScreen()" style="margin:40px">Cancel
|
|
</button>PATIENT DOE, JANE</div>
|
|
<div id="cover" style="position:fixed;left:0;top:0;width:400px;
|
|
height:200px;z-index:5">grid</div>
|
|
""")
|
|
lines = claim_form.describe_shown(page, claim_form.sel.SEL_CANCEL_HCFA)
|
|
assert len(lines) == 1, lines
|
|
line = lines[0]
|
|
assert "opacity 0" in line, line
|
|
assert "div#ghost" in line, line
|
|
assert "hits div#cover" in line, line
|
|
assert "DOE" not in line and "JANE" not in line, line
|
|
|
|
|
|
# ── a claim window is Cancelled, never Escaped ───────────────────────────────
|
|
# Escape drops the HCFA window without closeClaimScreen() (the fixture's own
|
|
# keydown handler does what uib does live); `closeClaimScreenRan` tells a
|
|
# real Cancel apart from that.
|
|
|
|
class _Stop(Exception):
|
|
pass
|
|
|
|
|
|
def _clear_with_cancel_failing(page):
|
|
orig = claim_form._close_leftover_form
|
|
|
|
def will_not_close(page, claim_id):
|
|
raise RuntimeError("would not close")
|
|
claim_form._close_leftover_form = will_not_close
|
|
try:
|
|
return _quiet(claim_form.clear_screen, page, "9103000",
|
|
page.context.pages)
|
|
finally:
|
|
claim_form._close_leftover_form = orig
|
|
|
|
|
|
@case
|
|
def clear_screen_cancels_a_claim_form_and_does_not_escape_it(page):
|
|
page.goto(HCFA.as_uri())
|
|
done, out = _quiet(claim_form.clear_screen, page, "9103000",
|
|
page.context.pages)
|
|
assert "a claim form" in done, (done, out)
|
|
assert page.evaluate("window.closeClaimScreenRan") is True, out
|
|
assert "LAST RESORT" not in out, out
|
|
|
|
|
|
@case
|
|
def an_hcfa_form_cancel_will_not_close_is_escaped_as_the_last_resort(page):
|
|
"""Operator, 2026-10-14: left up, it covered the search page and every
|
|
later claim of the run went to Review."""
|
|
page.goto(HCFA.as_uri())
|
|
done, out = _clear_with_cancel_failing(page)
|
|
assert "LAST RESORT" in out and "still up (would not close)" in out, out
|
|
assert "closeClaimScreen() does NOT run" in out, out
|
|
assert "a claim form" in done, (done, out)
|
|
assert not claim_form.any_claim_form_open(page), out
|
|
assert page.evaluate("window.closeClaimScreenRan") is False
|
|
|
|
|
|
@case
|
|
def a_ub_form_cancel_will_not_close_is_not_escaped(page):
|
|
page.goto((FIXTURES / "ub_claim.html").as_uri())
|
|
assert claim_form.any_claim_form_open(page)
|
|
done, out = _clear_with_cancel_failing(page)
|
|
assert "LAST RESORT" not in out, out
|
|
assert "Could not close a claim form" in out, out
|
|
assert "a claim form" not in done, (done, out)
|
|
assert claim_form.any_claim_form_open(page), out
|
|
|
|
|
|
@case
|
|
def open_claim_cancels_a_form_already_up_and_never_escapes_it(page):
|
|
"""eCW opens the next HCFA claim's form by itself after the Lookup (live
|
|
run 2026-10-13): open_claim()'s stray-panel check came first and would
|
|
have Escaped it."""
|
|
page.goto(HCFA.as_uri())
|
|
orig = claim_form._close_leftover_form
|
|
|
|
def then_stop(page, claim_id):
|
|
orig(page, claim_id)
|
|
raise _Stop()
|
|
claim_form._close_leftover_form = then_stop
|
|
try:
|
|
_quiet(claim_form.open_claim, page, "9103000")
|
|
except _Stop:
|
|
pass
|
|
finally:
|
|
claim_form._close_leftover_form = orig
|
|
assert page.evaluate("window.closeClaimScreenRan") is True, \
|
|
"the form was gone before _close_leftover_form() could Cancel it"
|
|
assert not claim_form.any_claim_form_open(page)
|
|
|
|
|
|
# ── the loop ─────────────────────────────────────────────────────────────────
|
|
|
|
def _run(ids, search, events=None):
|
|
"""
|
|
`_process_claims` with the page stubbed out. Clears are appended to
|
|
`events` (the search can append to the same list); returns it and the
|
|
outcomes.
|
|
"""
|
|
events = [] if events is None else events
|
|
outcomes = {}
|
|
orig = (cfg.SHOW_EACH_CLAIM, cfg.SHOW_CLAIM_IDS, cfg.CLAIM_PROCESS_LIMIT,
|
|
flow.claim_results.search_claim, flow._read_claim_form,
|
|
flow.claim_filters.clear_claim_no, flow._write_charge_summary,
|
|
flow._save_results_db, flow.claim_form.clear_screen)
|
|
cfg.SHOW_EACH_CLAIM = False
|
|
cfg.SHOW_CLAIM_IDS = False
|
|
cfg.CLAIM_PROCESS_LIMIT = 0
|
|
flow.claim_results.search_claim = search
|
|
flow._read_claim_form = lambda page, cid, head: {
|
|
"claim_no": cid,
|
|
"cpt": [{"code": "99213", "checked": True, "fee": "100.00"}],
|
|
"icd": [{"code": "Z00.00"}],
|
|
"progress_notes": {"opened": "modal", "text": "Office visit."},
|
|
"_conditions": {"fired": [], "saved": False, "failed": [],
|
|
"review": [], "comments": {}}}
|
|
flow.claim_filters.clear_claim_no = lambda page: None
|
|
flow._write_charge_summary = lambda info, o: outcomes.update(o)
|
|
flow._save_results_db = lambda info, o, d: None
|
|
flow.claim_form.clear_screen = (
|
|
lambda page, cid, keep=(): events.append(("clear", cid)) or [])
|
|
try:
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
# 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": list(ids), "file": None})
|
|
finally:
|
|
flow.recovery.recover = real_recover
|
|
finally:
|
|
(cfg.SHOW_EACH_CLAIM, cfg.SHOW_CLAIM_IDS, cfg.CLAIM_PROCESS_LIMIT,
|
|
flow.claim_results.search_claim, flow._read_claim_form,
|
|
flow.claim_filters.clear_claim_no, flow._write_charge_summary,
|
|
flow._save_results_db, flow.claim_form.clear_screen) = orig
|
|
return events, outcomes
|
|
|
|
|
|
@case
|
|
def every_claim_is_cleared_before_its_search_and_the_run_ends_clear():
|
|
events = []
|
|
|
|
def search(page, cid):
|
|
events.append(("search", cid))
|
|
return {"id": cid, "count": 1, "found": True, "seconds": 1.0}
|
|
_run(["1", "2"], search, events)
|
|
assert events == [("clear", "1"), ("search", "1"),
|
|
("clear", "2"), ("search", "2"),
|
|
("clear", "2")], events
|
|
|
|
|
|
@case
|
|
def a_page_that_will_not_clear_costs_one_claim_not_the_rest():
|
|
searched = []
|
|
|
|
def search(page, cid):
|
|
if cid == "1":
|
|
raise RuntimeError("The search page is not ready: Lookup is "
|
|
"covered by div#pnViewer.modal")
|
|
searched.append(cid)
|
|
return {"id": cid, "count": 1, "found": True, "seconds": 1.0}
|
|
_, outcomes = _run(["1", "2", "3"], search)
|
|
assert searched == ["2", "3"], searched
|
|
assert outcomes["1"]["Category"] == "Review", outcomes["1"]
|
|
assert "not ready" in outcomes["1"]["Comments"], outcomes["1"]
|
|
assert outcomes["2"]["Category"] == "Pass", outcomes["2"]
|
|
assert outcomes["3"]["Category"] == "Pass", outcomes["3"]
|
|
|
|
|
|
def _run_cases() -> int:
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
failures = 0
|
|
with sync_playwright() as pw:
|
|
browser = pw.chromium.launch(headless=True)
|
|
# A context of its own, so a test can open a second tab in it.
|
|
page = browser.new_context(
|
|
viewport={"width": 1600, "height": 1000}).new_page()
|
|
for fn in CASES:
|
|
name = fn.__name__.replace("_", " ")
|
|
try:
|
|
if len(inspect.signature(fn).parameters) == 0:
|
|
fn()
|
|
else:
|
|
fn(page)
|
|
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
|
|
browser.close()
|
|
|
|
print(f"\n{len(CASES) - failures}/{len(CASES)} passed")
|
|
return 1 if failures else 0
|
|
|
|
|
|
def main() -> int:
|
|
"""
|
|
shot() writes into the operator's dated Screenshots\\ folder; a test
|
|
run must not, so the cases run with it pointed at a temp folder.
|
|
"""
|
|
import tempfile
|
|
from ecw_bot import settings as _cfg
|
|
orig = _cfg.SCREENSHOTS
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
_cfg.SCREENSHOTS = pathlib.Path(tmp)
|
|
try:
|
|
return _run_cases()
|
|
finally:
|
|
_cfg.SCREENSHOTS = orig
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|