186 lines
6.6 KiB
Python
186 lines
6.6 KiB
Python
"""契约快照测试公共工具。
|
||
|
||
契约 = 前端/数据依赖的硬不变量：
|
||
- 路由 (METHOD /path)
|
||
- 权限字符串 (@permission_required("sys:..."))
|
||
- 数据库表名
|
||
|
||
用法：设置环境变量 UPDATE_GOLDEN=1 运行一次以重新生成 golden。
|
||
"""
|
||
import glob
|
||
import json
|
||
import os
|
||
import re
|
||
|
||
from sqlalchemy import BigInteger, DateTime, Date, Float, Integer, String, Text
|
||
|
||
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 ("src/api/v1/endpoints", "src/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))}"
|
||
)
|
||
|
||
|
||
# ============================================================
|
||
# 模型 ↔ SQL dump 建表结构一致性
|
||
# ============================================================
|
||
|
||
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
SQL_SCHEMA_PATH = os.path.join(PROJECT_ROOT, "document", "djangoadmin.fastapi.antdvue.sql")
|
||
|
||
|
||
def parse_sql_type(type_tail: str):
|
||
"""从列定义中解析 SQL 类型，归一化到与模型一致的 (类型族, 长度)。
|
||
|
||
SQL 方言与归一化规则：
|
||
int/int(0)/tinyint(1)/smallint/mediumint/bit(1) -> "int"
|
||
bigint -> "bigint"
|
||
datetime(0)/datetime(6)/timestamp -> "datetime"
|
||
date -> "date"
|
||
varchar(n)/char(n) -> ("varchar", n)
|
||
text/tinytext/mediumtext/longtext -> "text"
|
||
float/double/decimal -> "float"
|
||
"""
|
||
m = re.match(r"([a-z]+)(?:\((\d+)\))?", type_tail.strip().lower())
|
||
if not m:
|
||
raise ValueError(f"无法解析 SQL 类型: {type_tail!r}")
|
||
name, length = m.group(1), m.group(2)
|
||
if name in ("int", "tinyint", "smallint", "mediumint", "bit"):
|
||
return ("int", None)
|
||
if name == "bigint":
|
||
return ("bigint", None)
|
||
if name in ("datetime", "timestamp"):
|
||
return ("datetime", None)
|
||
if name == "date":
|
||
return ("date", None)
|
||
if name in ("varchar", "char"):
|
||
return ("varchar", int(length))
|
||
if name in ("text", "tinytext", "mediumtext", "longtext"):
|
||
return ("text", None)
|
||
if name in ("float", "double", "decimal"):
|
||
return ("float", None)
|
||
raise ValueError(f"无法解析 SQL 类型: {type_tail!r}")
|
||
|
||
|
||
def collect_sql_schema():
|
||
"""解析 SQL dump 中的 CREATE TABLE，产出 {表名: {列名: (类型族, 长度, 可空)}} 与 {表名: 主键集}。
|
||
|
||
兼容 Navicat 导出（Style A：`` `col` int(0) NULL DEFAULT NULL ``）与
|
||
手写（Style B：`` `col` int NOT NULL ``）两种格式。
|
||
"""
|
||
with open(SQL_SCHEMA_PATH, encoding="utf-8") as f:
|
||
content = f.read()
|
||
schema, pk_map = {}, {}
|
||
for m in re.finditer(r"CREATE TABLE\s+`([^`]+)`\s*\((.*?)\)\s*ENGINE", content, re.S):
|
||
table = m.group(1)
|
||
columns, pks = {}, set()
|
||
for line in m.group(2).splitlines():
|
||
s = line.strip().rstrip(",")
|
||
if not s or s.startswith("PRIMARY KEY"):
|
||
if s.startswith("PRIMARY KEY"):
|
||
pk_m = re.search(r"`([^`]+)`", s)
|
||
if pk_m:
|
||
pks.add(pk_m.group(1))
|
||
continue
|
||
if s.startswith(("KEY ", "UNIQUE KEY", "INDEX ", "FULLTEXT ")):
|
||
continue
|
||
col_m = re.match(r"`([^`]+)`\s+(.+)$", s)
|
||
if not col_m:
|
||
continue
|
||
col, type_tail = col_m.group(1), col_m.group(2)
|
||
# 忽略 COMMENT 子句后再判断可空，避免注释文本误含 NOT NULL
|
||
pre = type_tail.split("COMMENT")[0]
|
||
nullable = "NOT NULL" not in pre.upper()
|
||
columns[col] = (*parse_sql_type(type_tail), nullable)
|
||
schema[table] = columns
|
||
pk_map[table] = pks
|
||
return schema, pk_map
|
||
|
||
|
||
def model_type(col):
|
||
"""将 SQLAlchemy 列类型映射到与 parse_sql_type 相同的 (类型族, 长度)。"""
|
||
t = col.type
|
||
if isinstance(t, BigInteger):
|
||
return ("bigint", None)
|
||
if isinstance(t, Integer):
|
||
return ("int", None)
|
||
if isinstance(t, DateTime):
|
||
return ("datetime", None)
|
||
if isinstance(t, Date):
|
||
return ("date", None)
|
||
# 注意：Text 是 String 的子类，必须先判断 Text
|
||
if isinstance(t, Text):
|
||
return ("text", None)
|
||
if isinstance(t, String):
|
||
return ("varchar", t.length)
|
||
if isinstance(t, Float):
|
||
return ("float", None)
|
||
raise ValueError(f"无法映射模型列类型: {col} ({t!r})")
|
||
|
||
|
||
def collect_model_schema():
|
||
"""通过 create_app 强制加载全部模型，从 Base.metadata 产出与 collect_sql_schema 同构的数据。"""
|
||
from core.app import create_app
|
||
|
||
create_app()
|
||
|
||
from core.database import Base
|
||
|
||
schema, pk_map = {}, {}
|
||
for table_name, table in Base.metadata.tables.items():
|
||
columns, pks = {}, set()
|
||
for col in table.columns:
|
||
columns[col.name] = (*model_type(col), col.nullable)
|
||
if col.primary_key:
|
||
pks.add(col.name)
|
||
schema[table_name] = columns
|
||
pk_map[table_name] = pks
|
||
return schema, pk_map
|