VasanthandClaude Opus 5.5
344718007e
MODES, TEST_FILL, FAKE_BRIDGE, the stub loader, the page opener and the no-AI case list now live in one place; sample_data and push move to tests/conftest.py. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
85 lines
3.9 KiB
Python
85 lines
3.9 KiB
Python
"""A bot run as the page sees it: status, step and log envelopes in order, from a real bot
|
|
run (PAGE_CHECK_REPLAY, the JSON lines bot_runner.py's dry run prints) or a built-in one."""
|
|
import json
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from page_check_common import WIRED
|
|
from playwright.sync_api import expect
|
|
from test_contract import colour_of, kpi, open_tab
|
|
|
|
LOG_COLOURS = {"info": "blue", "success": "green", "warn": "amber", "error": "red"}
|
|
DOT_COLOURS = {"ready": "green", "running": "amber", "done": "green", "error": "red"}
|
|
|
|
|
|
def built_in_run(steps):
|
|
run = [{"event": "status", "data": {"state": "running", "msg": "Running…"}}]
|
|
for i, label in enumerate(steps):
|
|
run.append({"event": "step", "data": {"label": label, "pct": round((i + 1) / len(steps) * 100)}})
|
|
run.append({"event": "log", "data": {"msg": f"Finished {label}.", "type": "success"}})
|
|
run.insert(4, {"event": "log", "data": {"msg": "ERROR: portal timed out, retrying", "type": "error"}})
|
|
return run + [{"event": "status", "data": {"state": "done", "msg": "Completed"}}]
|
|
|
|
|
|
def recorded_run():
|
|
path = os.environ.get("PAGE_CHECK_REPLAY")
|
|
return [json.loads(x) for x in Path(path).read_text(encoding="utf-8-sig").splitlines() if x.strip()] if path else None
|
|
|
|
|
|
def expect_status(page, state, msg):
|
|
# The status indicator has no role of its own, so it's found by its documented ids.
|
|
expect(page.locator("#st-text")).to_have_text(msg or state)
|
|
assert colour_of(page.locator("#st-dot"), "backgroundColor") == DOT_COLOURS[state], state
|
|
run = page.get_by_role("button", name="▶ Run", exact=True)
|
|
expect(run).to_be_disabled() if state == "running" else expect(run).to_be_enabled()
|
|
|
|
|
|
def test_bot_run_shows_status_steps_and_log(page, push):
|
|
steps = page.evaluate("STEPS")
|
|
run = recorded_run() or built_in_run(steps)
|
|
open_tab(page, "Run")
|
|
expect_status(page, "ready", "System ready")
|
|
items = page.get_by_role("list", name="Steps").get_by_role("listitem")
|
|
log, last_step = page.get_by_role("log"), -1
|
|
for envelope in run:
|
|
event, d = envelope["event"], envelope["data"]
|
|
push(event, d)
|
|
if event == "status":
|
|
expect_status(page, d["state"], d.get("msg"))
|
|
elif event == "step":
|
|
assert d["label"] in steps, f"step label {d['label']!r} is not one of the page's STEPS {steps}"
|
|
idx = steps.index(d["label"])
|
|
assert idx >= last_step, f"progress went back from {steps[last_step]!r} to {d['label']!r}"
|
|
last_step = idx
|
|
expect(page.get_by_text(f"{d['pct']}%", exact=True)).to_be_visible()
|
|
if d["pct"] >= 100:
|
|
for i in range(len(steps)):
|
|
expect(items.nth(i)).to_have_text(re.compile(r"^\s*✓"))
|
|
else:
|
|
expect(items.nth(idx)).to_have_attribute("aria-current", "step")
|
|
elif event == "log":
|
|
line = log.get_by_text(d["msg"], exact=True).last
|
|
expect(line).to_be_visible()
|
|
assert colour_of(line) == LOG_COLOURS[d["type"]], d["msg"]
|
|
assert any(e["event"] == "step" for e in run), "the run never moved the progress bar"
|
|
|
|
|
|
def test_failed_run_shows_error_status(page, push):
|
|
push("status", {"state": "running", "msg": "Running…"})
|
|
expect_status(page, "running", "Running…")
|
|
push("status", {"state": "error", "msg": "Failed (exit code 1)"})
|
|
expect_status(page, "error", "Failed (exit code 1)")
|
|
|
|
|
|
@pytest.mark.skipif(not WIRED, reason="only for a page wired to a bot (PAGE_CHECK_WIRED=1)")
|
|
def test_wired_page_sends_its_actions_to_the_bot(page):
|
|
calls = lambda: page.evaluate("window.__calls") # noqa: E731
|
|
assert ["load_dashboard"] in calls()
|
|
kpi(page, "Total batches", 0) # no sample batches on a wired page
|
|
page.get_by_role("button", name="▶ Run", exact=True).click()
|
|
assert ["start_automation"] in calls()
|
|
open_tab(page, "Report")
|
|
assert ["load_latest_report"] in calls()
|