545 lines
20 KiB
Python
545 lines
20 KiB
Python
# -*- 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()
|