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>
315 lines
12 KiB
Python
315 lines
12 KiB
Python
r"""
|
|
conditions/medicare_wellness_history.py — document #27 G0402 (once per
|
|
lifetime), #28 G0438 (once in 3 years), #29 G0439 (once a year) on Medicare;
|
|
the history comes from Account Inquiry (pages/account_inquiry.py). Check
|
|
only: nothing on the claim changes (operator, 2026-10-20).
|
|
|
|
python tests\test_conditions_medicare_wellness_history.py
|
|
|
|
Browser cases use tests/fixtures/account_inquiry.html (made-up claims).
|
|
"""
|
|
|
|
import io
|
|
import sys
|
|
import inspect
|
|
import pathlib
|
|
import contextlib
|
|
from datetime import date
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT / "src"))
|
|
|
|
FIXTURE = (pathlib.Path(__file__).resolve().parent / "fixtures"
|
|
/ "account_inquiry.html")
|
|
|
|
from ecw_bot.conditions import medicare_wellness_history as cond # noqa: E402
|
|
from ecw_bot.conditions import CONDITIONS # noqa: E402
|
|
from ecw_bot.conditions.base import ConditionResult # noqa: E402
|
|
from ecw_bot.core import charge_summary as cs # noqa: E402
|
|
from ecw_bot.pages import account_inquiry # noqa: E402
|
|
|
|
CASES = []
|
|
DOS = date(2026, 10, 1)
|
|
|
|
|
|
def case(fn):
|
|
CASES.append(fn)
|
|
return fn
|
|
|
|
|
|
def _claim(payer, *lines, claim_no="9000001"):
|
|
"""lines: (code, fee, billed) or a bare code (10.00, billed)."""
|
|
cpt = []
|
|
for ln in lines:
|
|
code, fee, billed = (ln, "10.00", True) if isinstance(ln, str) else ln
|
|
cpt.append({"code": code, "billed_fee": fee, "billed": billed,
|
|
"sdos": "10/01/2026"})
|
|
return {"claim_no": claim_no, "screen": "ub",
|
|
"insurance": [{"Name": payer}] if payer else [], "cpt": cpt}
|
|
|
|
|
|
def _h(claim_no, when, *codes):
|
|
return {"claim_no": claim_no, "date": when, "codes": list(codes)}
|
|
|
|
|
|
# ── detect (no browser) ──────────────────────────────────────────────────────
|
|
|
|
@case
|
|
def detect_fires_on_medicare_with_a_wellness_code():
|
|
for code in ("G0402", "G0438", "G0439"):
|
|
assert cond.detect("C1", _claim("Medicare", "99213", code)), code
|
|
|
|
|
|
@case
|
|
def detect_ignores_other_payers_and_other_codes():
|
|
assert not cond.detect("C1", _claim("Medicaid", "G0438"))
|
|
assert not cond.detect("C1", _claim("", "G0438"))
|
|
assert not cond.detect("C1", _claim("Medicare", "99213", "G0008"))
|
|
assert not cond.detect("C1", {})
|
|
|
|
|
|
@case
|
|
def a_line_that_is_not_billed_does_not_count():
|
|
assert not cond.detect("C1", _claim("Medicare", ("G0438", "0.00", True)))
|
|
assert not cond.detect("C1", _claim("Medicare", ("G0438", "0.00", False)))
|
|
# unchecked with a fee: priced_line_check checks it at the end.
|
|
assert cond.detect("C1", _claim("Medicare", ("G0438", "175.00", False)))
|
|
|
|
|
|
@case
|
|
def the_service_date_is_the_wellness_lines():
|
|
assert cond.claim_service_date(_claim("Medicare", "G0439")) == DOS
|
|
data = _claim("Medicare", "G0439")
|
|
data["cpt"][0]["sdos"] = ""
|
|
assert cond.claim_service_date(data) is None
|
|
|
|
|
|
# ── the rule (no browser) ────────────────────────────────────────────────────
|
|
|
|
@case
|
|
def g0402_billed_before_at_any_time_is_reported():
|
|
out = cond.conflicts(["G0402"], DOS, "9000001",
|
|
[_h("8000001", date(2015, 3, 4), "99213", "G0402")])
|
|
assert out == ["G0402 already billed on 03/04/2015 (claim 8000001) — "
|
|
"once per lifetime"], out
|
|
assert cond.conflicts(["G0402"], DOS, "9000001",
|
|
[_h("8000001", date(2025, 3, 4), "99213")]) == []
|
|
|
|
|
|
@case
|
|
def g0438_inside_3_years_is_reported_older_is_fine():
|
|
inside = [_h("8000002", date(2024, 6, 1), "G0438")]
|
|
out = cond.conflicts(["G0438"], DOS, "9000001", inside)
|
|
assert out == ["G0438 already billed on 06/01/2024 (claim 8000002) — "
|
|
"once in 3 years"], out
|
|
older = [_h("8000003", date(2022, 6, 1), "G0438")]
|
|
assert cond.conflicts(["G0438"], DOS, "9000001", older) == []
|
|
|
|
|
|
@case
|
|
def exactly_3_years_or_12_months_away_is_outside_the_window():
|
|
assert cond.conflicts(["G0438"], DOS, "9000001",
|
|
[_h("8", date(2023, 10, 1), "G0438")]) == []
|
|
assert cond.conflicts(["G0439"], DOS, "9000001",
|
|
[_h("8", date(2025, 10, 1), "G0439")]) == []
|
|
assert cond.conflicts(["G0439"], DOS, "9000001",
|
|
[_h("8", date(2025, 10, 2), "G0439")]) != []
|
|
|
|
|
|
@case
|
|
def g0439_inside_12_months_is_reported_older_is_fine():
|
|
out = cond.conflicts(["G0439"], DOS, "9000001",
|
|
[_h("8000004", date(2026, 1, 15), "G0439")])
|
|
assert out == ["G0439 already billed on 01/15/2026 (claim 8000004) — "
|
|
"once a year"], out
|
|
assert cond.conflicts(["G0439"], DOS, "9000001",
|
|
[_h("8000005", date(2025, 8, 1), "G0439")]) == []
|
|
|
|
|
|
@case
|
|
def the_most_recent_earlier_claim_is_named():
|
|
hist = [_h("A", date(2024, 1, 1), "G0438"),
|
|
_h("B", date(2025, 5, 5), "G0438")]
|
|
out = cond.conflicts(["G0438"], DOS, "9000001", hist)
|
|
assert "05/05/2025 (claim B)" in out[0], out
|
|
|
|
|
|
@case
|
|
def this_claim_itself_is_left_out():
|
|
hist = [_h("9000001", DOS, "G0439", "99213")]
|
|
assert cond.conflicts(["G0439"], DOS, "9000001", hist) == []
|
|
assert cond.conflicts(["G0402"], DOS, " 9000001 ", hist) == []
|
|
|
|
|
|
@case
|
|
def a_later_claim_inside_the_window_counts_too():
|
|
out = cond.conflicts(["G0439"], DOS, "9000001",
|
|
[_h("8000006", date(2026, 12, 1), "G0439")])
|
|
assert out and "12/01/2026" in out[0], out
|
|
|
|
|
|
@case
|
|
def unread_dates_are_sent_to_a_person():
|
|
hist = [_h("8000007", None, "G0438")]
|
|
out = cond.conflicts(["G0438"], DOS, "9000001", hist)
|
|
assert out and "no service date read" in out[0] and "by hand" in out[0]
|
|
out = cond.conflicts(["G0438"], None, "9000001",
|
|
[_h("8000007", date(2025, 1, 1), "G0438")])
|
|
assert out and "service date was not read" in out[0], out
|
|
# G0402 is lifetime: no date needed either side.
|
|
out = cond.conflicts(["G0402"], None, "9000001",
|
|
[_h("8000008", None, "G0402")])
|
|
assert out == ["G0402 already billed on an unread date (claim 8000008) "
|
|
"— once per lifetime"], out
|
|
|
|
|
|
@case
|
|
def each_code_on_the_claim_is_checked_on_its_own():
|
|
hist = [_h("8000009", date(2026, 3, 1), "G0439"),
|
|
_h("8000010", date(2010, 3, 1), "G0402")]
|
|
out = cond.conflicts(["G0402", "G0439"], DOS, "9000001", hist)
|
|
assert len(out) == 2 and out[0].startswith("G0402") \
|
|
and out[1].startswith("G0439"), out
|
|
|
|
|
|
# ── Account Inquiry and apply (browser) ──────────────────────────────────────
|
|
|
|
def _load(page, pages, **flags):
|
|
page.goto(FIXTURE.as_uri())
|
|
page.evaluate("(a) => { window.HISTORY_PAGES = a.pages; "
|
|
"Object.assign(window, a.flags); }",
|
|
{"pages": pages, "flags": flags})
|
|
|
|
|
|
@case
|
|
def account_inquiry_reads_every_page_and_closes_with_the_x(page):
|
|
_load(page, [[{"id": "8000011", "date": "02/03/2024",
|
|
"codes": ["99213", "G0438"]}],
|
|
[{"id": "8000012", "date": "11/12/2019",
|
|
"codes": ["G0402"]},
|
|
{"id": "8000013", "date": "", "codes": []}]])
|
|
hist = account_inquiry.history(page)
|
|
assert hist == [
|
|
{"claim_no": "8000011", "date": date(2024, 2, 3),
|
|
"codes": ["99213", "G0438"]},
|
|
{"claim_no": "8000012", "date": date(2019, 11, 12),
|
|
"codes": ["G0402"]},
|
|
{"claim_no": "8000013", "date": None, "codes": []}], hist
|
|
assert page.evaluate("window.CLOSED_WITH") == "X"
|
|
assert not page.locator('input[data-f="cpt"]').is_checked(), "Show CPT Payments left as it was found"
|
|
assert not page.locator("#piModal").is_visible()
|
|
assert page.locator("#formCancel").is_visible(), "claim form stays"
|
|
|
|
|
|
@case
|
|
def apply_reports_a_repeat_and_changes_nothing(page):
|
|
_load(page, [[{"id": "8000014", "date": "01/15/2026",
|
|
"codes": ["G0439"]}]])
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
result = cond.apply(page, "9000001", _claim("Medicare", "G0439"))
|
|
assert isinstance(result, ConditionResult)
|
|
assert result.changed is False and result.changes == []
|
|
assert result.review == ("G0439 already billed on 01/15/2026 (claim "
|
|
"8000014) — once a year"), result.review
|
|
assert "G0439 already billed" in cond.describe(result)
|
|
|
|
|
|
@case
|
|
def apply_with_a_clean_history_is_nothing(page):
|
|
_load(page, [[{"id": "8000015", "date": "01/15/2024",
|
|
"codes": ["G0439"]}]])
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
result = cond.apply(page, "9000001", _claim("Medicare", "G0439"))
|
|
assert result.changed is False and result.review == "", result
|
|
|
|
|
|
@case
|
|
def account_inquiry_that_does_not_open_is_review_by_hand(page):
|
|
_load(page, [[]], NO_INQUIRY=True)
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
result = cond.apply(page, "9000001", _claim("Medicare", "G0438"))
|
|
assert result.changed is False
|
|
assert "Account Inquiry not read" in result.review, result.review
|
|
assert "check G0438 history by hand" in result.review, result.review
|
|
|
|
|
|
@case
|
|
def account_inquiry_that_will_not_close_is_review_too(page):
|
|
_load(page, [[]], STUCK_OPEN=True)
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
result = cond.apply(page, "9000001", _claim("Medicare", "G0402"))
|
|
assert "did not close" in result.review, result.review
|
|
|
|
|
|
@case
|
|
def apply_leaves_its_verdict_on_the_claim_for_the_pps_code(page):
|
|
"""#25 must not add G0468 for a preventive visit billed too often, or
|
|
one whose history was not read: the verdict stays on the claim."""
|
|
_load(page, [[{"id": "8000014", "date": "01/15/2026",
|
|
"codes": ["G0439"]}]])
|
|
data = _claim("Medicare", "G0439")
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
cond.apply(page, "9000001", data)
|
|
assert data["_wellness_history"] == {"read": True, "conflicts": [
|
|
"G0439 already billed on 01/15/2026 (claim 8000014) — once a "
|
|
"year"]}, data.get("_wellness_history")
|
|
|
|
_load(page, [[{"id": "8000015", "date": "01/15/2024",
|
|
"codes": ["G0439"]}]])
|
|
data = _claim("Medicare", "G0439")
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
cond.apply(page, "9000001", data)
|
|
assert data["_wellness_history"] == {"read": True, "conflicts": []}
|
|
|
|
_load(page, [[]], NO_INQUIRY=True)
|
|
data = _claim("Medicare", "G0438")
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
cond.apply(page, "9000001", data)
|
|
assert data["_wellness_history"] == {"read": False, "conflicts": []}
|
|
|
|
|
|
# ── registry and Charge Summary ──────────────────────────────────────────────
|
|
|
|
@case
|
|
def it_is_registered_before_priced_line_check():
|
|
names = [m.NAME for m in CONDITIONS]
|
|
assert "medicare_wellness_history" in names
|
|
assert names.index("medicare_wellness_history") < names.index(
|
|
"priced_line_check"), names
|
|
|
|
|
|
@case
|
|
def the_charge_summary_knows_it():
|
|
assert "medicare_wellness_history" in cs.COMMENTS
|
|
|
|
|
|
def main() -> int:
|
|
from playwright.sync_api import sync_playwright
|
|
|
|
failures = 0
|
|
with sync_playwright() as pw:
|
|
browser = pw.chromium.launch(headless=True)
|
|
page = browser.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
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|