xst
03e152baab
目录结构: tests/web/admin/ 管理端 (Playwright) tests/web/mini_program/ 司机端小程序 (Playwright/微信 WebView) tests/app/driver_android/ 司机安卓 (Appium) tests/app/shipper_android/ 货主安卓 (Appium) tests/app/driver_ios/ 司机苹果 (Appium) tests/app/shipper_ios/ 货主苹果 (Appium) 公共模块: tests/app/_shared.py driver+ensure_logged_in 工厂 tests/app/_base_page.py App Page Object 基类 配置: test_config.yaml 拆分为 6 个独立 section + 货主账号
127 lines
4.6 KiB
Python
127 lines
4.6 KiB
Python
"""管理端 fixture — Playwright 浏览器实例 + 登录态复用。
|
|
|
|
核心机制:
|
|
skip_login: true → 加载 storage_state，跳过验证码
|
|
skip_login: false → 执行完整 UI 登录
|
|
|
|
初始化:
|
|
python tests/web/admin/save_auth.py
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from playwright.sync_api import sync_playwright, Page
|
|
|
|
SCREENSHOTS_DIR = Path("output/screenshots")
|
|
|
|
|
|
@pytest.fixture(params=["chromium", "firefox"])
|
|
def browser_name(admin_config, request):
|
|
"""每个浏览器执行一轮。"""
|
|
browsers = admin_config.get("browsers", ["chromium", "firefox"])
|
|
request.param = browsers[len(getattr(request, "_index", 0))]
|
|
return request.param
|
|
|
|
|
|
@pytest.fixture
|
|
def page(admin_config, browser_name, request) -> Page:
|
|
"""Playwright Page，加载登录态，失败自动截图。"""
|
|
viewport = admin_config.get("viewport", {"width": 1920, "height": 1080})
|
|
timeout = admin_config.get("timeout_seconds", 120) * 1000
|
|
skip_login = admin_config.get("skip_login", True)
|
|
storage_state_dir = Path(admin_config.get("storage_state_dir", "tests/web/auth"))
|
|
|
|
with sync_playwright() as p:
|
|
browser_launcher = getattr(p, browser_name)
|
|
browser = browser_launcher.launch(headless=True)
|
|
|
|
context_kwargs = {
|
|
"viewport": {"width": viewport["width"], "height": viewport["height"]},
|
|
"locale": "zh-CN",
|
|
}
|
|
|
|
loaded_role = None
|
|
if skip_login:
|
|
loaded_role = _load_auth_state(context_kwargs, storage_state_dir, browser_name)
|
|
|
|
context = browser.new_context(**context_kwargs)
|
|
page = context.new_page()
|
|
page.set_default_timeout(timeout)
|
|
page._auth_role = loaded_role
|
|
|
|
yield page
|
|
|
|
if hasattr(request.node, "rep_call") and request.node.rep_call.failed:
|
|
scenario_dir = SCREENSHOTS_DIR / request.node.name
|
|
scenario_dir.mkdir(parents=True, exist_ok=True)
|
|
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
path = str(scenario_dir / f"FAIL_{browser_name}_{ts}.png")
|
|
page.screenshot(path=path, full_page=True)
|
|
|
|
context.close()
|
|
browser.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def ensure_login(admin_config, page, config):
|
|
"""确保管理端已登录。"""
|
|
|
|
def _ensure_login(role: str = "super_admin") -> None:
|
|
account = config["accounts"].get(role)
|
|
if not account:
|
|
raise ValueError(f"未知角色 '{role}'，可用: {list(config['accounts'].keys())}")
|
|
|
|
base_url = admin_config["base_url"]
|
|
skip_login = admin_config.get("skip_login", True)
|
|
|
|
if skip_login:
|
|
auth_role = getattr(page, "_auth_role", None)
|
|
if auth_role == role:
|
|
page.goto(base_url)
|
|
page.wait_for_timeout(2000)
|
|
if "#/login" not in page.url and "/login" not in page.url:
|
|
return
|
|
raise RuntimeError(
|
|
f"登录态已过期！请运行: python tests/web/admin/save_auth.py --role {role}"
|
|
)
|
|
if auth_role and auth_role != role:
|
|
raise RuntimeError(
|
|
f"当前 storage_state 角色为 '{auth_role}'，请求 '{role}'。\n"
|
|
f"请运行: python tests/web/admin/save_auth.py --role {role}"
|
|
)
|
|
raise RuntimeError(
|
|
f"未找到角色 '{role}' 的登录态文件！请运行:\n"
|
|
f" python tests/web/admin/save_auth.py --role {role}"
|
|
)
|
|
|
|
# skip_login: false
|
|
page.goto(f"{base_url}/#/login")
|
|
page.get_by_placeholder("请输入用户名").fill(account["phone"])
|
|
page.get_by_placeholder("请输入密码").fill(account["password"])
|
|
page.get_by_role("button", name="登录").click()
|
|
page.wait_for_timeout(3000)
|
|
if "#/login" in page.url or "/login" in page.url:
|
|
raise RuntimeError(
|
|
"登录失败！可能存在图形验证码。请改为手动登录: python tests/web/admin/save_auth.py"
|
|
)
|
|
|
|
return _ensure_login
|
|
|
|
|
|
def _load_auth_state(context_kwargs: dict, storage_dir: Path, browser: str) -> str | None:
|
|
auth_files = list(storage_dir.glob(f"*_{browser}.json")) if storage_dir.exists() else []
|
|
if not auth_files:
|
|
return None
|
|
auth_file = auth_files[0]
|
|
context_kwargs["storage_state"] = str(auth_file)
|
|
return auth_file.stem.replace(f"_{browser}", "")
|
|
|
|
|
|
@pytest.hookimpl(tryfirst=True, hookwrapper=True)
|
|
def pytest_runtest_makereport(item, call):
|
|
outcome = yield
|
|
rep = outcome.get_result()
|
|
setattr(item, f"rep_{rep.when}", rep)
|