215 lines
9.8 KiB
Python
215 lines
9.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
Feature: dataflow-field-completion, Property 3: TABLE_MAP 覆盖完整性
|
||
|
||
**Validates: Requirements 7.2, 8.2**
|
||
|
||
对于任意在 TABLE_MAP 中注册的 DWD 表，该表的所有非 SCD2 列要么在 FACT_MAPPINGS
|
||
中有显式映射，要么在对应 ODS 表中存在同名列（自动映射）。
|
||
|
||
本测试聚焦以下可静态验证的属性：
|
||
1. TABLE_MAP 所有条目的 ODS 源表非空
|
||
2. C 类表在 TABLE_MAP 中注册
|
||
3. C 类表在 FACT_MAPPINGS 中有条目
|
||
4. C 类表映射字段数量与期望一致
|
||
5. TABLE_MAP 与 FACT_MAPPINGS 交叉一致性
|
||
6. hypothesis 属性测试：随机 TABLE_MAP 条目结构验证
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
from hypothesis import given, settings, HealthCheck
|
||
import hypothesis.strategies as st
|
||
|
||
# ── 将 ETL 模块加入 sys.path ──
|
||
_ETL_ROOT = Path(__file__).resolve().parent.parent / "apps" / "etl" / "connectors" / "feiqiu"
|
||
if str(_ETL_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(_ETL_ROOT))
|
||
|
||
from tasks.dwd.dwd_load_task import DwdLoadTask
|
||
|
||
# ── C 类表定义 ──
|
||
C_CLASS_TABLES = [
|
||
"dwd.dwd_goods_stock_summary",
|
||
"dwd.dwd_goods_stock_movement",
|
||
]
|
||
|
||
# ── C 类表期望映射字段 ──
|
||
|
||
# goods_stock_summary → dwd_goods_stock_summary（14 个字段）
|
||
_GOODS_STOCK_SUMMARY_EXPECTED_COLS = {
|
||
"site_goods_id", "goods_name", "goods_unit", "goods_category_id",
|
||
"goods_category_second_id", "category_name", "range_start_stock",
|
||
"range_end_stock", "range_in", "range_out", "range_sale",
|
||
"range_sale_money", "range_inventory", "current_stock",
|
||
}
|
||
|
||
# goods_stock_movements → dwd_goods_stock_movement（19 个字段）
|
||
_GOODS_STOCK_MOVEMENT_EXPECTED_COLS = {
|
||
"site_goods_stock_id", "tenant_id", "site_id", "site_goods_id",
|
||
"goods_name", "goods_category_id", "goods_second_category_id",
|
||
"unit", "price", "stock_type", "change_num", "start_num", "end_num",
|
||
"change_num_a", "start_num_a", "end_num_a", "remark", "operator_name",
|
||
"create_time",
|
||
}
|
||
|
||
# ── 收集所有 TABLE_MAP 条目 ──
|
||
_ALL_TABLE_MAP_ENTRIES = list(DwdLoadTask.TABLE_MAP.items())
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# Property 3.1: TABLE_MAP 所有条目的 ODS 源表非空
|
||
# ══════════════════════════════════════════════════════════════════
|
||
|
||
def test_all_table_map_entries_have_ods_source():
|
||
"""TABLE_MAP 中每个 DWD 表都有非空的 ODS 源表名。"""
|
||
for dwd_table, ods_table in DwdLoadTask.TABLE_MAP.items():
|
||
assert ods_table and isinstance(ods_table, str) and ods_table.strip(), \
|
||
f"{dwd_table}: TABLE_MAP 中的 ODS 源表为空或无效"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# Property 3.2: C 类表在 TABLE_MAP 中注册
|
||
# ══════════════════════════════════════════════════════════════════
|
||
|
||
def test_c_class_tables_registered_in_table_map():
|
||
"""
|
||
**Validates: Requirements 7.2, 8.2**
|
||
|
||
C 类表（dwd_goods_stock_summary、dwd_goods_stock_movement）
|
||
必须在 TABLE_MAP 中注册。
|
||
"""
|
||
for table in C_CLASS_TABLES:
|
||
assert table in DwdLoadTask.TABLE_MAP, \
|
||
f"C 类表 {table} 未在 TABLE_MAP 中注册"
|
||
ods_table = DwdLoadTask.TABLE_MAP[table]
|
||
assert ods_table and isinstance(ods_table, str), \
|
||
f"C 类表 {table}: TABLE_MAP 中的 ODS 源表无效"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# Property 3.3: C 类表在 FACT_MAPPINGS 中有条目
|
||
# ══════════════════════════════════════════════════════════════════
|
||
|
||
def test_c_class_tables_have_fact_mappings():
|
||
"""
|
||
**Validates: Requirements 7.2, 8.2**
|
||
|
||
C 类表必须在 FACT_MAPPINGS 中有至少一个映射条目。
|
||
"""
|
||
for table in C_CLASS_TABLES:
|
||
entries = DwdLoadTask.FACT_MAPPINGS.get(table, [])
|
||
assert len(entries) > 0, \
|
||
f"C 类表 {table} 在 FACT_MAPPINGS 中无条目"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# Property 3.4: goods_stock_summary 14 个字段全覆盖
|
||
# ══════════════════════════════════════════════════════════════════
|
||
|
||
def test_goods_stock_summary_mapping_coverage():
|
||
"""
|
||
**Validates: Requirements 7.2**
|
||
|
||
dwd.dwd_goods_stock_summary 的 FACT_MAPPINGS 应覆盖全部 14 个期望字段。
|
||
"""
|
||
table = "dwd.dwd_goods_stock_summary"
|
||
entries = DwdLoadTask.FACT_MAPPINGS.get(table, [])
|
||
actual_cols = {e[0].lower() for e in entries}
|
||
|
||
# 验证数量
|
||
assert len(entries) == 14, \
|
||
f"{table}: 期望 14 个映射条目，实际 {len(entries)} 个"
|
||
|
||
# 验证字段覆盖
|
||
missing = _GOODS_STOCK_SUMMARY_EXPECTED_COLS - actual_cols
|
||
assert not missing, \
|
||
f"{table}: 缺少映射字段 {missing}"
|
||
|
||
extra = actual_cols - _GOODS_STOCK_SUMMARY_EXPECTED_COLS
|
||
assert not extra, \
|
||
f"{table}: 存在多余映射字段 {extra}"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# Property 3.5: goods_stock_movement 19 个字段全覆盖
|
||
# ══════════════════════════════════════════════════════════════════
|
||
|
||
def test_goods_stock_movement_mapping_coverage():
|
||
"""
|
||
**Validates: Requirements 8.2**
|
||
|
||
dwd.dwd_goods_stock_movement 的 FACT_MAPPINGS 应覆盖全部 19 个期望字段。
|
||
"""
|
||
table = "dwd.dwd_goods_stock_movement"
|
||
entries = DwdLoadTask.FACT_MAPPINGS.get(table, [])
|
||
actual_cols = {e[0].lower() for e in entries}
|
||
|
||
# 验证数量
|
||
assert len(entries) == 19, \
|
||
f"{table}: 期望 19 个映射条目，实际 {len(entries)} 个"
|
||
|
||
# 验证字段覆盖
|
||
missing = _GOODS_STOCK_MOVEMENT_EXPECTED_COLS - actual_cols
|
||
assert not missing, \
|
||
f"{table}: 缺少映射字段 {missing}"
|
||
|
||
extra = actual_cols - _GOODS_STOCK_MOVEMENT_EXPECTED_COLS
|
||
assert not extra, \
|
||
f"{table}: 存在多余映射字段 {extra}"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# Property 3.6: FACT_MAPPINGS 是 TABLE_MAP 的子集
|
||
# ══════════════════════════════════════════════════════════════════
|
||
|
||
def test_fact_mappings_subset_of_table_map():
|
||
"""FACT_MAPPINGS 中的所有 DWD 表都必须在 TABLE_MAP 中注册。"""
|
||
for table in DwdLoadTask.FACT_MAPPINGS:
|
||
assert table in DwdLoadTask.TABLE_MAP, \
|
||
f"FACT_MAPPINGS 中的 {table} 未在 TABLE_MAP 中注册"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# Hypothesis 属性测试：随机 TABLE_MAP 条目结构验证
|
||
# ══════════════════════════════════════════════════════════════════
|
||
|
||
_table_map_entry_strategy = st.sampled_from(_ALL_TABLE_MAP_ENTRIES)
|
||
|
||
|
||
@given(entry=_table_map_entry_strategy)
|
||
@settings(max_examples=200, suppress_health_check=[HealthCheck.function_scoped_fixture])
|
||
def test_random_table_map_entry_valid(entry):
|
||
"""
|
||
**Validates: Requirements 7.2, 8.2**
|
||
|
||
对于任意随机选取的 TABLE_MAP 条目，验证：
|
||
- DWD 表名为 "dwd." 前缀的非空字符串
|
||
- ODS 源表名为 "ods." 前缀的非空字符串
|
||
- 如果该表在 FACT_MAPPINGS 中有条目，每个条目都是合法的三元组
|
||
"""
|
||
dwd_table, ods_table = entry
|
||
|
||
# DWD 表名格式验证
|
||
assert isinstance(dwd_table, str) and dwd_table.startswith("dwd."), \
|
||
f"TABLE_MAP key {dwd_table!r} 不以 'dwd.' 开头"
|
||
|
||
# ODS 源表名格式验证
|
||
assert isinstance(ods_table, str) and ods_table.startswith("ods."), \
|
||
f"TABLE_MAP[{dwd_table}] = {ods_table!r} 不以 'ods.' 开头"
|
||
|
||
# 如果有 FACT_MAPPINGS 条目，验证结构
|
||
entries = DwdLoadTask.FACT_MAPPINGS.get(dwd_table, [])
|
||
for i, e in enumerate(entries):
|
||
assert isinstance(e, (tuple, list)) and len(e) == 3, \
|
||
f"{dwd_table}[{i}]: FACT_MAPPINGS 条目应为三元组"
|
||
dwd_col, ods_expr, cast_type = e
|
||
assert isinstance(dwd_col, str) and dwd_col.strip(), \
|
||
f"{dwd_table}[{i}]: dwd_col 为空"
|
||
assert isinstance(ods_expr, str) and ods_expr.strip(), \
|
||
f"{dwd_table}[{i}]: ods_expr 为空"
|
||
assert cast_type is None or isinstance(cast_type, str), \
|
||
f"{dwd_table}[{i}].{dwd_col}: cast_type 类型无效"
|