# -*- coding: utf-8 -*- """ 认证系统属性测试全量运行脚本（100 次迭代） 背景： Spec: 03-miniapp-auth-system（小程序用户认证系统） 任务 11: 属性测试全量运行 前面各任务中的属性测试仅用 5 次迭代快速验证逻辑正确性， 本脚本集中对所有 15 个属性测试执行 100 次迭代，确保健壮性。 系统概述： NeoZQYY 台球门店全栈数据平台的小程序认证系统，涵盖： - 微信登录（code2Session → openid → JWT） - 用户申请审核（site_code 映射、人员匹配、角色分配） - RBAC 权限控制（多店铺隔离、权限中间件） - JWT 令牌管理（受限令牌、店铺切换、过期拒绝） 数据库： 测试库 test_zqyy_app（通过 APP_DB_DSN 环境变量连接） 禁止连接正式库 zqyy_app 运行方式： cd C:\\NeoZQYY python scripts/ops/_run_auth_pbt_full.py [--concurrency N] [--only P1,P2,...] [--skip P3,P5] 设计要求： 1. 后台逐个运行每个属性测试（subprocess），前台定时监控 2. 控制数据库并发：同一时刻只运行 1 个测试（默认），避免占满连接 3. 每完成一个测试输出进度 4. 全部完成后生成详细 MD 格式报告 """ import os import sys import subprocess import time import json import argparse from datetime import datetime, timezone, timedelta from pathlib import Path # ── 环境初始化 ────────────────────────────────────────────────── from dotenv import load_dotenv PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent load_dotenv(PROJECT_ROOT / ".env") APP_DB_DSN = os.environ.get("APP_DB_DSN") if not APP_DB_DSN: print("❌ APP_DB_DSN 环境变量未设置，无法运行属性测试。") sys.exit(1) # ── 测试定义 ────────────────────────────────────────────────── # 每个属性测试的 pytest 节点 ID 和描述 PROPERTY_TESTS = [ { "id": "P1", "name": "迁移脚本幂等性", "class": "TestProperty1MigrationIdempotency", "validates": "Requirements 1.9, 2.4, 11.5", "description": "对认证系统迁移脚本（DDL + 种子数据）连续执行两次，验证幂等性", "db_write": True, # 标记是否写数据库 }, { "id": "P2", "name": "登录创建/查找用户", "class": "TestProperty2LoginCreateFindUser", "validates": "Requirements 3.2, 3.3", "description": "随机 openid 登录，验证新用户创建（pending）和已有用户查找", "db_write": True, }, { "id": "P3", "name": "disabled 用户登录拒绝", "class": "TestProperty3DisabledUserLoginRejection", "validates": "Requirements 3.5", "description": "disabled 状态用户登录返回 403，不签发 JWT", "db_write": True, }, { "id": "P4", "name": "申请创建正确性", "class": "TestProperty4ApplicationCreation", "validates": "Requirements 4.1, 4.2, 4.3, 4.4", "description": "随机合法申请数据，验证申请记录创建和 site_code 映射", "db_write": True, }, { "id": "P5", "name": "手机号格式验证", "class": "TestProperty5PhoneFormatValidation", "validates": "Requirements 4.5", "description": "随机非法手机号提交申请，验证 422 拒绝", "db_write": False, }, { "id": "P6", "name": "重复申请拒绝", "class": "TestProperty6DuplicateApplicationRejection", "validates": "Requirements 4.6", "description": "已有 pending 申请的用户再次提交，验证 409 拒绝", "db_write": True, }, { "id": "P7", "name": "人员匹配合并正确性", "class": "TestProperty7MatchingMerge", "validates": "Requirements 5.1, 5.2, 5.3, 5.4", "description": "随机 site_id + phone 组合，验证助教/员工匹配结果合并", "db_write": False, }, { "id": "P8", "name": "审核操作正确性", "class": "TestProperty8ReviewOperations", "validates": "Requirements 6.1, 6.2, 6.3, 6.4, 6.5", "description": "随机 pending 申请执行批准/拒绝，验证状态流转和绑定创建", "db_write": True, }, { "id": "P9", "name": "非 pending 审核拒绝", "class": "TestProperty9NonPendingReviewRejection", "validates": "Requirements 6.6", "description": "非 pending 状态申请执行审核，验证 409 拒绝", "db_write": True, }, { "id": "P10", "name": "用户状态查询完整性", "class": "TestProperty10UserStatusQueryCompleteness", "validates": "Requirements 7.1, 7.2", "description": "随机用户状态组合，验证查询返回完整的申请列表和店铺信息", "db_write": True, }, { "id": "P11", "name": "多店铺角色独立分配", "class": "TestProperty11MultiSiteRoleIndependence", "validates": "Requirements 8.1", "description": "随机用户 + 多 site_id，验证角色独立分配互不干扰", "db_write": True, }, { "id": "P12", "name": "店铺切换令牌正确性", "class": "TestProperty12SiteSwitchTokenCorrectness", "validates": "Requirements 8.2, 10.4", "description": "多店铺用户切换店铺，验证新 JWT 中 site_id 和 roles 正确", "db_write": True, }, { "id": "P13", "name": "权限中间件拦截正确性", "class": "TestProperty13PermissionMiddleware", "validates": "Requirements 8.3, 9.1, 9.2, 9.3", "description": "随机用户 + 权限组合，验证中间件拦截/放行逻辑", "db_write": True, }, { "id": "P14", "name": "JWT payload 结构一致性", "class": "TestProperty14JwtPayloadStructure", "validates": "Requirements 10.1, 10.2, 10.3", "description": "随机用户状态签发 JWT，验证 payload 字段与状态一致", "db_write": False, }, { "id": "P15", "name": "JWT 过期/无效拒绝", "class": "TestProperty15JwtExpiredInvalidRejection", "validates": "Requirements 9.4", "description": "随机过期/篡改/错密钥/垃圾 JWT，验证 401 拒绝", "db_write": False, }, ] TEST_FILE = "tests/test_auth_system_properties.py" MAX_EXAMPLES = 100 # ── 时区 ────────────────────────────────────────────────────── CST = timezone(timedelta(hours=8)) def _now_cst() -> datetime: return datetime.now(CST) def _fmt(dt: datetime) -> str: return dt.strftime("%Y-%m-%d %H:%M:%S") # ── 运行单个测试 ───────────────────────────────────────────── def run_single_test(prop: dict, max_examples: int = MAX_EXAMPLES) -> dict: """ 运行单个属性测试，返回结果字典。 通过 HYPOTHESIS_MAX_EXAMPLES 环境变量覆盖迭代次数。 """ node_id = f"{TEST_FILE}::{prop['class']}" start = _now_cst() print(f"\n{'='*60}") print(f"▶ [{prop['id']}] {prop['name']} (max_examples={max_examples})") print(f" 节点: {node_id}") print(f" 开始: {_fmt(start)}") print(f"{'='*60}") env = os.environ.copy() # hypothesis 通过 settings profile 或环境变量控制 # 我们用 --override-ini 传递 hypothesis settings cmd = [ sys.executable, "-m", "pytest", node_id, "-v", "--tb=short", f"--hypothesis-seed=0", # 固定种子保证可复现 "-x", # 遇到第一个失败就停止（节省时间） f"-o", f"hypothesis_settings_max_examples={max_examples}", ] # hypothesis 不支持 pytest -o 覆盖 max_examples， # 改用环境变量 + conftest 或直接 patch # 实际方案：通过 HYPOTHESIS_MAX_EXAMPLES 环境变量 env["HYPOTHESIS_MAX_EXAMPLES"] = str(max_examples) result = subprocess.run( cmd, capture_output=True, text=True, cwd=str(PROJECT_ROOT), env=env, timeout=600, # 单个测试最多 10 分钟 ) end = _now_cst() duration = (end - start).total_seconds() # 解析结果 passed = result.returncode == 0 stdout = result.stdout or "" stderr = result.stderr or "" # 提取测试计数 test_count = _extract_test_count(stdout) outcome = { "id": prop["id"], "name": prop["name"], "class": prop["class"], "validates": prop["validates"], "description": prop["description"], "db_write": prop["db_write"], "passed": passed, "returncode": result.returncode, "duration_sec": round(duration, 1), "start_time": _fmt(start), "end_time": _fmt(end), "test_count": test_count, "stdout_tail": _tail(stdout, 30), "stderr_tail": _tail(stderr, 10), } # 实时进度输出 status = "✅ PASSED" if passed else "❌ FAILED" print(f"\n 结果: {status} 耗时: {duration:.1f}s 测试数: {test_count}") if not passed: print(f" --- stdout 尾部 ---") print(outcome["stdout_tail"]) if stderr.strip(): print(f" --- stderr 尾部 ---") print(outcome["stderr_tail"]) return outcome def _extract_test_count(stdout: str) -> str: """从 pytest 输出中提取测试计数，如 '3 passed' 或 '2 passed, 1 failed'""" for line in reversed(stdout.splitlines()): line = line.strip() if "passed" in line or "failed" in line or "error" in line: # 去掉 ANSI 颜色码 import re clean = re.sub(r'\x1b\[[0-9;]*m', '', line) if "=" in clean: return clean.split("=")[-1].strip().rstrip("=").strip() return "unknown" def _tail(text: str, n: int) -> str: """取文本最后 n 行""" lines = text.splitlines() return "\n".join(lines[-n:]) if len(lines) > n else text # ── 生成 MD 报告 ───────────────────────────────────────────── def generate_report(results: list[dict], total_start: datetime, total_end: datetime) -> str: """生成详细的 MD 格式测试报告""" total_duration = (total_end - total_start).total_seconds() passed_count = sum(1 for r in results if r["passed"]) failed_count = len(results) - passed_count all_passed = failed_count == 0 lines = [] lines.append(f"# 认证系统属性测试全量报告") lines.append(f"") lines.append(f"- Spec: `03-miniapp-auth-system`（小程序用户认证系统）") lines.append(f"- 任务: 11. 属性测试全量运行（{MAX_EXAMPLES} 次迭代）") lines.append(f"- 测试文件: `{TEST_FILE}`") lines.append(f"- 数据库: `test_zqyy_app`（通过 `APP_DB_DSN`）") lines.append(f"- 运行时间: {_fmt(total_start)} → {_fmt(total_end)}（共 {total_duration:.0f}s）") lines.append(f"- 总体结果: {'✅ 全部通过' if all_passed else f'❌ {failed_count} 个失败'}") lines.append(f"- 通过/总数: {passed_count}/{len(results)}") lines.append(f"") # 汇总表 lines.append(f"## 汇总") lines.append(f"") lines.append(f"| # | 属性 | 验证需求 | 结果 | 耗时 | 测试数 | 写库 |") lines.append(f"|---|------|---------|------|------|--------|------|") for r in results: status = "✅" if r["passed"] else "❌" db = "是" if r["db_write"] else "否" lines.append( f"| {r['id']} | {r['name']} | {r['validates']} | " f"{status} | {r['duration_sec']}s | {r['test_count']} | {db} |" ) lines.append(f"") # 失败详情 failed = [r for r in results if not r["passed"]] if failed: lines.append(f"## 失败详情") lines.append(f"") for r in failed: lines.append(f"### {r['id']} {r['name']}") lines.append(f"") lines.append(f"- 验证需求: {r['validates']}") lines.append(f"- 描述: {r['description']}") lines.append(f"- 返回码: {r['returncode']}") lines.append(f"- 耗时: {r['duration_sec']}s") lines.append(f"") lines.append(f"```") lines.append(r["stdout_tail"]) lines.append(f"```") if r["stderr_tail"].strip(): lines.append(f"") lines.append(f"stderr:") lines.append(f"```") lines.append(r["stderr_tail"]) lines.append(f"```") lines.append(f"") # 各测试详情 lines.append(f"## 各属性测试详情") lines.append(f"") for r in results: status = "✅ PASSED" if r["passed"] else "❌ FAILED" lines.append(f"### {r['id']} {r['name']} — {status}") lines.append(f"") lines.append(f"- 描述: {r['description']}") lines.append(f"- 验证需求: {r['validates']}") lines.append(f"- 测试类: `{r['class']}`") lines.append(f"- 开始: {r['start_time']} 结束: {r['end_time']}") lines.append(f"- 耗时: {r['duration_sec']}s") lines.append(f"- 测试数: {r['test_count']}") lines.append(f"- 写库: {'是' if r['db_write'] else '否'}") lines.append(f"") # 数据库带宽说明 lines.append(f"## 数据库资源控制") lines.append(f"") lines.append(f"- 串行执行：同一时刻仅运行 1 个属性测试，避免数据库连接争用") lines.append(f"- 写库测试（{sum(1 for r in results if r['db_write'])} 个）每个测试内部自行清理测试数据") lines.append(f"- 纯读/内存测试（{sum(1 for r in results if not r['db_write'])} 个）不产生数据库写入") lines.append(f"- 测试间无并发，为其他调试任务保留数据库带宽") lines.append(f"") return "\n".join(lines) # ── conftest 补丁 ───────────────────────────────────────────── CONFTEST_PATCH = '''# -*- coding: utf-8 -*- """ 临时 conftest：通过环境变量覆盖 hypothesis max_examples。 由 _run_auth_pbt_full.py 自动生成，测试完成后自动删除。 原理：在 pytest 收集完测试后，遍历所有 hypothesis 测试项， 替换其 settings 中的 max_examples 为环境变量指定的值。 """ import os _max_env = os.environ.get("HYPOTHESIS_MAX_EXAMPLES") def pytest_collection_modifyitems(items): """收集完测试后，覆盖 hypothesis settings 中的 max_examples""" if not _max_env: return forced_max = int(_max_env) from hypothesis import settings as _settings for item in items: # hypothesis 测试会在 item 上挂 hypothesis_settings if hasattr(item, "_hypothesis_internal_use_settings"): old = item._hypothesis_internal_use_settings item._hypothesis_internal_use_settings = _settings( old, max_examples=forced_max, ) ''' def _ensure_conftest(): """确保 tests/conftest.py 中有 hypothesis max_examples 覆盖逻辑""" conftest_path = PROJECT_ROOT / "tests" / "conftest.py" marker = "HYPOTHESIS_MAX_EXAMPLES" if conftest_path.exists(): content = conftest_path.read_text(encoding="utf-8") if marker in content: return False # 已有，不需要补丁 # 追加 with open(conftest_path, "a", encoding="utf-8") as f: f.write("\n\n" + CONFTEST_PATCH) return True else: conftest_path.write_text(CONFTEST_PATCH, encoding="utf-8") return True def _cleanup_conftest(): """清理临时 conftest 补丁""" conftest_path = PROJECT_ROOT / "tests" / "conftest.py" if not conftest_path.exists(): return content = conftest_path.read_text(encoding="utf-8") if "由 _run_auth_pbt_full.py 自动生成" in content: # 如果整个文件都是我们生成的，删除 if content.strip().startswith("# -*- coding: utf-8 -*-\n") and "HYPOTHESIS_MAX_EXAMPLES" in content: lines = content.split("HYPOTHESIS_MAX_EXAMPLES") # 简单判断：如果文件很短且主要是我们的内容 if len(content) < 500: conftest_path.unlink() return # 否则只移除我们追加的部分 marker = "\n\n# -*- coding: utf-8 -*-\n\"\"\"\n临时 conftest" if marker in content: content = content[:content.index(marker)] conftest_path.write_text(content, encoding="utf-8") # ── 主流程 ──────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser(description="认证系统属性测试全量运行") parser.add_argument( "--max-examples", type=int, default=MAX_EXAMPLES, help=f"每个属性测试的迭代次数（默认 {MAX_EXAMPLES}）", ) parser.add_argument( "--only", type=str, default=None, help="仅运行指定属性（逗号分隔，如 P1,P2,P14）", ) parser.add_argument( "--skip", type=str, default=None, help="跳过指定属性（逗号分隔，如 P3,P5）", ) parser.add_argument( "--report-dir", type=str, default=None, help="报告输出目录（默认 export/ 下）", ) args = parser.parse_args() # 筛选测试 tests_to_run = PROPERTY_TESTS[:] if args.only: only_ids = {x.strip().upper() for x in args.only.split(",")} tests_to_run = [t for t in tests_to_run if t["id"] in only_ids] if args.skip: skip_ids = {x.strip().upper() for x in args.skip.split(",")} tests_to_run = [t for t in tests_to_run if t["id"] not in skip_ids] if not tests_to_run: print("❌ 没有要运行的测试。") sys.exit(1) max_ex = args.max_examples print(f"╔{'═'*58}╗") print(f"║ 认证系统属性测试全量运行 ║") print(f"║ 迭代次数: {max_ex:<5} 测试数: {len(tests_to_run):<3} ║") print(f"║ 数据库: test_zqyy_app（串行执行，控制带宽） ║") print(f"╚{'═'*58}╝") # 确保 conftest 补丁 patched = _ensure_conftest() if patched: print("📝 已注入 conftest.py hypothesis max_examples 覆盖") total_start = _now_cst() results = [] try: for i, prop in enumerate(tests_to_run, 1): print(f"\n📊 进度: {i}/{len(tests_to_run)}") outcome = run_single_test(prop, max_examples=max_ex) results.append(outcome) # 测试间短暂休息，释放数据库连接 if i < len(tests_to_run): time.sleep(1) except KeyboardInterrupt: print("\n\n⚠️ 用户中断，生成已完成部分的报告...") except Exception as e: print(f"\n\n❌ 运行异常: {e}") finally: total_end = _now_cst() # 生成报告 if results: report = generate_report(results, total_start, total_end) # 确定报告路径 report_dir = Path(args.report_dir) if args.report_dir else PROJECT_ROOT / "export" / "reports" report_dir.mkdir(parents=True, exist_ok=True) timestamp = total_start.strftime("%Y%m%d_%H%M%S") report_path = report_dir / f"auth_pbt_full_{timestamp}.md" report_path.write_text(report, encoding="utf-8") passed = sum(1 for r in results if r["passed"]) failed = len(results) - passed print(f"\n{'='*60}") print(f"📋 报告已生成: {report_path}") print(f" 通过: {passed} 失败: {failed} 总计: {len(results)}") print(f" 总耗时: {(total_end - total_start).total_seconds():.0f}s") print(f"{'='*60}") # 清理 conftest 补丁 if patched: _cleanup_conftest() print("🧹 已清理临时 conftest 补丁") if __name__ == "__main__": main()