Files
workphone-sdk/sdk/scripts/console_redesign_visual_qa.py
2026-08-09 14:04:58 +08:00

515 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""总控平台页面视觉与性能验收工具。
默认只读:只访问 BASE_URL 的 GET 页面和接口,不点击业务按钮、不调用手机、不执行写接口。
如需登录,使用 CONSOLE_QA_USERNAME/CONSOLE_QA_PASSWORD或 USERNAME/PASSWORD环境变量。
"""
from __future__ import annotations
import argparse
import asyncio
import json
import os
import re
import shutil
import sys
import time
from collections import Counter
from dataclasses import asdict, dataclass
from datetime import datetime
from pathlib import Path
from typing import Any
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
try:
from PIL import Image, ImageStat
except ImportError as exc: # pragma: no cover - 运行环境缺依赖时给出明确提示
Image = None # type: ignore[assignment]
ImageStat = None # type: ignore[assignment]
_PIL_IMPORT_ERROR = exc
else:
_PIL_IMPORT_ERROR = None
try:
from playwright.async_api import Browser, BrowserContext, Page, async_playwright
except ImportError as exc: # pragma: no cover - 运行环境缺依赖时给出明确提示
Browser = BrowserContext = Page = Any # type: ignore[misc,assignment]
async_playwright = None # type: ignore[assignment]
_PLAYWRIGHT_IMPORT_ERROR = exc
else:
_PLAYWRIGHT_IMPORT_ERROR = None
ROOT = Path(__file__).resolve().parents[2]
DEFAULT_OUTPUT_DIR = ROOT / "开发文档/10、项目管理/02-测试报告/20260809_WP_CONSOLE_13_VISUAL_QA"
VIEWPORTS = {
1440: 1200,
1024: 1100,
768: 1024,
375: 900,
}
WRITE_METHODS = {"POST", "PUT", "PATCH", "DELETE"}
AUTH_MARKERS = ("登录总控平台", "登录并加载平台", "账号", "密码", "退出登录")
BUSINESS_MARKERS = ("数据总览", "手机设备", "智能引擎", "统一接入中心")
@dataclass(frozen=True)
class ScreenshotSpec:
id: str
title: str
tab: str
width: int
query: tuple[tuple[str, str], ...] = ()
required_tokens: tuple[str, ...] = ()
SCREENSHOTS = (
ScreenshotSpec("SS-01", "数据总览 1440px", "overview", 1440, required_tokens=("数据总览",)),
ScreenshotSpec("SS-02", "数据总览接口异常/数据过期", "overview", 1440, (("qa_state", "expired"),), ("数据总览", "数据已过期")),
ScreenshotSpec("SS-03", "手机设备含下载与扫码按钮", "devices", 1024, required_tokens=("手机设备",)),
ScreenshotSpec("SS-04", "添加设备 APK 下载步骤", "devices", 1024, (("qa_step", "apk"),), ("手机设备",)),
ScreenshotSpec("SS-05", "添加设备原扫码绑定步骤", "devices", 1024, (("qa_step", "scan"),), ("手机设备",)),
ScreenshotSpec("SS-06", "设备卡片三种状态", "devices", 1024, (("qa_state", "statuses"),), ("手机设备",)),
ScreenshotSpec("SS-07", "单机控制台概览", "devices", 1024, (("view", "device"),), ("手机设备",)),
ScreenshotSpec("SS-08", "设备分组管理", "devices", 1024, (("view", "groups"),), ("手机设备",)),
ScreenshotSpec("SS-09", "在线设备禁止删除", "devices", 1024, (("qa_state", "delete-protected"),), ("手机设备",)),
ScreenshotSpec("SS-10", "智能引擎四模块", "engine", 1024, required_tokens=("智能引擎",)),
ScreenshotSpec("SS-11", "统一接入中心接口目录", "integrations", 1024, (("view", "api"),), ("统一接入中心",)),
ScreenshotSpec("SS-12", "知识库文档可打开", "integrations", 1024, (("view", "knowledge"),), ("统一接入中心",)),
ScreenshotSpec("SS-13", "375px 手机设备页", "devices", 375, required_tokens=("手机设备",)),
ScreenshotSpec("SS-14", "真机写操作四证汇总", "devices", 1440, (("view", "evidence"),), ("手机设备",)),
ScreenshotSpec("SS-15", "第三方应用与凭证列表", "integrations", 1024, (("view", "apps"),), ("统一接入中心",)),
ScreenshotSpec("SS-16", "接口设备文档授权详情", "integrations", 1024, (("view", "grants"),), ("统一接入中心",)),
ScreenshotSpec("SS-17", "第三方用量与额度看板", "integrations", 1024, (("view", "usage"),), ("统一接入中心",)),
ScreenshotSpec("SS-18", "首页设备与微信运行摘要", "overview", 1440, (("view", "wechat-summary"),), ("数据总览",)),
)
def env_first(*names: str) -> str | None:
for name in names:
value = os.environ.get(name)
if value:
return value
return None
def build_url(base_url: str, tab: str, query: tuple[tuple[str, str], ...]) -> str:
parts = urlsplit(base_url)
existing = dict(parse_qsl(parts.query, keep_blank_values=True))
existing["tab"] = tab
existing.update(query)
return urlunsplit((parts.scheme, parts.netloc, parts.path or "/", urlencode(existing), parts.fragment))
def redact_url(url: str) -> str:
parts = urlsplit(url)
return urlunsplit((parts.scheme, parts.netloc, parts.path, "", ""))
def timestamp() -> str:
return datetime.now().astimezone().strftime("%Y%m%dT%H%M%S%z")
def safe_filename(value: str) -> str:
return re.sub(r"[^A-Za-z0-9_.-]+", "-", value)
def find_browser(executable_path: str | None) -> str | None:
if executable_path:
return executable_path
candidates = [
"/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
"/Applications/Chromium.app/Contents/MacOS/Chromium",
]
for candidate in candidates:
if Path(candidate).is_file() and os.access(candidate, os.X_OK):
return candidate
for command in ("google-chrome", "google-chrome-stable", "chromium", "chromium-browser"):
found = shutil.which(command)
if found and Path(found).is_file() and os.access(found, os.X_OK):
# Homebrew/旧环境可能留下只负责转发到已删除 App 的壳脚本。
try:
if Path(found).read_text(encoding="utf-8", errors="ignore").startswith("#!") and "exec '" in Path(found).read_text(encoding="utf-8", errors="ignore"):
continue
except OSError:
pass
return found
return None
def is_auth_request(url: str, login_path: str) -> bool:
return urlsplit(url).path.rstrip("/") == login_path.rstrip("/")
def image_guard(path: Path) -> tuple[bool, str]:
"""挡住全黑、全白或几乎没有内容的截图。"""
if Image is None:
return False, f"Pillow 不可用: {_PIL_IMPORT_ERROR}"
try:
with Image.open(path) as image:
rgb = image.convert("RGB")
stat = ImageStat.Stat(rgb.resize((64, 64)))
mean = sum(stat.mean) / 3
variance = sum(stat.var) / 3
if mean < 3:
return False, "black_page: 截图几乎全黑"
if mean > 252 and variance < 3:
return False, "blank_page: 截图几乎全白"
if variance < 1 and (mean < 8 or mean > 247):
return False, "blank_page: 截图近乎纯色"
return True, ""
except (OSError, ValueError) as exc:
return False, f"screenshot_unreadable: {exc}"
async def navigation_timing(page: Page, wall_ms: float) -> dict[str, Any]:
timing = await page.evaluate(
"""() => {
const navigation = performance.getEntriesByType('navigation')[0];
const paints = Object.fromEntries(
performance.getEntriesByType('paint').map(entry => [entry.name, entry.startTime])
);
return {
dom_content_loaded_ms: navigation?.domContentLoadedEventEnd || null,
load_event_ms: navigation?.loadEventEnd || null,
response_start_ms: navigation?.responseStart || null,
first_paint_ms: paints['first-paint'] || null,
first_contentful_paint_ms: paints['first-contentful-paint'] || null,
};
}"""
)
candidates = [
timing.get("first_contentful_paint_ms"),
timing.get("first_paint_ms"),
timing.get("dom_content_loaded_ms"),
timing.get("response_start_ms"),
wall_ms,
]
timing["first_screen_ms"] = round(float(next(value for value in candidates if value is not None)), 2)
return timing
async def page_text(page: Page) -> str:
try:
return (await page.locator("body").inner_text(timeout=1500)).strip()
except Exception:
return ""
async def validate_business_page(page: Page, spec: ScreenshotSpec) -> tuple[bool, str, str]:
text = await page_text(page)
lowered = text.lower()
if any(marker in text for marker in AUTH_MARKERS) and (await page.locator("input[type='password']").count() > 0):
return False, "login_page: 当前仍是登录页,未算业务页", text
if not text or len(text) < 80:
return False, "blank_page: 页面可见文本不足,未算业务页", text
if not any(marker in text for marker in BUSINESS_MARKERS):
return False, "not_business_page: 未找到总控平台业务页标识", text
missing = [token for token in spec.required_tokens if token not in text]
if missing:
return False, f"business_token_missing: 缺少页面标识 {', '.join(missing)}", text
if "error" in lowered and "登录服务暂时不可用" in text:
return False, "page_error: 页面显示登录服务异常", text
return True, "", text
async def maybe_login(page: Page, username: str | None, password: str | None, login_path: str) -> dict[str, Any]:
"""仅在显式提供账号密码时登录;不点任何业务操作。"""
result: dict[str, Any] = {"attempted": False, "succeeded": False, "reason": ""}
if not username or not password:
result["reason"] = "未提供登录环境变量"
return result
result["attempted"] = True
text = await page_text(page)
if "登录总控平台" not in text and await page.locator("input[type='password']").count() == 0:
result["succeeded"] = True
result["reason"] = "初始页面已是业务页"
return result
try:
user_input = page.locator("input[name='username'], input[autocomplete='username'], input[type='text']").first
password_input = page.locator("input[name='password'], input[autocomplete='current-password'], input[type='password']").first
await user_input.fill(username)
await password_input.fill(password)
await page.locator("button[type='submit'], button:has-text('登录')").first.click()
await page.wait_for_timeout(150)
result["succeeded"] = (await page.locator("input[type='password']").count()) == 0
result["reason"] = "登录完成" if result["succeeded"] else "登录后仍停留在登录页"
except Exception as exc:
result["reason"] = f"登录失败: {type(exc).__name__}: {exc}"
return result
async def capture_spec(
context: BrowserContext,
browser: Browser,
base_url: str,
spec: ScreenshotSpec,
output_dir: Path,
run_stamp: str,
timeout_ms: int,
failure_state: dict[str, Any],
) -> dict[str, Any]:
page = await context.new_page()
target_url = build_url(base_url, spec.tab, spec.query)
screenshot_dir = output_dir / "screenshots"
error_dir = output_dir / "errors"
screenshot_dir.mkdir(parents=True, exist_ok=True)
error_dir.mkdir(parents=True, exist_ok=True)
base_name = f"{spec.id}_{spec.width}x{VIEWPORTS[spec.width]}_{run_stamp}"
screenshot_path = screenshot_dir / f"{safe_filename(base_name)}.png"
error_path = error_dir / f"{safe_filename(base_name)}_error.png"
network_failures: list[dict[str, Any]] = []
response_failures: list[dict[str, Any]] = []
def record_failed_request(request: Any) -> None:
if request.resource_type in {"xhr", "fetch"} or "/api/" in request.url:
network_failures.append({"url": redact_url(request.url), "error": request.failure})
def record_response(response: Any) -> None:
if response.request.resource_type in {"xhr", "fetch"} or "/api/" in response.url:
if response.status >= 400:
response_failures.append({"url": redact_url(response.url), "status": response.status})
page.on("requestfailed", record_failed_request)
page.on("response", record_response)
started = time.perf_counter()
navigation_error = ""
try:
await page.goto(target_url, wait_until="domcontentloaded", timeout=timeout_ms)
await page.wait_for_timeout(120)
try:
await page.evaluate("document.fonts && document.fonts.ready")
except Exception:
pass
except Exception as exc:
navigation_error = f"navigation_error: {type(exc).__name__}: {exc}"
wall_ms = round((time.perf_counter() - started) * 1000, 2)
timing: dict[str, Any] = {"first_screen_ms": wall_ms}
if not navigation_error:
try:
timing = await navigation_timing(page, wall_ms)
except Exception as exc:
timing["timing_error"] = f"{type(exc).__name__}: {exc}"
ok, reason, text = (False, navigation_error, "") if navigation_error else await validate_business_page(page, spec)
try:
await page.screenshot(path=str(error_path if not ok else screenshot_path), full_page=True, timeout=timeout_ms)
except Exception as exc:
reason = f"{reason}; screenshot_error: {type(exc).__name__}: {exc}" if reason else f"screenshot_error: {type(exc).__name__}: {exc}"
image_path = error_path if not ok else screenshot_path
if image_path.exists():
image_ok, image_reason = image_guard(image_path)
if not image_ok:
ok = False
reason = f"{reason}; {image_reason}" if reason else image_reason
if image_path == screenshot_path:
try:
await page.screenshot(path=str(error_path), full_page=True, timeout=timeout_ms)
except Exception:
pass
interface_failures = network_failures + response_failures
if failure_state.get("readonly_violations"):
ok = False
reason = f"readonly_violation: 发现非认证写请求 {len(failure_state['readonly_violations'])}"
result = {
"id": spec.id,
"title": spec.title,
"url": redact_url(target_url),
"viewport": {"width": spec.width, "height": VIEWPORTS[spec.width]},
"status": "passed" if ok else "failed",
"reason": reason or "业务页标识、内容和截图检查通过",
"screenshot_path": str(screenshot_path.resolve()) if ok and screenshot_path.exists() else None,
"error_screenshot_path": str(error_path.resolve()) if (not ok and error_path.exists()) else None,
"screenshot_filename_contains_id_size_time": spec.id in image_path.name and str(spec.width) in image_path.name and run_stamp in image_path.name,
"visible_text_length": len(text),
"first_screen_ms": timing.get("first_screen_ms"),
"timing": timing,
"interface_failure_count": len(interface_failures),
"interface_failures": interface_failures,
"readonly_violation_count": len(failure_state.get("readonly_violations", [])),
}
await page.close()
return result
async def run_qa(args: argparse.Namespace) -> dict[str, Any]:
if _PLAYWRIGHT_IMPORT_ERROR:
raise RuntimeError(f"Playwright 不可用: {_PLAYWRIGHT_IMPORT_ERROR}")
output_dir = Path(args.output_dir).expanduser().resolve()
output_dir.mkdir(parents=True, exist_ok=True)
run_stamp = timestamp()
base_url = args.base_url
username = args.username
password = args.password
token = args.auth_token
login_path = args.login_path
failure_state: dict[str, Any] = {"readonly_violations": [], "auth_requests": []}
async with async_playwright() as playwright:
executable_path = find_browser(args.executable_path)
browser = await playwright.chromium.launch(headless=not args.headed, executable_path=executable_path)
contexts: dict[int, BrowserContext] = {}
try:
for width, height in VIEWPORTS.items():
headers = {"Authorization": f"Bearer {token}"} if token else None
context = await browser.new_context(viewport={"width": width, "height": height}, extra_http_headers=headers)
async def guard_route(route: Any, request: Any) -> None:
if request.method in WRITE_METHODS:
if is_auth_request(request.url, login_path):
failure_state["auth_requests"].append({"method": request.method, "url": redact_url(request.url)})
await route.continue_()
return
failure_state["readonly_violations"].append({"method": request.method, "url": redact_url(request.url)})
await route.abort(error_code="blockedbyclient")
return
await route.continue_()
await context.route("**/*", guard_route)
contexts[width] = context
login_page = await context.new_page()
try:
await login_page.goto(base_url, wait_until="domcontentloaded", timeout=args.timeout_ms)
await login_page.wait_for_timeout(80)
await maybe_login(login_page, username, password, login_path)
finally:
await login_page.close()
results: list[dict[str, Any]] = []
for spec in SCREENSHOTS:
result = await capture_spec(
contexts[spec.width],
browser,
base_url,
spec,
output_dir,
run_stamp,
args.timeout_ms,
failure_state,
)
results.append(result)
finally:
for context in contexts.values():
await context.close()
await browser.close()
performance_by_viewport: dict[str, dict[str, Any]] = {}
for width in VIEWPORTS:
rows = [row for row in results if row["viewport"]["width"] == width]
timings = [row["first_screen_ms"] for row in rows if isinstance(row.get("first_screen_ms"), (int, float))]
performance_by_viewport[str(width)] = {
"viewport": {"width": width, "height": VIEWPORTS[width]},
"sample_count": len(rows),
"first_screen_ms": round(sum(timings) / len(timings), 2) if timings else None,
"first_screen_ms_max": max(timings) if timings else None,
"interface_failure_count": sum(row["interface_failure_count"] for row in rows),
}
counts = Counter(row["status"] for row in results)
report = {
"task_id": "WP-CONSOLE-13",
"task_name": "阿端·页面验收工具·一次跑完四种尺寸、性能和18张截图",
"run_mode": "local_read_only",
"generated_at": datetime.now().astimezone().isoformat(),
"base_url": redact_url(base_url),
"read_only": True,
"write_methods_blocked": sorted(WRITE_METHODS),
"auth_request_count": len(failure_state["auth_requests"]),
"readonly_violation_count": len(failure_state["readonly_violations"]),
"viewport_widths": list(VIEWPORTS),
"screenshot_spec_count": len(SCREENSHOTS),
"screenshot_success_count": counts["passed"],
"screenshot_failure_count": counts["failed"],
"business_page_rejection_count": sum(1 for row in results if row["status"] == "failed" and any(key in row["reason"] for key in ("blank_page", "black_page", "login_page", "not_business_page"))),
"interface_failure_count": sum(row["interface_failure_count"] for row in results),
"performance": performance_by_viewport,
"results": results,
}
(output_dir / "visual_qa_report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
performance_report = {
"task_id": report["task_id"],
"generated_at": report["generated_at"],
"first_screen_time_definition": "优先 First Contentful Paint其次 First Paint、DOMContentLoaded、responseStart 或导航墙钟时间",
"interface_failure_count": report["interface_failure_count"],
"by_viewport": performance_by_viewport,
}
(output_dir / "performance_report.json").write_text(json.dumps(performance_report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
manifest = {
"task_id": report["task_id"],
"generated_at": report["generated_at"],
"screenshots": [
{
"id": row["id"],
"title": row["title"],
"viewport": row["viewport"],
"status": row["status"],
"screenshot_path": row["screenshot_path"],
"error_screenshot_path": row["error_screenshot_path"],
"reason": row["reason"],
}
for row in results
],
}
(output_dir / "screenshot_manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
md_lines = [
"# WP-CONSOLE-13 页面视觉与性能验收报告",
"",
"- 运行模式:本地只读;写方法已拦截,未调用手机,未执行业务写接口。",
f"- 页面入口:`{report['base_url']}`",
f"- 四档尺寸:{', '.join(f'{w}px' for w in VIEWPORTS)};截图规格:{report['screenshot_spec_count']} 张。",
f"- 截图结果:成功 {report['screenshot_success_count']},失败 {report['screenshot_failure_count']};接口失败数:{report['interface_failure_count']}",
"",
"## 性能",
"",
"| 宽度 | 样本数 | 平均首屏(ms) | 最大首屏(ms) | 接口失败数 |",
"|---:|---:|---:|---:|---:|",
]
for width, item in performance_by_viewport.items():
md_lines.append(f"| {width} | {item['sample_count']} | {item['first_screen_ms']} | {item['first_screen_ms_max']} | {item['interface_failure_count']} |")
md_lines.extend(["", "## 截图清单", "", "| 编号 | 尺寸 | 结果 | 成功截图 | 失败截图 | 原因 |", "|---|---:|---|---|---|---|"])
for row in results:
md_lines.append(f"| {row['id']} | {row['viewport']['width']}x{row['viewport']['height']} | {row['status']} | {row['screenshot_path'] or ''} | {row['error_screenshot_path'] or ''} | {row['reason']} |")
(output_dir / "visual_qa_report.md").write_text("\n".join(md_lines) + "\n", encoding="utf-8")
return report
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description="工作手机总控平台四档页面视觉与性能验收")
parser.add_argument("--base-url", default=env_first("BASE_URL", "CONSOLE_QA_BASE_URL") or "http://127.0.0.1:8899/")
parser.add_argument("--output-dir", type=Path, default=Path(env_first("CONSOLE_QA_OUTPUT_DIR") or DEFAULT_OUTPUT_DIR))
parser.add_argument("--username", default=env_first("CONSOLE_QA_USERNAME", "CONSOLE_USERNAME", "USERNAME"))
parser.add_argument("--password", default=env_first("CONSOLE_QA_PASSWORD", "CONSOLE_PASSWORD", "PASSWORD"))
parser.add_argument("--auth-token", default=env_first("CONSOLE_QA_AUTH_TOKEN", "CONSOLE_AUTH_TOKEN", "AUTH_TOKEN"))
parser.add_argument("--login-path", default=env_first("CONSOLE_QA_LOGIN_PATH", "LOGIN_PATH") or "/api/v3/console/login")
parser.add_argument("--timeout-ms", type=int, default=int(env_first("CONSOLE_QA_TIMEOUT_MS") or "15000"))
parser.add_argument("--executable-path", default=env_first("CONSOLE_QA_BROWSER") or None)
parser.add_argument("--headed", action="store_true", help="显示浏览器窗口;默认无界面")
parser.add_argument("--strict", action="store_true", help="存在截图失败、写请求或接口失败时返回非零")
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
try:
report = asyncio.run(run_qa(args))
except Exception as exc:
print(f"视觉验收工具执行失败:{type(exc).__name__}: {exc}", file=sys.stderr)
return 3
summary = {
"task_id": report["task_id"],
"run_mode": report["run_mode"],
"viewport_widths": report["viewport_widths"],
"screenshot_spec_count": report["screenshot_spec_count"],
"screenshot_success_count": report["screenshot_success_count"],
"screenshot_failure_count": report["screenshot_failure_count"],
"interface_failure_count": report["interface_failure_count"],
"readonly_violation_count": report["readonly_violation_count"],
"output_dir": str(Path(args.output_dir).expanduser().resolve()),
}
print(json.dumps(summary, ensure_ascii=False, indent=2))
if args.strict and (report["screenshot_failure_count"] or report["interface_failure_count"] or report["readonly_violation_count"]):
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())