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>
474 lines
18 KiB
Python
474 lines
18 KiB
Python
r"""
|
|
core/results_db.py — Charge Summary rows to SQL Server — and the flow's
|
|
`_save_results_db` hook.
|
|
|
|
python tests\test_results_db.py
|
|
|
|
No real database: a fake connection records what would be sent.
|
|
"""
|
|
|
|
import io
|
|
import sys
|
|
import json
|
|
import pathlib
|
|
import contextlib
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
|
|
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 charge_summary as cs # noqa: E402
|
|
from ecw_bot.core import results_db as rdb # noqa: E402
|
|
from ecw_bot.flows import export_medcor_ready as flow # noqa: E402
|
|
|
|
CASES = []
|
|
|
|
|
|
def case(fn):
|
|
CASES.append(fn)
|
|
return fn
|
|
|
|
|
|
class FakeCursor:
|
|
def __init__(self, log):
|
|
self.log = log
|
|
|
|
def executemany(self, sql, params):
|
|
self.log.append(("executemany", sql, [list(p) for p in params]))
|
|
|
|
|
|
class FakeConn:
|
|
def __init__(self):
|
|
self.log = []
|
|
|
|
def cursor(self):
|
|
return FakeCursor(self.log)
|
|
|
|
def commit(self):
|
|
self.log.append(("commit",))
|
|
|
|
def close(self):
|
|
self.log.append(("close",))
|
|
|
|
|
|
def _row(**over):
|
|
row = {h: "" for h in cs.HEADERS}
|
|
row.update({"S.No": 1, "CLAIMS#": "9117000", "SERVICE DATE": "7/31/2026",
|
|
"PAYER": "Medicaid", "Group Name": "Medicaid",
|
|
"CHARGES": "$1,365.00", "Balance": "365",
|
|
"Recevied date": "10/08/2026", "Comments": "ICD pointer issue",
|
|
"Current Status": "Corrections pending (not saved)",
|
|
"Category": "Fail"})
|
|
row.update(over)
|
|
return row
|
|
|
|
|
|
def _sheet(*rows):
|
|
return [list(cs.HEADERS)] + [[r[h] for h in cs.HEADERS] for r in rows]
|
|
|
|
|
|
RUN_AT = datetime(2026, 10, 8, 15, 10, 0)
|
|
|
|
|
|
# ── column map ───────────────────────────────────────────────────────────────
|
|
|
|
@case
|
|
def the_table_carries_the_projects_cv_prefix():
|
|
"""Operator, 2026-10-12: every table of this project starts with CV_."""
|
|
assert rdb.TABLE == "dbo.CV_ChargeSummary", rdb.TABLE
|
|
sql = (ROOT / "tools" / "sql" / "create_tables.sql").read_text(
|
|
encoding="utf-8")
|
|
assert "CREATE TABLE dbo.CV_ChargeSummary" in sql
|
|
assert "IX_CV_ChargeSummary_run_at" in sql
|
|
assert "IX_CV_ChargeSummary_claim_no" in sql
|
|
|
|
|
|
@case
|
|
def every_charge_summary_header_has_a_db_column():
|
|
assert [h for h, _ in rdb.COLUMNS] == cs.HEADERS
|
|
assert len(rdb.COLUMNS) == 24
|
|
cols = [c for _, c in rdb.COLUMNS]
|
|
assert len(set(cols)) == 24, cols
|
|
assert rdb.DB_COLUMNS == ["run_at", "run_mode"] + cols + ["details_json"]
|
|
assert dict(rdb.COLUMNS)["Assigned To "] == "assigned_to"
|
|
assert dict(rdb.COLUMNS)["Recevied date"] == "received_date"
|
|
|
|
|
|
# ── conversion ───────────────────────────────────────────────────────────────
|
|
|
|
@case
|
|
def money_strips_dollar_and_commas():
|
|
assert rdb.to_money("$1,234.50") == Decimal("1234.50")
|
|
assert rdb.to_money(" 365 ") == Decimal("365.00")
|
|
assert rdb.to_money(365) == Decimal("365.00")
|
|
assert rdb.to_money(12.5) == Decimal("12.50")
|
|
assert rdb.to_money("(12.00)") == Decimal("-12.00")
|
|
assert rdb.to_money("-5") == Decimal("-5.00")
|
|
|
|
|
|
@case
|
|
def money_that_is_blank_or_junk_is_null():
|
|
for v in ("", None, "abc", "NaN", "inf", "1e20", True):
|
|
assert rdb.to_money(v) is None, v
|
|
|
|
|
|
@case
|
|
def dates_accept_us_iso_and_date_objects():
|
|
assert rdb.to_date("7/31/2026") == date(2026, 7, 31)
|
|
assert rdb.to_date("07/31/2026") == date(2026, 7, 31)
|
|
assert rdb.to_date("2026-07-31") == date(2026, 7, 31)
|
|
assert rdb.to_date(datetime(2026, 7, 31, 9, 0)) == date(2026, 7, 31)
|
|
assert rdb.to_date(date(2026, 7, 31)) == date(2026, 7, 31)
|
|
for v in ("", None, "13/45/2026", "soon"):
|
|
assert rdb.to_date(v) is None, v
|
|
|
|
|
|
@case
|
|
def to_record_types_every_column():
|
|
rec = rdb.to_record(_row(), run_at=RUN_AT, run_mode="VERIFY",
|
|
details={"fired": ["depression_dx_pointer"]})
|
|
assert set(rec) == set(rdb.DB_COLUMNS), set(rec) ^ set(rdb.DB_COLUMNS)
|
|
assert rec["run_at"] == RUN_AT and rec["run_mode"] == "VERIFY"
|
|
assert rec["s_no"] == 1
|
|
assert rec["claim_no"] == "9117000"
|
|
assert rec["service_date"] == date(2026, 7, 31)
|
|
assert rec["received_date"] == date(2026, 10, 8)
|
|
assert rec["charges"] == Decimal("1365.00")
|
|
assert rec["balance"] == Decimal("365.00")
|
|
assert rec["pvdr"] is None, "blank text is NULL"
|
|
assert rec["category"] == "Fail"
|
|
assert json.loads(rec["details_json"]) == {"fired": ["depression_dx_pointer"]}
|
|
|
|
|
|
@case
|
|
def to_record_without_details_stores_null():
|
|
rec = rdb.to_record(_row(), run_at=RUN_AT, run_mode="TEST", details=None)
|
|
assert rec["details_json"] is None
|
|
|
|
|
|
# ── details ──────────────────────────────────────────────────────────────────
|
|
|
|
@case
|
|
def details_for_copies_conditions_dental_and_errors():
|
|
data = {
|
|
"_conditions": {"fired": ["dummy_codes"], "saved": False,
|
|
"failed": [], "review": [{"name": "x", "reason": "y"}],
|
|
"comments": {}, "changes": {"dummy_codes": [{"row": 3}]}},
|
|
"high_risk": {"high_risk": True, "match": "HIGH RISK"},
|
|
"sliding_fee": {"verdict": "active", "reason": "DOS inside",
|
|
"from_date": "01/01/2026"},
|
|
"_not_evaluated": "form did not load",
|
|
"cpt": [{"code": "99213"}], "patient": {"name": "never copied"},
|
|
}
|
|
d = rdb.details_for(data)
|
|
assert d["fired"] == ["dummy_codes"]
|
|
assert d["changes"] == {"dummy_codes": [{"row": 3}]}
|
|
assert d["review"] == [{"name": "x", "reason": "y"}]
|
|
# Verdicts only (review, 2026-10-12): not the note text, not the dates.
|
|
assert d["dental"] == {"high_risk": True,
|
|
"sliding_fee": {"verdict": "active",
|
|
"reason": "DOS inside"}}
|
|
assert d["not_evaluated"] == "form did not load"
|
|
assert "patient" not in d and "cpt" not in d
|
|
assert rdb.details_for({}) == {} and rdb.details_for(None) == {}
|
|
|
|
|
|
# ── insert ───────────────────────────────────────────────────────────────────
|
|
|
|
@case
|
|
def insert_rows_sends_one_parameterized_statement_and_commits():
|
|
conn = FakeConn()
|
|
recs = [rdb.to_record(_row(), run_at=RUN_AT, run_mode="VERIFY", details=None),
|
|
rdb.to_record(_row(**{"CLAIMS#": "9118000", "S.No": 2}),
|
|
run_at=RUN_AT, run_mode="VERIFY", details=None)]
|
|
assert rdb.insert_rows(conn, recs) == 2
|
|
(kind, sql, params), commit = conn.log
|
|
assert kind == "executemany" and commit == ("commit",)
|
|
assert sql.startswith("INSERT INTO dbo.CV_ChargeSummary (run_at, run_mode, s_no,")
|
|
assert sql.count("?") == len(rdb.DB_COLUMNS)
|
|
assert "9117000" not in sql, "values never go into the SQL text"
|
|
assert params[1][rdb.DB_COLUMNS.index("claim_no")] == "9118000"
|
|
assert len(params[0]) == len(rdb.DB_COLUMNS)
|
|
|
|
|
|
@case
|
|
def insert_rows_with_nothing_sends_nothing():
|
|
conn = FakeConn()
|
|
assert rdb.insert_rows(conn, []) == 0
|
|
assert conn.log == []
|
|
|
|
|
|
@case
|
|
def save_summary_skips_rows_the_run_did_not_fill_in():
|
|
conn = FakeConn()
|
|
seen = {}
|
|
|
|
def connect(conn_str, timeout):
|
|
seen["conn_str"], seen["timeout"] = conn_str, timeout
|
|
return conn
|
|
|
|
sheet = _sheet(_row(), _row(**{"CLAIMS#": "999", "Category": "",
|
|
"Comments": "", "Current Status": ""}))
|
|
with contextlib.redirect_stdout(io.StringIO()) as out:
|
|
n = rdb.save_summary(sheet, {"9117000": {"fired": ["a"]}},
|
|
run_at=RUN_AT, run_mode="VERIFY",
|
|
conn_str="DSN=x", connect=connect)
|
|
assert n == 1, n
|
|
assert seen == {"conn_str": "DSN=x", "timeout": 15}, seen
|
|
params = conn.log[0][2]
|
|
assert [p[rdb.DB_COLUMNS.index("claim_no")] for p in params] == ["9117000"]
|
|
assert json.loads(params[0][rdb.DB_COLUMNS.index("details_json")]) == \
|
|
{"fired": ["a"]}
|
|
assert conn.log[-1] == ("close",)
|
|
assert "[RESULTS-DB] 1 row(s) saved" in out.getvalue(), out.getvalue()
|
|
|
|
|
|
@case
|
|
def save_summary_never_raises_on_a_db_error():
|
|
def connect(conn_str, timeout):
|
|
raise RuntimeError("Login failed for user\nmore detail")
|
|
|
|
with contextlib.redirect_stdout(io.StringIO()) as out:
|
|
n = rdb.save_summary(_sheet(_row()), {}, run_at=RUN_AT,
|
|
run_mode="VERIFY", conn_str="DSN=x",
|
|
connect=connect)
|
|
assert n is None
|
|
text = out.getvalue()
|
|
assert "[RESULTS-DB] Could not save" in text and "RuntimeError" in text, \
|
|
text
|
|
assert "Login failed" not in text, "the driver's message is never logged"
|
|
|
|
|
|
@case
|
|
def save_summary_with_no_filled_rows_does_not_connect():
|
|
def connect(conn_str, timeout):
|
|
raise AssertionError("must not connect")
|
|
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
n = rdb.save_summary(_sheet(_row(Category="")), {}, run_at=RUN_AT,
|
|
run_mode="VERIFY", conn_str="DSN=x",
|
|
connect=connect)
|
|
assert n == 0
|
|
|
|
|
|
# ── hardening (review, 2026-10-12) ───────────────────────────────────────────
|
|
|
|
@case
|
|
def money_that_rounds_past_the_column_is_null():
|
|
assert rdb.to_money("9999999999.995") is None
|
|
assert rdb.to_money("9999999999.99") == Decimal("9999999999.99")
|
|
|
|
|
|
@case
|
|
def dates_with_a_time_or_as_an_excel_serial_still_convert():
|
|
assert rdb.to_date("07/31/2026 10:30:00") == date(2026, 7, 31)
|
|
assert rdb.to_date("46234") == date(2026, 7, 31)
|
|
assert rdb.to_date(46234) == date(2026, 7, 31)
|
|
assert rdb.to_date("46234.0") == date(2026, 7, 31)
|
|
for v in ("123", "9117000", "99999", "07/31/2026x"):
|
|
assert rdb.to_date(v) is None, v
|
|
|
|
|
|
@case
|
|
def text_longer_than_its_column_is_cut_to_fit():
|
|
rec = rdb.to_record(_row(PVDR="P" * 80, PATIENT="N" * 300,
|
|
Comments="C" * 5000),
|
|
run_at=RUN_AT, run_mode="TEST LIVE-WRITE",
|
|
details=None)
|
|
assert rec["pvdr"] == "P" * 50
|
|
assert rec["patient"] == "N" * 200
|
|
assert rec["comments"] == "C" * 5000, "NVARCHAR(MAX) is not cut"
|
|
assert rec["run_mode"] == "TEST LIVE-WRITE"
|
|
|
|
|
|
@case
|
|
def details_never_carry_patient_text_from_the_changes():
|
|
data = {"_conditions": {
|
|
"fired": ["vaccine_admin_codes"], "changed": ["vaccine_admin_codes"],
|
|
"detected": ["vaccine_admin_codes", "zero_fee_uncheck"],
|
|
"saved": False, "save_unconfirmed": True,
|
|
"changes": {"vaccine_admin_codes": [
|
|
{"row": 4, "code": "90460", "age": 7,
|
|
"age_source": "DOB 01/02/2019"}]}}}
|
|
d = rdb.details_for(data)
|
|
change = d["changes"]["vaccine_admin_codes"][0]
|
|
assert change == {"row": 4, "code": "90460", "age": 7}, change
|
|
assert "01/02/2019" not in json.dumps(d)
|
|
assert d["changed"] == ["vaccine_admin_codes"]
|
|
assert d["detected"] == ["vaccine_admin_codes", "zero_fee_uncheck"]
|
|
assert d["save_unconfirmed"] is True
|
|
|
|
|
|
class _DbError(Exception):
|
|
pass
|
|
|
|
|
|
@case
|
|
def a_db_error_is_logged_by_sqlstate_without_the_value():
|
|
"""Msg 2628 quotes the value that did not fit — a patient name."""
|
|
class Conn(FakeConn):
|
|
def cursor(self):
|
|
raise _DbError("22001", "[Microsoft][ODBC Driver 18] String or "
|
|
"binary data would be truncated in column "
|
|
"'patient', truncated value: 'Smith, John'")
|
|
|
|
with contextlib.redirect_stdout(io.StringIO()) as out:
|
|
n = rdb.save_summary(_sheet(_row()), {}, run_at=RUN_AT,
|
|
run_mode="VERIFY", conn_str="DSN=x",
|
|
connect=lambda cs_, timeout: Conn())
|
|
assert n is None
|
|
text = out.getvalue()
|
|
assert "SQLSTATE 22001" in text and "too long" in text, text
|
|
assert "Smith" not in text, text
|
|
|
|
|
|
@case
|
|
def save_summary_sets_a_query_timeout():
|
|
conn = FakeConn()
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
rdb.save_summary(_sheet(_row()), {}, run_at=RUN_AT,
|
|
run_mode="VERIFY", conn_str="DSN=x",
|
|
connect=lambda cs_, timeout: conn)
|
|
assert getattr(conn, "timeout", None) == 30
|
|
|
|
|
|
# ── settings ─────────────────────────────────────────────────────────────────
|
|
|
|
@case
|
|
def settings_default_off_with_the_windows_login_connection():
|
|
"""
|
|
The code's defaults, read from settings.py itself: the loaded values
|
|
follow this machine's .env, which may well turn RESULTS_DB on.
|
|
"""
|
|
assert isinstance(cfg.RESULTS_DB, bool)
|
|
src = (ROOT / "src" / "ecw_bot" / "settings.py").read_text(
|
|
encoding="utf-8")
|
|
assert '_env_bool("RESULTS_DB", False)' in src
|
|
start = src.index('"RESULTS_DB_CONN",')
|
|
default = src[start:src.index(")", start)]
|
|
for part in ("ODBC Driver 18 for SQL Server", "MECARE331\\\\MECARE331",
|
|
"DATABASE=Charge_Validation", "Trusted_Connection=yes"):
|
|
assert part in default, (part, default)
|
|
assert "PWD" not in default.upper() and "UID" not in default.upper()
|
|
|
|
|
|
# ── flow hook ────────────────────────────────────────────────────────────────
|
|
|
|
def _with_cfg(**values):
|
|
saved = {k: getattr(cfg, k) for k in values}
|
|
for k, v in values.items():
|
|
setattr(cfg, k, v)
|
|
return saved
|
|
|
|
|
|
@case
|
|
def run_mode_names_test_live_write_and_verify():
|
|
for test_file, save, want in (("C:/t.xlsx", True, "TEST LIVE-WRITE"),
|
|
("C:/t.xlsx", False, "TEST VERIFY"),
|
|
("", True, "LIVE-WRITE"),
|
|
("", False, "VERIFY")):
|
|
saved = _with_cfg(TEST_CLAIM_IDS_FILE=test_file, SAVE_CORRECTIONS=save)
|
|
try:
|
|
assert flow._run_mode() == want, (test_file, save)
|
|
finally:
|
|
_with_cfg(**saved)
|
|
|
|
|
|
def _hooked(results_db_on, build=None):
|
|
calls = []
|
|
orig = (cs.build, rdb.save_summary)
|
|
saved = _with_cfg(RESULTS_DB=results_db_on, TEST_CLAIM_IDS_FILE="",
|
|
SAVE_CORRECTIONS=False)
|
|
try:
|
|
cs.build = build or (lambda *a: calls.append(("build", a)) or ["SHEET"])
|
|
rdb.save_summary = lambda *a, **k: calls.append(("save", a, k)) or 1
|
|
with contextlib.redirect_stdout(io.StringIO()) as out:
|
|
flow._save_results_db({"file": pathlib.Path("r.xlsx")},
|
|
{"1": {"Category": "Pass"}}, {"1": {"x": 1}})
|
|
finally:
|
|
cs.build, rdb.save_summary = orig
|
|
_with_cfg(**saved)
|
|
return calls, out.getvalue()
|
|
|
|
|
|
@case
|
|
def flow_hook_does_nothing_when_results_db_is_off():
|
|
calls, _ = _hooked(False)
|
|
assert calls == [], calls
|
|
|
|
|
|
@case
|
|
def flow_hook_builds_the_sheet_and_saves_it():
|
|
calls, _ = _hooked(True)
|
|
assert calls[0] == ("build", (pathlib.Path("r.xlsx"),
|
|
{"1": {"Category": "Pass"}},
|
|
cfg.SUMMARY_ASSIGNED_TO)), calls
|
|
kind, args, kwargs = calls[1]
|
|
assert kind == "save" and args == (["SHEET"], {"1": {"x": 1}}), calls
|
|
assert kwargs["run_mode"] == "VERIFY"
|
|
assert kwargs["conn_str"] == cfg.RESULTS_DB_CONN
|
|
assert isinstance(kwargs["run_at"], datetime)
|
|
assert kwargs["run_at"].microsecond == 0
|
|
|
|
|
|
@case
|
|
def flow_hook_keeps_one_run_at_for_every_claim_of_a_run():
|
|
"""Saved claim by claim, the rows still group as one run on the page."""
|
|
seen = []
|
|
ticks = iter([datetime(2026, 10, 12, 9, 0, 0),
|
|
datetime(2026, 10, 12, 9, 5, 0)])
|
|
|
|
class _Clock(datetime):
|
|
@classmethod
|
|
def now(cls, tz=None):
|
|
return next(ticks)
|
|
|
|
orig = (cs.build, rdb.save_summary, flow.datetime)
|
|
saved = _with_cfg(RESULTS_DB=True, TEST_CLAIM_IDS_FILE="",
|
|
SAVE_CORRECTIONS=False)
|
|
info = {"file": pathlib.Path("r.xlsx")}
|
|
try:
|
|
flow.datetime = _Clock
|
|
cs.build = lambda *a: ["SHEET"]
|
|
rdb.save_summary = lambda *a, **k: seen.append(k["run_at"]) or 1
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
flow._save_results_db(info, {"1": {"Category": "Pass"}}, {})
|
|
flow._save_results_db(info, {"2": {"Category": "Fail"}}, {})
|
|
finally:
|
|
cs.build, rdb.save_summary, flow.datetime = orig
|
|
_with_cfg(**saved)
|
|
assert len(seen) == 2 and seen[0] == seen[1], seen
|
|
|
|
|
|
@case
|
|
def flow_hook_survives_a_build_error():
|
|
def boom(*a):
|
|
raise OSError("report locked")
|
|
|
|
calls, out = _hooked(True, build=boom)
|
|
assert calls == [], calls
|
|
assert "[RESULTS-DB]" in out and "report locked" in out, out
|
|
|
|
|
|
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())
|