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>
281 lines
9.3 KiB
Python
281 lines
9.3 KiB
Python
"""
|
|
Where the flow writes its Run events — `flows/export_medcor_ready.py`
|
|
(spec 2026-10-22 Stage 1, EEWA-31). No browser, no eCW: the login, guard,
|
|
run lock and disk are stand-ins.
|
|
|
|
python tests\\test_run_events_flow.py
|
|
|
|
* **Every claim** gets a `claim_result`; a search that raised, a form not
|
|
read and a form that never loaded are Claim Errors; not found is a
|
|
Warning.
|
|
* **Each eCW Login Attempt** has a start and an end with its result —
|
|
OK, 2FA timeout (and the retry is a second attempt), restricted,
|
|
captcha failed, failed.
|
|
* **The disk check**: under DISK_STOP_MB stops before login (Run Error,
|
|
step disk); under DISK_WARN_MB is a Warning; enough space, nothing.
|
|
* **main()** always ends the file with `run_end` and its exit code, and a
|
|
raise that escapes `_main` is a Run Error with its traceback.
|
|
|
|
Claim numbers here are made up.
|
|
"""
|
|
|
|
import sys
|
|
import json
|
|
import pathlib
|
|
import tempfile
|
|
from collections import namedtuple
|
|
|
|
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 run_events # noqa: E402
|
|
from ecw_bot.flows import export_medcor_ready as flow # noqa: E402
|
|
from ecw_bot.auth.dead_end import LoginTimedOut # noqa: E402
|
|
from ecw_bot.auth.restricted import Restricted # noqa: E402
|
|
|
|
CASES = []
|
|
|
|
|
|
def case(fn):
|
|
CASES.append(fn)
|
|
return fn
|
|
|
|
|
|
def _started():
|
|
run_events._reset()
|
|
d = pathlib.Path(tempfile.mkdtemp())
|
|
log = d / "run_20261022_141530.log"
|
|
run_events.start(log, mode="VERIFY")
|
|
return log
|
|
|
|
|
|
def _events(log, kind=None):
|
|
lines = run_events.path_for(log).read_text(encoding="utf-8").splitlines()
|
|
ev = [json.loads(l) for l in lines]
|
|
return [e for e in ev if kind is None or e["kind"] == kind]
|
|
|
|
|
|
class _Patch:
|
|
"""Set attributes for one case and put them back."""
|
|
def __init__(self, *triples):
|
|
self.triples = triples
|
|
|
|
def __enter__(self):
|
|
self.saved = [(o, n, getattr(o, n)) for o, n, _ in self.triples]
|
|
for o, n, v in self.triples:
|
|
setattr(o, n, v)
|
|
|
|
def __exit__(self, *exc):
|
|
for o, n, v in self.saved:
|
|
setattr(o, n, v)
|
|
|
|
|
|
# ── Claims ────────────────────────────────────────────────────────────────────
|
|
|
|
@case
|
|
def a_read_claim_has_only_its_result():
|
|
log = _started()
|
|
flow._claim_events(1, "9001", {"found": True}, {"cpt": []},
|
|
{"Category": "Pass"})
|
|
assert [e["kind"] for e in _events(log)][1:] == ["claim_result"]
|
|
assert _events(log, "claim_result")[0]["category"] == "Pass"
|
|
|
|
|
|
@case
|
|
def a_search_that_raised_is_a_claim_error_not_a_not_found():
|
|
log = _started()
|
|
flow._claim_events(1, "9001", {"found": False, "search_error": "timeout"},
|
|
{"_search_error": "timeout"}, {"Category": "Review"})
|
|
errs = _events(log, "claim_error")
|
|
assert [(e["claim"], e["step"]) for e in errs] == [("9001", "search")]
|
|
assert not _events(log, "warning")
|
|
|
|
|
|
@case
|
|
def a_claim_not_found_is_a_warning():
|
|
log = _started()
|
|
flow._claim_events(2, "9002", {"found": False}, {}, {})
|
|
assert _events(log, "claim_result")[0]["category"] == "Not found"
|
|
w = _events(log, "warning")
|
|
assert [(e["code"], e["claim"]) for e in w] == [("not_found", "9002")]
|
|
|
|
|
|
@case
|
|
def a_form_not_read_and_a_form_not_loaded_are_claim_errors():
|
|
log = _started()
|
|
flow._claim_events(1, "9001", {"found": True}, {"_error": "boom"},
|
|
{"Category": "Review"})
|
|
flow._claim_events(2, "9002", {"found": True},
|
|
{"_not_evaluated": "rows never came"},
|
|
{"Category": "Review"})
|
|
errs = _events(log, "claim_error")
|
|
assert [(e["claim"], e["step"]) for e in errs] == [
|
|
("9001", "form"), ("9002", "load")], errs
|
|
|
|
|
|
# ── Disk ──────────────────────────────────────────────────────────────────────
|
|
|
|
Usage = namedtuple("Usage", "total used free")
|
|
|
|
|
|
def _disk(free_mb):
|
|
return lambda drive: Usage(0, 0, free_mb * 1024 * 1024)
|
|
|
|
|
|
@case
|
|
def a_nearly_full_disk_stops_before_login():
|
|
log = _started()
|
|
with _Patch((flow.shutil, "disk_usage", _disk(100))):
|
|
assert flow._disk_check() is False
|
|
e = _events(log, "run_error")[0]
|
|
assert e["step"] == "disk" and e["exit_code"] == 2, e
|
|
assert _events(log, "disk")[0]["free_mb"] == 100
|
|
|
|
|
|
@case
|
|
def a_low_disk_is_a_warning_and_goes_on():
|
|
log = _started()
|
|
with _Patch((flow.shutil, "disk_usage", _disk(1000))):
|
|
assert flow._disk_check() is True
|
|
assert [e["code"] for e in _events(log, "warning")] == ["disk_low"]
|
|
assert not _events(log, "run_error")
|
|
|
|
|
|
@case
|
|
def enough_disk_is_no_warning():
|
|
log = _started()
|
|
with _Patch((flow.shutil, "disk_usage", _disk(50_000))):
|
|
assert flow._disk_check() is True
|
|
assert not _events(log, "warning") and not _events(log, "run_error")
|
|
|
|
|
|
# ── eCW Login Attempts ────────────────────────────────────────────────────────
|
|
|
|
def _login_patches(login):
|
|
g = flow.guard
|
|
return _Patch((g, "wait_if_cooling", lambda: None),
|
|
(g, "check", lambda: None),
|
|
(g, "record_attempt", lambda: None),
|
|
(flow, "login", login),
|
|
(cfg, "LOGIN_RETRY_ON_TIMEOUT", 1))
|
|
|
|
|
|
def _attempt_ends(log):
|
|
return [(e["attempt"], e["result"])
|
|
for e in _events(log, "login_attempt_end")]
|
|
|
|
|
|
@case
|
|
def a_timed_out_attempt_then_an_ok_one():
|
|
log = _started()
|
|
calls = []
|
|
|
|
def login(pw, name):
|
|
calls.append(1)
|
|
if len(calls) == 1:
|
|
raise LoginTimedOut("no mail")
|
|
return ("b", "c", "p")
|
|
|
|
with _login_patches(login):
|
|
assert flow._login_with_retry(None) == ("b", "c", "p")
|
|
assert _attempt_ends(log) == [(1, "twofa_timeout"), (2, "ok")]
|
|
assert [e["attempt"] for e in _events(log, "login_attempt_start")] == [1, 2]
|
|
|
|
|
|
@case
|
|
def restricted_captcha_and_other_failures_end_their_attempt():
|
|
for exc, result in ((Restricted("blocked"), "restricted"),
|
|
(RuntimeError("CAPTCHA not solved"), "captcha_failed"),
|
|
(RuntimeError("Login failed"), "failed")):
|
|
log = _started()
|
|
|
|
def login(pw, name, exc=exc):
|
|
raise exc
|
|
|
|
with _login_patches(login):
|
|
try:
|
|
flow._login_with_retry(None)
|
|
raise AssertionError("did not raise")
|
|
except type(exc):
|
|
pass
|
|
assert _attempt_ends(log) == [(1, result)], (result, _attempt_ends(log))
|
|
|
|
|
|
# ── main() ────────────────────────────────────────────────────────────────────
|
|
|
|
def _main_patches(log, inner):
|
|
return _Patch((flow.runlog, "start", lambda: log),
|
|
(flow.run_lock, "acquire", lambda p: {}),
|
|
(flow.run_lock, "finish", lambda c: None),
|
|
(flow, "_main", inner))
|
|
|
|
|
|
@case
|
|
def main_ends_the_file_with_the_exit_code():
|
|
run_events._reset()
|
|
log = pathlib.Path(tempfile.mkdtemp()) / "run_20261022_141530.log"
|
|
with _main_patches(log, lambda p: 3):
|
|
assert flow.main() == 3
|
|
ev = _events(log)
|
|
assert ev[0]["kind"] == "run_start" and ev[-1]["kind"] == "run_end", ev
|
|
assert ev[-1]["exit_code"] == 3
|
|
|
|
|
|
@case
|
|
def a_raise_escaping_main_is_a_run_error_with_its_traceback():
|
|
run_events._reset()
|
|
log = pathlib.Path(tempfile.mkdtemp()) / "run_20261022_141530.log"
|
|
|
|
def boom(p):
|
|
raise KeyError("lost")
|
|
|
|
with _main_patches(log, boom):
|
|
try:
|
|
flow.main()
|
|
raise AssertionError("did not raise")
|
|
except KeyError:
|
|
pass
|
|
err = _events(log, "run_error")[0]
|
|
assert err["step"] == "unexpected" and err["exit_code"] == 1, err
|
|
assert "KeyError" in err["traceback"], err
|
|
assert _events(log)[-1] == {**_events(log)[-1], "kind": "run_end",
|
|
"exit_code": 1}
|
|
|
|
|
|
@case
|
|
def a_busy_run_lock_is_a_run_error_and_an_end():
|
|
run_events._reset()
|
|
log = pathlib.Path(tempfile.mkdtemp()) / "run_20261022_141530.log"
|
|
|
|
def busy(p):
|
|
raise flow.run_lock.Busy("A bot run is already going")
|
|
|
|
with _Patch((flow.runlog, "start", lambda: log),
|
|
(flow.run_lock, "acquire", busy)):
|
|
assert flow.main() == 2
|
|
kinds = [e["kind"] for e in _events(log)]
|
|
assert kinds == ["run_start", "run_error", "run_end"], kinds
|
|
|
|
|
|
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
|
|
run_events._reset()
|
|
print(f"\n{len(CASES) - failures}/{len(CASES)} passed")
|
|
return 1 if failures else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|