"""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 self.tables: list[Any] = [] self.validations: dict[str, str] = {} self.reject_next = False def query_selector(self, selector: str) -> FakeCell | None: if selector == "#seacher": return FakeCell("") if selector == "#update": return FakeCell("", page=self) # 已知选择器要返回带 input_value 之类的元素,不能返回裸 object() if selector in self._existing: return FakeCell( "已填内容", page=self, selector=selector, validation=self.validations.get(selector, ""), ) return None def click(self, selector: str, timeout: int | None = None) -> None: """页面级点击:#cjtj 会把弹窗关掉(除非 reject_next=True 模拟必填被挡)。""" self.calls.append(("click", selector)) if selector == "#cjtj" and not self.reject_next: self._existing.discard("#fjlx") if selector == "#insertAcc": self._existing.add("#fjlx") def query_selector_all(self, selector: str) -> list[Any]: if selector == "#table tr": return list(self.rows) if selector == "table": return list(self.tables) 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, selector: str = "", validation: str = "", ) -> None: self._text = text self._radio_of = radio_of self._page = page self._selector = selector self._validation = validation self.clicks = 0 def inner_text(self) -> str: return self._text def input_value(self) -> str: return self._text def is_visible(self) -> bool: return True def evaluate(self, _script: str) -> str: return self._validation def get_attribute(self, name: str) -> str | None: return None 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 self._page.notify_click(self._selector) def __getattr__(self, name: str) -> Any: def _noop(*args: Any, **kwargs: Any) -> None: return None return _noop class FakeTable: """页面的一张表:inner_text 是整表文本,rows 是若干 FakeRow。""" def __init__(self, header: str, rows: list[str] | None = None) -> None: self._header = header self._rows = [FakeRow(r, has_radio=False) for r in (rows or [])] def inner_text(self) -> str: return self._header def query_selector_all(self, selector: str) -> list[FakeRow]: return list(self._rows) 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", "reimburse_url": "http://x", "reimburse_page": "/expen/common/common?v=4.0", "travel_page": "/expen/travel/travel?v=4.0", } 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.TAB_MARKERS = BaseBot.TAB_MARKERS FakeBot.TAB_LABELS = BaseBot.TAB_LABELS FakeBot.tab_ready = BaseBot.tab_ready # type: ignore[method-assign] FakeBot.open_tab = BaseBot.open_tab # type: ignore[method-assign] FakeBot.click_if_visible = BaseBot.click_if_visible # type: ignore[method-assign] FakeBot.reimburse_host = BaseBot.reimburse_host # type: ignore[method-assign] FakeBot._on_reimburse = BaseBot._on_reimburse # type: ignore[method-assign] FakeBot.on_target_page = BaseBot.on_target_page # type: ignore[method-assign] FakeBot.attachment_dialog_open = BaseBot.attachment_dialog_open # type: ignore[method-assign] FakeBot._validation_hint = BaseBot._validation_hint # type: ignore[method-assign] FakeBot.confirm_attachment = BaseBot.confirm_attachment # type: ignore[method-assign] FakeBot.upload_attachment = BaseBot.upload_attachment # type: ignore[method-assign] FakeBot.attachment_rows = BaseBot.attachment_rows # type: ignore[method-assign] FakeBot.STEP_TABLE_HINTS = BaseBot.STEP_TABLE_HINTS FakeBot._row_amount = staticmethod(BaseBot._row_amount) # type: ignore[assignment] FakeBot.self_check_step = BaseBot.self_check_step # type: ignore[method-assign] FakeBot._read_step_table = BaseBot._read_step_table # type: ignore[method-assign] FakeBot._find_step_table = BaseBot._find_step_table # type: ignore[method-assign] 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["ok"] is True, second assert second["filled"] == {"单据数": 3, "总金额": 464.0} third = browser.browser_fill(env["id"], "payment") assert third["ok"] is True, third 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 "近三个月" 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_resumed_fill_allows_missing_prerequisites_with_warning(env: dict[str, Any]) -> None: """续填时前置步骤在财务系统里已填,内存没记录也必须能继续填,但要留警告。""" _write_info( env, "normal", { "reimbursement_details": {"total_invoices": 2, "total_amount": 10.0}, "payment_methods": [], "attachments": [], }, ) bot = _open(env, portal=True) bot.page.rows = _rows_with([DRAFT]) browser.browser_open_form(env["id"], "元器件采购") result = browser.browser_fill(env["id"], "details") assert result["ok"] is True, result assert result["warnings"], "续填跳步要给警告" assert result["filled"] == {"单据数": 2, "总金额": 10.0} assert result["done"] == ["details"] 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"] def test_tab_ready_accepts_travel_basic_marker(env: dict[str, Any]) -> None: """差旅单的基本信息页没有 #EXPENEXPLAIN,只有 #CAUSE;多候选标记才不白等。""" bot = _open(env, portal=False, human=False) bot.page._existing = {"#CAUSE"} assert bot.tab_ready("basic") is True bot.page._existing = set() assert bot.tab_ready("basic") is False def test_on_target_page_distinguishes_two_lists(env: dict[str, Any]) -> None: """日常报销与差旅是两个独立列表页,停在日常页不能算到了差旅页。""" bot = _open(env, portal=False, human=False) bot.page.url = "http://x/expen/common/common?v=4.0" bot.page._existing = {"#table"} assert bot.on_target_page("reimburse_page") is True assert bot.on_target_page("travel_page") is False bot.page.url = "http://elsewhere/expen/travel/travel?v=4.0" assert bot.on_target_page("travel_page") is False # 不在报销系统主机上 def test_confirm_attachment_raises_with_browser_hint(env: dict[str, Any]) -> None: """必填没填时浏览器只弹校验气泡、不发请求;必须报错而不是假装成功。""" bot = _open(env, portal=True) bot.page._existing = {"#fjlx", "#fpsmxx"} bot.page.validations = {"#fpsmxx": "请填写此字段。"} bot.page.reject_next = True with pytest.raises(RuntimeError) as exc: bot.confirm_attachment(timeout_ms=600) assert "弹窗没有关闭" in str(exc.value) assert "请填写此字段" in str(exc.value) def test_upload_attachment_defaults_description(env: dict[str, Any]) -> None: """附件说明是必填:info 里没给时要兜底成「发票:文件名」,否则提交会被挡。""" bot = _open(env, portal=True) target = env["case"] / "发票1.pdf" target.write_bytes(b"%PDF") bot.page._existing = {"#insertAcc", "#fjlx", "#file", "#submit2"} bot.upload_attachment("发票1.pdf", "invoice") fills = [c for c in bot.page.calls if c[0] == "fill" and c[1][0] == "#fpsmxx"] assert fills and fills[0][1][1] == "发票:发票1" assert bot.attachment_dialog_open() is False def test_upload_attachment_rejects_bad_paths(env: dict[str, Any]) -> None: bot = _open(env, portal=True) with pytest.raises(FileNotFoundError): bot.upload_attachment("不存在的.pdf") outside = Path(env["case"]).parent / "别处.pdf" outside.write_bytes(b"%PDF") with pytest.raises(RuntimeError, match="会话目录"): bot.upload_attachment(str(outside)) def test_attachment_rows_finds_the_attachment_table(env: dict[str, Any]) -> None: """附件清单不是 #table:早先按 #table 数,附件传成功了却报 0 行。""" bot = _open(env, portal=True) bot.page.tables = [ FakeTable("序号 状态 报销说明", ["1 未提交 别的"]), FakeTable( "序号 类型 附件说明 附件备注 附件文件名", ["1 1 发票:a 发票 /imgupload/a.pdf", "2 2 其他 b /imgupload/b.pdf"], ), ] assert bot.attachment_rows() == 2 bot.page.tables = [FakeTable("序号 类型 附件说明 附件文件名")] assert bot.attachment_rows() == 0