ManikandanandClaude Opus 5.5
120fd6808a
Header, breadcrumb, page and footer share one 1180px column; the header marks the current page; a footer sits at the bottom of every page. Dashboard: pass rate on Checked today, a Last 7 days Pass / Fail / Review chart, today's Fail and Review reasons grouped with counts, and three equal cards (Claim status, Runs, Find a claim). Motion is movement only and off under prefers-reduced-motion. Context.md brought up to date. Snapshot of local 9d5ac63 (tree scanned: no real claim ids, no patient data). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
848 lines
32 KiB
Python
848 lines
32 KiB
Python
r"""
|
|
web/app.py and web/repo.py — the results portal, against a fake repository
|
|
and a fake database connection (no SQL Server needed). Spec:
|
|
docs/superpowers/specs/2026-10-19-results-portal-redesign-design.md.
|
|
|
|
python tests\test_web_app.py
|
|
|
|
Claim numbers here are made up.
|
|
"""
|
|
|
|
import io
|
|
import re
|
|
import sys
|
|
import json
|
|
import pathlib
|
|
import tempfile
|
|
import contextlib
|
|
from datetime import date, datetime, timedelta
|
|
from decimal import Decimal
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT))
|
|
sys.path.insert(0, str(ROOT / "src"))
|
|
|
|
from fastapi.testclient import TestClient # noqa: E402
|
|
|
|
from web import app as web_app # noqa: E402
|
|
from web.repo import SqlResultsRepo # noqa: E402
|
|
from ecw_bot.core import results_db as rdb # noqa: E402
|
|
from web import users as portal_users # noqa: E402
|
|
|
|
sys.path.insert(0, str(ROOT / "tests"))
|
|
from test_portal_users import FakeUserStore # noqa: E402
|
|
|
|
CASES = []
|
|
|
|
|
|
def case(fn):
|
|
CASES.append(fn)
|
|
return fn
|
|
|
|
|
|
DAY = date(2026, 10, 16)
|
|
RUN_AM = datetime(2026, 10, 16, 10, 30, 0) # 10 claims run
|
|
RUN_PM = datetime(2026, 10, 16, 14, 15, 0) # 20 claims run, same day
|
|
RUN_OLD = datetime(2026, 10, 15, 11, 0, 0)
|
|
|
|
|
|
def _row(claim, category, s_no, run_at=RUN_PM, **over):
|
|
row = {c: None for _, c in rdb.COLUMNS}
|
|
row.update({"id": s_no, "run_at": run_at, "run_mode": "VERIFY",
|
|
"s_no": s_no, "claim_no": claim, "category": category,
|
|
"service_date": date(2026, 7, 31), "payer": "Medicaid",
|
|
"charges": Decimal("1365.00"), "comments": f"comment {claim}",
|
|
"current_status": "Corrections pending (not saved)",
|
|
"details_json": json.dumps({"fired": ["dummy_codes"]})})
|
|
row.update(over)
|
|
return row
|
|
|
|
|
|
def _day(day, claims, runs, p, f, r):
|
|
return {"day": day, "claims": claims, "runs": runs, "pass_count": p,
|
|
"fail_count": f, "review_count": r}
|
|
|
|
|
|
class FakeRepo:
|
|
def __init__(self, days=None, rows=None, fail=False):
|
|
self._days = days if days is not None else [
|
|
_day(DAY, 3, 2, 1, 1, 1), _day(RUN_OLD.date(), 2, 1, 2, 0, 0)]
|
|
self._rows = rows if rows is not None else [
|
|
_row("9000001", "Pass", 1, RUN_AM),
|
|
_row("9000002", "Fail", 1, details_json=None),
|
|
_row("9000003", "Review", 2, payer="HCFA Medicare")]
|
|
self.fail = fail
|
|
self.calls = []
|
|
|
|
def _maybe_fail(self):
|
|
if self.fail:
|
|
raise self.fail if isinstance(self.fail, Exception) \
|
|
else RuntimeError("Login failed for user")
|
|
|
|
def runs(self):
|
|
self._maybe_fail()
|
|
return [{"run_at": RUN_PM, "run_mode": "VERIFY", "claims": 2,
|
|
"pass_count": 0, "fail_count": 1, "review_count": 1}]
|
|
|
|
def results(self, run_at, category=None, claim_no=None):
|
|
self.calls.append(("results", run_at, category, claim_no))
|
|
self._maybe_fail()
|
|
return [r for r in self._rows
|
|
if (not category or r["category"] == category)]
|
|
|
|
def days(self, limit=90):
|
|
self.calls.append(("days",))
|
|
self._maybe_fail()
|
|
return self._days
|
|
|
|
def day_results(self, day):
|
|
self.calls.append(("day_results", day))
|
|
self._maybe_fail()
|
|
return [dict(r) for r in self._rows if r["run_at"].date() == day]
|
|
|
|
def claim_history(self, claim_no):
|
|
self.calls.append(("claim_history", claim_no))
|
|
self._maybe_fail()
|
|
return [dict(r) for r in self._rows if claim_no in r["claim_no"]]
|
|
|
|
|
|
TESTER_PASSWORD = "tester-password-1"
|
|
|
|
|
|
def signed_in(repo, runner, base_url="http://127.0.0.1:8000", role="Admin",
|
|
**kw):
|
|
"""(client, app): the Portal with one Portal User, signed in — every
|
|
page needs a Sign-in since 2026-10-22 (EEWA-32). A foreign host gets
|
|
its 400 before any Sign-in, so there it is not tried."""
|
|
store = FakeUserStore()
|
|
portal_users.Accounts(store).create_user("tester", "Tester", role,
|
|
TESTER_PASSWORD)
|
|
# No run events unless a test brings its own: the real Logs\ would
|
|
# make the Dashboard's last-run banner depend on the PC.
|
|
kw.setdefault("logs_dir", tempfile.mkdtemp())
|
|
app = web_app.create_app(repo, runner=runner, users_store=store, **kw)
|
|
client = TestClient(app, base_url=base_url)
|
|
if base_url in web_app.ALLOWED_ORIGINS:
|
|
r = client.post("/signin", headers={"Origin": base_url}, data={
|
|
"username": "tester", "password": TESTER_PASSWORD,
|
|
"token": app.state.run_token, "next": "/account/password"})
|
|
# A page that reads no database: FakeRepo(fail=True) must not
|
|
# break the Sign-in itself.
|
|
assert r.status_code == 200, r.status_code
|
|
assert r.url.path == "/account/password", r.url
|
|
return client, app
|
|
|
|
|
|
def _client(repo=None, base_url="http://127.0.0.1:8000"):
|
|
repo = repo or FakeRepo()
|
|
client, _ = signed_in(repo, FakeRunner(), base_url=base_url)
|
|
return client, repo
|
|
|
|
|
|
class FakeRunner:
|
|
def __init__(self, running=False, stop_fails=None):
|
|
self.running = running
|
|
self.stop_fails = stop_fails
|
|
self.calls = []
|
|
|
|
def _st(self):
|
|
return {"running": self.running, "starting": False, "pid": 1,
|
|
"started_at": "2026-10-14T10:00:00", "ended_at": None,
|
|
"exit_code": None, "stopped_by_operator": False,
|
|
"source": "web", "log_path": None, "out_path": None}
|
|
|
|
def start(self, started_by=""):
|
|
self.calls.append("start")
|
|
self.started_by = started_by
|
|
if self.running:
|
|
raise web_app.AlreadyRunning("A run is already going (started x)")
|
|
self.running = True
|
|
return self._st()
|
|
|
|
def stop(self, stopped_by=""):
|
|
self.calls.append("stop")
|
|
self.stopped_by = stopped_by
|
|
if self.stop_fails:
|
|
raise web_app.StopFailed(self.stop_fails)
|
|
if not self.running:
|
|
raise web_app.NotRunning("No run is going")
|
|
self.running = False
|
|
return self._st()
|
|
|
|
def status(self):
|
|
return self._st()
|
|
|
|
|
|
def _bot_client(runner=None):
|
|
runner = runner or FakeRunner()
|
|
client, app = signed_in(FakeRepo(), runner)
|
|
return (client, runner,
|
|
app.state.run_token)
|
|
|
|
|
|
_GOOD_ORIGIN = {"Origin": "http://127.0.0.1:8000"}
|
|
_PAGES = ("/", "/days", f"/day/{DAY.isoformat()}", "/claim?claim_no=9000")
|
|
|
|
|
|
def _quiet(fn, *a, **k):
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
return fn(*a, **k)
|
|
|
|
|
|
def _css_rule(html, selector):
|
|
"""What `selector { ... }` in the page's <style> declares."""
|
|
css = html[html.index("<style>"):html.index("</style>")]
|
|
rule = css[css.index("\n " + selector + " {"):]
|
|
return rule[:rule.index("}")]
|
|
|
|
|
|
# ── Dashboard ────────────────────────────────────────────────────────────────
|
|
|
|
@case
|
|
def the_dashboard_has_no_claims_table():
|
|
"""Operator, 2026-10-19: the claims table (three quarters of the page)
|
|
is not on the home page — the Bot run card, today's counts, Find a
|
|
claim and one Claim status button are."""
|
|
client, repo = _client()
|
|
html = client.get("/").text
|
|
assert "<table" not in html, "no claims table on the Dashboard"
|
|
assert 'id="br-start"' in html and 'id="live-open"' in html
|
|
assert 'href="/days" id="claim-status"' in html
|
|
assert 'action="/claim"' in html, "Find a claim"
|
|
assert repo.calls == [("days",)], repo.calls
|
|
|
|
|
|
@case
|
|
def the_dashboard_tiles_are_todays_counts():
|
|
today = date.today()
|
|
client, _ = _client(FakeRepo(days=[_day(today, 30, 2, 24, 5, 1)]))
|
|
html = client.get("/").text
|
|
assert '<span class="n">30</span><span class="l">Checked today</span>' \
|
|
in html, html
|
|
assert f'href="/day/{today.isoformat()}?category=Fail"><span class="n">5<' \
|
|
in html
|
|
# No run today: zeros, not yesterday's numbers.
|
|
html = _client(FakeRepo(days=[_day(today - timedelta(days=1), 9, 1, 9,
|
|
0, 0)]))[0].get("/").text
|
|
assert '<span class="n">0</span><span class="l">Checked today</span>' \
|
|
in html
|
|
|
|
|
|
@case
|
|
def the_dashboard_shows_todays_pass_rate():
|
|
today = date.today()
|
|
html = _client(FakeRepo(days=[_day(today, 30, 2, 24, 5, 1)]))[0] \
|
|
.get("/").text
|
|
assert '<span class="rate">80% pass</span>' in html
|
|
html = _client(FakeRepo(days=[]))[0].get("/").text
|
|
assert 'class="rate"' not in html, "no claims today: no rate"
|
|
|
|
|
|
@case
|
|
def the_dashboard_shows_the_last_seven_days_oldest_first():
|
|
today = date.today()
|
|
days = [_day(today - timedelta(days=i), 10 + i, 1, 8, 2, i % 2)
|
|
for i in range(9)] # newest first, as repo
|
|
html = _client(FakeRepo(days=days))[0].get("/").text
|
|
assert 'id="week"' in html
|
|
shown = re.findall(r'data-day="([\d-]+)"', html)
|
|
assert shown == [(today - timedelta(days=i)).isoformat()
|
|
for i in range(6, -1, -1)], shown
|
|
assert f'href="/day/{(today - timedelta(days=3)).isoformat()}"' in html
|
|
assert 'data-n="13"' in html, "a day's claims count"
|
|
|
|
|
|
@case
|
|
def the_dashboard_lists_todays_top_fail_and_review_reasons():
|
|
run = datetime.combine(date.today(), datetime.min.time()) \
|
|
+ timedelta(hours=10)
|
|
tele = "Telehealth (POS 02, modifier 93 on 99213)"
|
|
rows = ([_row(f"90000{i}", "Fail", i, run, comments=tele)
|
|
for i in range(1, 4)]
|
|
+ [_row("9000005", "Fail", 5, run, comments="ICD <pointer>"),
|
|
_row("9000006", "Review", 6, run, comments="Could not search"),
|
|
_row("9000007", "Pass", 7, run, comments="Clean claims")])
|
|
html = _client(FakeRepo(days=[_day(date.today(), 6, 1, 1, 4, 1)],
|
|
rows=rows))[0].get("/").text
|
|
box = html.split('id="reasons"', 1)[1].split("</section>", 1)[0]
|
|
assert box.index(tele) < box.index("ICD <pointer>") \
|
|
< box.index("Could not search"), box
|
|
assert '<b class="Fail">3</b>' in box, "grouped, with its count"
|
|
assert "Clean claims" not in box, "Pass is not a reason"
|
|
html = _client(FakeRepo(rows=[]))[0].get("/").text
|
|
assert "No Fail or Review today" in html
|
|
|
|
|
|
@case
|
|
def every_page_has_the_footer_and_marks_where_you_are():
|
|
client, _ = _client()
|
|
for path in _PAGES:
|
|
html = client.get(path).text
|
|
foot = html.split('<footer class="foot">', 1)[1]
|
|
assert "Ecare India" in foot and "patient data" in foot, path
|
|
html = client.get("/days").text
|
|
assert '<a href="/days" class="on" aria-current="page">Claim status</a>' \
|
|
in html, "the page you are on is marked in the header"
|
|
|
|
|
|
@case
|
|
def an_old_one_run_link_goes_to_that_runs_day():
|
|
client, repo = _client()
|
|
r = client.get("/", params={"run_at": RUN_OLD.isoformat(),
|
|
"category": "Fail", "claim_no": " 90-00 "},
|
|
follow_redirects=False)
|
|
assert r.status_code == 303, r.status_code
|
|
assert r.headers["location"] == \
|
|
"/day/2026-10-15?category=Fail&claim_no=9000", r.headers
|
|
assert repo.calls == []
|
|
|
|
|
|
@case
|
|
def the_dashboard_rejects_a_bad_filter():
|
|
client, _ = _client()
|
|
assert client.get("/", params={"category": "Bad"}).status_code == 400
|
|
assert client.get("/", params={"run_at": "not-a-date"}).status_code == 400
|
|
|
|
|
|
# ── Claim status: date cards ─────────────────────────────────────────────────
|
|
|
|
@case
|
|
def claim_status_lists_every_day_newest_first():
|
|
"""Operator, 2026-10-19: a second run hid the first — every day with
|
|
a run is a card, all its runs together."""
|
|
client, _ = _client()
|
|
html = client.get("/days").text
|
|
first = html.index('href="/day/2026-10-16"')
|
|
second = html.index('href="/day/2026-10-15"')
|
|
assert first < second
|
|
assert "Fri, 16 Oct 2026" in html and "3 claims checked · 2 runs" in html
|
|
assert "2 claims checked · 1 run<" in html, "one run, not 'runs'"
|
|
assert 'data-n="1"' in html, "the colour bar's counts"
|
|
|
|
|
|
@case
|
|
def claim_status_with_no_runs_says_so():
|
|
html = _client(FakeRepo(days=[], rows=[]))[0].get("/days").text
|
|
assert "No results saved yet" in html
|
|
|
|
|
|
# ── One day ──────────────────────────────────────────────────────────────────
|
|
|
|
@case
|
|
def a_day_shows_every_run_of_that_day_together():
|
|
client, repo = _client()
|
|
html = client.get("/day/2026-10-16").text
|
|
for claim in ("9000001", "9000002", "9000003"):
|
|
assert f'data-claim="{claim}"' in html, claim
|
|
assert "10:30 AM" in html and "2:15 PM" in html, "each row's run time"
|
|
assert "3 claims checked · 2\n runs" in html, html
|
|
assert repo.calls == [("day_results", DAY)], repo.calls
|
|
# The short table only — no 24-column sideways table.
|
|
for header in ("Claim #", "Time", "Payer", "Result", "Comments"):
|
|
assert f"<th>{header}</th>" in html, header
|
|
assert "SERVICE DATE" not in html[html.index("<table"):html.index("</thead>")]
|
|
|
|
|
|
@case
|
|
def a_days_tiles_filter_but_still_count_the_whole_day():
|
|
client, _ = _client()
|
|
html = client.get("/day/2026-10-16", params={"category": "Fail"}).text
|
|
assert 'data-claim="9000002"' in html
|
|
assert 'data-claim="9000001"' not in html
|
|
assert 'class="tile Fail on"' in html
|
|
assert '<span class="n">3</span><span class="l">All</span>' in html, \
|
|
"the All tile counts the whole day"
|
|
html = client.get("/day/2026-10-16", params={"claim_no": "0003"}).text
|
|
assert 'data-claim="9000003"' in html and 'data-claim="9000002"' not in html
|
|
assert "?category=Fail&claim_no=0003" in html, "tiles keep the search"
|
|
|
|
|
|
@case
|
|
def a_day_row_carries_every_field_for_its_side_panel():
|
|
html = _client()[0].get("/day/2026-10-16").text
|
|
row = html[html.index('data-claim="9000003"'):]
|
|
row = row[:row.index("</tr>")]
|
|
fields = json.loads(row[row.index("data-fields='") + 13:
|
|
row.index("' data-details")]
|
|
.replace("'", "'"))
|
|
assert ["CLAIMS#", "9000003"] in fields, fields
|
|
assert ["SERVICE DATE", "07/31/2026"] in fields
|
|
assert ["Charges", "1,365.00"] in fields or any(
|
|
v == "1,365.00" for _, v in fields), fields
|
|
assert ["Mode", "VERIFY"] in fields
|
|
assert "dummy_codes" in row, "the conditions JSON is on the row"
|
|
|
|
|
|
@case
|
|
def a_day_opens_a_claim_from_the_link():
|
|
html = _client()[0].get("/day/2026-10-16",
|
|
params={"claim": "9000002"}).text
|
|
assert 'const want = "9000002";' in html
|
|
assert "showClaim(allRows[i]);" in html
|
|
|
|
|
|
@case
|
|
def a_day_with_nothing_says_so_and_a_bad_day_is_400():
|
|
client, _ = _client()
|
|
assert "No claims were checked on this day" in \
|
|
client.get("/day/2026-01-01").text
|
|
html = client.get("/day/2026-10-16", params={"category": "Pass",
|
|
"claim_no": "0002"}).text
|
|
assert "No claims match this filter" in html
|
|
for bad in ("/day/16-10-2026", "/day/2026-13-01", "/day/x"):
|
|
assert client.get(bad).status_code == 400, bad
|
|
assert client.get("/day/2026-10-16",
|
|
params={"category": "Bad"}).status_code == 400
|
|
|
|
|
|
@case
|
|
def claim_text_reaches_the_page_escaped():
|
|
evil = "<script>alert(1)</script>'"
|
|
rows = [_row("9000009", "Fail", 1, comments=evil, payer=evil)]
|
|
html = _client(FakeRepo(rows=rows))[0].get("/day/2026-10-16").text
|
|
assert "<script>alert(1)" not in html, "escaped in cells and attributes"
|
|
assert "<script>" in html
|
|
|
|
|
|
# ── Find a claim ─────────────────────────────────────────────────────────────
|
|
|
|
@case
|
|
def find_a_claim_lists_every_check_with_a_link_to_its_day():
|
|
client, repo = _client()
|
|
html = client.get("/claim", params={"claim_no": " 9000-002 "}).text
|
|
assert repo.calls == [("claim_history", "9000002")], repo.calls
|
|
assert 'href="/day/2026-10-16?claim=9000002"' in html, html
|
|
assert "No saved check" not in html
|
|
html = client.get("/claim", params={"claim_no": "1234567"}).text
|
|
assert "No saved check of a claim matching “1234567”" in html
|
|
|
|
|
|
@case
|
|
def find_a_claim_with_no_number_shows_the_box_only():
|
|
client, repo = _client()
|
|
html = client.get("/claim").text
|
|
assert 'name="claim_no"' in html and "<table" not in html
|
|
assert repo.calls == []
|
|
client.get("/claim", params={"claim_no": "1" * 500})
|
|
assert repo.calls == [("claim_history", "1" * 20)], repo.calls
|
|
|
|
|
|
# ── every page ───────────────────────────────────────────────────────────────
|
|
|
|
@case
|
|
def every_page_has_the_navy_header_breadcrumb_and_mode():
|
|
client, _ = _client()
|
|
for path in _PAGES:
|
|
html = client.get(path).text
|
|
assert '<header class="top">' in html, path
|
|
assert 'class="pill VERIFY"' in html, path
|
|
assert '<nav class="crumbs">' in html, path
|
|
assert "--navy: #12355b" in html, path
|
|
html = client.get("/day/2026-10-16").text
|
|
assert '<a href="/days">Claim status</a>' in html
|
|
assert '<span class="here">Fri, 16 Oct 2026</span>' in html
|
|
|
|
|
|
@case
|
|
def an_open_side_panel_stops_the_page_scrolling():
|
|
"""One scroll bar (operator, 2026-10-15): the page under an open panel
|
|
does not scroll; Escape and every Close button shut it."""
|
|
html = _client()[0].get("/day/2026-10-16").text
|
|
assert "overflow: hidden" in _css_rule(html, "body.drawer-open")
|
|
assert "classList.toggle('drawer-open', id !== null)" in html
|
|
assert "e.target.closest('.drawer .close')" in html
|
|
assert "if (e.key === 'Escape') closePanels();" in html
|
|
assert "openPanel('claim-panel')" in html
|
|
|
|
|
|
@case
|
|
def the_tables_page_ten_claims_and_never_scroll_sideways():
|
|
html = _client()[0].get("/day/2026-10-16").text
|
|
assert "const PAGE_SIZE = 10;" in html
|
|
assert 'id="pager-top"' in html and 'id="pager-bottom"' in html
|
|
assert "if (id === 'pager-bottom')" in html
|
|
assert "table-layout: fixed" in _css_rule(html, "table.list")
|
|
assert "text-overflow: ellipsis" in html
|
|
|
|
|
|
@case
|
|
def the_panel_writes_db_text_with_text_content_only():
|
|
html = _client()[0].get("/day/2026-10-16").text
|
|
script = html[html.index("function showClaim"):]
|
|
script = script[:script.index("openPanel('claim-panel')")]
|
|
assert "innerHTML" not in script
|
|
assert "dd.textContent = value;" in script
|
|
assert "CATEGORIES.includes(cat)" in script, "class names allowlisted"
|
|
|
|
|
|
@case
|
|
def no_page_carries_an_inline_style_attribute():
|
|
client, _ = _client()
|
|
for path in _PAGES:
|
|
assert " style=" not in client.get(path).text, path
|
|
|
|
|
|
@case
|
|
def the_live_claims_panel_opens_a_claim_on_its_day():
|
|
"""The Bot run card's Live claims list a run's claims; a saved result
|
|
opens its day page with the claim's panel; a run that saves nothing, or
|
|
a claim Not found, opens nothing."""
|
|
html = _bot_client()[0].get("/").text
|
|
assert 'id="live-panel"' in html and "s.claims" in html
|
|
assert "s.results_run_at.slice(0, 10)" in html
|
|
assert "'/day/' + day + '?claim=' +" in html
|
|
assert "const result = c.done && c.status !== 'Not found';" in html
|
|
assert "if (result && day) {" in html
|
|
|
|
|
|
# ── api ──────────────────────────────────────────────────────────────────────
|
|
|
|
@case
|
|
def api_runs_is_json_with_iso_times():
|
|
client, _ = _client()
|
|
data = client.get("/api/runs").json()
|
|
assert data[0]["run_at"] == "2026-10-16T14:15:00", data
|
|
assert data[0]["review_count"] == 1
|
|
|
|
|
|
@case
|
|
def api_results_returns_rows_with_details_parsed():
|
|
client, repo = _client()
|
|
data = client.get("/api/results", params={"category": "Fail"}).json()
|
|
assert data["run_at"] == "2026-10-16T14:15:00", data
|
|
assert [r["claim_no"] for r in data["rows"]] == ["9000002"]
|
|
row = data["rows"][0]
|
|
assert row["details"] == {}
|
|
assert "details_json" not in row
|
|
assert row["service_date"] == "2026-07-31"
|
|
assert float(row["charges"]) == 1365.0
|
|
assert repo.calls == [("results", RUN_PM, "Fail", None)]
|
|
|
|
|
|
@case
|
|
def a_page_or_api_says_the_database_is_not_reachable():
|
|
client, _ = _client(FakeRepo(fail=True))
|
|
for path in _PAGES:
|
|
r = _quiet(client.get, path)
|
|
assert r.status_code == 503, path
|
|
assert "Results database not reachable" in r.text, path
|
|
assert "Traceback" not in r.text
|
|
r = _quiet(client.get, "/api/results")
|
|
assert r.status_code == 503
|
|
assert r.json() == {"detail": "Results database not reachable"}
|
|
|
|
|
|
@case
|
|
def the_app_has_no_public_docs_pages():
|
|
client, _ = _client()
|
|
assert client.get("/docs").status_code == 404
|
|
assert client.get("/openapi.json").status_code == 404
|
|
|
|
|
|
# ── hardening (review, 2026-10-12) ───────────────────────────────────────────
|
|
|
|
@case
|
|
def a_request_for_another_host_name_is_refused():
|
|
"""DNS rebinding: a page on evil.example resolving to 127.0.0.1."""
|
|
for base in ("http://evil.example:8000", "http://192.168.1.5:8000"):
|
|
client, repo = _client(base_url=base)
|
|
for path in _PAGES + ("/api/runs",):
|
|
assert client.get(path).status_code == 400, (base, path)
|
|
assert repo.calls == [], "no data read for a foreign host"
|
|
client, _ = _client(base_url="http://localhost:8000")
|
|
assert client.get("/").status_code == 200
|
|
|
|
|
|
@case
|
|
def responses_are_never_cached():
|
|
client, _ = _client()
|
|
for path in _PAGES + ("/api/runs", "/api/results"):
|
|
r = client.get(path)
|
|
assert r.headers.get("cache-control") == "no-store", (path, r.headers)
|
|
assert r.headers.get("x-frame-options") == "DENY", path
|
|
assert r.headers.get("x-content-type-options") == "nosniff", path
|
|
r = _quiet(_client(FakeRepo(fail=True))[0].get, "/days")
|
|
assert r.headers.get("cache-control") == "no-store", "error pages too"
|
|
|
|
|
|
@case
|
|
def the_error_page_names_what_went_wrong():
|
|
class DbError(Exception):
|
|
pass
|
|
|
|
for state, text in (("42S02", "table is missing"),
|
|
("28000", "login"), ("42000", "permission"),
|
|
("08001", "not reachable")):
|
|
client, _ = _client(FakeRepo(fail=DbError(state, "[Microsoft] "
|
|
"Smith, John")))
|
|
r = _quiet(client.get, "/days")
|
|
assert r.status_code == 503, state
|
|
assert text in r.text, (state, r.text)
|
|
assert "Smith" not in r.text, "the driver's message is never shown"
|
|
|
|
|
|
# ── repo SQL ─────────────────────────────────────────────────────────────────
|
|
|
|
class FakeCursor:
|
|
def __init__(self, log, rows, cols):
|
|
self.log, self.rows = log, rows
|
|
self.description = [(c,) for c in cols]
|
|
|
|
def execute(self, sql, params):
|
|
self.log.append((sql, list(params)))
|
|
return self
|
|
|
|
def fetchall(self):
|
|
return self.rows
|
|
|
|
|
|
class FakeConn:
|
|
def __init__(self, rows=(), cols=()):
|
|
self.log, self.rows, self.cols, self.closed = [], list(rows), cols, False
|
|
|
|
def cursor(self):
|
|
return FakeCursor(self.log, self.rows, self.cols)
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
def _repo(conn):
|
|
return SqlResultsRepo("DSN=x", connect=lambda cs, timeout: conn)
|
|
|
|
|
|
@case
|
|
def repo_turns_a_text_run_at_or_day_into_a_datetime_or_date():
|
|
"""The legacy {SQL Server} driver returns DATETIME2 / DATE as text."""
|
|
conn = FakeConn(rows=[("2026-10-16 14:15:00.0000000", "VERIFY", 3, 1, 1,
|
|
1)],
|
|
cols=("run_at", "run_mode", "claims", "pass_count",
|
|
"fail_count", "review_count"))
|
|
assert _repo(conn).runs()[0]["run_at"] == RUN_PM
|
|
conn = FakeConn(rows=[("2026-10-16", 3, 2, 1, 1, 1)],
|
|
cols=("day", "claims", "runs", "pass_count",
|
|
"fail_count", "review_count"))
|
|
assert _repo(conn).days()[0]["day"] == DAY
|
|
|
|
|
|
@case
|
|
def repo_results_is_a_parameterized_select():
|
|
conn = FakeConn(rows=[("9000002", "Fail")], cols=("claim_no", "category"))
|
|
rows = _repo(conn).results(RUN_PM, category="Fail", claim_no="9000")
|
|
assert rows == [{"claim_no": "9000002", "category": "Fail"}]
|
|
(sql, params), = conn.log
|
|
assert sql.lstrip().upper().startswith("SELECT"), sql
|
|
assert "FROM dbo.CV_ChargeSummary" in sql
|
|
assert "run_at = ?" in sql and "category = ?" in sql
|
|
assert "claim_no LIKE ?" in sql
|
|
assert params == [RUN_PM, "Fail", "%9000%"], params
|
|
assert "9000" not in sql, "values never go into the SQL text"
|
|
assert conn.closed
|
|
|
|
|
|
@case
|
|
def repo_runs_groups_by_run_newest_first():
|
|
conn = FakeConn()
|
|
assert _repo(conn).runs() == []
|
|
(sql, params), = conn.log
|
|
assert "GROUP BY run_at" in sql and "ORDER BY run_at DESC" in sql
|
|
assert params == []
|
|
|
|
|
|
@case
|
|
def repo_days_groups_by_the_day_a_run_started():
|
|
conn = FakeConn()
|
|
_repo(conn).days()
|
|
(sql, params), = conn.log
|
|
assert sql.lstrip().upper().startswith("SELECT TOP 90 "), sql
|
|
assert "GROUP BY CAST(run_at AS DATE)" in sql
|
|
assert "ORDER BY CAST(run_at AS DATE) DESC" in sql
|
|
assert "COUNT(DISTINCT run_at) AS runs" in sql
|
|
assert params == []
|
|
|
|
|
|
@case
|
|
def repo_day_results_takes_the_whole_day():
|
|
conn = FakeConn()
|
|
_repo(conn).day_results(DAY)
|
|
(sql, params), = conn.log
|
|
assert "WHERE run_at >= ? AND run_at < ?" in sql, sql
|
|
assert "ORDER BY run_at, s_no, id" in sql
|
|
assert params == [datetime(2026, 10, 16), datetime(2026, 10, 17)], params
|
|
|
|
|
|
@case
|
|
def repo_claim_history_is_newest_first_and_capped():
|
|
conn = FakeConn()
|
|
_repo(conn).claim_history("9000002")
|
|
(sql, params), = conn.log
|
|
assert sql.lstrip().upper().startswith("SELECT TOP 200 "), sql
|
|
assert "WHERE claim_no LIKE ?" in sql
|
|
assert "ORDER BY run_at DESC" in sql
|
|
assert params == ["%9000002%"] and "9000002" not in sql
|
|
|
|
|
|
# ── start / stop the bot from the page ───────────────────────────────────────
|
|
|
|
@case
|
|
def the_dashboard_carries_the_run_token_and_the_start_button():
|
|
client, _, token = _bot_client()
|
|
page = client.get("/").text
|
|
assert f'content="{token}"' in page
|
|
assert 'id="br-start"' in page and "VERIFY mode" in page
|
|
|
|
|
|
@case
|
|
def the_bot_run_card_shows_even_with_no_runs():
|
|
client, _ = signed_in(FakeRepo(days=[], rows=[]), FakeRunner())
|
|
page = client.get("/").text
|
|
assert 'id="br-start"' in page and "No results saved yet" in page
|
|
|
|
|
|
@case
|
|
def start_without_the_token_or_origin_is_forbidden():
|
|
client, runner, token = _bot_client()
|
|
for headers in ({}, {"X-Run-Token": token},
|
|
{**_GOOD_ORIGIN},
|
|
{**_GOOD_ORIGIN, "X-Run-Token": "wrong"},
|
|
{"Origin": "http://evil.example", "X-Run-Token": token},
|
|
{"Referer": "http://evil.example/x",
|
|
"X-Run-Token": token}):
|
|
r = client.post("/api/bot/start", headers=headers)
|
|
assert r.status_code == 403, (headers, r.status_code)
|
|
assert runner.calls == [], runner.calls
|
|
|
|
|
|
@case
|
|
def stop_without_the_token_or_origin_is_forbidden():
|
|
client, runner, token = _bot_client(FakeRunner(running=True))
|
|
for headers in ({}, {"X-Run-Token": token}, {**_GOOD_ORIGIN},
|
|
{"Origin": "http://evil.example", "X-Run-Token": token}):
|
|
r = client.post("/api/bot/stop", headers=headers)
|
|
assert r.status_code == 403, (headers, r.status_code)
|
|
assert runner.calls == [], runner.calls
|
|
|
|
|
|
@case
|
|
def start_and_stop_with_origin_and_token():
|
|
client, runner, token = _bot_client()
|
|
h = {**_GOOD_ORIGIN, "X-Run-Token": token}
|
|
r = client.post("/api/bot/start", headers=h)
|
|
assert r.status_code == 200 and r.json()["running"] is True, r.text
|
|
r = client.post("/api/bot/stop", headers=h)
|
|
assert r.status_code == 200 and r.json()["running"] is False, r.text
|
|
assert runner.calls == ["start", "stop"]
|
|
|
|
|
|
@case
|
|
def a_referer_from_the_page_counts_when_origin_is_absent():
|
|
client, runner, token = _bot_client()
|
|
r = client.post("/api/bot/start", headers={
|
|
"Referer": "http://localhost:8000/days", "X-Run-Token": token})
|
|
assert r.status_code == 200, r.text
|
|
|
|
|
|
@case
|
|
def start_while_running_is_409_and_stop_of_nothing_is_409():
|
|
client, runner, token = _bot_client(FakeRunner(running=True))
|
|
h = {**_GOOD_ORIGIN, "X-Run-Token": token}
|
|
r = client.post("/api/bot/start", headers=h)
|
|
assert r.status_code == 409 and "already going" in r.json()["detail"]
|
|
client.post("/api/bot/stop", headers=h)
|
|
r = client.post("/api/bot/stop", headers=h)
|
|
assert r.status_code == 409 and "No run" in r.json()["detail"]
|
|
|
|
|
|
@case
|
|
def a_stop_that_cannot_kill_the_run_is_500_with_the_message():
|
|
msg = "Could not stop the run (pid 1 is still running) — end it by hand"
|
|
client, runner, token = _bot_client(FakeRunner(running=True,
|
|
stop_fails=msg))
|
|
r = client.post("/api/bot/stop",
|
|
headers={**_GOOD_ORIGIN, "X-Run-Token": token})
|
|
assert r.status_code == 500, r.status_code
|
|
assert r.json() == {"detail": msg}, r.text
|
|
|
|
|
|
@case
|
|
def a_non_ascii_token_or_a_null_origin_is_forbidden_not_a_crash():
|
|
client, runner, token = _bot_client()
|
|
r = client.post("/api/bot/start", headers={
|
|
**_GOOD_ORIGIN, "X-Run-Token": "tokén".encode("latin-1")})
|
|
assert r.status_code == 403, (r.status_code, r.text)
|
|
r = client.post("/api/bot/start",
|
|
headers={"Origin": "null", "X-Run-Token": token})
|
|
assert r.status_code == 403, (r.status_code, r.text)
|
|
assert runner.calls == [], runner.calls
|
|
|
|
|
|
@case
|
|
def the_card_says_when_the_run_was_started_in_the_terminal():
|
|
"""The page's JS hides Start and Stop for a terminal run (its Stop
|
|
would be refused) and says where to stop it."""
|
|
page = _bot_client()[0].get("/").text
|
|
assert "Run started in the terminal" in page
|
|
assert "s.source !== 'web'" in page
|
|
|
|
|
|
@case
|
|
def hidden_really_hides_inside_the_bot_run_card():
|
|
"""`.botrun .row { display: flex }` beats the browser's [hidden] rule:
|
|
without this, #br-progress (a .row) never hides."""
|
|
page = _bot_client()[0].get("/").text
|
|
assert ".botrun [hidden] { display: none; }" in page
|
|
assert page.index(".botrun [hidden]") > page.index(".botrun .row {")
|
|
|
|
|
|
@case
|
|
def the_card_script_polls_once_and_shows_the_exit_code():
|
|
page = _bot_client()[0].get("/").text
|
|
assert "clearTimeout(timer)" in page, "one poll loop after a Start"
|
|
assert "— exit ${s.exit_code}" in page, "ended message has the code"
|
|
assert "$('br-stop').disabled = true" in page, "Stop off while posting"
|
|
assert "location.reload()" in page, "today's counts after a run ends"
|
|
|
|
|
|
@case
|
|
def a_start_or_poll_with_the_portal_gone_says_so():
|
|
"""run_web.bat closed: the fetch throws. A Start used to do nothing at
|
|
all after its confirm (live, 2026-10-22)."""
|
|
page = _bot_client()[0].get("/").text
|
|
assert "is run_web.bat still" in page
|
|
assert "alert(UNREACHABLE);" in page, "a Start that cannot reach it"
|
|
assert "textContent = UNREACHABLE;" in page, "the status poll too"
|
|
|
|
|
|
@case
|
|
def status_has_the_page_fields_and_no_file_paths():
|
|
client, _, _ = _bot_client(FakeRunner(running=True))
|
|
s = client.get("/api/bot/status").json()
|
|
for key in ("running", "phase", "progress", "counts", "tail",
|
|
"message", "started_at", "exit_code", "claims",
|
|
"results_run_at", "started_by"):
|
|
assert key in s, (key, s)
|
|
assert s["results_run_at"] is None, "no log, no saved run"
|
|
assert "log_path" not in s and "out_path" not in s, s
|
|
assert "pid" not in s, s
|
|
|
|
|
|
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())
|