94c49138bf
open_menu 时分组参数化、fill_filter :text-is 精确匹配、messagebox 带 :visible（base_data 实现为准）——两脚本此前各持一份且匹配精度漂移 （has_text 子串 vs :text-is），使用者误以为两者等价；登录/系统管理/ 基础数据/业务数据四链路 e2e 复跑全过。 Co-Authored-By: Claude Code <noreply@anthropic.com>
222 lines
8.6 KiB
Python
222 lines
8.6 KiB
Python
"""端到端测试公共工具。
|
||
|
||
抽取自 `login_flow.py`，供各流程脚本复用：环境常量、Redis 读取（取验证码
|
||
答案，仅自动化测试可用）、完整登录动作、失败处理与控制台错误收集。
|
||
|
||
约定：断言失败抛 `E2EFailure`（消息格式 `步骤：原因`），由各脚本的 `main()`
|
||
统一截图并返回退出码 1。
|
||
"""
|
||
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from urllib.parse import urlparse
|
||
|
||
from playwright.sync_api import Page
|
||
|
||
BASE_URL = "http://127.0.0.1:5173"
|
||
REDIS_CONTAINER = "ys-redis-dev"
|
||
|
||
# Windows 终端默认 cp936，✓/✗ 等符号会抛 UnicodeEncodeError 而非打印失败原因。
|
||
# 与后端 core/logging.py 的处理一致，显式重配标准输出编码。
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
|
||
|
||
class E2EFailure(Exception):
|
||
"""步骤断言失败。"""
|
||
|
||
|
||
def redis_password() -> str:
|
||
"""从 docker/.env 读取 Redis 密码。
|
||
|
||
测试脚本无法用应用配置（那需要后端 venv），故直接读环境文件。
|
||
"""
|
||
if value := os.environ.get("REDIS_PASSWORD"):
|
||
return value
|
||
env_file = os.path.join(os.path.dirname(__file__), "..", "..", "docker", ".env")
|
||
with open(env_file, encoding="utf-8") as handle:
|
||
for line in handle:
|
||
if line.startswith("REDIS_PASSWORD="):
|
||
return line.split("=", 1)[1].strip()
|
||
raise RuntimeError("未能从 docker/.env 读取 REDIS_PASSWORD")
|
||
|
||
|
||
def redis_get(key: str) -> str | None:
|
||
"""读取 Redis 键值。仅用于自动化测试 —— 生产当然不能这样取验证码答案。"""
|
||
result = subprocess.run(
|
||
["docker", "exec", REDIS_CONTAINER, "redis-cli", "-a", redis_password(),
|
||
"--no-auth-warning", "get", key],
|
||
capture_output=True, text=True, encoding="utf-8", check=False,
|
||
)
|
||
return result.stdout.strip() or None
|
||
|
||
|
||
def fail(step: str, detail: str) -> None:
|
||
"""断言失败：抛异常而非进程退出 —— 由 main() 统一截图后再退出。"""
|
||
raise E2EFailure(f"{step}：{detail}")
|
||
|
||
|
||
def current_path(page: Page) -> str:
|
||
"""当前 URL 的路径部分。
|
||
|
||
比较路径**不能**用 `**/dashboard` 这类 glob —— 它会把
|
||
`/login?redirect=/dashboard` 也匹配上，"登录失败"会被误判为成功。
|
||
"""
|
||
return urlparse(page.url).path
|
||
|
||
|
||
def collect_console_errors(page: Page, sink: list[str] | None = None) -> list[str]:
|
||
"""注册控制台错误收集（失败诊断时打印最近几条）。
|
||
|
||
:param sink: 可选的外部列表（如脚本的模块级变量，供 main() 的失败路径读取）
|
||
"""
|
||
errors = sink if sink is not None else []
|
||
page.on("console", lambda msg: errors.append(msg.text) if msg.type == "error" else None)
|
||
page.on("pageerror", lambda exc: errors.append(str(exc)))
|
||
return errors
|
||
|
||
|
||
def toast_text(page: Page) -> str:
|
||
"""当前显示的全局提示文本（el-message），无提示时为空串。"""
|
||
locator = page.locator(".el-message")
|
||
return locator.first.text_content().strip() if locator.count() else ""
|
||
|
||
|
||
def wait_toast(page: Page, expected: str, timeout_ms: int = 5000) -> str:
|
||
"""等待包含 expected 的全局提示出现，返回其文本（超时返回最后见到的非空提示）。
|
||
|
||
直接读 `toast_text` 会在「上一条提示尚未消失、新提示尚未出现」的窗口里
|
||
读到旧文本，产生误判；轮询等待把这类时序问题挡在断言之外。
|
||
返回值语义：匹配成功返回该条；超时返回**见过的最后一条非空提示** ——
|
||
toast 只有 3 秒生命周期，等到超时才回来断言的场景（如「请求已 400、
|
||
错误 toast 一闪而过」）若返回最后一次的空串，诊断信息会全部丢失。
|
||
"""
|
||
deadline = time.monotonic() + timeout_ms / 1000
|
||
last_seen = ""
|
||
while time.monotonic() < deadline:
|
||
text = toast_text(page)
|
||
if text:
|
||
last_seen = text
|
||
if expected in text:
|
||
return text
|
||
page.wait_for_timeout(200)
|
||
return last_seen
|
||
|
||
|
||
def expect_toast(page: Page, expected: str, step: str, timeout_ms: int = 5000) -> None:
|
||
"""断言出现了包含指定文本的全局提示。"""
|
||
text = wait_toast(page, expected, timeout_ms)
|
||
if expected not in text:
|
||
fail(step, f"未出现提示「{expected}」，当前提示：{text or '（无）'}")
|
||
|
||
|
||
def login(page: Page, username: str, password: str) -> None:
|
||
"""完整登录：打开登录页 → 从 Redis 取验证码答案 → 提交 → 等待进入首页。
|
||
|
||
验证码 id 经响应监听捕获（前端把它内联在响应体里，不是 URL 参数）。
|
||
"""
|
||
captcha_ids: list[str] = []
|
||
|
||
def on_response(response) -> None:
|
||
if "/api/v1/auth/captcha" not in response.url or response.status != 200:
|
||
return
|
||
try:
|
||
captcha_ids.append(response.json()["data"]["captcha_id"])
|
||
except Exception:
|
||
# 页面导航后响应体已不可读（会抛协议错误），属可忽略的噪声
|
||
pass
|
||
|
||
page.on("response", on_response)
|
||
try:
|
||
page.goto(f"{BASE_URL}/login", wait_until="networkidle")
|
||
for _ in range(50):
|
||
if captcha_ids:
|
||
break
|
||
page.wait_for_timeout(100)
|
||
if not captcha_ids:
|
||
fail("登录", "未捕获到验证码请求")
|
||
|
||
captcha_id = captcha_ids[-1]
|
||
answer = redis_get(f"ys:captcha:{captcha_id}")
|
||
if not answer:
|
||
fail("登录", f"Redis 中未找到验证码 ys:captcha:{captcha_id}")
|
||
|
||
page.fill('input[autocomplete="username"]', username)
|
||
page.fill('input[autocomplete="current-password"]', password)
|
||
page.fill('input[autocomplete="off"]', answer)
|
||
page.click("button:has-text('登 录')")
|
||
page.wait_for_timeout(2000)
|
||
if current_path(page) != "/dashboard":
|
||
detail = f"未跳转到首页，当前 {page.url}"
|
||
if message := toast_text(page):
|
||
detail += f"，提示：{message}"
|
||
fail("登录", detail)
|
||
page.wait_for_load_state("networkidle")
|
||
finally:
|
||
page.remove_listener("response", on_response)
|
||
|
||
|
||
# ==================== 列表页 UI 操作（base_data/system_management 共用） ====================
|
||
|
||
|
||
def open_menu(page: Page, title: str, group: str) -> None:
|
||
"""点击侧边菜单项，必要时先展开指定分组。
|
||
|
||
分组（el-sub-menu）已展开时再次点击会收拢，因此先判断菜单项是否可见。
|
||
"""
|
||
menu = page.locator(".app-menu")
|
||
item = menu.locator(".el-menu-item", has_text=title).first
|
||
if not item.is_visible():
|
||
menu.locator(".el-sub-menu__title", has_text=group).click()
|
||
page.wait_for_timeout(400)
|
||
item.click()
|
||
page.wait_for_timeout(800)
|
||
|
||
|
||
def fill_filter(page: Page, label: str, value: str) -> None:
|
||
"""在**当前可见**的筛选表单中按标签填写输入框。
|
||
|
||
标签用 `:text-is` 精确匹配 —— 子串匹配会让「操作」命中「操作人」；
|
||
选择器带 `:visible` —— 同页存在多个表单实例（v-show 隐藏未销毁）时
|
||
会命中隐藏元素。
|
||
"""
|
||
form_item = page.locator(
|
||
f'.filter-form:visible .el-form-item:has(.el-form-item__label:text-is("{label}"))'
|
||
).first
|
||
form_item.locator("input").first.fill(value)
|
||
|
||
|
||
def select_filter_option(page: Page, label: str, option: str) -> None:
|
||
"""展开可见筛选表单中指定标签的下拉并选择选项。"""
|
||
item = page.locator(
|
||
f'.filter-form:visible .el-form-item:has(.el-form-item__label:text-is("{label}"))'
|
||
)
|
||
item.locator(".el-select").click()
|
||
page.wait_for_timeout(300)
|
||
page.locator(".el-select-dropdown:visible .el-select-dropdown__item", has_text=option).first.click()
|
||
page.wait_for_timeout(200)
|
||
|
||
|
||
def click_query(page: Page) -> None:
|
||
"""点击可见筛选表单中的查询按钮。"""
|
||
page.locator(".filter-form:visible button:has-text('查询')").first.click()
|
||
page.wait_for_timeout(800)
|
||
|
||
|
||
def table_rows(page: Page):
|
||
"""当前可见的表格行（选择器带 :visible，避开隐藏实例）。"""
|
||
return page.locator(".el-table__body .el-table__row:visible")
|
||
|
||
|
||
def visible_dialog(page: Page):
|
||
"""当前可见的对话框（同页可能存在多个 dialog 实例，按可见性取最后一个）。"""
|
||
return page.locator(".el-dialog:visible").last
|
||
|
||
|
||
def confirm_message_box(page: Page, button_text: str) -> None:
|
||
"""点击确认框（ElMessageBox）中的指定按钮。"""
|
||
page.locator(".el-message-box:visible").locator(f"button:has-text('{button_text}')").click()
|
||
page.wait_for_timeout(800)
|