r""" export_medcor_ready._claims_phase - the TEST_CLAIM_IDS_FILE override. python tests\test_test_claim_ids_file.py TEMPORARY testing feature: a hand-maintained "standard" sheet lists one known claim per implemented condition, so a run against it exercises every condition against a real claim without waiting on the full live-queue bulk search and export. `TEST_CLAIM_IDS_FILE` blank is the permanent, default behaviour and must be provably untouched; set, it must skip exactly the bulk-lookup/export/categorize steps and nothing else — the claim-status/ date/balance filter setup and the per-claim search/open/apply pipeline stay identical either way, since only the *source of the id list* changes. Everything `_claims_phase` touches is a real Playwright page action or a live network call, so every dependency is stubbed here; this proves the branching and wiring, not any one page module's own behaviour (those have their own test files). Review item D1 (2026-10-05) made a test run say so and fail loudly: a `[TEST MODE]` line names the sheet and the `_TEST_` summary file, a path that does not exist is a `validate()` problem (exit 2 before login), and a sheet that cannot be read raises, which `main()` turns into exit 5 — it used to return quietly and exit 0. """ import io import sys import pathlib import tempfile import types 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 import claim_file # noqa: E402 from ecw_bot.pages import leftnav, claim_filters, claim_results, claim_report # noqa: E402 from ecw_bot.flows import export_medcor_ready as flow # noqa: E402 CASES = [] def case(fn): CASES.append(fn) return fn class _Called(Exception): """Raised by a stub to prove it was reached, without needing a mock lib.""" def _forbidden(*a, **k): raise _Called("must not be called") # Every attribute any case below replaces, restored after each case. The # real .env is loaded at import, so a case that depends on a setting sets it # itself rather than trusting what the operator's .env happens to say. _PATCHED = [ (cfg, "TEST_CLAIM_IDS_FILE"), (cfg, "SHOW_CLAIM_IDS"), (cfg, "WRITE_CHARGE_SUMMARY"), (cfg, "CLAIM_STATUS_LABEL"), (cfg, "SAVE_CORRECTIONS"), (cfg, "ECW_USERNAME"), (cfg, "validate"), (cfg, "PROXY_IP"), (cfg, "PROXY_PORT"), (cfg, "BROWSER_CDP_URL"), (leftnav, "open_claims"), (claim_filters, "set_claim_status"), (claim_filters, "set_date_type"), (claim_filters, "set_date_range"), (claim_filters, "clear_balance"), (claim_results, "run_lookup"), (claim_report, "export_claims_report"), (claim_file, "claim_ids"), (msgbox, "info_box"), (flow, "_process_claims"), (flow, "sync_playwright"), (flow, "_login_with_retry"), (flow, "KEEP_OPEN"), (flow.runlog, "start"), (flow.run_lock, "acquire"), (flow.run_lock, "finish"), (flow.guard, "record_success"), ] _ORIG = [(obj, name, getattr(obj, name)) for obj, name in _PATCHED] def _restore(): for obj, name, value in _ORIG: setattr(obj, name, value) def _stub_filters(calls): """Every filter-setup step _claims_phase always runs, live path or not.""" leftnav.open_claims = lambda page: calls.append("open_claims") or True claim_filters.set_claim_status = \ lambda page, *a, **k: calls.append("set_claim_status") or True claim_filters.set_date_type = \ lambda page, *a, **k: calls.append("set_date_type") or True claim_filters.set_date_range = \ lambda page, *a, **k: calls.append("set_date_range") or True claim_filters.clear_balance = \ lambda page: calls.append("clear_balance") or True @case def blank_file_setting_leaves_the_live_path_untouched(): """ The regression check that matters most: TEST_CLAIM_IDS_FILE unset (the default, "" from settings.py) must still run the live bulk search and export exactly as before this feature existed. """ calls = [] try: cfg.TEST_CLAIM_IDS_FILE = "" _stub_filters(calls) claim_results.run_lookup = \ lambda page: calls.append("run_lookup") or True claim_report.export_claims_report = \ lambda page: calls.append("export_claims_report") or None claim_file.claim_ids = _forbidden flow._process_claims = lambda page, info: calls.append("process_claims") # _read_back_claim_ids(None) short-circuits (saved is None) and # _categorize_by_payer(None) does too — both real, both no-ops here. flow._claims_phase(types.SimpleNamespace(url='fake://claims')) finally: _restore() assert calls == ["open_claims", "set_claim_status", "set_date_type", "set_date_range", "clear_balance", "run_lookup", "export_claims_report"], calls @case def set_file_setting_skips_the_live_bulk_search_and_export(): calls = [] fake_info = {"file": pathlib.Path("standard.xlsx"), "header": "CLAIMS#", "column": 1, "ids": ["9103000"], "duplicates": [], "blank_rows": 0, "total_rows": 1} try: cfg.TEST_CLAIM_IDS_FILE = str(ROOT / "standard.xlsx") cfg.SHOW_CLAIM_IDS = False _stub_filters(calls) claim_results.run_lookup = _forbidden claim_report.export_claims_report = _forbidden claim_file.claim_ids = \ lambda path: calls.append(f"claim_ids:{path}") or fake_info flow._process_claims = \ lambda page, info: calls.append(f"process_claims:{info['ids']}") flow._claims_phase(types.SimpleNamespace(url='fake://claims')) finally: _restore() assert calls == ["open_claims", "set_claim_status", "set_date_type", "set_date_range", "clear_balance", f"claim_ids:{ROOT / 'standard.xlsx'}", "process_claims:['9103000']"], calls @case def set_file_setting_still_shows_the_summary_box_when_configured_to(): calls = [] fake_info = {"file": pathlib.Path("standard.xlsx"), "header": "CLAIMS#", "column": 1, "ids": ["9103000"], "duplicates": [], "blank_rows": 0, "total_rows": 1} try: cfg.TEST_CLAIM_IDS_FILE = str(ROOT / "standard.xlsx") cfg.SHOW_CLAIM_IDS = True _stub_filters(calls) claim_results.run_lookup = _forbidden claim_report.export_claims_report = _forbidden claim_file.claim_ids = lambda path: fake_info msgbox.info_box = lambda t, m: calls.append(f"box:{t}") flow._process_claims = lambda page, info: calls.append("process_claims") flow._claims_phase(types.SimpleNamespace(url='fake://claims')) finally: _restore() assert "box:eCW Bot — Test claim ids" in calls, calls assert "process_claims" in calls, calls @case def a_bad_file_shows_an_error_box_then_raises_and_never_calls_process_claims(): """ `claim_file.claim_ids` is left real here (not stubbed) so this proves the actual FileNotFoundError a missing path raises is caught and reported, not just that some exception would be. Changed 2026-10-05 (review item D1): this used to pin a quiet return, which made `main()` exit 0 after processing nothing. The box is still shown first; then it raises, so `main()` exits 5. """ calls = [] path = ROOT / "does-not-exist.xlsx" raised = None try: cfg.TEST_CLAIM_IDS_FILE = str(path) _stub_filters(calls) claim_results.run_lookup = _forbidden claim_report.export_claims_report = _forbidden msgbox.info_box = lambda t, m: calls.append(f"box:{t}") flow._process_claims = _forbidden with contextlib.redirect_stdout(io.StringIO()): flow._claims_phase(types.SimpleNamespace(url='fake://claims')) except RuntimeError as exc: raised = exc finally: _restore() assert raised is not None, "an unreadable test sheet must raise" assert str(raised).startswith( f"Could not read the test claim sheet {path}: "), raised assert isinstance(raised.__cause__, FileNotFoundError), raised.__cause__ assert "box:eCW Bot — Could not read the test claim sheet" in calls, calls assert "process_claims" not in calls, calls @case def an_unreadable_test_sheet_makes_main_exit_5(): """ End to end through `main()`: login stubbed to succeed, the claims phase real down to `claim_file.claim_ids`, which fails on a sheet that is there but is not a spreadsheet — so `validate()`'s existence check passes and the read failure is what stops the run. """ printed = io.StringIO() with tempfile.TemporaryDirectory() as tmp: sheet = pathlib.Path(tmp) / "Standard Claims.xlsx" sheet.write_bytes(b"\xd0\xcf\x11\xe0 legacy OLE2, unreadable here") try: cfg.TEST_CLAIM_IDS_FILE = str(sheet) cfg.SAVE_CORRECTIONS = False # VERIFY: no start box # Nothing from the operator's .env in the captured output. cfg.ECW_USERNAME = "test-user" cfg.PROXY_IP, cfg.PROXY_PORT = "", "" # Never the operator's Chrome: with BROWSER_CDP_URL from .env the # run would open a real one — here with no proxy (2026-10-22). cfg.BROWSER_CDP_URL = "" cfg.validate = lambda: [] # In tmp: main() writes the Run's events beside the log. flow.runlog.start = lambda: pathlib.Path(tmp) / "unused.log" flow.run_lock.acquire = lambda log_path, source=None: {} flow.run_lock.finish = lambda code: None flow.sync_playwright = lambda: contextlib.nullcontext(None) page = types.SimpleNamespace(url="fake://landing") flow._login_with_retry = lambda pw: (None, None, page) flow.guard.record_success = lambda: None flow.KEEP_OPEN = False _stub_filters([]) claim_results.run_lookup = _forbidden claim_report.export_claims_report = _forbidden msgbox.info_box = lambda t, m: None flow._process_claims = _forbidden with contextlib.redirect_stdout(printed): code = flow.main() finally: _restore() out = printed.getvalue() assert code == 5, (code, out) assert "[CLAIMS] Failed: Could not read the test claim sheet" in out, out _FAKE_INFO = {"file": pathlib.Path("standard.xlsx"), "header": "CLAIMS#", "column": 1, "ids": ["9103000"], "duplicates": [], "blank_rows": 0, "total_rows": 1} @case def test_mode_line_names_the_sheet_and_the_TEST_summary_file(): printed = io.StringIO() try: cfg.TEST_CLAIM_IDS_FILE = str(ROOT / "standard.xlsx") cfg.SHOW_CLAIM_IDS = False cfg.WRITE_CHARGE_SUMMARY = True cfg.CLAIM_STATUS_LABEL = "Medcor Ready" _stub_filters([]) claim_results.run_lookup = _forbidden claim_report.export_claims_report = _forbidden claim_file.claim_ids = lambda path: _FAKE_INFO flow._process_claims = lambda page, info: None with contextlib.redirect_stdout(printed): flow._claims_phase(types.SimpleNamespace(url='fake://claims')) finally: _restore() want = (f"[TEST MODE] Claim ids come from the test sheet standard.xlsx, " f"not the live Medcor Ready queue — the charge summary goes to " f"Charge Summary_TEST_{cfg.TODAY}.xlsx, not the live " f"Charge Summary_{cfg.TODAY}.xlsx") assert want in printed.getvalue(), printed.getvalue() @case def test_mode_line_says_when_no_summary_is_written(): printed = io.StringIO() try: cfg.TEST_CLAIM_IDS_FILE = str(ROOT / "standard.xlsx") cfg.SHOW_CLAIM_IDS = False cfg.WRITE_CHARGE_SUMMARY = False _stub_filters([]) claim_file.claim_ids = lambda path: _FAKE_INFO flow._process_claims = lambda page, info: None with contextlib.redirect_stdout(printed): flow._claims_phase(types.SimpleNamespace(url='fake://claims')) finally: _restore() out = printed.getvalue() assert "[TEST MODE] Claim ids come from the test sheet standard.xlsx" \ in out, out assert "no charge summary is written (WRITE_CHARGE_SUMMARY=0)" in out, out assert "Charge Summary_TEST_" not in out, out @case def validate_reports_a_test_claim_ids_file_that_does_not_exist(): missing = str(ROOT / "does-not-exist.xlsx") try: cfg.TEST_CLAIM_IDS_FILE = missing problems = cfg.validate() finally: _restore() assert f"TEST_CLAIM_IDS_FILE={missing} does not exist" in problems, \ problems @case def validate_accepts_a_test_claim_ids_file_that_exists_or_is_blank(): with tempfile.TemporaryDirectory() as tmp: sheet = pathlib.Path(tmp) / "Standard Claims.xlsx" sheet.write_bytes(b"present") for value in (str(sheet), ""): try: cfg.TEST_CLAIM_IDS_FILE = value problems = cfg.validate() finally: _restore() assert not any("TEST_CLAIM_IDS_FILE" in p for p in problems), \ (value, problems) 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 print(f"\n{len(CASES) - failures}/{len(CASES)} passed") return 1 if failures else 0 if __name__ == "__main__": sys.exit(main())