"""根级 conftest — 全局 fixture：配置加载、日志、中文 HTML 报告生成。 Usage: def test_xxx(admin_config): # 管理端配置 def test_xxx(driver_android_config): # 司机安卓配置 """ from pathlib import Path import sys import json import pytest sys.path.insert(0, str(Path(__file__).parent.parent)) @pytest.fixture(scope="session") def config(): """全局配置（session 级，只加载一次）。""" from tests.config.test_config import load_config return load_config() @pytest.fixture(scope="function") def logger(request): """每个测试函数的 logger。""" from tests.utils.logger import get_logger module_name = request.module.__name__.replace("tests.", "") return get_logger(module_name) # ═══════════════════════════════════════════════════════════════════ # 各端配置 fixture（从 config 中摘取对应 section） # ═══════════════════════════════════════════════════════════════════ @pytest.fixture(scope="session") def admin_config(config): """管理端 (Playwright) 配置。""" return config["admin"] @pytest.fixture(scope="session") def mini_program_config(config): """司机端小程序 (Playwright) 配置。""" return config["mini_program"] @pytest.fixture(scope="session") def driver_android_config(config): """司机安卓端 (Appium) 配置。""" return config["driver_android"] @pytest.fixture(scope="session") def shipper_android_config(config): """货主安卓端 (Appium) 配置。""" return config["shipper_android"] @pytest.fixture(scope="session") def driver_ios_config(config): """司机苹果端 (Appium) 配置。""" return config["driver_ios"] @pytest.fixture(scope="session") def shipper_ios_config(config): """货主苹果端 (Appium) 配置。""" return config["shipper_ios"] # ═══════════════════════════════════════════════════════════════════ # 测试步骤收集 (conftest hook，用于报告中展示步骤明细) # ═══════════════════════════════════════════════════════════════════ _STEP_BUF = [] def _step_context(test_name: str): """获取或创建当前测试的步骤缓冲。""" for entry in _STEP_BUF: if entry["test"] == test_name: return entry entry = {"test": test_name, "steps": []} _STEP_BUF.append(entry) return entry @pytest.hookimpl(hookwrapper=True) def pytest_runtest_makereport(item, call): """拦截测试执行过程中的 log 输出，提取阶段信息。""" outcome = yield rep = outcome.get_result() setattr(item, f"rep_{rep.when}", rep) # ═══════════════════════════════════════════════════════════════════ # 中文 HTML 报告生成 (pytest_sessionfinish hook) # ═══════════════════════════════════════════════════════════════════ def pytest_sessionfinish(session): """测试会话结束后，生成详细的中文 HTML 报告。""" from datetime import datetime from xml.etree import ElementTree as ET report_dir = Path("output/reports") report_dir.mkdir(parents=True, exist_ok=True) junit_path = report_dir / "junit.xml" if not junit_path.exists(): return try: tree = ET.parse(junit_path) root = tree.getroot() except Exception: return # ── 基础统计 ── total = 0 passed = 0 failed = 0 skipped = 0 cases = [] for suite in root: suite_name = suite.attrib.get("name", "") suite_time = suite.attrib.get("time", "0") suite_ts = suite.attrib.get("timestamp", "") err = suite.attrib.get("errors", "0") suite_fail = suite.attrib.get("failures", "0") suite_skip = suite.attrib.get("skipped", "0") suite_tests = suite.attrib.get("tests", "0") for case in suite: total += 1 classname = case.attrib.get("classname", "") func_name = case.attrib.get("name", "") cn_name = _resolve_cn_name(session, func_name) time_elapsed = case.attrib.get("time", "0") # 提取用例元信息（用例ID、优先级、标签、测试覆盖） meta = _extract_case_meta(session, func_name) failure = case.find("failure") error = case.find("error") system_out = case.find("system-out") system_err = case.find("system-err") log_text = (system_out.text or "") if system_out is not None else "" err_text = (system_err.text or "") if system_err is not None else "" # 从日志中提取阶段时间线 phases = _extract_phases(log_text) if failure is not None: failed += 1 msg = failure.attrib.get("message", "")[:500] cases.append((cn_name, "失败", time_elapsed, msg, meta, phases)) elif case.find("skipped") is not None: skipped += 1 skip_msg = (case.find("skipped").attrib.get("message", ""))[:200] cases.append((cn_name, "跳过", time_elapsed, skip_msg, meta, phases)) else: passed += 1 cases.append((cn_name, "通过", time_elapsed, "", meta, phases)) pass_rate = (passed / total * 100) if total > 0 else 0 now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") # ── 收集日志文件 ── log_contents = _collect_logs(report_dir) # ── 收集截图列表 ── screenshots = _collect_screenshots() html = _build_cn_report_html( total, passed, failed, skipped, pass_rate, cases, now, suite_time, suite_ts, log_contents, screenshots ) (report_dir / "report_cn.html").write_text(html, encoding="utf-8-sig") # ═══════════════════════════════════════════════════════════════════ # 辅助函数 # ═══════════════════════════════════════════════════════════════════ def _resolve_cn_name(session, func_name: str) -> str: """从用例 docstring 中提取中文名称。 优先级: 1) "用例名称: xxx" 2) display_name 属性 3) docstring 第一行去 【P0】 前缀 """ for item in session.items: item_func_name = getattr(item, "originalname", None) or item.name.rsplit("::", 1)[-1] if item_func_name == func_name: test_cls = item.cls if hasattr(item, "cls") else None test_func = getattr(test_cls, item_func_name, None) if test_cls else None doc = (test_func.__doc__ or "") if test_func else "" import re m = re.search(r"用例名称[:：]\s*(.+)", doc) if m: return m.group(1).strip() if test_func and hasattr(test_func, "display_name"): return test_func.display_name if doc: first_line = doc.strip().split("\n")[0].strip() if first_line.startswith("【"): idx = first_line.find("】") if idx > 0: first_line = first_line[idx + 1:] if first_line and len(first_line) > 2: return first_line break return func_name def _extract_case_meta(session, func_name: str) -> dict: """从 docstring 中提取用例元信息。""" meta = {} for item in session.items: item_func_name = getattr(item, "originalname", None) or item.name.rsplit("::", 1)[-1] if item_func_name == func_name: test_cls = item.cls if hasattr(item, "cls") else None test_func = getattr(test_cls, item_func_name, None) if test_cls else None doc = (test_func.__doc__ or "") if test_func else "" import re meta["case_id"] = _re_first(r"用例ID[:：]\s*(.+)", doc) or "" meta["case_name"] = _re_first(r"用例名称[:：]\s*(.+)", doc) or "" meta["priority"] = _re_first(r"优先级[:：]\s*(.+)", doc) or "" meta["tags"] = _re_first(r"标签[:：]\s*(.+)", doc) or "" # 取 docstring 中除元信息外的描述段落 desc_lines = [] in_meta = True for line in doc.split("\n"): stripped = line.strip() if not stripped: in_meta = False continue if in_meta and any(stripped.startswith(k) for k in ["用例ID", "用例名称", "优先级", "标签"]): continue if not in_meta: desc_lines.append(stripped) meta["description"] = " ".join(desc_lines) if desc_lines else "" break return meta def _re_first(pattern, text): import re m = re.search(pattern, text) return m.group(1).strip() if m else "" def _extract_phases(log_text: str) -> list: """从日志文本中提取阶段信息和时间点。""" import re phases = [] # 匹配 "Phase 1: ..." 和 "-> xxx完成" 等模式 phase_pattern = re.findall(r"Phase (\d+)[：:]\s*(.+?)(?:\n|$)", log_text) for num, desc in phase_pattern: phases.append({"phase": num, "desc": desc.strip(), "type": "phase"}) # 匹配完成标记 done_pattern = re.findall(r"-> (.+?)(?:\n|$)", log_text) for d in done_pattern: phases.append({"phase": "", "desc": d.strip(), "type": "done"}) return phases def _collect_logs(report_dir: Path) -> list: """收集最近的测试日志内容。""" logs = [] log_dir = Path("output/logs") if log_dir.exists(): log_files = sorted(log_dir.glob("test_run_*.log"), reverse=True)[:3] for lf in log_files: try: content = lf.read_text(encoding="utf-8") # 只取最后 500 行 lines = content.strip().split("\n") if len(lines) > 500: content = "\n".join(lines[-500:]) logs.append({"name": lf.name, "content": content}) except Exception: pass return logs def _collect_screenshots() -> list: """收集测试过程中生成的截图，转为 base64 内嵌到 HTML 中。""" import base64 shots = [] ss_dir = Path("output/screenshots") if ss_dir.exists(): for p in sorted(ss_dir.rglob("*.png"), reverse=True)[:30]: try: data = p.read_bytes() b64 = base64.b64encode(data).decode("ascii") shots.append({ "name": p.name, "path": str(p).replace("\\", "/"), "b64": b64, }) except Exception: pass return shots # ═══════════════════════════════════════════════════════════════════ # HTML 构建 # ═══════════════════════════════════════════════════════════════════ def _build_cn_report_html(total, passed, failed, skipped, rate, cases, now, suite_time, suite_ts, log_contents, screenshots): """构建包含完整细节的中文 HTML 报告。""" # ── 概览卡片 ── summary_html = f"""
{total}
总计
{passed}
通过
{failed}
失败
执行时间: {now} 套件耗时: {float(suite_time):.1f}s 通过率: {rate:.1f}%
""" # ── 用例明细表格 ── rows_html = "" for i, (name, status, elapsed, msg, meta, phases) in enumerate(cases, 1): cls = {"通过": "s-pass", "失败": "s-fail", "跳过": "s-skip"}.get(status, "") case_id = meta.get("case_id", "") priority = meta.get("priority", "") desc = meta.get("description", "") tags = meta.get("tags", "") # 折叠的错误信息 msg_html = "" if msg: msg_html = f"""
错误详情 
{msg}
""" # 阶段明细 phase_html = "" if phases: phase_items = "".join( f'{p.get("phase","")} {p["desc"]}' for p in phases[:20] ) phase_html = f'
{phase_items}
' rows_html += f""" {i}
{name}
{f'
{desc}
' if desc else ''} {status} {case_id} {priority} {tags} {float(elapsed):.1f} {phase_html} {msg_html} """ # ── 日志面板 ── log_html = "" for lf in log_contents: safe_content = lf["content"].replace("&", "&").replace("<", "<").replace(">", ">") log_html += f"""
📄 {lf['name']} 
{safe_content}
""" # ── 截图面板 ── screenshot_html = "" if screenshots: shot_cards = "" for s in screenshots[:20]: shot_cards += f"""
{s['name']}
{s['name']}
""" screenshot_html = f"""
📸 测试截图 ({len(screenshots)} 张) — 点击可放大
{shot_cards}
""" return f""" 自动化测试报告 - {now}

🚀 自动化测试报告

运八网络货运平台 · QaAutomationHub · {now}
{summary_html}
{rows_html}
#用例名称 / 描述状态 用例ID优先级标签耗时(s)阶段详情 / 备注
📋 测试日志
{log_html if log_html else '

暂无日志

'}
📸 测试截图
{screenshot_html if screenshot_html else '

暂无截图

'}
报告自动生成于 {now} · QaAutomationHub 自动化测试框架
"""