Neo-ZQYY/tests/test_property_site_id_existence.py

# -*- 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:\Project\NeoZQYY", "db", "etl_feiqiu", "schemas")
ZQYY_APP_DIR = os.path.join(r"C:\Project\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 以支持多门店隔离。"
    )