wandering
1725244faf
填报从「一把梭」变成分步：Agent 可以开浏览器、登录、建单、按步填、看页面、关，每步有截图， 中途失败不必从头重来。提交不在本步——提交不可逆，留给 P4 的 submit_preview/submit_execute。 - src/infra/browser/actor.py（新）：Playwright 单线程 actor（sync API 的 greenlet 线程亲和要求所有 调用回到创建线程）+ 会话注册表 + cases/<id>/.browser.lock（PID/started_at/heartbeat/url， 心跳 180 秒过期才允许接管；锁文件损坏按无锁处理；atexit 尽力关浏览器）。与 COM actor 分开。 - base.py：login_portal 拆成 submit_login + wait_for_portal（返回布尔不抛异常），新增 portal_reached/ human_required；_screenshot 支持 screenshot_dir 并按 001-步骤名.png 编号、返回路径、失败只告警 - src/mcp/tools/browser.py（新）：browser_open/login/new_form/fill/steps/page_state/close，工具 17 → 24 · need_human 用 ok:true + logged_in:false 返回（扫码不是失败，避免宿主当错误重试或另开窗口） · 步骤顺序硬约束，跳步拒绝并用 next_action 指回该补的一步；步骤报错回传 detail.screenshot · browser_close 明确不提交不存草稿；travel 五步、normal 四步（无 subsidy） - tests/test_mcp_browser.py（新，17 例，FakeBot/FakePage 不起浏览器）：锁四种情形、need_human 分支与 截图落点、重复登录幂等、未打开时各工具拒绝、类型推断与两 info 并存时拒猜、未登录拒建单、跳步拦截、 步骤推进与 remaining、未知步骤、报错截图、close 释放会话与锁、open 复用、steps 两种流程 - 文档：docs/MCP.md 工具段按 wire schema 重生成（24 工具）；AGENTS/SKILL/操作指南/README/浏览器 README 同步分步流程、锁与心跳、有头默认、提交必须交人工；方案文档 §13 记录实现、取舍与未完成项 验证： - 联机跑通 open（3.1s 起有头 Chromium）→ 锁写入（pid/heartbeat）→ page_state（about:blank， 001-state.png 4721 字节）→ 重复 open 复用（already_open=true，不多开窗）→ close（锁释放、注册表清空） - 登录与填报的联机验证需要用户在场（SSO 可能扫码），本轮未做 - 门禁：ruff / format --check / mypy src / deptry / pytest 129 passed / mcp_smoke 24 工具 instructions 2109 字节。仍用 --no-verify 提交并手跑等价检查（原因见 AGENTS 提交纪律）。
328 lines
12 KiB
Python
328 lines
12 KiB
Python
"""MCP 分步浏览器工具的守卫逻辑测试（不启动真实浏览器）
|
|
|
|
真实 Chromium 交互没法在测试里跑，这里覆盖的是本次新写的那部分：need_human 分支、
|
|
单据类型推断、跳步拦截、步骤推进与报错截图、锁的占用与接管、会话与锁的释放。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from src.infra.browser import actor
|
|
from src.mcp.tools import browser
|
|
|
|
|
|
class FakePage:
|
|
"""记录调用；query_selector 只在预置的选择器上返回真值。"""
|
|
|
|
def __init__(self, existing: set[str] | None = None) -> None:
|
|
self._existing = existing or set()
|
|
self.calls: list[Any] = []
|
|
self.url = "http://sso.test/login"
|
|
|
|
def query_selector(self, selector: str) -> object | None:
|
|
return object() if selector in self._existing else None
|
|
|
|
def title(self) -> str:
|
|
return "统一认证"
|
|
|
|
def screenshot(self, target: str) -> None:
|
|
Path(target).write_bytes(b"PNG")
|
|
|
|
def wait_for_timeout(self, _ms: int) -> None:
|
|
return None
|
|
|
|
def __getattr__(self, name: str) -> Any:
|
|
def _record(*args: Any, **kwargs: Any) -> None:
|
|
self.calls.append((name, args))
|
|
|
|
return _record
|
|
|
|
|
|
class FakeBot:
|
|
"""只实现工具真正依赖的接口，行为由构造参数决定。"""
|
|
|
|
def __init__(self, case_dir: Path, *, portal: bool = False, human: bool = True) -> None:
|
|
self.work_dir = case_dir
|
|
self.screenshot_dir = case_dir / "screenshots"
|
|
self._shot_seq = 0
|
|
self.headless = False
|
|
self.config = {
|
|
"default_name": "示例报销人",
|
|
"default_card_no": "6282880000000000",
|
|
"default_person_id": "200000001",
|
|
"sso_login_url": "http://sso.test/login",
|
|
}
|
|
self.page = FakePage({'text="微信扫码登录"'} if human else set())
|
|
self._portal = portal
|
|
self._human = human
|
|
self.closed = False
|
|
self.navigated: list[str] = []
|
|
self.created = 0
|
|
|
|
def _mark_logged_in(self) -> None:
|
|
self.page.url = "https://tyrz.fynu.edu.cn/oshall"
|
|
self.page._existing = {"#EXPENEXPLAIN", "#insertDetail", "#insertPay", "#insertAcc", "#submit"}
|
|
|
|
def portal_reached(self) -> bool:
|
|
if self._portal:
|
|
self._mark_logged_in()
|
|
return self._portal
|
|
|
|
def human_required(self) -> bool:
|
|
return self._human and not self._portal
|
|
|
|
def submit_login(self) -> None:
|
|
self._portal = self._portal or not self._human
|
|
|
|
def wait_for_portal(self, timeout: int = 30) -> bool:
|
|
if self._portal:
|
|
self._mark_logged_in()
|
|
return self._portal
|
|
|
|
def _wait_for(self, selector: str, timeout: int | None = None) -> None:
|
|
self.page.wait_for_selector(selector, timeout=timeout)
|
|
|
|
def navigate_to_reimburse(self, page_key: str = "reimburse_page") -> None:
|
|
self.navigated.append(page_key)
|
|
|
|
def create_new_form(self) -> None:
|
|
self.created += 1
|
|
|
|
def _screenshot(self, name: str) -> str:
|
|
self._shot_seq += 1
|
|
target_dir = self.screenshot_dir
|
|
assert target_dir is not None
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
target = target_dir / (str(self._shot_seq).zfill(3) + "-" + name + ".png")
|
|
self.page.screenshot(str(target))
|
|
return str(target)
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
class FakeStore:
|
|
"""让工具以为 case_id 指向 tmp 目录。"""
|
|
|
|
def __init__(self, root_dir: Path) -> None:
|
|
self.root = root_dir
|
|
|
|
def resolve(self, case_id: str) -> Path | None:
|
|
target = self.root / case_id
|
|
return target if target.exists() else None
|
|
|
|
|
|
@pytest.fixture()
|
|
def env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Any:
|
|
case = tmp_path / "20261019-测试会话"
|
|
(case / ".invoice_cache").mkdir(parents=True)
|
|
monkeypatch.setattr(browser, "_store", FakeStore(tmp_path)) # type: ignore[arg-type]
|
|
yield {"case": case, "id": case.name}
|
|
actor.close_all()
|
|
for leftover in tmp_path.glob("*/.browser.lock"):
|
|
leftover.unlink(missing_ok=True)
|
|
|
|
|
|
def _open(env: dict[str, Any], **kwargs: Any) -> FakeBot:
|
|
bot = FakeBot(env["case"], **kwargs)
|
|
actor.put_session(env["id"], bot)
|
|
return bot
|
|
|
|
|
|
def _write_info(env: dict[str, Any], kind: str, payload: dict[str, Any]) -> None:
|
|
name = browser.INFO_FILE[kind]
|
|
path = env["case"] / ".invoice_cache" / name
|
|
path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
|
|
|
|
|
|
def test_lock_acquires_when_free(tmp_path: Path) -> None:
|
|
got, info = actor.acquire(tmp_path)
|
|
assert got is True
|
|
assert info["pid"] == os.getpid()
|
|
assert actor.lock_path(tmp_path).exists()
|
|
actor.release(tmp_path)
|
|
assert not actor.lock_path(tmp_path).exists()
|
|
|
|
|
|
def test_lock_blocks_other_live_process(tmp_path: Path) -> None:
|
|
now = time.time()
|
|
actor.lock_path(tmp_path).write_text(
|
|
json.dumps({"pid": 999999, "heartbeat": now, "started_at": now}), encoding="utf-8"
|
|
)
|
|
got, info = actor.acquire(tmp_path)
|
|
assert got is False
|
|
assert info.get("pid") == 999999
|
|
|
|
|
|
def test_lock_takes_over_when_stale(tmp_path: Path) -> None:
|
|
dead = time.time() - actor.STALE_AFTER_SECONDS - 5
|
|
actor.lock_path(tmp_path).write_text(
|
|
json.dumps({"pid": 999999, "heartbeat": dead, "started_at": dead}), encoding="utf-8"
|
|
)
|
|
got, info = actor.acquire(tmp_path)
|
|
assert got is True
|
|
assert info["pid"] == os.getpid()
|
|
|
|
|
|
def test_broken_lock_file_is_ignored(tmp_path: Path) -> None:
|
|
actor.lock_path(tmp_path).write_text("{坏文件", encoding="utf-8")
|
|
got, _info = actor.acquire(tmp_path)
|
|
assert got is True
|
|
|
|
|
|
def test_login_requires_human(env: dict[str, Any]) -> None:
|
|
_open(env)
|
|
result = browser.browser_login(env["id"])
|
|
assert result["ok"] is True
|
|
assert result["need_human"] is True
|
|
assert "browser_login" in result["next_action"]
|
|
shots = list((env["case"] / "screenshots").glob("*need_human*.png"))
|
|
assert shots, "need_human 的截图要落在会话目录里"
|
|
|
|
|
|
def test_login_idempotent_without_captcha(env: dict[str, Any]) -> None:
|
|
bot = _open(env, human=False)
|
|
first = browser.browser_login(env["id"])
|
|
assert first["ok"] and first["logged_in"] is True and first["already"] is False
|
|
again = browser.browser_login(env["id"])
|
|
assert again["logged_in"] is True and again["already"] is True
|
|
assert bot.created == 0
|
|
|
|
|
|
def test_login_and_state_refuse_before_open(env: dict[str, Any]) -> None:
|
|
assert browser.browser_login(env["id"])["ok"] is False
|
|
state = browser.browser_page_state(env["id"])
|
|
assert state["ok"] is False
|
|
assert "browser_open" in state["next_action"]
|
|
filled = browser.browser_fill(env["id"], "basic")
|
|
assert filled["ok"] is False
|
|
closed = browser.browser_close(env["id"])
|
|
assert closed["ok"] and closed["closed"] is False
|
|
|
|
|
|
def test_new_form_infers_kind(env: dict[str, Any]) -> None:
|
|
_write_info(env, "normal", {"basic_info": {"reimbursement_description": "元件采购"}})
|
|
bot = _open(env, portal=True)
|
|
result = browser.browser_new_form(env["id"])
|
|
assert result["ok"] is True
|
|
assert result["kind"] == "normal"
|
|
assert result["steps"] == ["basic", "details", "payment", "attachments"]
|
|
assert bot.navigated == ["reimburse_page"]
|
|
assert bot.created == 1
|
|
|
|
|
|
def test_new_form_ambiguous_when_both_infos(env: dict[str, Any]) -> None:
|
|
_write_info(env, "normal", {})
|
|
_write_info(env, "travel", {})
|
|
_open(env, portal=True)
|
|
result = browser.browser_new_form(env["id"])
|
|
assert result["ok"] is False
|
|
assert "kind" in result["next_action"]
|
|
|
|
|
|
def test_new_form_requires_login(env: dict[str, Any]) -> None:
|
|
_write_info(env, "travel", {})
|
|
_open(env, portal=False, human=False)
|
|
result = browser.browser_new_form(env["id"])
|
|
assert result["ok"] is False
|
|
assert "browser_login" in result["next_action"]
|
|
|
|
|
|
def test_fill_blocks_missing_step(env: dict[str, Any]) -> None:
|
|
_write_info(env, "normal", {"payment_methods": [{}, {}]})
|
|
_open(env, portal=True)
|
|
browser.browser_new_form(env["id"])
|
|
result = browser.browser_fill(env["id"], "payment")
|
|
assert result["ok"] is False
|
|
assert "basic" in result["next_action"]
|
|
assert "details" in result["reason"]
|
|
|
|
|
|
def test_fill_walks_steps_and_reports_remaining(env: dict[str, Any]) -> None:
|
|
_write_info(
|
|
env,
|
|
"normal",
|
|
{
|
|
"basic_info": {"reimbursement_description": "元件采购"},
|
|
"reimbursement_details": {"total_invoices": 3, "total_amount": 464.0},
|
|
"payment_methods": [{}, {}],
|
|
"attachments": [],
|
|
},
|
|
)
|
|
_open(env, portal=True)
|
|
browser.browser_new_form(env["id"])
|
|
first = browser.browser_fill(env["id"], "basic")
|
|
assert first["ok"] is True
|
|
assert first["done"] == ["basic"]
|
|
assert first["remaining"] == ["details", "payment", "attachments"]
|
|
second = browser.browser_fill(env["id"], "details")
|
|
assert second["filled"] == {"单据数": 3, "总金额": 464.0}
|
|
third = browser.browser_fill(env["id"], "payment")
|
|
assert third["filled"] == {"支付笔数": 2}
|
|
assert third["remaining"] == ["attachments"]
|
|
state = browser.browser_page_state(env["id"])
|
|
assert state["ok"] is True
|
|
assert state["steps_done"] == ["basic", "details", "payment"]
|
|
assert state["markers"]["支付页"] is True
|
|
assert state["recent_screenshots"]
|
|
|
|
|
|
def test_fill_rejects_unknown_step(env: dict[str, Any]) -> None:
|
|
_write_info(env, "normal", {})
|
|
_open(env, portal=True)
|
|
browser.browser_new_form(env["id"])
|
|
result = browser.browser_fill(env["id"], "subsidy")
|
|
assert result["ok"] is False
|
|
assert result["detail"]["steps"] == ["basic", "details", "payment", "attachments"]
|
|
|
|
|
|
def test_fill_surfaces_step_error_with_screenshot(env: dict[str, Any], monkeypatch: pytest.MonkeyPatch) -> None:
|
|
_write_info(env, "normal", {"basic_info": {"reimbursement_description": "元件采购"}})
|
|
_open(env, portal=True)
|
|
browser.browser_new_form(env["id"])
|
|
|
|
def boom(*_args: Any, **_kwargs: Any) -> None:
|
|
raise RuntimeError("元素找不到")
|
|
|
|
monkeypatch.setattr("src.infra.browser.normal.fill_basic_info", boom)
|
|
result = browser.browser_fill(env["id"], "basic")
|
|
assert result["ok"] is False
|
|
assert "元素找不到" in result["reason"]
|
|
shot = result["detail"]["screenshot"]
|
|
assert Path(shot).exists()
|
|
assert Path(shot).name.endswith("basic_error.png")
|
|
|
|
|
|
def test_close_releases_session_and_lock(env: dict[str, Any]) -> None:
|
|
bot = _open(env, portal=True)
|
|
actor.lock_path(env["case"]).write_text(
|
|
json.dumps({"pid": os.getpid(), "heartbeat": time.time()}), encoding="utf-8"
|
|
)
|
|
result = browser.browser_close(env["id"])
|
|
assert result["ok"] and result["closed"] is True
|
|
assert bot.closed is True
|
|
assert actor.session(env["id"]) is None
|
|
assert not actor.lock_path(env["case"]).exists()
|
|
|
|
|
|
def test_open_reuses_existing_browser(env: dict[str, Any]) -> None:
|
|
_open(env, portal=True)
|
|
result = browser.browser_open(env["id"])
|
|
assert result["ok"] and result["already_open"] is True
|
|
|
|
|
|
def test_steps_lists_both_flows_then_one(env: dict[str, Any]) -> None:
|
|
result = browser.browser_steps(env["id"])
|
|
assert result["ok"] and set(result["kinds"]) == {"travel", "normal"}
|
|
_write_info(env, "travel", {})
|
|
picked = browser.browser_steps(env["id"])
|
|
assert picked["kind"] == "travel"
|
|
assert [s["step"] for s in picked["steps"]] == ["basic", "details", "payment", "subsidy", "attachments"]
|