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())