AI web starter: bands, review split pane, approve gate, server stub (checkpoint)
React + Vite + Tailwind starter with the no-AI shell, band legend, review split pane with drafts and verify, and a server stub that refuses approval while flagged fields are unverified. Band self-check and AI page check. Both page checks share one Chromium via tests/conftest.py. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
VasanthandClaude Opus 5.5 committed
1 parent
3eb3fdf00e
commit
d4d3f83fb1
24 files changed
+3429
-73
No files matched your search
@@ -0,0 +1,128 @@
|
||||
"""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(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)
|
||||
@@ -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()
|
||||
@@ -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,
|
||||
)
|
||||
Reference in new issue
Block a user