No-AI desktop path: PyQt bot wired to the page
- templates/bot_runner.py (new, no Qt): runs the bot as a child process (unbuffered, UTF-8), tails its output, keeps raw lines in bot_raw.log, and turns lines into step/log/status events. Step markers and friendly rules use specific substrings, first match wins; unmatched error lines are always shown, other unmatched lines hidden. Dry run: python dashboard/bot_runner.py prints the events the page would get. - bridge.py: delegates to bot_runner, one run at a time, restores the saved dark mode on every page load (fixes sample-data mode in PyQt), adds load_latest_report / download_report slots the page calls. - SKILL.md: skip the stack question for a PyQt project; desktop wiring steps from the bot's real output; verify steps for the host. - tests: assert-based self-check for the line logic and run loop; Seam 1 run replay (status dot, Run button, steps, log) and wired-page mode; fixture b-pyqt-bot and its Seam 2 eval (all 14 checks pass). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
3eb3fdf00e
commit
162edd41df
17 files changed
+753
-47
No files matched your search
@@ -0,0 +1,84 @@
|
||||
"""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 conftest 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()
|
||||
Reference in new issue
Block a user