- core/redis_client.py 移除 scheduler 耦合,lifespan 由 core/app.py 组合(Redis→调度器→Redis) - main.py 改从 core.app 导入 create_app - 删除 apps/、extensions/ 目录;utils 删除已迁移垫片与死代码 formatter.py - 新增 deploy/(nginx、supervisor)、scripts/(init_db、seed)、logs/(结构化日志落盘) - core/app.py 接入 core.logger setup_logging - 更新 CLAUDE.md 反映 core/shared/modules 新结构 Co-Authored-By: Claude <noreply@anthropic.com>
67 lines
2.0 KiB
Python
67 lines
2.0 KiB
Python
"""契约快照测试公共工具。
|
|
|
|
契约 = 前端/数据依赖的硬不变量:
|
|
- 路由 (METHOD /path)
|
|
- 权限字符串 (@permission_required("sys:..."))
|
|
- 数据库表名
|
|
|
|
用法:设置环境变量 UPDATE_GOLDEN=1 运行一次以重新生成 golden。
|
|
"""
|
|
import glob
|
|
import json
|
|
import os
|
|
import re
|
|
|
|
GOLDEN_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "golden")
|
|
|
|
PERMISSION_RE = re.compile(r'@permission_required\(\s*["\'](sys:[^"\']+)["\']')
|
|
|
|
UPDATE_GOLDEN = os.getenv("UPDATE_GOLDEN") == "1"
|
|
|
|
|
|
def collect_routes():
|
|
"""通过 openapi schema 收集全部 (METHOD /path),去重排序。"""
|
|
from core.app import create_app
|
|
|
|
app = create_app()
|
|
spec = app.openapi()
|
|
return sorted(
|
|
f"{method.upper()} {path}"
|
|
for path, methods in spec["paths"].items()
|
|
for method in methods
|
|
)
|
|
|
|
|
|
def collect_permissions():
|
|
"""静态扫描源码中的 @permission_required("sys:...") 权限串。"""
|
|
perms = set()
|
|
for root in ("apps", "modules"):
|
|
for path in glob.glob(os.path.join(root, "**", "*.py"), recursive=True):
|
|
with open(path, encoding="utf-8") as f:
|
|
perms.update(PERMISSION_RE.findall(f.read()))
|
|
return sorted(perms)
|
|
|
|
|
|
def collect_tables():
|
|
"""从 SQLAlchemy metadata 收集全部表名。"""
|
|
from core.database import Base
|
|
|
|
return sorted(Base.metadata.tables.keys())
|
|
|
|
|
|
def compare_snapshot(name: str, actual: list):
|
|
"""与 golden 对比;UPDATE_GOLDEN=1 时重新生成并跳过断言。"""
|
|
path = os.path.join(GOLDEN_DIR, name)
|
|
if UPDATE_GOLDEN:
|
|
os.makedirs(GOLDEN_DIR, exist_ok=True)
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump(actual, f, ensure_ascii=False, indent=2)
|
|
return
|
|
with open(path, encoding="utf-8") as f:
|
|
golden = json.load(f)
|
|
assert actual == golden, (
|
|
f"契约快照 [{name}] 不一致!\n"
|
|
f"新增: {sorted(set(actual) - set(golden))}\n"
|
|
f"缺失: {sorted(set(golden) - set(actual))}"
|
|
)
|