签入版本

2026-09-12 18:24:41 +08:00
commit fb2b7d9538
852 changed files with 115466 additions and 0 deletions
@@ -0,0 +1,185 @@
+"""契约快照测试公共工具。
+
+契约 = 前端/数据依赖的硬不变量：
+- 路由 (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.naivevue.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