PayrollManagementSystem/frontend/e2e/common.py

"""端到端测试公共工具。

抽取自 `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)