# -*- coding: utf-8 -*- """ 测试数据库结构一致性属性测试 **Validates: Requirements 9.1, 9.2** Property 8: 对于任意生产数据库(etl_feiqiu、zqyy_app)中的 schema 和表定义, 对应的测试数据库(test_etl_feiqiu、test_zqyy_app)中应存在相同的 schema 和表结构。 测试逻辑:测试数据库创建脚本通过 \\i 引用生产 DDL 文件, 结构一致性可以通过验证脚本引用的完整性来保证—— 即每个 \\i 引用的 DDL 文件在磁盘上实际存在。 """ import os import re from hypothesis import given, settings from hypothesis.strategies import sampled_from # ── 路径常量 ────────────────────────────────────────────── MONOREPO_ROOT = r"C:\NeoZQYY" DB_CONFIGS = { "etl_feiqiu": { "script": os.path.join( MONOREPO_ROOT, "db", "etl_feiqiu", "scripts", "create_test_db.sql" ), "base_dir": os.path.join( MONOREPO_ROOT, "db", "etl_feiqiu", "scripts" ), }, "zqyy_app": { "script": os.path.join( MONOREPO_ROOT, "db", "zqyy_app", "scripts", "create_test_db.sql" ), "base_dir": os.path.join( MONOREPO_ROOT, "db", "zqyy_app", "scripts" ), }, } # ── 解析 \\i 引用 ───────────────────────────────────────── # 匹配注释中的 \i 指令(psql 元命令),如: # \i ../schemas/meta.sql # \i ../seeds/*.sql ← 通配符引用,跳过 _PSQL_INCLUDE_RE = re.compile(r"\\i\s+(\S+)") def _extract_ddl_refs(script_path: str, base_dir: str) -> list[tuple[str, str]]: """ 从 create_test_db.sql 中提取所有 \\i 引用的 DDL 文件路径。 返回 [(相对路径, 绝对路径), ...] 列表。 跳过包含通配符的引用(如 ../seeds/*.sql)。 """ with open(script_path, encoding="utf-8") as f: content = f.read() refs = [] for m in _PSQL_INCLUDE_RE.finditer(content): rel_path = m.group(1) # 跳过通配符引用,无法逐文件验证 if "*" in rel_path: continue abs_path = os.path.normpath(os.path.join(base_dir, rel_path)) refs.append((rel_path, abs_path)) return refs # ── 预加载所有引用(模块级,只解析一次) ───────────────────── DDL_REFERENCES: list[tuple[str, str, str]] = [] """每个元素: (数据库名, 相对路径, 绝对路径)""" for db_name, cfg in DB_CONFIGS.items(): assert os.path.isfile(cfg["script"]), ( f"测试数据库创建脚本不存在: {cfg['script']}" ) for rel_path, abs_path in _extract_ddl_refs(cfg["script"], cfg["base_dir"]): DDL_REFERENCES.append((db_name, rel_path, abs_path)) assert len(DDL_REFERENCES) > 0, ( "未从 create_test_db.sql 中提取到任何 \\i DDL 引用,请检查脚本内容" ) # ── 属性测试 ────────────────────────────────────────────── @given(ref=sampled_from(DDL_REFERENCES)) @settings(max_examples=100) def test_test_db_ddl_references_exist(ref: tuple[str, str, str]): """ Property 8: 测试数据库结构一致性 对于任意生产数据库的测试数据库创建脚本中 \\i 引用的 DDL 文件, 该文件在磁盘上应实际存在。这保证了测试数据库能完整复用生产 DDL, 从而确保结构一致性。 **Validates: Requirements 9.1, 9.2** """ db_name, rel_path, abs_path = ref assert os.path.isfile(abs_path), ( f"[{db_name}] create_test_db.sql 引用的 DDL 文件不存在: " f"{rel_path} → {abs_path}" )