Neo
6f8f12314f
包含多个会话的累积代码变更： - backend: AI 聊天服务、触发器调度、认证增强、WebSocket、调度器最小间隔 - admin-web: ETL 状态页、任务管理、调度配置、登录优化 - miniprogram: 看板页面、聊天集成、UI 组件、导航更新 - etl: DWS 新任务（finance_area_daily/board_cache）、连接器增强 - tenant-admin: 项目初始化 - db: 19 个迁移脚本（etl_feiqiu 11 + zqyy_app 8） - packages/shared: 枚举和工具函数更新 - tools: 数据库工具、报表生成、健康检查 - docs: PRD/架构/部署/合约文档更新 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
157 lines
5.3 KiB
Python
157 lines
5.3 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
业务表 site_id 字段存在性属性测试
|
||
|
||
**Validates: Requirements 13.1**
|
||
|
||
Property 10: 对于任意 app schema 中的业务视图和 dws/core schema 中的业务表，
|
||
其定义中应包含 site_id 字段。
|
||
"""
|
||
import os
|
||
import re
|
||
|
||
import pytest
|
||
from hypothesis import given, settings
|
||
from hypothesis.strategies import sampled_from
|
||
|
||
# ── 路径常量 ──────────────────────────────────────────────
|
||
SCHEMAS_DIR = os.path.join(r"C:\NeoZQYY", "db", "etl_feiqiu", "schemas")
|
||
ZQYY_APP_DIR = os.path.join(r"C:\NeoZQYY", "db", "zqyy_app", "schemas")
|
||
|
||
APP_SQL = os.path.join(SCHEMAS_DIR, "app.sql")
|
||
DWS_SQL = os.path.join(SCHEMAS_DIR, "dws.sql")
|
||
CORE_SQL = os.path.join(SCHEMAS_DIR, "core.sql")
|
||
ZQYY_INIT_SQL = os.path.join(ZQYY_APP_DIR, "init.sql")
|
||
|
||
# DDL 基线文件在 ETL schema 重构后已删除，跳过整个模块
|
||
_required = [APP_SQL, DWS_SQL, CORE_SQL, ZQYY_INIT_SQL]
|
||
if not all(os.path.exists(f) for f in _required):
|
||
pytest.skip(
|
||
"DDL 基线文件不存在（ETL schema 重构后已删除）",
|
||
allow_module_level=True,
|
||
)
|
||
|
||
# ── 全局排除表 ────────────────────────────────────────────
|
||
# permissions / role_permissions 是全局表，不需要 site_id
|
||
# cfg_* 是 dws 层的配置表，属于全局/租户级配置
|
||
# dim_goods_category 是商品分类维度，属于租户级全局参照表
|
||
GLOBAL_TABLES = {
|
||
"permissions",
|
||
"role_permissions",
|
||
"dim_goods_category",
|
||
}
|
||
|
||
# dws 配置表前缀（全局配置，不按门店隔离）
|
||
CFG_PREFIX = "cfg_"
|
||
|
||
|
||
# ── 解析工具 ──────────────────────────────────────────────
|
||
|
||
# 匹配 CREATE TABLE [IF NOT EXISTS] [schema.]table_name(
|
||
_CREATE_TABLE_RE = re.compile(
|
||
r"CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?"
|
||
r"(?:[\w]+\.)?(\w+)\s*\(",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
# 匹配 CREATE [OR REPLACE] VIEW [schema.]view_name AS
|
||
_CREATE_VIEW_RE = re.compile(
|
||
r"CREATE\s+(?:OR\s+REPLACE\s+)?VIEW\s+(?:[\w]+\.)?(\w+)\s+AS",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
def _extract_definitions(sql_path: str) -> dict[str, str]:
|
||
"""
|
||
从 SQL 文件中提取所有 CREATE TABLE / CREATE VIEW 定义。
|
||
返回 {name: definition_text} 字典。
|
||
"""
|
||
with open(sql_path, encoding="utf-8") as f:
|
||
content = f.read()
|
||
|
||
markers: list[tuple[int, str]] = []
|
||
|
||
for m in _CREATE_TABLE_RE.finditer(content):
|
||
markers.append((m.start(), m.group(1).lower()))
|
||
|
||
for m in _CREATE_VIEW_RE.finditer(content):
|
||
markers.append((m.start(), m.group(1).lower()))
|
||
|
||
markers.sort(key=lambda x: x[0])
|
||
|
||
result: dict[str, str] = {}
|
||
for i, (pos, name) in enumerate(markers):
|
||
end = markers[i + 1][0] if i + 1 < len(markers) else len(content)
|
||
result[name] = content[pos:end]
|
||
|
||
return result
|
||
|
||
|
||
def _has_site_id(definition: str) -> bool:
|
||
"""检查定义文本中是否包含 site_id 字段。"""
|
||
return bool(re.search(r"\bsite_id\b", definition, re.IGNORECASE))
|
||
|
||
|
||
def _is_business_object(name: str) -> bool:
|
||
"""判断是否为业务表/视图（排除全局表和配置表）。"""
|
||
if name in GLOBAL_TABLES:
|
||
return False
|
||
if name.startswith(CFG_PREFIX):
|
||
return False
|
||
return True
|
||
|
||
|
||
# ── 预加载定义（模块级，只解析一次） ────────────────────────
|
||
|
||
_app_defs = _extract_definitions(APP_SQL)
|
||
_dws_defs = _extract_definitions(DWS_SQL)
|
||
_core_defs = _extract_definitions(CORE_SQL)
|
||
_zqyy_defs = _extract_definitions(ZQYY_INIT_SQL)
|
||
|
||
# 构建业务对象列表：(name, source, definition)
|
||
BUSINESS_OBJECTS: list[tuple[str, str, str]] = []
|
||
|
||
for name, defn in _app_defs.items():
|
||
if _is_business_object(name):
|
||
BUSINESS_OBJECTS.append((name, "app", defn))
|
||
|
||
for name, defn in _dws_defs.items():
|
||
if _is_business_object(name):
|
||
BUSINESS_OBJECTS.append((name, "dws", defn))
|
||
|
||
for name, defn in _core_defs.items():
|
||
if _is_business_object(name):
|
||
BUSINESS_OBJECTS.append((name, "core", defn))
|
||
|
||
for name, defn in _zqyy_defs.items():
|
||
if _is_business_object(name):
|
||
BUSINESS_OBJECTS.append((name, "zqyy_app", defn))
|
||
|
||
# 排除 dws 中的函数定义（不是表/视图）
|
||
BUSINESS_OBJECTS = [
|
||
(n, s, d) for n, s, d in BUSINESS_OBJECTS
|
||
if not n.startswith("get_")
|
||
]
|
||
|
||
assert len(BUSINESS_OBJECTS) > 0, "未找到任何业务表/视图定义，请检查 DDL 文件路径"
|
||
|
||
|
||
# ── 属性测试 ──────────────────────────────────────────────
|
||
|
||
@given(obj=sampled_from(BUSINESS_OBJECTS))
|
||
@settings(max_examples=100)
|
||
def test_business_object_has_site_id(obj: tuple[str, str, str]):
|
||
"""
|
||
Property 10: 业务表 site_id 字段存在性
|
||
|
||
对于任意 app schema 中的业务视图和 dws/core/zqyy_app schema 中的业务表，
|
||
其定义中应包含 site_id 字段。
|
||
|
||
**Validates: Requirements 13.1**
|
||
"""
|
||
name, source, definition = obj
|
||
assert _has_site_id(definition), (
|
||
f"{source}.{name} 缺少 site_id 字段。"
|
||
f"Requirements 13.1 要求所有业务表包含 site_id 以支持多门店隔离。"
|
||
)
|