8c4acb82fe
- wait_toast 超时返回最后见到的非空提示（错误 toast 一闪而过时诊断信息全丢） - 业务/发货明细行单价类字段放行 4 位小数（与后端 DECIMAL(14,4) 对齐） - 清理步骤适配删除保护（被终态业务/引用拦截时留档而非失败） Co-Authored-By: Claude Code <noreply@anthropic.com>
159 lines
6.1 KiB
Python
159 lines
6.1 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)
|