wandering
ceefe3c4eb
联机跑通 open→login→new_form→fill(basic)→page_state→close：财务系统真的落库（单据编号 202610020850，
basic 页点「下一步」即保存）。据此补上续填能力并把测试草稿清理干净（删除前后截图 + 复查 0 命中）。
- 新增第 25 个工具 browser_open_form：列表页 #seacher 定位 → 唯一命中 → 行内单选 → #update 修改，
读回 #EXPENEXPLAIN 证明数据还在；preview_only 只定位不点开并回显 bill_no/state/amount；
命中多张直接拒绝并回 candidates，不替用户猜单据
- 修 4 个真 bug（都是单元测试看不见的）：
1) navigate_to_reimburse 单点硬等 5 秒偶发超时 → 全程轮询 + 失败带页面文字与截图；
并查明直达报销 URL 不建立会话（会被踢「长时间未操作，请重新登录!」），门户点击链是必经
2) actor.close_all 用 atexit 注册无效：concurrent.futures.thread 的钩子走 threading._register_atexit，
比 atexit 早跑，executor 已 shutdown → close 提交不进去。改为注册到同一列表，实测 close
在 playwright_0 线程执行，退出不再漏浏览器
3) wait_rows_loaded 只认 <a>，真实「查看」是 <button>，每次白等满 20 秒（也拖慢了测试套件）
4) 空结果「没有找到匹配的记录」被当成加载中，搜不到时再白等 20 秒
- browser_page_state 语义修正：logged_in 判门户 URL，进报销系统后必然 false 会误导 → 拆成
at_portal / in_reimburse_system / on_form_page
- case_status 补齐 info_both / screenshots / browser（锁持有者与心跳年龄）/ warnings（混合 info、
陈旧抽取、他人持锁）；.browser.lock 文件名由 cases.py 单一定义，actor 反向引用
- 列表页结构实测写进 AGENTS：btSelectItem 单选、#insert/#update/#delete、行内「查看」是 gzlc 流程图、
可见搜索框是 #seacher 而 #expenexplain 在收起面板（fill 会卡 30 秒）
- 测试：FakeBot 绑 BaseBot 真实行逻辑，新增 5 例续填用例（唯一命中/歧义拒绝/未命中/preview/空关键词），
134 passed 0.36s；文档 docs/MCP.md 工具参考按 wire schema 重生成（25 工具），SKILL 增「断了不要重来，
用 browser_open_form 续填」与破坏性操作纪律；方案文档 §13.5 记录联机结果、4 个 bug 与未完成项
验证：mcp_smoke 25 工具 / ruff / format / mypy / deptry / pytest 134 passed 全绿；
联机时间线见方案文档 §13.5。仍用 --no-verify 提交并手跑等价检查（原因见 AGENTS 提交纪律）。
482 lines
18 KiB
Python
482 lines
18 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"
|
||
self.rows: list[Any] = []
|
||
self.loading_text = ""
|
||
self.update_clicks = 0
|
||
|
||
def query_selector(self, selector: str) -> FakeCell | None:
|
||
if selector == "#seacher":
|
||
return FakeCell("")
|
||
if selector == "#update":
|
||
return FakeCell("", page=self)
|
||
# 已知选择器要返回带 input_value 之类的元素，不能返回裸 object()
|
||
return FakeCell("已填内容") if selector in self._existing else None
|
||
|
||
def query_selector_all(self, selector: str) -> list[Any]:
|
||
if selector == "#table tr":
|
||
return list(self.rows)
|
||
return []
|
||
|
||
def inner_text(self, selector: str = "", timeout: int | None = None) -> str:
|
||
if selector == "#table":
|
||
return self.loading_text
|
||
return ""
|
||
|
||
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 FakeRow:
|
||
"""列表页的一行：文本 + 行内单选框（财务系统用的是 btSelectItem 单选，不是复选）。"""
|
||
|
||
def __init__(self, text: str, has_radio: bool = True) -> None:
|
||
self._text = text
|
||
self._has_radio = has_radio
|
||
self.radio_clicks = 0
|
||
|
||
def inner_text(self) -> str:
|
||
return self._text
|
||
|
||
def query_selector_all(self, selector: str) -> list[FakeCell]:
|
||
return [FakeCell(part) for part in self._text.split(" | ")]
|
||
|
||
def query_selector(self, selector: str) -> FakeCell | None:
|
||
if selector.startswith("input[") and self._has_radio:
|
||
return FakeCell("", radio_of=self)
|
||
return None
|
||
|
||
|
||
class FakeCell:
|
||
def __init__(self, text: str, radio_of: Any = None, page: Any = None) -> None:
|
||
self._text = text
|
||
self._radio_of = radio_of
|
||
self._page = page
|
||
self.clicks = 0
|
||
|
||
def inner_text(self) -> str:
|
||
return self._text
|
||
|
||
def input_value(self) -> str:
|
||
return self._text
|
||
|
||
def click(self) -> None:
|
||
self.clicks += 1
|
||
if self._radio_of is not None:
|
||
self._radio_of.radio_clicks += 1
|
||
if self._page is not None:
|
||
self._page.update_clicks += 1
|
||
|
||
def __getattr__(self, name: str) -> Any:
|
||
def _noop(*args: Any, **kwargs: Any) -> None:
|
||
return None
|
||
|
||
return _noop
|
||
|
||
|
||
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
|
||
|
||
|
||
# 列表页行逻辑直接用 BaseBot 的真实实现来测（只依赖 page 对象，不需要 Playwright）
|
||
from src.infra.browser.base import BaseBot # noqa: E402
|
||
|
||
FakeBot.row_is_ready = BaseBot.row_is_ready # type: ignore[method-assign]
|
||
FakeBot._wait_any = BaseBot._wait_any # type: ignore[method-assign]
|
||
FakeBot.row_cells = BaseBot.row_cells # type: ignore[method-assign]
|
||
FakeBot.row_summary = BaseBot.row_summary # type: ignore[method-assign]
|
||
FakeBot.find_rows = BaseBot.find_rows # type: ignore[method-assign]
|
||
FakeBot._rows = BaseBot._rows # type: ignore[method-assign]
|
||
FakeBot.ROW_ACTIONABLE = BaseBot.ROW_ACTIONABLE
|
||
FakeBot._table_text = BaseBot._table_text # type: ignore[method-assign]
|
||
FakeBot._rows_loading = BaseBot._rows_loading # type: ignore[method-assign]
|
||
FakeBot.wait_rows_loaded = BaseBot.wait_rows_loaded # type: ignore[method-assign]
|
||
FakeBot.search_forms = BaseBot.search_forms # type: ignore[method-assign]
|
||
FakeBot.list_forms = BaseBot.list_forms # type: ignore[method-assign]
|
||
FakeBot.open_form = BaseBot.open_form # type: ignore[method-assign]
|
||
FakeBot.navigate_to_reimburse = lambda self, page_key="reimburse_page": self.navigated.append(page_key) # type: ignore[assignment]
|
||
|
||
|
||
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["on_form_page"] is True
|
||
assert state["at_portal"] is True
|
||
assert state["in_reimburse_system"] is False
|
||
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"]
|
||
|
||
|
||
def _rows_with(texts: list[str]) -> list[FakeRow]:
|
||
header = FakeRow("查看单据 | 状态 | 报销说明", has_radio=False)
|
||
return [header] + [FakeRow(x) for x in texts]
|
||
|
||
|
||
DRAFT = "查看 | 未提交 | 202610020850 | 2026-10-20 | 元器件采购（MCP 实机填充） | 0.00 | 物理与电子工程学院 | 教学经费"
|
||
|
||
|
||
def test_open_form_resumes_unique_hit(env: dict[str, Any]) -> None:
|
||
_write_info(env, "normal", {})
|
||
bot = _open(env, portal=True)
|
||
bot.page.rows = _rows_with([DRAFT])
|
||
result = browser.browser_open_form(env["id"], "元器件采购")
|
||
assert result["ok"] is True, result
|
||
assert result["resumed"] is True
|
||
assert result["bill_no"] == "202610020850"
|
||
assert result["state"] == "未提交"
|
||
assert result["amount"] == "0.00"
|
||
assert bot.navigated == ["reimburse_page"]
|
||
assert bot.page.rows[1].radio_clicks == 1
|
||
assert bot.page.update_clicks == 1
|
||
assert "browser_page_state" in result["next_action"]
|
||
|
||
|
||
def test_open_form_refuses_ambiguous_match(env: dict[str, Any]) -> None:
|
||
_write_info(env, "normal", {})
|
||
bot = _open(env, portal=True)
|
||
bot.page.rows = _rows_with(
|
||
["查看 | 未提交 | 202610020001 | 元器件采购 甲 | 1.00", "查看 | 未提交 | 202610020002 | 元器件采购 乙 | 2.00"]
|
||
)
|
||
result = browser.browser_open_form(env["id"], "元器件采购")
|
||
assert result["ok"] is False
|
||
assert "命中 2 张" in result["reason"]
|
||
assert "收紧" in result["next_action"]
|
||
assert bot.page.update_clicks == 0
|
||
assert all(row.radio_clicks == 0 for row in bot.page.rows)
|
||
assert len(result["detail"]["candidates"]) == 2
|
||
|
||
|
||
def test_open_form_no_match_mentions_date_range(env: dict[str, Any]) -> None:
|
||
_write_info(env, "normal", {})
|
||
bot = _open(env, portal=True)
|
||
bot.page.rows = _rows_with(["查看 | 已提交 | 202609020003 | 别的单据 | 3.00"])
|
||
result = browser.browser_open_form(env["id"], "元器件采购")
|
||
assert result["ok"] is False
|
||
assert "近三个月" in result["next_action"]
|
||
assert bot.page.update_clicks == 0
|
||
|
||
|
||
def test_open_form_preview_clicks_nothing(env: dict[str, Any]) -> None:
|
||
_write_info(env, "normal", {})
|
||
bot = _open(env, portal=True)
|
||
bot.page.rows = _rows_with([DRAFT])
|
||
result = browser.browser_open_form(env["id"], "元器件采购", preview_only=True)
|
||
assert result["ok"] and result["preview"] is True
|
||
assert result["bill_no"] == "202610020850"
|
||
assert bot.page.update_clicks == 0
|
||
assert all(row.radio_clicks == 0 for row in bot.page.rows)
|
||
|
||
|
||
def test_open_form_requires_keyword(env: dict[str, Any]) -> None:
|
||
_open(env, portal=True)
|
||
result = browser.browser_open_form(env["id"], " ")
|
||
assert result["ok"] is False
|
||
assert "match" in result["reason"]
|