wandering
6aec1bc837
两条真实草稿走完全流程并逐张截图核对：普通报销 4 步（明细 3010109/3张/464、支付 231 工行公务卡、 附件入库，单据 202610020884）；差旅报销 5 步（明细 交通费/2张/233 火车 阜阳→合肥、补助 3×80+3×100=540、 附件入库，单据 202610030245 总额 773.00）。验完两条都按守卫删除并复查 0 命中。 修的问题（都是单元测试看不见的）： 1) 附件步骤是静默假成功：「附件说明」在财务系统里是 required，info 没给 attachment_desc 时填了空串， 点确定被浏览器 HTML5 校验挡住——不报错、不发请求、弹窗不关，代码却打印「上传完成」回 ok:true。 新增 base.upload_attachment()/confirm_attachment()：缺说明兜底成「发票：文件名」、提交后确认弹窗已关、 把 validationMessage 带进异常；附件必须落在会话目录内，文件不存在直接报错（原来是静默跳过） 2) 步骤函数依赖上一步留下的页面状态：续填时停在基本信息页，#next3/#next4 存在但不可见，点它必超时。 所有步骤改成显式 open_tab(...)，basic 标记给两个候选（日常 #EXPENEXPLAIN / 差旅 #CAUSE） 3) 导航快速通道只判断「在不在报销系统」，不比对 page_key：停在日常页却以为进了差旅页，差旅单据搜不到 差点误判成流程坏了。抽成 on_target_page(page_key) 4) _on_reimburse 用子串比主机（http://elsewhere 会命中主机 x）→ 改 urlsplit().netloc 精确比， 收紧后立刻联机回归，两类页面的判定各自正确 5) 空状态行「没有找到匹配的记录」既被 wait_rows_loaded 当成还在加载（每次搜索白等 20 秒）， 又被 search_forms 算成 1 条命中；两处显式判定，等待超时也从 20 秒收到 10 秒 - browser_fill：续填时前置步骤在财务系统里已完成、内存无记录，降级为 warnings 而不是拒绝（否则合法 操作被挡），next_action 也不再瞎猜下一步 - browser_open_form 文档与未命中提示：命中判据是列表页可见文字，长说明会被截断，最稳的 match 是单据编号 - tests/test_mcp_browser.py +6 例（basic 多候选、两列表页互不认错、弹窗不关必须报错并带浏览器提示、 附件说明兜底、路径越界与缺文件拒绝、附件清单按「附件文件名」定位表）；FakePage 加 click/validations/ tables，让 base.py 真实逻辑不依赖 Playwright 也能测。148 passed / 1.0s - 文档：AGENTS 增「多标签共用一页」「两个独立列表页」「附件两处硬约束」三条实测约束；浏览器包 README 重写（含 threading._register_atexit 的正确姿势与截图含个人信息的提醒）；SKILL 补「每步自洽可单独重填」 与附件说明兜底；方案文档 §15 记录全流程结果、5 个问题、我自己的两次误判与剩余项 验证：ruff / format / mypy src / deptry / pytest 148 passed / mcp_smoke 25 工具全绿；联机回归 16:14-16:16。 删除与提交类动作仍不进工具层。--no-verify 提交并手跑等价检查。
642 lines
24 KiB
Python
642 lines
24 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
|
||
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.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 "近三个月" 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
|