148 lines
5.0 KiB
Python
148 lines
5.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
业务表 site_id 字段存在性属性测试
|
|
|
|
**Validates: Requirements 13.1**
|
|
|
|
Property 10: 对于任意 app schema 中的业务视图和 dws/core schema 中的业务表，
|
|
其定义中应包含 site_id 字段。
|
|
"""
|
|
import os
|
|
import re
|
|
|
|
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")
|
|
|
|
# ── 全局排除表 ────────────────────────────────────────────
|
|
# 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 以支持多门店隔离。"
|
|
)
|