Merge integration/rpa-dashboard-skills (d89dce3) into ticket 04

Brings in tickets 05-07 (AI skill, cost page, AI desktop sections). tests/README.md
keeps both the optional-section notes and the AI page-check notes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
VasanthandClaude Opus 5.5 committed 2026-10-16 15:35:44 +05:30
commit 73adc74637
67 files changed
+4826 -100

No files matched your search

+21
View File
@@ -27,6 +27,23 @@ uv run --with pytest --with playwright pytest tests
the login section is signed in (`login_result ok`) before each case.
- `test_sections.py`: the optional-section strip tool keeps only the chosen
sections, and a page with none of them still passes the page check.
- `page_check_ai/`: **Seam 1, AI web page.** Builds the React starter with npm
(filled with test values, in a temp folder that keeps `node_modules`), serves it with
its own `server_stub.py`, and checks bands, band filters, drafts, the approve gate
"page X of Y" and the cost page (spend tiles, coverage pill, month breakdown,
by-hand assumptions), plus every no-AI case. Tests set run and review state on the stub
directly (the `stub` fixture). To check a generated dashboard folder instead, set
`PAGE_CHECK_AI_DIR=path/to/dashboard`. Needs npm on PATH, else it skips.
- `page_check_ai_desktop/`: **Seam 1, AI desktop page.** The no-AI `dashboard.html` with
`ai_sections.html` pasted in, loaded from `file://` with a fake Qt bridge whose
`ai_request` slot goes to `server_stub.dispatch()` in this process (what `ai_bridge.py`
does). Runs the AI web page's review and cost cases and the no-AI cases unchanged, plus
a tampered page calling the slot directly (still refused). To check a generated folder,
set `PAGE_CHECK_AI_DESKTOP_DIR=path/to/dashboard`; `PAGE_CHECK_WIRED=1` as above.
- `test_pricing.py`: the AI cost function (normal, long-context, thinking tokens,
unknown model).
- `test_reference_scan.py`: no patient data, internal IPs, credentials or keys in any
bundled file, and every Superbill extract carries an `EXAMPLE ONLY` banner.
- `evals/`: **Seam 2.** Runs the skills on a fixture project with `claude -p`,
then grades the reply and runs the page check on the page it built:
@@ -38,6 +55,10 @@ uv run --with pytest --with playwright pytest tests
`FIXTURES` in `run_eval.py`. An entry with a `"fixture"` key reuses another
fixture folder with different answers: `c-keep` and `c-restyle` both run on
`c-web-existing-frontend`, and also check what changed in the files it already had.
An `"ai": True` entry (`d-ai-extraction`) expects the AI skill's React starter and
runs `page_check_ai/` on the folder it built. An `"ai_desktop": True` entry
(`e-pyqt-ai`) expects the desktop AI page and runs `page_check/` (with the bot replay)
and `page_check_ai_desktop/` on it.
For a bot fixture (`b-pyqt-bot`), the eval also dry-runs the generated
`bot_runner.py`, saves its events to `evals/results/<fixture>.run.jsonl`, and
replays them in the page check.
+12
View File
@@ -0,0 +1,12 @@
"""Shared by both page checks: one headless Chromium for the whole run
(Playwright's sync API allows only one instance per process)."""
import pytest
from playwright.sync_api import sync_playwright
@pytest.fixture(scope="session")
def browser():
with sync_playwright() as p:
b = p.chromium.launch()
yield b
b.close()
+39 -1
View File
@@ -81,6 +81,28 @@ FIXTURES = {
"page_has": ["--accent:#ff3049"],
"existing": "#ff3049", # only token lines may change, and the theme now uses this colour
},
# (d) web service with an LLM extraction call and no UI: the AI skill's web path.
"d-ai-extraction": {
"prompt": "Build me an RPA dashboard for this project. I can't answer questions while you "
"work, so here are my answers to anything the skill asks. Colours: Superbill Ember. "
"Desktop or web: web. Optional sections: none.",
"frontend": r"^\s*(none|no\b)",
"asks": [r"colou?r", r"desktop|web", r"section"],
"never_asks": [r"\bkeep\b|restyle"],
"page_has": ["--accent:#ff3049"],
"ai": True,
},
# (e) PyQt bot with an LLM extraction call: the AI skill's desktop path, no web server.
"e-pyqt-ai": {
"prompt": "Build me an RPA dashboard for this project. I can't answer questions while you "
"work, so here are my answers to anything the skill asks. Colours: Tebra Sky. "
"Optional sections: none.",
"frontend": r"^\s*(none|no\b)",
"asks": [r"colou?r", r"section"],
"never_asks": [r"\bkeep\b|restyle", r"desktop|web"],
"bot": True,
"ai_desktop": True,
},
}
# A line a restyle may change: a custom property, a comment, a blank, or a web-font link.
TOKEN_LINE = r"^\s*(--[\w-]+\s*:.*|/\*.*\*/|<link[^>]*fonts\.googleapis\.com.*|@import.*fonts\.googleapis\.com.*)?\s*$"
@@ -167,6 +189,16 @@ def main(fixture):
results.mkdir(exist_ok=True)
if pages:
html = pages[0].read_text(encoding="utf-8")
if cfg.get("ai"): # the React starter: its theme lives in src/index.css
css = pages[0].parent / "src" / "index.css"
html += css.read_text(encoding="utf-8") if css.exists() else ""
checks["AI path: review starter and server stub built"] = all(
(pages[0].parent / f).exists() for f in ["server_stub.py", "bands.py", "src/components/Review.tsx"])
if cfg.get("ai_desktop"): # the no-AI page with the plain-HTML AI sections, hosted by ai_bridge.py
host = pages[0].with_name("ai_bridge.py")
checks["AI desktop path: sections, ai_bridge and review backend built"] = 'id="ai-review-tab"' in html and all(
pages[0].with_name(f).exists() for f in ["ai_bridge.py", "server_stub.py", "bands.py", "pricing.py"])
checks["no placeholders left in ai_bridge.py"] = host.exists() and "{{" not in host.read_text(encoding="utf-8")
shutil.copy(pages[0], results / f"{fixture}.dashboard.html")
checks["no placeholders left"] = "{{" not in html
if cfg.get("brand"):
@@ -187,7 +219,13 @@ def main(fixture):
env = {**os.environ, "PAGE_CHECK_HTML": str(pages[0])}
if cfg.get("bot"):
env.update(bot_checks(project, html, checks, results / f"{fixture}.run.jsonl"))
pc = subprocess.run([sys.executable, "-m", "pytest", str(ROOT / "tests" / "page_check"), "-q"],
if cfg.get("ai"):
env["PAGE_CHECK_AI_DIR"] = str(pages[0].parent)
check_dirs = [ROOT / "tests" / ("page_check_ai" if cfg.get("ai") else "page_check")]
if cfg.get("ai_desktop"):
env["PAGE_CHECK_AI_DESKTOP_DIR"] = str(pages[0].parent)
check_dirs.append(ROOT / "tests" / "page_check_ai_desktop")
pc = subprocess.run([sys.executable, "-m", "pytest", *map(str, check_dirs), "-q"],
env=env, capture_output=True, text=True)
checks["page check passes"] = pc.returncode == 0
page_check = pc.stdout.strip().splitlines()[-1:] if pc.stdout else [pc.stderr[-500:]]
+8
View File
@@ -0,0 +1,8 @@
# Invoice extraction service
A small web service. It sends a scanned invoice to an LLM and gets back the
invoice fields, each with a confidence score per word. There is no UI yet.
```
uvicorn app:app --port 8000
```
+11
View File
@@ -0,0 +1,11 @@
"""Upload a scanned invoice, get its fields back."""
from fastapi import FastAPI, UploadFile
from extractor import extract_invoice
app = FastAPI()
@app.post("/extract")
async def extract(file: UploadFile):
return extract_invoice(await file.read(), file.content_type or "application/pdf")
+34
View File
@@ -0,0 +1,34 @@
"""The LLM extraction call: one scanned invoice in, its fields out, each with word scores."""
import json
from google import genai
from google.genai import types
MODEL = "gemini-2.5-flash"
FIELDS = { # field name -> mandatory
"Invoice number": True,
"Invoice date": True,
"Vendor": True,
"PO number": False,
"Due date": True,
"Total": True,
}
PROMPT = (
"Read this invoice. Return JSON: a list of {field, value, word_scores}, one per field in "
+ ", ".join(FIELDS)
+ ". word_scores is your 0-1 legibility score for each word of the value. Use an empty "
"value and no scores when a field isn't on the page."
)
def extract_invoice(document: bytes, mime_type: str) -> list[dict]:
client = genai.Client() # credentials come from the environment
response = client.models.generate_content(
model=MODEL,
contents=[types.Part.from_bytes(data=document, mime_type=mime_type), PROMPT],
config=types.GenerateContentConfig(response_mime_type="application/json"),
)
rows = json.loads(response.text)
for row in rows:
row["mandatory"] = FIELDS.get(row["field"], False)
return rows
+4
View File
@@ -0,0 +1,4 @@
fastapi
uvicorn
google-genai
python-multipart
+11
View File
@@ -0,0 +1,11 @@
# Invoice intake bot
A desktop RPA bot. It reads each scanned invoice in `inbox/` with an LLM
(`extractor.py`) and queues the fields for a person to review.
The operator window will be a PyQt6 app.
Run the bot on its own:
```
python -m bot
```
+31
View File
@@ -0,0 +1,31 @@
"""Invoice intake bot: picks up scanned invoices and extracts their fields with an LLM.
Start it with: python -m bot"""
import time
from pathlib import Path
INBOX = Path(__file__).with_name("inbox")
def say(line):
print(line, flush=True)
time.sleep(0.2)
def main():
say("[step 1] Initializing")
docs = sorted(INBOX.glob("*.pdf"))
say(f"[inbox] {len(docs)} invoices waiting")
say("[step 2] Extracting fields")
for i, doc in enumerate(docs, 1):
try:
from extractor import extract_invoice # the LLM call
rows = extract_invoice(doc.read_bytes(), "application/pdf")
say(f"[extract] {doc.name}: {len(rows)} fields ({i}/{len(docs)})")
except Exception as e: # one bad document must not stop the batch
say(f"ERROR: extraction failed for {doc.name}: {type(e).__name__}")
say("[step 3] Queueing for review")
say(f"[all done] {len(docs)} invoices processed")
if __name__ == "__main__":
main()
+34
View File
@@ -0,0 +1,34 @@
"""The LLM extraction call: one scanned invoice in, its fields out, each with word scores."""
import json
from google import genai
from google.genai import types
MODEL = "gemini-2.5-flash"
FIELDS = { # field name -> mandatory
"Invoice number": True,
"Invoice date": True,
"Vendor": True,
"PO number": False,
"Due date": True,
"Total": True,
}
PROMPT = (
"Read this invoice. Return JSON: a list of {field, value, word_scores}, one per field in "
+ ", ".join(FIELDS)
+ ". word_scores is your 0-1 legibility score for each word of the value. Use an empty "
"value and no scores when a field isn't on the page."
)
def extract_invoice(document: bytes, mime_type: str) -> list[dict]:
client = genai.Client() # credentials come from the environment
response = client.models.generate_content(
model=MODEL,
contents=[types.Part.from_bytes(data=document, mime_type=mime_type), PROMPT],
config=types.GenerateContentConfig(response_mime_type="application/json"),
)
rows = json.loads(response.text)
for row in rows:
row["mandatory"] = FIELDS.get(row["field"], False)
return rows
+2
View File
@@ -0,0 +1,2 @@
%PDF-1.4
% sample invoice placeholder
+2
View File
@@ -0,0 +1,2 @@
%PDF-1.4
% sample invoice placeholder
+3
View File
@@ -0,0 +1,3 @@
PyQt6
PyQt6-WebEngine
google-genai
-9
View File
@@ -13,7 +13,6 @@ import re
from pathlib import Path
import pytest
from playwright.sync_api import sync_playwright
ROOT = Path(__file__).resolve().parents[2]
TEMPLATE = ROOT / "rpa-dashboard-no-ai" / "templates" / "dashboard.html"
@@ -57,14 +56,6 @@ def page_url(tmp_path_factory):
return out.as_uri()
@pytest.fixture(scope="session")
def browser():
with sync_playwright() as p:
b = p.chromium.launch()
yield b
b.close()
@pytest.fixture(params=list(MODES))
def page(request, browser, page_url):
width, height, dark = MODES[request.param]
+135
View File
@@ -0,0 +1,135 @@
"""Seam 1 page check for the AI web dashboard (React + Vite), served by its server stub.
Page under test: set PAGE_CHECK_AI_DIR to a generated dashboard folder (the one with
package.json, index.html and server_stub.py). Without it, the AI web starter template
is filled with TEST_FILL and built in a cache folder. Either way the folder is built
with npm, and its server_stub.py runs in this process, so tests can set run and
review state directly through the `stub` fixture.
Contract envelopes go in through window.__pyMsg, as in the no-AI check. Each action
the page POSTs to {API_BASE}/<slot> is recorded in window.__calls as [slot, ...args].
Every page test runs in light, dark and narrow mode.
"""
import hashlib
import importlib.util
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import threading
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
TEMPLATE = ROOT / "rpa-dashboard-ai" / "templates" / "web"
TEST_FILL = {
"APP_NAME": "Test Extractor",
"APP_SUBTITLE": "Page check",
"LOGO_SRC": "",
"API_BASE": "/api",
"STEPS_JSON": json.dumps(["Initializing", "Login", "Process items", "Completed"]),
"REPORT_COLUMNS_JSON": json.dumps([{"key": "item", "label": "Item"}, {"key": "status", "label": "Status"}]),
}
RECORD_ACTIONS = """
window.__calls = [];
const realFetch = window.fetch;
window.fetch = (url, opts = {}) => {
const m = String(url).match(/\\/api\\/([a-z-]+)$/);
if (m && opts.method === 'POST') window.__calls.push([m[1].replace(/-/g, '_'), ...JSON.parse(opts.body || '[]')]);
return realFetch(url, opts);
};
"""
MODES = {"light": (1280, 820, False), "dark": (1280, 820, True), "narrow": (800, 900, False)}
SKIP_COPY = shutil.ignore_patterns("node_modules", "dist", "__pycache__")
@pytest.fixture
def sample_data():
"""Shared no-AI cases that need the page's own sample data skip on a bot-wired page."""
if os.environ.get("PAGE_CHECK_WIRED") == "1":
pytest.skip("page is wired to a bot, so it has no sample data")
@pytest.fixture(scope="session")
def site():
"""The dashboard folder under test: server_stub.py, index.html and the sources."""
if os.environ.get("PAGE_CHECK_AI_DIR"):
return Path(os.environ["PAGE_CHECK_AI_DIR"]).resolve()
pkg = (TEMPLATE / "package.json").read_bytes() + (TEMPLATE / "package-lock.json").read_bytes()
out = Path(tempfile.gettempdir()) / f"rpa-dashboard-ai-web-{hashlib.sha1(pkg).hexdigest()[:10]}"
shutil.rmtree(out / "src", ignore_errors=True) # node_modules is kept between runs
shutil.copytree(TEMPLATE, out, dirs_exist_ok=True, ignore=SKIP_COPY)
index = out / "index.html"
index.write_text(re.sub(r"\{\{(\w+)\}\}", lambda m: TEST_FILL[m[1]], index.read_text(encoding="utf-8")),
encoding="utf-8")
return out
@pytest.fixture(scope="session")
def built(site):
npm = shutil.which("npm") or pytest.skip("npm not on PATH")
if not (site / "node_modules").exists():
subprocess.run([npm, "ci" if (site / "package-lock.json").exists() else "install"], cwd=site, check=True)
r = subprocess.run([npm, "run", "build"], cwd=site, capture_output=True, text=True)
assert r.returncode == 0, r.stdout + r.stderr
return site
@pytest.fixture(scope="session")
def stub_module(site):
sys.path.insert(0, str(site)) # the stub imports bands.py from its own folder
spec = importlib.util.spec_from_file_location("server_stub", site / "server_stub.py")
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
@pytest.fixture(scope="session")
def server(stub_module):
srv = stub_module.make_server("127.0.0.1", 0)
threading.Thread(target=srv.serve_forever, daemon=True).start()
yield f"http://127.0.0.1:{srv.server_address[1]}"
srv.shutdown()
@pytest.fixture
def stub(stub_module, server):
"""The running stub, reset to its sample data. Tests change stub.RUNS / stub.ROWS directly."""
stub_module.reset()
return stub_module
@pytest.fixture(scope="session")
def page_url(built, server):
return server + "/"
@pytest.fixture(params=list(MODES))
def page(request, browser, page_url, stub):
width, height, dark = MODES[request.param]
ctx = browser.new_context(viewport={"width": width, "height": height})
ctx.add_init_script(RECORD_ACTIONS)
pg = ctx.new_page()
pg.set_default_timeout(5000)
errors = []
pg.on("pageerror", lambda e: errors.append(e))
pg.goto(page_url)
assert not errors, f"page script errors: {errors}"
if dark:
push_to(pg, "apply_dark", {"on": True})
yield pg
ctx.close()
def push_to(pg, event, data):
pg.evaluate("([event, data]) => window.__pyMsg(JSON.stringify({event, data}))", [event, data])
@pytest.fixture
def push(page):
"""push(event, data): send one contract envelope to the page, as the bot would."""
return lambda event, data: push_to(page, event, data)
+85
View File
@@ -0,0 +1,85 @@
"""AI cost page: spend tiles, coverage pill, month breakdown and the by-hand assumptions.
ROI runs are set on the server stub (stub.ROI_RUNS); the stub prices them with pricing.py.
The test pins its own pricing table, so it holds on a generated dashboard whose table
lists only the project's models. See rpa-dashboard-ai/templates/contract.md.
"""
import sys
from datetime import datetime, timedelta, timezone
import pytest
from playwright.sync_api import expect
NOW = datetime.now(timezone.utc)
LAST = NOW.replace(day=1) - timedelta(days=1) # a day in last month
def roi_run(job_id, when, model, inp=None, out=None, thinking=0):
return {"job_id": job_id, "document": f"{job_id}.pdf", "created_at": when.strftime("%Y-%m-%dT%H:%M:%SZ"),
"model": model, "input_tokens": inp, "output_tokens": out, "thinking_tokens": thinking}
TABLE = {
"test-flash": {"input": 0.30, "output": 2.50},
"test-pro": {"input": 1.25, "output": 10.00, "long_threshold": 200_000, "input_long": 2.50, "output_long": 15.00},
"test-lite": {"input": 0.10, "output": 0.40},
}
AS_OF = "2030-01-31"
# Costs worked by hand from TABLE:
RUNS = [
roi_run("r1", NOW, "test-flash", 1_000_000, 200_000), # 0.30 + 0.50 = 0.80
roi_run("r2", NOW, "test-pro", 400_000, 60_000, 40_000), # long: 1.00 + 1.50 = 2.50
roi_run("r3", LAST, "test-lite", 1_000_000, 1_000_000), # 0.10 + 0.40 = 0.50
roi_run("r4", LAST, "test-flash"), # no token usage recorded
]
@pytest.fixture
def cost_page(page, stub, monkeypatch):
monkeypatch.setattr(sys.modules["pricing"], "PRICING", TABLE) # the stub's own pricing module
monkeypatch.setattr(stub, "PRICING_AS_OF", AS_OF)
stub.ROI_RUNS[:] = RUNS
page.get_by_role("navigation").get_by_role("button", name="Cost", exact=True).click()
return page
def tile(page, name):
return page.get_by_role("group", name=name, exact=True)
def test_spend_tiles_sum_the_priced_runs(cost_page):
expect(tile(cost_page, "Total spend")).to_contain_text("$3.80")
expect(tile(cost_page, "This month")).to_contain_text("$3.30")
def test_coverage_pill_counts_priced_of_all_documents(cost_page):
expect(cost_page.get_by_text("3 of 4 documents priced")).to_be_visible()
def test_month_breakdown_groups_runs_by_month(cost_page):
rows = cost_page.get_by_role("table", name="Spend by month").get_by_role("row")
expect(rows).to_have_count(3) # header + two months, newest first
expect(rows.nth(1)).to_contain_text(NOW.strftime("%b %Y"))
expect(rows.nth(1)).to_contain_text("2 of 2")
expect(rows.nth(1)).to_contain_text("$3.30")
expect(rows.nth(2)).to_contain_text(LAST.strftime("%b %Y"))
expect(rows.nth(2)).to_contain_text("1 of 2")
expect(rows.nth(2)).to_contain_text("$0.50")
def test_runs_show_tokens_and_cost_and_unpriced_is_not_zero(cost_page):
runs = cost_page.get_by_role("table", name="Cost per run")
r2 = runs.get_by_role("row").filter(has_text="r2.pdf")
for text in ("test-pro", "400,000", "60,000", "40,000", "$2.50"):
expect(r2).to_contain_text(text)
expect(runs.get_by_role("row").filter(has_text="r4.pdf")).to_contain_text("not priced")
def test_by_hand_panel_is_labelled_assumptions_and_recomputes(cost_page):
panel = cost_page.get_by_role("region", name="Against doing it by hand")
expect(panel).to_contain_text("assumptions")
# 3 priced documents x 10 min = 0.5 h at $30/h = $15.00
panel.get_by_label("Minutes per document").fill("10")
panel.get_by_label("Hourly rate").fill("30")
expect(panel).to_contain_text("$15.00")
expect(cost_page.get_by_text(AS_OF)).to_be_visible() # pricing as-of date shown
+140
View File
@@ -0,0 +1,140 @@
"""AI contract cases: confidence bands, review drafts, the approve gate and run progress.
Run and review state is set on the server stub (the page's backend); what the page
shows is asserted through the browser. See rpa-dashboard-ai/templates/contract.md.
"""
import json
import re
import urllib.error
import urllib.request
from playwright.sync_api import expect
# (field, value, word scores, mandatory) -> the band each one must land in
FIELDS = [
("Invoice number", "INV-2207", [1.0], True), # green
("Vendor", "Northwind Traders", [0.99, 0.97], True), # green
("Invoice date", "2026-10-02", [0.93], True), # amber
("Total", "1,284.50", [0.98, 0.71], True), # red: worst word wins
("Due date", "", [1.0], True), # blank mandatory
("PO number", "PO-118", [], False), # unscored
]
EXPECTED = {"green": ["Invoice number", "Vendor"], "amber": ["Invoice date"], "red": ["Total"],
"blank": ["Due date"], "unscored": ["PO number"]}
FLAGGED = 4 # everything but green
def api(server, method, path, body=None):
req = urllib.request.Request(server + "/api" + path, method=method,
data=None if body is None else json.dumps(body).encode(),
headers={"Content-Type": "application/json"})
try:
with urllib.request.urlopen(req) as r:
return r.status, json.loads(r.read())
except urllib.error.HTTPError as e:
return e.code, json.loads(e.read())
def test_server_refuses_approval_while_flagged_fields_are_unverified(stub, server):
"""The gate lives on the server: no request from any page can skip it."""
stub.put_run("job-gate", "gate.pdf", FIELDS)
status, body = api(server, "POST", "/runs/job-gate/approve")
assert status == 409 and str(FLAGGED) in body["detail"]
_, rows = api(server, "GET", "/runs/job-gate/review")
first_flagged = next(r for r in rows["rows"] if r["band"] != "green")
api(server, "PATCH", f"/review/{first_flagged['row_id']}", {"verified": True})
status, body = api(server, "POST", "/runs/job-gate/approve")
assert status == 409 and str(FLAGGED - 1) in body["detail"]
api(server, "POST", "/runs/job-gate/verify-remaining")
status, _ = api(server, "POST", "/runs/job-gate/approve")
assert status == 200
assert api(server, "GET", "/runs/job-gate")[1]["approved"] is True
def open_review(page, job_id):
page.get_by_role("navigation").get_by_role("button", name="Review", exact=True).click()
page.get_by_label("Document", exact=True).select_option(job_id)
def fields_list(page):
return page.get_by_role("list", name="Extracted fields").get_by_role("listitem")
LEGEND = {"green": "Clean", "amber": "Check", "red": "Review", "blank": "Missing", "unscored": "Unscored"}
def test_band_counts_and_legend_match(page, stub):
stub.put_run("job-t", "bands.pdf", FIELDS)
open_review(page, "job-t")
legend = page.get_by_role("list", name="Confidence bands").get_by_role("listitem")
expect(legend).to_have_count(len(LEGEND))
for i, (band, word) in enumerate(LEGEND.items()):
expect(legend.nth(i)).to_have_text(re.compile(rf"^\s*{word}\b.*· {len(EXPECTED[band])}\s*$"))
expect(page.get_by_text(re.compile("print clarity, not correctness", re.I))).to_be_visible()
def test_band_filters_show_the_right_rows(page, stub):
stub.put_run("job-t", "bands.pdf", FIELDS)
open_review(page, "job-t")
expect(fields_list(page)).to_have_count(len(FIELDS))
filters = page.get_by_role("group", name="Filter by band")
for name, band in [("Red", "red"), ("Amber", "amber"), ("Missing", "blank"), ("Unscored", "unscored")]:
filters.get_by_role("button", name=name, exact=True).click()
expect(fields_list(page)).to_have_count(len(EXPECTED[band]))
for field in EXPECTED[band]:
expect(page.get_by_role("textbox", name=field, exact=True)).to_be_visible()
filters.get_by_role("button", name="All", exact=True).click()
expect(fields_list(page)).to_have_count(len(FIELDS))
def row(stub, job_id, field):
return next(r for r in stub.ROWS[job_id] if r["field"] == field)
def test_draft_edit_is_not_saved_until_save(page, stub):
stub.put_run("job-t", "bands.pdf", FIELDS)
open_review(page, "job-t")
box = page.get_by_role("textbox", name="Vendor", exact=True)
box.fill("Northwind Traders Ltd")
expect(page.get_by_text("1 unsaved change", exact=True)).to_be_visible()
assert row(stub, "job-t", "Vendor")["corrected_value"] is None
asked = []
page.once("dialog", lambda d: (asked.append(d.message), d.dismiss()))
page.get_by_role("navigation").get_by_role("button", name="Report", exact=True).click()
expect(box).to_have_value("Northwind Traders Ltd") # leaving was refused, the draft is still there
assert asked and "unsaved" in asked[0].lower()
page.get_by_role("button", name="Save", exact=True).click()
expect(page.get_by_text("1 unsaved change", exact=True)).to_be_hidden()
saved = row(stub, "job-t", "Vendor")
assert saved["corrected_value"] == "Northwind Traders Ltd" and saved["verified"] is True
def test_approve_stays_blocked_while_rows_are_unverified(page, stub):
stub.put_run("job-t", "bands.pdf", FIELDS)
open_review(page, "job-t")
page.get_by_role("button", name="Verify Total", exact=True).click()
expect(page.get_by_role("button", name="Verify Total", exact=True)).to_have_attribute("aria-pressed", "true")
page.get_by_role("button", name="Approve", exact=True).click()
expect(page.get_by_role("alert")).to_contain_text(f"{FLAGGED - 1} flagged")
assert stub.RUNS["job-t"]["approved"] is False
page.get_by_role("button", name="Verify remaining", exact=True).click()
page.get_by_role("button", name="Approve", exact=True).click()
expect(page.get_by_text("✓ Approved", exact=True)).to_be_visible()
expect(page.get_by_role("alert")).to_be_hidden()
assert stub.RUNS["job-t"]["approved"] is True
def test_page_x_of_y_updates_while_running(page, stub):
stub.put_run("job-busy", "busy.pdf", FIELDS, status="running", pages_done=1, pages_total=4)
open_review(page, "job-busy")
expect(page.get_by_text("page 1 of 4", exact=True)).to_be_visible()
stub.RUNS["job-busy"]["pages_done"] = 3
expect(page.get_by_text("page 3 of 4", exact=True)).to_be_visible()
stub.RUNS["job-busy"].update(status="done", pages_done=4)
expect(fields_list(page)).to_have_count(len(FIELDS))
expect(page.get_by_text(re.compile(r"^page \d+ of \d+$"))).to_be_hidden()
+19
View File
@@ -0,0 +1,19 @@
"""Everything from the no-AI page: the no-AI contract cases, run unchanged against the AI page.
They use this folder's `page` and `push` fixtures, so they check the React page.
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "page_check"))
from test_contract import ( # noqa: E402,F401
test_dark_toggle_flips_theme,
test_donut_and_legend_match_batch_states,
test_kpi_counts_match_batch_states,
test_log_lines_show_their_type,
test_no_bot_page_runs_on_sample_data,
test_plain_browser_first_open_shows_sample_data,
test_report_search_filters_rows_and_shows_bucket_pill,
test_step_list_follows_step_label,
)
+123
View File
@@ -0,0 +1,123 @@
"""Seam 1 page check for the AI desktop page: the no-AI dashboard.html with the plain-HTML
AI sections (rpa-dashboard-ai/templates/desktop/ai_sections.html) pasted in before </body>.
Page under test: set PAGE_CHECK_AI_DESKTOP_DIR to a generated dashboard folder (the one with
dashboard.html and server_stub.py). Without it, the no-AI template is filled with the no-AI
check's TEST_FILL and the AI sections are pasted in, the way the skill does it.
The page sees a fake Qt bridge, as in the no-AI check: each slot call lands in window.__calls.
The AI slot, ai_request(id, method, path, body), goes to the review backend's dispatch() in
this process, and the reply comes back as an `ai_response` envelope, exactly what
ai_bridge.py's slot does. So the approve gate under test is the Python one. Tests set run and
review state on the backend directly (the `stub` fixture). Light, dark and narrow for every case.
"""
import importlib.util
import json
import os
import re
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
NO_AI = ROOT / "rpa-dashboard-no-ai" / "templates" / "dashboard.html"
AI = ROOT / "rpa-dashboard-ai" / "templates"
TEST_FILL = {
"APP_NAME": "Test Extractor",
"APP_SUBTITLE": "Page check",
"LOGO_SRC": "",
"API_BASE": "",
"STEPS_JSON": json.dumps(["Initializing", "Login", "Process items", "Completed"]),
"REPORT_COLUMNS_JSON": json.dumps([{"key": "item", "label": "Item"}, {"key": "status", "label": "Status"}]),
}
FAKE_BRIDGE = """
window.__calls = [];
window.qt = {webChannelTransport: {}};
window.QWebChannel = function (_transport, ready) {
const pyBridge = new Proxy({}, {get: (_o, slot) => (...args) => {
window.__calls.push([slot, ...args]);
if (slot === 'ai_request') window.__aiRequest(...args).then(window.__pyMsg);
}});
ready({objects: {pyBridge}});
};
"""
MODES = {"light": (1280, 820, False), "dark": (1280, 820, True), "narrow": (800, 900, False)}
WIRED = os.environ.get("PAGE_CHECK_WIRED") == "1"
@pytest.fixture
def sample_data():
if WIRED:
pytest.skip("page is wired to a bot, so it has no sample data")
@pytest.fixture(scope="session")
def site(tmp_path_factory):
"""The dashboard folder under test: dashboard.html, server_stub.py and bands.py."""
if os.environ.get("PAGE_CHECK_AI_DESKTOP_DIR"):
return Path(os.environ["PAGE_CHECK_AI_DESKTOP_DIR"]).resolve()
html = re.sub(r"\{\{(\w+)\}\}", lambda m: TEST_FILL[m[1]], NO_AI.read_text(encoding="utf-8"))
if WIRED:
html = html.replace("const SAMPLE_DATA = true", "const SAMPLE_DATA = false")
sections = (AI / "desktop" / "ai_sections.html").read_text(encoding="utf-8")
out = tmp_path_factory.mktemp("desktop")
(out / "dashboard.html").write_text(html.replace("</body>", sections + "\n</body>"), encoding="utf-8")
for f in ["server_stub.py", "bands.py", "pricing.py"]:
(out / f).write_bytes((AI / "web" / f).read_bytes())
return out
@pytest.fixture(scope="session")
def page_url(site):
return (site / "dashboard.html").as_uri()
@pytest.fixture(scope="session")
def stub_module(site):
sys.path.insert(0, str(site)) # the stub imports bands.py from its own folder
spec = importlib.util.spec_from_file_location("desktop_server_stub", site / "server_stub.py")
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
@pytest.fixture
def stub(stub_module):
"""The review backend the bridge talks to, reset to its sample data."""
stub_module.reset()
return stub_module
def ai_request(stub, req_id, method, path, body):
"""What ai_bridge.py's ai_request slot does: dispatch, then answer with one envelope."""
status, payload = stub.dispatch(method, path, json.loads(body or "{}"))
return json.dumps({"event": "ai_response", "data": {"id": req_id, "status": status, "body": payload}})
@pytest.fixture(params=list(MODES))
def page(request, browser, page_url, stub):
width, height, dark = MODES[request.param]
ctx = browser.new_context(viewport={"width": width, "height": height})
ctx.expose_function("__aiRequest", lambda i, method, path, body: ai_request(stub, i, method, path, body))
ctx.add_init_script(FAKE_BRIDGE)
pg = ctx.new_page()
pg.set_default_timeout(5000)
errors = []
pg.on("pageerror", lambda e: errors.append(e))
pg.goto(page_url)
assert not errors, f"page script errors: {errors}"
if dark:
push_to(pg, "apply_dark", {"on": True})
yield pg
assert not errors, f"page script errors: {errors}"
ctx.close()
def push_to(pg, event, data):
pg.evaluate("([event, data]) => window.__pyMsg(JSON.stringify({event, data}))", [event, data])
@pytest.fixture
def push(page):
return lambda event, data: push_to(page, event, data)
@@ -0,0 +1,54 @@
"""The AI web page's review cases, run unchanged against the desktop page.
They use this folder's `page` and `stub` fixtures: the page talks to the review backend
over the fake Qt bridge instead of HTTP.
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "page_check_ai"))
from test_ai_review import ( # noqa: E402,F401
test_approve_stays_blocked_while_rows_are_unverified,
test_band_counts_and_legend_match,
test_band_filters_show_the_right_rows,
test_draft_edit_is_not_saved_until_save,
test_page_x_of_y_updates_while_running,
)
from test_ai_review import FIELDS, FLAGGED, open_review # noqa: E402
from test_ai_cost import ( # noqa: E402,F401
cost_page,
test_by_hand_panel_is_labelled_assumptions_and_recomputes,
test_coverage_pill_counts_priced_of_all_documents,
test_month_breakdown_groups_runs_by_month,
test_runs_show_tokens_and_cost_and_unpriced_is_not_zero,
test_spend_tiles_sum_the_priced_runs,
)
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "page_check"))
from test_contract import ( # noqa: E402,F401
test_dark_toggle_flips_theme,
test_donut_and_legend_match_batch_states,
test_kpi_counts_match_batch_states,
test_log_lines_show_their_type,
test_no_bot_page_runs_on_sample_data,
test_plain_browser_first_open_shows_sample_data,
test_report_search_filters_rows_and_shows_bucket_pill,
test_step_list_follows_step_label,
)
TAMPER = """async (path) => {
const reply = new Promise(done => { const base = window.__pyMsg;
window.__pyMsg = m => { base(m); if (JSON.parse(m).event === 'ai_response') done(JSON.parse(m).data); }; });
bridge.ai_request(999, 'POST', path, '{}');
return await reply;
}"""
def test_a_changed_page_cannot_approve_over_the_bridge(page, stub):
"""The gate is Python's: a page that skips its own buttons and calls the slot is still refused."""
stub.put_run("job-t", "bands.pdf", FIELDS)
open_review(page, "job-t")
reply = page.evaluate(TAMPER, "/runs/job-t/approve")
assert reply["id"] == 999 and reply["status"] == 409 and str(FLAGGED) in reply["body"]["detail"]
assert stub.RUNS["job-t"]["approved"] is False
+41
View File
@@ -0,0 +1,41 @@
"""Confidence bands: thresholds, the worst-word rule and the blank-mandatory flag (spec: Banding)."""
import importlib.util
from pathlib import Path
BANDS = Path(__file__).resolve().parents[1] / "rpa-dashboard-ai" / "templates" / "web" / "bands.py"
spec = importlib.util.spec_from_file_location("bands", BANDS)
bands = importlib.util.module_from_spec(spec)
spec.loader.exec_module(bands)
def test_thresholds():
assert bands.band_of(1.0) == "green"
assert bands.band_of(0.95) == "green"
assert bands.band_of(0.9499) == "amber"
assert bands.band_of(0.90) == "amber"
assert bands.band_of(0.8999) == "red"
assert bands.band_of(0.0) == "red"
def test_field_takes_its_worst_word():
assert bands.field_band([0.99, 0.97, 0.98], "ACME Office Supply") == "green"
assert bands.field_band([0.99, 0.93, 0.98], "ACME Office Supply") == "amber"
assert bands.field_band([0.99, 0.97, 0.62], "ACME Office Supply") == "red"
def test_blank_mandatory_field_is_always_flagged():
assert bands.field_band([1.0], "", mandatory=True) == "blank"
assert bands.field_band([], " ", mandatory=True) == "blank"
assert bands.field_band(None, None, mandatory=True) == "blank"
assert bands.needs_review("blank")
# An optional blank is graded like any other field.
assert bands.field_band([1.0], "", mandatory=False) == "green"
def test_no_scores_means_unscored():
assert bands.field_band([], "INV-1001") == "unscored"
def test_only_green_skips_review():
assert not bands.needs_review("green")
assert all(bands.needs_review(b) for b in ["amber", "red", "blank", "unscored"])
+50
View File
@@ -0,0 +1,50 @@
"""Self-check for the AI cost function: rates per million tokens, thinking bills as output,
long-context rates above the threshold, and an unknown model is an error, never zero."""
import importlib.util
from pathlib import Path
import pytest
SRC = Path(__file__).resolve().parents[1] / "rpa-dashboard-ai" / "templates" / "web" / "pricing.py"
spec = importlib.util.spec_from_file_location("pricing", SRC)
pricing = importlib.util.module_from_spec(spec)
spec.loader.exec_module(pricing)
TABLE = {
"flat-model": {"input": 0.30, "output": 2.50},
"long-model": {"input": 1.25, "output": 10.00, "long_threshold": 200_000, "input_long": 2.50, "output_long": 15.00},
}
@pytest.fixture(autouse=True)
def table(monkeypatch):
monkeypatch.setattr(pricing, "PRICING", TABLE)
def test_normal_call_prices_input_and_output_per_million():
# 1M in at 0.30 + 0.2M out at 2.50 = 0.30 + 0.50
assert pricing.cost_for_call("flat-model", 1_000_000, 200_000) == pytest.approx(0.80)
def test_thinking_tokens_bill_at_the_output_rate():
# 0.1M in at 0.30 + (0.1M out + 0.3M thinking) at 2.50 = 0.03 + 1.00
assert pricing.cost_for_call("flat-model", 100_000, 100_000, 300_000) == pytest.approx(1.03)
def test_long_context_rates_apply_above_the_threshold_only():
# at the threshold: 0.2M at 1.25 + 0.1M at 10 = 0.25 + 1.00
assert pricing.cost_for_call("long-model", 200_000, 100_000) == pytest.approx(1.25)
# above it: 0.4M at 2.50 + 0.1M at 15 = 1.00 + 1.50
assert pricing.cost_for_call("long-model", 400_000, 100_000) == pytest.approx(2.50)
def test_unknown_model_raises_and_names_the_known_models():
with pytest.raises(ValueError) as e:
pricing.cost_for_call("mystery-model", 1000, 1000)
assert "mystery-model" in str(e.value)
assert "flat-model" in str(e.value) and "long-model" in str(e.value)
def test_bundled_table_records_its_source_and_as_of_date():
assert pricing.PRICING_SOURCE.startswith("https://")
assert len(pricing.PRICING_AS_OF) == 10 and pricing.PRICING_AS_OF[4] == "-"
+45
View File
@@ -0,0 +1,45 @@
"""The skills are shared outside the source projects: no patient data, internal IPs,
credentials or keys in any bundled file, and the Superbill extracts are marked as examples."""
import re
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
SKILLS = [ROOT / "rpa-dashboard-ai", ROOT / "rpa-dashboard-no-ai"]
AI_REFERENCE = ROOT / "rpa-dashboard-ai" / "templates" / "reference"
FORBIDDEN = {
"internal IP": r"\b(10\.\d{1,3}|192\.168|172\.(1[6-9]|2\d|3[01]))\.\d{1,3}\.\d{1,3}\b",
"Google API key": r"AIza[0-9A-Za-z_-]{30,}",
"OpenAI/Anthropic key": r"\bsk-[A-Za-z0-9_-]{20,}",
"AWS key": r"\bAKIA[0-9A-Z]{16}\b",
"private key": r"-----BEGIN [A-Z ]*PRIVATE KEY-----",
"hard-coded secret": r"""(?i)\b(password|passwd|api_?key|secret|client_secret)\b\s*[:=]\s*['"][^'"{}\s]{6,}['"]""",
"SSN": r"\b\d{3}-\d{2}-\d{4}\b",
"quoted roster name ('SURNAME, FIRST')": r"""["'][A-Z]{2,}, [A-Z]""",
"client name": r"Gulf Coast",
}
MARKER = "EXAMPLE ONLY"
def text_files():
for skill in SKILLS:
for f in skill.rglob("*"):
if f.is_file() and "node_modules" not in f.parts and f.name != "package-lock.json":
yield f
@pytest.mark.parametrize("what", FORBIDDEN)
def test_no_sensitive_data_in_bundled_files(what):
hits = [f"{f.relative_to(ROOT)}:{n}" for f in text_files()
for n, line in enumerate(f.read_text(encoding="utf-8", errors="ignore").splitlines(), 1)
if re.search(FORBIDDEN[what], line)]
assert not hits, f"{what} found: {hits}"
def test_superbill_extracts_are_marked_as_examples():
files = [f for f in AI_REFERENCE.rglob("*") if f.is_file()]
assert files
unmarked = [str(f.relative_to(ROOT)) for f in files
if MARKER not in "".join(f.read_text(encoding="utf-8").splitlines(keepends=True)[:5])]
assert not unmarked, f"no '{MARKER}' banner in the first lines: {unmarked}"
+13 -1
View File
@@ -1,4 +1,5 @@
"""Python templates must stay valid Python before their placeholders are filled."""
import re
import shutil
import subprocess
from pathlib import Path
@@ -10,7 +11,8 @@ HOOK_RULES = "E9,F63,F7,F82,F811" # same rule set as the PostToolUse lint hook
@pytest.mark.parametrize("template", ["rpa-dashboard-no-ai/templates/bridge.py",
"rpa-dashboard-no-ai/templates/bot_runner.py"])
"rpa-dashboard-no-ai/templates/bot_runner.py",
"rpa-dashboard-ai/templates/desktop/ai_bridge.py"])
def test_template_passes_lint_hook_unfilled(template):
ruff = shutil.which("ruff") or pytest.skip("ruff not on PATH")
r = subprocess.run([ruff, "check", "--select", HOOK_RULES, "--no-cache", str(ROOT / template)],
@@ -23,3 +25,13 @@ def test_tebra_sky_preset_is_the_template_default():
themes = (templates / "themes.md").read_text(encoding="utf-8")
sky = themes.split("## 1. Tebra Sky")[1].split("```css\n")[1].split("```")[0]
assert sky in (templates / "dashboard.html").read_text(encoding="utf-8")
def test_ai_skill_references_files_that_exist():
skill = ROOT / "rpa-dashboard-ai"
text = (skill / "SKILL.md").read_text(encoding="utf-8")
paths = re.findall(r"`((?:templates/|<no-ai>)[\w./-]+)`", text)
assert len(paths) > 5
missing = [p for p in paths
if not (skill / p.replace("<no-ai>", "../rpa-dashboard-no-ai/templates/")).exists()]
assert not missing