1898 lines
75 KiB
Python
1898 lines
75 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
数据流结构分析 — 属性测试
|
||
|
||
Feature: dataflow-structure-audit
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
# scripts/ops 不是 Python 包，通过 sys.path 导入
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts" / "ops"))
|
||
|
||
from dataflow_analyzer import ( # noqa: E402
|
||
AnalyzerConfig,
|
||
ColumnInfo,
|
||
FieldInfo,
|
||
TableCollectionResult,
|
||
collect_all_tables,
|
||
dump_collection_results,
|
||
flatten_json_tree,
|
||
)
|
||
|
||
from hypothesis import given, settings, assume
|
||
import hypothesis.strategies as st
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 策略：生成任意嵌套 JSON（dict / list / 标量组合）
|
||
# ---------------------------------------------------------------------------
|
||
_scalar = st.one_of(
|
||
st.none(),
|
||
st.booleans(),
|
||
st.integers(min_value=-10_000, max_value=10_000),
|
||
st.floats(allow_nan=False, allow_infinity=False),
|
||
st.text(min_size=0, max_size=20),
|
||
)
|
||
|
||
# 限制 key 为非空 ASCII 字母数字（避免含 '.' / '[]' 等干扰路径解析的字符）
|
||
_json_key = st.from_regex(r"[A-Za-z][A-Za-z0-9_]{0,9}", fullmatch=True)
|
||
|
||
|
||
def _json_value(max_depth: int = 3):
|
||
"""递归策略：生成嵌套 JSON 值，限制深度防止爆炸。"""
|
||
if max_depth <= 0:
|
||
return _scalar
|
||
return st.one_of(
|
||
_scalar,
|
||
st.dictionaries(_json_key, st.deferred(lambda: _json_value(max_depth - 1)),
|
||
min_size=0, max_size=4),
|
||
st.lists(st.deferred(lambda: _json_value(max_depth - 1)),
|
||
min_size=0, max_size=3),
|
||
)
|
||
|
||
|
||
# 顶层必须是 dict（与 flatten_json_tree 的输入契约一致）
|
||
_json_record = st.dictionaries(_json_key, _json_value(3), min_size=0, max_size=5)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 参考实现：收集所有叶子节点路径（用于对比验证）
|
||
# ---------------------------------------------------------------------------
|
||
def _collect_leaves(obj, prefix: str = "") -> set[str]:
|
||
"""
|
||
独立于被测代码的参考叶子收集器。
|
||
返回所有标量叶子节点的路径集合。
|
||
"""
|
||
if isinstance(obj, dict):
|
||
paths: set[str] = set()
|
||
for key, val in obj.items():
|
||
child = f"{prefix}.{key}" if prefix else key
|
||
paths |= _collect_leaves(val, child)
|
||
return paths
|
||
elif isinstance(obj, list):
|
||
arr = f"{prefix}[]" if prefix else "[]"
|
||
paths = set()
|
||
for item in obj:
|
||
paths |= _collect_leaves(item, arr)
|
||
return paths
|
||
else:
|
||
# 标量叶子 — 空前缀的裸标量不会被记录（与实现一致）
|
||
return {prefix} if prefix else set()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 1: JSON 递归展开路径正确性
|
||
# Validates: Requirements 1.1, 1.3, 1.4
|
||
# ---------------------------------------------------------------------------
|
||
@given(record=_json_record)
|
||
@settings(max_examples=200)
|
||
def test_flatten_json_tree_path_correctness(record: dict):
|
||
"""
|
||
Feature: dataflow-structure-audit, Property 1: JSON 递归展开路径正确性
|
||
|
||
对于任意嵌套 JSON 对象，flatten_json_tree 的输出应满足：
|
||
1. 路径使用 '.' 分隔层级
|
||
2. 数组层级使用 '[]' 标记
|
||
3. depth == 路径去掉 '[]' 后 '.' 的数量
|
||
4. 不遗漏任何叶子节点
|
||
|
||
**Validates: Requirements 1.1, 1.3, 1.4**
|
||
"""
|
||
result = flatten_json_tree([record])
|
||
|
||
# --- 参考叶子集合 ---
|
||
expected_leaves = _collect_leaves(record)
|
||
|
||
# --- 属性 4：不遗漏叶子节点 ---
|
||
actual_paths = set(result.keys())
|
||
assert actual_paths == expected_leaves, (
|
||
f"叶子节点不一致。\n"
|
||
f" 遗漏: {expected_leaves - actual_paths}\n"
|
||
f" 多余: {actual_paths - expected_leaves}"
|
||
)
|
||
|
||
for path, fi in result.items():
|
||
# --- 属性 1：路径使用 '.' 分隔层级 ---
|
||
# 路径中不应出现连续的 '.'，且不以 '.' 开头或结尾
|
||
assert not path.startswith("."), f"路径不应以 '.' 开头: {path}"
|
||
stripped = path.replace("[]", "")
|
||
assert not stripped.endswith("."), f"路径不应以 '.' 结尾: {path}"
|
||
assert ".." not in stripped, f"路径不应含连续 '..': {path}"
|
||
|
||
# --- 属性 2：数组标记格式 ---
|
||
# '[]' 标记合法位置：紧跟在 key 之后或另一个 '[]' 之后
|
||
# 合法模式：key[]、key[][].child（嵌套数组）、key[].child
|
||
if "[]" in path:
|
||
# 路径不应以 '[]' 开头（第一个 '[]' 前必须有 key）
|
||
assert not path.startswith("[]"), (
|
||
f"路径不应以 '[]' 开头: {path}"
|
||
)
|
||
|
||
# --- 属性 3：depth 一致性 ---
|
||
expected_depth = stripped.count(".")
|
||
assert fi.depth == expected_depth, (
|
||
f"depth 不一致: path={path}, "
|
||
f"expected={expected_depth}, actual={fi.depth}"
|
||
)
|
||
|
||
# --- 基本字段完整性 ---
|
||
assert fi.path == path
|
||
assert fi.occurrence >= 1
|
||
assert fi.total_records == 1 # 单条记录
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 2: 多记录字段合并完整性与出现频率准确性
|
||
# Validates: Requirements 1.2, 1.5
|
||
# ---------------------------------------------------------------------------
|
||
@given(records=st.lists(_json_record, min_size=1, max_size=5))
|
||
@settings(max_examples=200)
|
||
def test_flatten_json_tree_field_merge_completeness(records: list[dict]):
|
||
"""
|
||
Feature: dataflow-structure-audit, Property 2: 多记录字段合并完整性与出现频率准确性
|
||
|
||
对于任意一组 JSON 记录列表，flatten_json_tree 的输出应满足：
|
||
1. 合并后的字段路径集合 == 所有单条记录字段路径集合的并集
|
||
2. 每个字段的 occurrence == 实际包含该字段的记录数
|
||
3. 每个字段的 occurrence <= total_records
|
||
4. total_records == len(records)
|
||
|
||
**Validates: Requirements 1.2, 1.5**
|
||
"""
|
||
result = flatten_json_tree(records)
|
||
|
||
# --- 参考计算：每条记录的叶子路径集合 ---
|
||
per_record_leaves: list[set[str]] = [
|
||
_collect_leaves(rec) for rec in records
|
||
]
|
||
expected_union = set().union(*per_record_leaves) if per_record_leaves else set()
|
||
|
||
# --- 属性 1：并集完整性 ---
|
||
actual_paths = set(result.keys())
|
||
assert actual_paths == expected_union, (
|
||
f"字段路径并集不一致。\n"
|
||
f" 遗漏: {expected_union - actual_paths}\n"
|
||
f" 多余: {actual_paths - expected_union}"
|
||
)
|
||
|
||
# --- 属性 2：occurrence 准确性 ---
|
||
for path, fi in result.items():
|
||
expected_occ = sum(1 for leaves in per_record_leaves if path in leaves)
|
||
assert fi.occurrence == expected_occ, (
|
||
f"occurrence 不一致: path={path}, "
|
||
f"expected={expected_occ}, actual={fi.occurrence}"
|
||
)
|
||
|
||
# --- 属性 3：occurrence <= total_records ---
|
||
assert fi.occurrence <= fi.total_records, (
|
||
f"occurrence 超过 total_records: path={path}, "
|
||
f"occurrence={fi.occurrence}, total_records={fi.total_records}"
|
||
)
|
||
|
||
# --- 属性 4：total_records 一致性 ---
|
||
for fi in result.values():
|
||
assert fi.total_records == len(records), (
|
||
f"total_records 不一致: expected={len(records)}, "
|
||
f"actual={fi.total_records}"
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 单元测试：flatten_json_tree 边界情况
|
||
# Requirements: 1.1, 1.2, 1.3, 1.4
|
||
# ---------------------------------------------------------------------------
|
||
from collections import OrderedDict
|
||
|
||
|
||
def test_empty_records():
|
||
"""空记录列表返回空 OrderedDict。
|
||
|
||
Requirements: 1.2 边界
|
||
"""
|
||
result = flatten_json_tree([])
|
||
assert isinstance(result, OrderedDict)
|
||
assert len(result) == 0
|
||
|
||
|
||
def test_flat_json():
|
||
"""单层 JSON（无嵌套），所有字段 depth=0。
|
||
|
||
Requirements: 1.1
|
||
"""
|
||
record = {"name": "Alice", "age": 30, "active": True}
|
||
result = flatten_json_tree([record])
|
||
|
||
assert set(result.keys()) == {"name", "age", "active"}
|
||
for path, fi in result.items():
|
||
assert fi.depth == 0, f"顶层字段 {path} 的 depth 应为 0，实际为 {fi.depth}"
|
||
assert "." not in fi.path, f"顶层字段路径不应含 '.': {fi.path}"
|
||
assert fi.occurrence == 1
|
||
assert fi.total_records == 1
|
||
|
||
assert result["name"].json_type == "string"
|
||
assert result["name"].sample == "Alice"
|
||
assert result["age"].json_type == "integer"
|
||
assert result["age"].sample == "30"
|
||
assert result["active"].json_type == "boolean"
|
||
|
||
|
||
def test_deep_nesting():
|
||
"""深层嵌套（5+ 层），验证路径和 depth 正确。
|
||
|
||
Requirements: 1.1, 1.4
|
||
"""
|
||
# 构造 6 层嵌套：a.b.c.d.e.f = "leaf"
|
||
record = {
|
||
"a": {
|
||
"b": {
|
||
"c": {
|
||
"d": {
|
||
"e": {
|
||
"f": "leaf"
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
result = flatten_json_tree([record])
|
||
|
||
assert "a.b.c.d.e.f" in result
|
||
fi = result["a.b.c.d.e.f"]
|
||
assert fi.depth == 5, f"6 层嵌套叶子 depth 应为 5，实际为 {fi.depth}"
|
||
assert fi.json_type == "string"
|
||
assert fi.sample == "leaf"
|
||
assert fi.occurrence == 1
|
||
assert fi.total_records == 1
|
||
# 确保中间 dict 节点不出现在结果中（只记录叶子）
|
||
assert len(result) == 1
|
||
|
||
|
||
def test_array_with_nested_objects():
|
||
"""数组内嵌套对象，验证 [] 标记和路径格式。
|
||
|
||
Requirements: 1.3
|
||
"""
|
||
record = {
|
||
"items": [
|
||
{"name": "apple", "price": 3.5},
|
||
{"name": "banana", "tags": ["fruit", "yellow"]},
|
||
]
|
||
}
|
||
result = flatten_json_tree([record])
|
||
|
||
# items[].name 和 items[].price 应存在
|
||
assert "items[].name" in result
|
||
assert "items[].price" in result
|
||
|
||
# items[].name: depth = "items.name".count(".") = 1
|
||
fi_name = result["items[].name"]
|
||
assert fi_name.depth == 1
|
||
assert fi_name.json_type == "string"
|
||
|
||
fi_price = result["items[].price"]
|
||
assert fi_price.depth == 1
|
||
assert fi_price.json_type == "number"
|
||
|
||
# 嵌套数组内的标量：items[].tags[] → 叶子路径
|
||
assert "items[].tags[]" in result
|
||
fi_tags = result["items[].tags[]"]
|
||
assert fi_tags.depth == 1 # "items.tags".count(".") = 1
|
||
assert fi_tags.json_type == "string"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 单元测试：collect_all_tables 编排逻辑
|
||
# Requirements: 2.1, 2.2, 2.3, 2.4
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
# 测试用 specs（精简版，只保留 collect_all_tables 需要的字段）
|
||
_TEST_SPECS = [
|
||
{
|
||
"code": "ODS_TABLE_A",
|
||
"table": "table_a",
|
||
"endpoint": "/api/table_a",
|
||
"description": "测试表 A",
|
||
},
|
||
{
|
||
"code": "ODS_TABLE_B",
|
||
"table": "table_b",
|
||
"endpoint": "/api/table_b",
|
||
"description": "测试表 B",
|
||
},
|
||
{
|
||
"code": "ODS_TABLE_C",
|
||
"table": "table_c",
|
||
"endpoint": "/api/table_c",
|
||
"description": "测试表 C",
|
||
},
|
||
]
|
||
|
||
|
||
def _fake_fetch(spec: dict, limit: int) -> list[dict]:
|
||
"""返回简单的假数据，用于验证编排逻辑。"""
|
||
return [{"id": 1, "name": "test", "value": 42}]
|
||
|
||
|
||
def test_collect_all_tables_happy_path():
|
||
"""全部表采集成功，无数据库连接。
|
||
|
||
Requirements: 2.1, 2.2
|
||
"""
|
||
config = AnalyzerConfig(limit=10)
|
||
results = collect_all_tables(config, _TEST_SPECS, fetch_fn=_fake_fetch)
|
||
|
||
assert len(results) == 3
|
||
for r in results:
|
||
assert isinstance(r, TableCollectionResult)
|
||
assert r.error is None
|
||
assert r.record_count == 1
|
||
assert len(r.json_fields) > 0
|
||
# 无 DB 连接时 ODS/DWD 列为空
|
||
assert r.ods_columns == []
|
||
assert r.dwd_columns == []
|
||
|
||
|
||
def test_collect_all_tables_filter_by_name():
|
||
"""config.tables 过滤只采集指定表。
|
||
|
||
Requirements: 2.2
|
||
"""
|
||
config = AnalyzerConfig(tables=["table_b"])
|
||
results = collect_all_tables(config, _TEST_SPECS, fetch_fn=_fake_fetch)
|
||
|
||
assert len(results) == 1
|
||
assert results[0].table_name == "table_b"
|
||
assert results[0].task_code == "ODS_TABLE_B"
|
||
|
||
|
||
def test_collect_all_tables_filter_case_insensitive():
|
||
"""表名过滤不区分大小写。
|
||
|
||
Requirements: 2.2
|
||
"""
|
||
config = AnalyzerConfig(tables=["TABLE_A", "Table_C"])
|
||
results = collect_all_tables(config, _TEST_SPECS, fetch_fn=_fake_fetch)
|
||
|
||
assert len(results) == 2
|
||
names = {r.table_name for r in results}
|
||
assert names == {"table_a", "table_c"}
|
||
|
||
|
||
def test_collect_all_tables_filter_no_match():
|
||
"""过滤条件无匹配时返回空列表。
|
||
|
||
Requirements: 2.2
|
||
"""
|
||
config = AnalyzerConfig(tables=["nonexistent"])
|
||
results = collect_all_tables(config, _TEST_SPECS, fetch_fn=_fake_fetch)
|
||
|
||
assert results == []
|
||
|
||
|
||
def test_collect_all_tables_single_table_error_isolation():
|
||
"""单表 fetch 失败不中断其余表的采集。
|
||
|
||
Requirements: 2.4
|
||
"""
|
||
call_count = 0
|
||
|
||
def _failing_fetch(spec: dict, limit: int) -> list[dict]:
|
||
nonlocal call_count
|
||
call_count += 1
|
||
if spec["table"] == "table_b":
|
||
raise RuntimeError("模拟 API 超时")
|
||
return [{"id": 1}]
|
||
|
||
config = AnalyzerConfig(limit=5)
|
||
results = collect_all_tables(config, _TEST_SPECS, fetch_fn=_failing_fetch)
|
||
|
||
assert len(results) == 3
|
||
assert call_count == 3
|
||
|
||
# table_a 成功
|
||
assert results[0].table_name == "table_a"
|
||
assert results[0].error is None
|
||
assert results[0].record_count == 1
|
||
|
||
# table_b 失败但有记录
|
||
assert results[1].table_name == "table_b"
|
||
assert results[1].error is not None
|
||
assert "模拟 API 超时" in results[1].error
|
||
assert results[1].record_count == 0
|
||
|
||
# table_c 成功（未被 table_b 的失败影响）
|
||
assert results[2].table_name == "table_c"
|
||
assert results[2].error is None
|
||
assert results[2].record_count == 1
|
||
|
||
|
||
def test_collect_all_tables_empty_specs():
|
||
"""空 specs 列表返回空结果。
|
||
|
||
Requirements: 2.2
|
||
"""
|
||
config = AnalyzerConfig()
|
||
results = collect_all_tables(config, [], fetch_fn=_fake_fetch)
|
||
|
||
assert results == []
|
||
|
||
|
||
def test_collect_all_tables_preserves_metadata():
|
||
"""采集结果保留 spec 中的元数据字段。
|
||
|
||
Requirements: 2.2
|
||
"""
|
||
config = AnalyzerConfig(limit=50)
|
||
results = collect_all_tables(config, _TEST_SPECS[:1], fetch_fn=_fake_fetch)
|
||
|
||
r = results[0]
|
||
assert r.table_name == "table_a"
|
||
assert r.task_code == "ODS_TABLE_A"
|
||
assert r.description == "测试表 A"
|
||
assert r.endpoint == "/api/table_a"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# dump_collection_results 单元测试
|
||
# Requirements: 5.4, 5.6
|
||
# ---------------------------------------------------------------------------
|
||
import json as _json
|
||
|
||
|
||
def _make_result(
|
||
table: str = "test_table",
|
||
task_code: str = "ODS_TEST",
|
||
record_count: int = 2,
|
||
error: str | None = None,
|
||
) -> TableCollectionResult:
|
||
"""构造一个最小化的 TableCollectionResult 用于测试。"""
|
||
fields = OrderedDict()
|
||
if not error:
|
||
fields["id"] = FieldInfo(
|
||
path="id", json_type="integer", sample="1",
|
||
depth=0, occurrence=2, total_records=record_count,
|
||
samples=["1", "2"],
|
||
)
|
||
fields["name"] = FieldInfo(
|
||
path="name", json_type="string", sample="测试",
|
||
depth=0, occurrence=1, total_records=record_count,
|
||
samples=["测试"],
|
||
)
|
||
ods_cols = [
|
||
ColumnInfo(name="id", data_type="bigint", is_nullable=False,
|
||
column_default=None, comment="主键", ordinal_position=1),
|
||
] if not error else []
|
||
dwd_cols_list = [
|
||
ColumnInfo(name="id", data_type="bigint", is_nullable=False,
|
||
column_default=None, comment="主键", ordinal_position=1),
|
||
ColumnInfo(name="store_id", data_type="integer", is_nullable=False,
|
||
column_default=None, comment="门店", ordinal_position=2),
|
||
] if not error else []
|
||
# dwd_tables: 模拟一张 DWD 表（表名 = "dwd_{table}"）
|
||
dwd_tables_dict: dict[str, list[ColumnInfo]] = {}
|
||
if not error and dwd_cols_list:
|
||
dwd_tables_dict[f"dwd_{table}"] = dwd_cols_list
|
||
return TableCollectionResult(
|
||
table_name=table,
|
||
task_code=task_code,
|
||
description="测试表",
|
||
endpoint="/api/test",
|
||
record_count=record_count,
|
||
json_fields=fields,
|
||
ods_columns=ods_cols,
|
||
dwd_columns=dwd_cols_list,
|
||
dwd_tables=dwd_tables_dict,
|
||
error=error,
|
||
)
|
||
|
||
|
||
def test_dump_creates_subdirectories(tmp_path):
|
||
"""dump_collection_results 自动创建 json_trees/、db_schemas/、field_mappings/ 和 bd_descriptions/ 子目录。
|
||
|
||
Requirements: 5.4
|
||
"""
|
||
results = [_make_result()]
|
||
paths = dump_collection_results(results, tmp_path)
|
||
|
||
assert (tmp_path / "json_trees").is_dir()
|
||
assert (tmp_path / "db_schemas").is_dir()
|
||
assert (tmp_path / "field_mappings").is_dir()
|
||
assert (tmp_path / "bd_descriptions").is_dir()
|
||
assert paths["json_trees"] == tmp_path / "json_trees"
|
||
assert paths["db_schemas"] == tmp_path / "db_schemas"
|
||
assert paths["field_mappings"] == tmp_path / "field_mappings"
|
||
assert paths["bd_descriptions"] == tmp_path / "bd_descriptions"
|
||
assert paths["manifest"] == tmp_path
|
||
|
||
|
||
def test_dump_json_trees_format(tmp_path):
|
||
"""json_trees/{table}.json 包含正确的字段结构（含 samples）。
|
||
|
||
Requirements: 5.4
|
||
"""
|
||
results = [_make_result(table="orders", record_count=10)]
|
||
dump_collection_results(results, tmp_path)
|
||
|
||
tree_file = tmp_path / "json_trees" / "orders.json"
|
||
assert tree_file.exists()
|
||
data = _json.loads(tree_file.read_text(encoding="utf-8"))
|
||
|
||
assert data["table"] == "orders"
|
||
assert data["total_records"] == 10
|
||
assert isinstance(data["fields"], list)
|
||
assert len(data["fields"]) == 2
|
||
# 验证字段结构
|
||
f0 = data["fields"][0]
|
||
assert f0["path"] == "id"
|
||
assert f0["json_type"] == "integer"
|
||
assert f0["depth"] == 0
|
||
# 验证 samples 字段存在
|
||
assert "samples" in f0
|
||
assert isinstance(f0["samples"], list)
|
||
|
||
|
||
def test_dump_db_schemas_format(tmp_path):
|
||
"""db_schemas/ods_{table}.json 和 dwd_{dwd_short}.json 包含正确的列结构。
|
||
|
||
Requirements: 5.4
|
||
"""
|
||
results = [_make_result(table="members")]
|
||
dump_collection_results(results, tmp_path)
|
||
|
||
ods_file = tmp_path / "db_schemas" / "ods_members.json"
|
||
# DWD 文件名现在使用 DWD 表名（dwd_members），不再使用 ODS 表名
|
||
dwd_file = tmp_path / "db_schemas" / "dwd_dwd_members.json"
|
||
assert ods_file.exists()
|
||
assert dwd_file.exists()
|
||
|
||
ods = _json.loads(ods_file.read_text(encoding="utf-8"))
|
||
assert ods["schema"] == "ods"
|
||
assert ods["table"] == "members"
|
||
assert len(ods["columns"]) == 1
|
||
assert ods["columns"][0]["name"] == "id"
|
||
|
||
dwd = _json.loads(dwd_file.read_text(encoding="utf-8"))
|
||
assert dwd["schema"] == "dwd"
|
||
assert dwd["table"] == "dwd_members"
|
||
assert dwd["ods_source"] == "members"
|
||
assert len(dwd["columns"]) == 2
|
||
|
||
|
||
def test_dump_manifest_format(tmp_path):
|
||
"""collection_manifest.json 包含时间戳和表清单。
|
||
|
||
Requirements: 5.6
|
||
"""
|
||
results = [
|
||
_make_result(table="t1", task_code="ODS_T1", record_count=100),
|
||
_make_result(table="t2", task_code="ODS_T2", record_count=0, error="连接失败"),
|
||
]
|
||
dump_collection_results(results, tmp_path)
|
||
|
||
manifest_file = tmp_path / "collection_manifest.json"
|
||
assert manifest_file.exists()
|
||
manifest = _json.loads(manifest_file.read_text(encoding="utf-8"))
|
||
|
||
assert "timestamp" in manifest
|
||
assert isinstance(manifest["tables"], list)
|
||
assert len(manifest["tables"]) == 2
|
||
|
||
t1 = manifest["tables"][0]
|
||
assert t1["table"] == "t1"
|
||
assert t1["task_code"] == "ODS_T1"
|
||
assert t1["record_count"] == 100
|
||
assert t1["json_field_count"] == 2 # id + name
|
||
assert t1["ods_column_count"] == 1
|
||
assert t1["dwd_column_count"] == 2
|
||
assert t1["dwd_tables"] == ["dwd_t1"]
|
||
assert t1["error"] is None
|
||
|
||
t2 = manifest["tables"][1]
|
||
assert t2["error"] == "连接失败"
|
||
assert t2["record_count"] == 0
|
||
|
||
|
||
def test_dump_utf8_chinese_content(tmp_path):
|
||
"""JSON 文件使用 UTF-8 编码，中文字符不被转义。
|
||
|
||
Requirements: 5.4
|
||
"""
|
||
results = [_make_result()]
|
||
dump_collection_results(results, tmp_path)
|
||
|
||
tree_file = tmp_path / "json_trees" / "test_table.json"
|
||
raw = tree_file.read_text(encoding="utf-8")
|
||
# ensure_ascii=False 意味着中文直接写入，不会出现 \uXXXX
|
||
assert "测试" in raw
|
||
assert "\\u" not in raw
|
||
|
||
|
||
def test_dump_empty_results(tmp_path):
|
||
"""空结果列表仍生成 manifest 和子目录。
|
||
|
||
Requirements: 5.4
|
||
"""
|
||
paths = dump_collection_results([], tmp_path)
|
||
|
||
assert (tmp_path / "json_trees").is_dir()
|
||
assert (tmp_path / "db_schemas").is_dir()
|
||
assert (tmp_path / "field_mappings").is_dir()
|
||
assert (tmp_path / "bd_descriptions").is_dir()
|
||
manifest = _json.loads(
|
||
(tmp_path / "collection_manifest.json").read_text(encoding="utf-8")
|
||
)
|
||
assert manifest["tables"] == []
|
||
assert "timestamp" in manifest
|
||
assert "table_map" in manifest
|
||
assert paths["json_trees"] == tmp_path / "json_trees"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CLI 入口：build_parser / resolve_output_dir 单元测试
|
||
# ---------------------------------------------------------------------------
|
||
from analyze_dataflow import build_parser, resolve_output_dir
|
||
|
||
|
||
class TestBuildParser:
|
||
"""build_parser 参数解析测试。Requirements: 5.1, 5.2, 5.3"""
|
||
|
||
def test_defaults(self):
|
||
"""无参数时使用默认值。"""
|
||
parser = build_parser()
|
||
args = parser.parse_args([])
|
||
assert args.date_from is None
|
||
assert args.date_to is None
|
||
assert args.limit == 200
|
||
assert args.tables is None
|
||
|
||
def test_date_range(self):
|
||
"""--date-from / --date-to 正确解析。"""
|
||
parser = build_parser()
|
||
args = parser.parse_args(["--date-from", "2025-01-01", "--date-to", "2025-01-15"])
|
||
assert args.date_from == "2025-01-01"
|
||
assert args.date_to == "2025-01-15"
|
||
|
||
def test_limit(self):
|
||
"""--limit 解析为 int。"""
|
||
parser = build_parser()
|
||
args = parser.parse_args(["--limit", "50"])
|
||
assert args.limit == 50
|
||
|
||
def test_tables(self):
|
||
"""--tables 解析为逗号分隔字符串。"""
|
||
parser = build_parser()
|
||
args = parser.parse_args(["--tables", "settlement_records,payment_transactions"])
|
||
assert args.tables == "settlement_records,payment_transactions"
|
||
# 调用方可通过 split(',') 拆分为列表
|
||
assert args.tables.split(",") == ["settlement_records", "payment_transactions"]
|
||
|
||
def test_all_params_combined(self):
|
||
"""所有参数同时传入。"""
|
||
parser = build_parser()
|
||
args = parser.parse_args([
|
||
"--date-from", "2025-06-01",
|
||
"--date-to", "2025-06-30",
|
||
"--limit", "100",
|
||
"--tables", "orders,members",
|
||
])
|
||
assert args.date_from == "2025-06-01"
|
||
assert args.date_to == "2025-06-30"
|
||
assert args.limit == 100
|
||
assert args.tables == "orders,members"
|
||
|
||
|
||
class TestResolveOutputDir:
|
||
"""resolve_output_dir 输出目录回退测试。Requirements: 5.4, 5.5"""
|
||
|
||
def test_env_var_takes_priority(self, tmp_path, monkeypatch):
|
||
"""SYSTEM_ANALYZE_ROOT 存在时优先使用。"""
|
||
target = tmp_path / "custom_output"
|
||
monkeypatch.setenv("SYSTEM_ANALYZE_ROOT", str(target))
|
||
result = resolve_output_dir()
|
||
assert result == target
|
||
assert target.is_dir()
|
||
|
||
def test_fallback_raises_when_env_missing(self, monkeypatch):
|
||
"""SYSTEM_ANALYZE_ROOT 未设置时抛出 KeyError。"""
|
||
monkeypatch.delenv("SYSTEM_ANALYZE_ROOT", raising=False)
|
||
with pytest.raises(KeyError):
|
||
resolve_output_dir()
|
||
|
||
def test_creates_directory(self, tmp_path, monkeypatch):
|
||
"""目录不存在时自动创建。"""
|
||
target = tmp_path / "nested" / "deep" / "dir"
|
||
monkeypatch.setenv("SYSTEM_ANALYZE_ROOT", str(target))
|
||
result = resolve_output_dir()
|
||
assert result.is_dir()
|
||
|
||
def test_existing_directory_no_error(self, tmp_path, monkeypatch):
|
||
"""目录已存在时不报错。"""
|
||
target = tmp_path / "already_exists"
|
||
target.mkdir()
|
||
monkeypatch.setenv("SYSTEM_ANALYZE_ROOT", str(target))
|
||
result = resolve_output_dir()
|
||
assert result == target
|
||
assert target.is_dir()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Property 3: 输出文件名格式正确性
|
||
# Feature: dataflow-structure-audit, Property 3: 输出文件名格式正确性
|
||
# Validates: Requirements 5.6
|
||
# ---------------------------------------------------------------------------
|
||
import re
|
||
from datetime import datetime
|
||
|
||
from analyze_dataflow import generate_output_filename
|
||
|
||
|
||
@settings(max_examples=100)
|
||
@given(dt=st.datetimes(
|
||
min_value=datetime(1970, 1, 1),
|
||
max_value=datetime(9999, 12, 31, 23, 59, 59),
|
||
))
|
||
def test_output_filename_format(dt: datetime):
|
||
"""
|
||
Property 3: 输出文件名格式正确性
|
||
|
||
对于任意合法 datetime，generate_output_filename 生成的文件名应：
|
||
1. 匹配模式 dataflow_YYYY-MM-DD_HHMMSS.md
|
||
2. 文件名中的日期时间与输入 datetime 一致
|
||
|
||
Validates: Requirements 5.6
|
||
"""
|
||
filename = generate_output_filename(dt)
|
||
|
||
# 1. 格式匹配
|
||
pattern = r"^dataflow_\d{4}-\d{2}-\d{2}_\d{6}\.md$"
|
||
assert re.match(pattern, filename), (
|
||
f"文件名 '{filename}' 不匹配模式 {pattern}"
|
||
)
|
||
|
||
# 2. 日期时间一致性：从文件名中提取并与输入比对
|
||
match = re.match(
|
||
r"^dataflow_(\d{4})-(\d{2})-(\d{2})_(\d{2})(\d{2})(\d{2})\.md$",
|
||
filename,
|
||
)
|
||
assert match, f"无法从文件名 '{filename}' 中提取日期时间"
|
||
|
||
year, month, day, hour, minute, second = (int(g) for g in match.groups())
|
||
assert year == dt.year
|
||
assert month == dt.month
|
||
assert day == dt.day
|
||
assert hour == dt.hour
|
||
assert minute == dt.minute
|
||
assert second == dt.second
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 单元测试：Hook 配置文件格式验证
|
||
# Validates: Requirements 6.1
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestHookConfigFile:
|
||
"""验证 .kiro/hooks/dataflow-analyze.kiro.hook 配置文件格式正确性"""
|
||
|
||
HOOK_PATH = Path(__file__).resolve().parent.parent / ".kiro" / "hooks" / "dataflow-analyze.kiro.hook"
|
||
|
||
def test_hook_file_is_valid_json(self):
|
||
"""Hook 文件必须是合法 JSON"""
|
||
text = self.HOOK_PATH.read_text(encoding="utf-8")
|
||
data = json.loads(text) # 若非法 JSON 则抛 JSONDecodeError
|
||
assert isinstance(data, dict)
|
||
|
||
def test_hook_required_fields_exist(self):
|
||
"""Hook 文件必须包含 name、version、when.type、then.type、then.prompt"""
|
||
data = json.loads(self.HOOK_PATH.read_text(encoding="utf-8"))
|
||
assert "name" in data
|
||
assert "version" in data
|
||
assert "when" in data and "type" in data["when"]
|
||
assert "then" in data and "type" in data["then"]
|
||
assert "prompt" in data["then"]
|
||
|
||
def test_hook_when_type_is_user_triggered(self):
|
||
"""when.type 必须为 userTriggered"""
|
||
data = json.loads(self.HOOK_PATH.read_text(encoding="utf-8"))
|
||
assert data["when"]["type"] == "userTriggered"
|
||
|
||
def test_hook_then_type_is_ask_agent(self):
|
||
"""then.type 必须为 askAgent"""
|
||
data = json.loads(self.HOOK_PATH.read_text(encoding="utf-8"))
|
||
assert data["then"]["type"] == "askAgent"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 新增测试：parse_table_map / parse_fact_mappings / build_field_mappings
|
||
# Validates: 三层字段映射构建正确性
|
||
# ---------------------------------------------------------------------------
|
||
from dataflow_analyzer import (
|
||
build_field_mappings,
|
||
parse_table_map,
|
||
parse_fact_mappings,
|
||
)
|
||
|
||
|
||
class TestParseTableMap:
|
||
"""从 ETL 源码解析 TABLE_MAP。"""
|
||
|
||
def test_parse_real_file(self):
|
||
"""解析真实的 dwd_load_task.py，应返回非空字典。"""
|
||
result = parse_table_map()
|
||
# 真实文件存在时应有映射
|
||
if Path("apps/etl/connectors/feiqiu/tasks/dwd/dwd_load_task.py").exists():
|
||
assert len(result) > 0
|
||
# 所有 key 应以 "dwd." 开头，所有 value 应以 "ods." 开头
|
||
for dwd_t, ods_t in result.items():
|
||
assert dwd_t.startswith("dwd."), f"DWD 表名应以 dwd. 开头: {dwd_t}"
|
||
assert ods_t.startswith("ods."), f"ODS 表名应以 ods. 开头: {ods_t}"
|
||
|
||
def test_parse_nonexistent_file(self, tmp_path):
|
||
"""文件不存在时返回空字典。"""
|
||
result = parse_table_map(tmp_path / "nonexistent.py")
|
||
assert result == {}
|
||
|
||
def test_parse_file_without_table_map(self, tmp_path):
|
||
"""文件存在但无 TABLE_MAP 定义时返回空字典。"""
|
||
py_file = tmp_path / "empty.py"
|
||
py_file.write_text("# 空文件\n", encoding="utf-8")
|
||
result = parse_table_map(py_file)
|
||
assert result == {}
|
||
|
||
def test_parse_synthetic_table_map(self, tmp_path):
|
||
"""解析合成的 TABLE_MAP 定义。"""
|
||
py_file = tmp_path / "task.py"
|
||
py_file.write_text(
|
||
'TABLE_MAP: dict[str, str] = {\n'
|
||
' "dwd.dim_foo": "ods.foo_master",\n'
|
||
' "dwd.dwd_bar": "ods.bar_records",\n'
|
||
'}\n',
|
||
encoding="utf-8",
|
||
)
|
||
result = parse_table_map(py_file)
|
||
assert result == {
|
||
"dwd.dim_foo": "ods.foo_master",
|
||
"dwd.dwd_bar": "ods.bar_records",
|
||
}
|
||
|
||
|
||
class TestParseFactMappings:
|
||
"""从 ETL 源码解析 FACT_MAPPINGS。"""
|
||
|
||
def test_parse_real_file(self):
|
||
"""解析真实的 dwd_load_task.py，应返回非空字典。"""
|
||
result = parse_fact_mappings()
|
||
if Path("apps/etl/connectors/feiqiu/tasks/dwd/dwd_load_task.py").exists():
|
||
assert len(result) > 0
|
||
for dwd_t, mappings in result.items():
|
||
assert dwd_t.startswith("dwd."), f"DWD 表名应以 dwd. 开头: {dwd_t}"
|
||
for entry in mappings:
|
||
assert len(entry) == 3, f"映射条目应为 3 元组: {entry}"
|
||
|
||
def test_parse_nonexistent_file(self, tmp_path):
|
||
"""文件不存在时返回空字典。"""
|
||
result = parse_fact_mappings(tmp_path / "nonexistent.py")
|
||
assert result == {}
|
||
|
||
|
||
class TestBuildFieldMappings:
|
||
"""build_field_mappings 三层映射构建。"""
|
||
|
||
def _make_test_data(self):
|
||
"""构造测试用的 TableCollectionResult + TABLE_MAP + FACT_MAPPINGS。"""
|
||
fields = OrderedDict()
|
||
fields["id"] = FieldInfo(
|
||
path="id", json_type="integer", sample="123",
|
||
depth=0, occurrence=10, total_records=10,
|
||
samples=["123", "456", "789"],
|
||
)
|
||
fields["name"] = FieldInfo(
|
||
path="name", json_type="string", sample="测试",
|
||
depth=0, occurrence=10, total_records=10,
|
||
samples=["测试", "示例"],
|
||
)
|
||
fields["extra.nested"] = FieldInfo(
|
||
path="extra.nested", json_type="string", sample="嵌套值",
|
||
depth=1, occurrence=5, total_records=10,
|
||
samples=["嵌套值"],
|
||
)
|
||
|
||
ods_cols = [
|
||
ColumnInfo(name="id", data_type="BIGINT", is_nullable=False,
|
||
column_default=None, comment="主键", ordinal_position=1),
|
||
ColumnInfo(name="name", data_type="TEXT", is_nullable=True,
|
||
column_default=None, comment="名称", ordinal_position=2),
|
||
]
|
||
|
||
dwd_cols_main = [
|
||
ColumnInfo(name="foo_id", data_type="BIGINT", is_nullable=False,
|
||
column_default=None, comment="主键", ordinal_position=1),
|
||
ColumnInfo(name="name", data_type="TEXT", is_nullable=True,
|
||
column_default=None, comment="名称", ordinal_position=2),
|
||
ColumnInfo(name="scd2_start_time", data_type="TIMESTAMPTZ", is_nullable=True,
|
||
column_default=None, comment=None, ordinal_position=3),
|
||
]
|
||
|
||
result = TableCollectionResult(
|
||
table_name="foo_master",
|
||
task_code="ODS_FOO",
|
||
description="测试表",
|
||
endpoint="/api/foo",
|
||
record_count=10,
|
||
json_fields=fields,
|
||
ods_columns=ods_cols,
|
||
dwd_columns=dwd_cols_main,
|
||
dwd_tables={"dim_foo": dwd_cols_main},
|
||
)
|
||
|
||
table_map = {"dwd.dim_foo": "ods.foo_master"}
|
||
fact_mappings = {
|
||
"dwd.dim_foo": [("foo_id", "id", None)],
|
||
}
|
||
all_dwd_cols = {"dim_foo": dwd_cols_main}
|
||
|
||
return result, table_map, fact_mappings, all_dwd_cols
|
||
|
||
def test_anchors_generated(self):
|
||
"""锚点 ID 格式正确。"""
|
||
result, tm, fm, dwd = self._make_test_data()
|
||
mapping = build_field_mappings(result, tm, fm, dwd)
|
||
|
||
assert mapping["anchors"]["api"] == "api-foo-master"
|
||
assert mapping["anchors"]["ods"] == "ods-foo-master"
|
||
assert "dim_foo" in mapping["anchors"]["dwd"]
|
||
assert mapping["anchors"]["dwd"]["dim_foo"] == "dwd-dim-foo"
|
||
|
||
def test_json_to_ods_mapping(self):
|
||
"""JSON→ODS 映射：已映射和未映射字段正确标注。"""
|
||
result, tm, fm, dwd = self._make_test_data()
|
||
mapping = build_field_mappings(result, tm, fm, dwd)
|
||
|
||
j2o = mapping["json_to_ods"]
|
||
assert len(j2o) == 3
|
||
|
||
# id → 已映射
|
||
id_entry = next(e for e in j2o if e["json_path"] == "id")
|
||
assert id_entry["ods_col"] == "id"
|
||
assert id_entry["match_type"] in ("exact", "case_insensitive")
|
||
|
||
# name → 已映射
|
||
name_entry = next(e for e in j2o if e["json_path"] == "name")
|
||
assert name_entry["ods_col"] == "name"
|
||
|
||
# extra.nested → 未映射（ODS 无此列）
|
||
nested_entry = next(e for e in j2o if e["json_path"] == "extra.nested")
|
||
assert nested_entry["ods_col"] is None
|
||
assert nested_entry["match_type"] == "unmapped"
|
||
|
||
def test_ods_to_dwd_mapping(self):
|
||
"""ODS→DWD 映射：id 列映射到 dim_foo.foo_id。"""
|
||
result, tm, fm, dwd = self._make_test_data()
|
||
mapping = build_field_mappings(result, tm, fm, dwd)
|
||
|
||
o2d = mapping["ods_to_dwd"]
|
||
# id → foo_id（通过 FACT_MAPPINGS）
|
||
assert "id" in o2d
|
||
targets = o2d["id"]
|
||
assert any(t["dwd_col"] == "foo_id" and t["dwd_table"] == "dim_foo" for t in targets)
|
||
|
||
# name → name（同名直传）
|
||
assert "name" in o2d
|
||
name_targets = o2d["name"]
|
||
assert any(t["dwd_col"] == "name" and t["dwd_table"] == "dim_foo" for t in name_targets)
|
||
|
||
def test_dwd_to_ods_mapping(self):
|
||
"""DWD→ODS 映射：包含直接映射和 SCD2 列。"""
|
||
result, tm, fm, dwd = self._make_test_data()
|
||
mapping = build_field_mappings(result, tm, fm, dwd)
|
||
|
||
d2o = mapping["dwd_to_ods"]
|
||
assert "dim_foo" in d2o
|
||
cols = d2o["dim_foo"]
|
||
|
||
# foo_id ← id（FACT_MAPPINGS 显式映射）
|
||
foo_id_entry = next(c for c in cols if c["dwd_col"] == "foo_id")
|
||
assert foo_id_entry["ods_source"] == "id"
|
||
assert foo_id_entry["mapping_type"] == "直接"
|
||
|
||
# name ← name（同名直传）
|
||
name_entry = next(c for c in cols if c["dwd_col"] == "name")
|
||
assert name_entry["ods_source"] == "name"
|
||
assert name_entry["mapping_type"] == "直接"
|
||
assert name_entry["note"] == "同名直传"
|
||
|
||
# scd2_start_time（SCD2 元数据）
|
||
scd2_entry = next(c for c in cols if c["dwd_col"] == "scd2_start_time")
|
||
assert scd2_entry["mapping_type"] == "SCD2"
|
||
assert scd2_entry["ods_source"] == "—"
|
||
|
||
def test_field_mappings_dumped_to_file(self, tmp_path):
|
||
"""dump_collection_results 生成 field_mappings JSON 文件。"""
|
||
result, _, _, _ = self._make_test_data()
|
||
dump_collection_results([result], tmp_path)
|
||
|
||
fm_file = tmp_path / "field_mappings" / "foo_master.json"
|
||
assert fm_file.exists()
|
||
data = _json.loads(fm_file.read_text(encoding="utf-8"))
|
||
assert data["ods_table"] == "foo_master"
|
||
assert "anchors" in data
|
||
assert "json_to_ods" in data
|
||
assert "ods_to_dwd" in data
|
||
assert "dwd_to_ods" in data
|
||
|
||
def test_no_dwd_tables_produces_empty_mappings(self):
|
||
"""ODS 表无 DWD 映射时，ods_to_dwd 和 dwd_to_ods 为空。"""
|
||
fields = OrderedDict()
|
||
fields["id"] = FieldInfo(
|
||
path="id", json_type="integer", sample="1",
|
||
depth=0, occurrence=1, total_records=1,
|
||
samples=["1"],
|
||
)
|
||
ods_cols = [
|
||
ColumnInfo(name="id", data_type="BIGINT", is_nullable=False,
|
||
column_default=None, comment=None, ordinal_position=1),
|
||
]
|
||
result = TableCollectionResult(
|
||
table_name="orphan_table",
|
||
task_code="ODS_ORPHAN",
|
||
description="无 DWD 映射的表",
|
||
endpoint="/api/orphan",
|
||
record_count=1,
|
||
json_fields=fields,
|
||
ods_columns=ods_cols,
|
||
dwd_columns=[],
|
||
dwd_tables={},
|
||
)
|
||
mapping = build_field_mappings(result, {}, {}, {})
|
||
assert mapping["ods_to_dwd"] == {}
|
||
assert mapping["dwd_to_ods"] == {}
|
||
assert mapping["anchors"]["dwd"] == {}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 新增测试：多示例值采集（samples 字段）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_flatten_json_tree_collects_multiple_samples():
|
||
"""flatten_json_tree 为每个字段收集多个不同的示例值。"""
|
||
records = [
|
||
{"status": 1, "name": "Alice"},
|
||
{"status": 2, "name": "Bob"},
|
||
{"status": 1, "name": "Charlie"},
|
||
{"status": 3, "name": "Alice"},
|
||
]
|
||
result = flatten_json_tree(records)
|
||
|
||
# status 有 3 个不同值
|
||
assert "status" in result
|
||
fi_status = result["status"]
|
||
assert set(fi_status.samples) == {"1", "2", "3"}
|
||
|
||
# name 有 3 个不同值
|
||
fi_name = result["name"]
|
||
assert set(fi_name.samples) == {"Alice", "Bob", "Charlie"}
|
||
|
||
|
||
def test_flatten_json_tree_samples_limit():
|
||
"""samples 数量不超过 MAX_SAMPLES。"""
|
||
from dataflow_analyzer import MAX_SAMPLES
|
||
|
||
# 生成超过 MAX_SAMPLES 个不同值
|
||
records = [{"val": i} for i in range(MAX_SAMPLES + 5)]
|
||
result = flatten_json_tree(records)
|
||
|
||
fi = result["val"]
|
||
assert len(fi.samples) <= MAX_SAMPLES
|
||
|
||
|
||
def test_flatten_json_tree_samples_skip_none():
|
||
"""None 值不被收集到 samples 中。"""
|
||
records = [
|
||
{"x": None},
|
||
{"x": 42},
|
||
{"x": None},
|
||
{"x": 99},
|
||
]
|
||
result = flatten_json_tree(records)
|
||
|
||
fi = result["x"]
|
||
assert "None" not in fi.samples
|
||
assert set(fi.samples) == {"42", "99"}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 新增测试：parse_bd_manual_fields / load_bd_descriptions / dump_bd_descriptions
|
||
# ---------------------------------------------------------------------------
|
||
from dataflow_analyzer import (
|
||
parse_bd_manual_fields,
|
||
dump_bd_descriptions,
|
||
)
|
||
|
||
|
||
class TestParseBdManualFields:
|
||
"""从 BD_manual Markdown 文档解析字段说明。"""
|
||
|
||
def test_parse_standard_format(self, tmp_path):
|
||
"""标准 BD_manual 格式解析。"""
|
||
doc = tmp_path / "BD_manual_test.md"
|
||
doc.write_text(
|
||
"# test 测试表\n\n"
|
||
"## 字段说明\n\n"
|
||
"| 序号 | 字段名 | 类型 | 可空 | 说明 |\n"
|
||
"|------|--------|------|------|------|\n"
|
||
"| 1 | id | BIGINT | NO | 主键 ID |\n"
|
||
"| 2 | name | TEXT | YES | 名称字段 |\n"
|
||
"| 3 | status | INTEGER | YES | 状态：1=启用，0=禁用 |\n",
|
||
encoding="utf-8",
|
||
)
|
||
result = parse_bd_manual_fields(doc)
|
||
assert result["id"] == "主键 ID"
|
||
assert result["name"] == "名称字段"
|
||
assert result["status"] == "状态：1=启用，0=禁用"
|
||
|
||
def test_parse_nonexistent_file(self, tmp_path):
|
||
"""文件不存在时返回空字典。"""
|
||
result = parse_bd_manual_fields(tmp_path / "nonexistent.md")
|
||
assert result == {}
|
||
|
||
def test_parse_no_field_table(self, tmp_path):
|
||
"""文件存在但无字段说明表格时返回空字典。"""
|
||
doc = tmp_path / "empty.md"
|
||
doc.write_text("# 空文档\n\n没有字段说明。\n", encoding="utf-8")
|
||
result = parse_bd_manual_fields(doc)
|
||
assert result == {}
|
||
|
||
def test_parse_dwd_format_with_pk_column(self, tmp_path):
|
||
"""DWD 格式（含主键列）也能正确解析。"""
|
||
doc = tmp_path / "BD_manual_dim_test.md"
|
||
doc.write_text(
|
||
"# dim_test 测试维表\n\n"
|
||
"## 字段说明\n\n"
|
||
"| 序号 | 字段名 | 类型 | 可空 | 主键 | 说明 |\n"
|
||
"|------|--------|------|------|------|------|\n"
|
||
"| 1 | test_id | BIGINT | NO | PK | 唯一标识 |\n"
|
||
"| 2 | label | TEXT | YES | | 标签名 |\n",
|
||
encoding="utf-8",
|
||
)
|
||
result = parse_bd_manual_fields(doc)
|
||
assert result["test_id"] == "唯一标识"
|
||
assert result["label"] == "标签名"
|
||
|
||
def test_parse_real_ods_doc(self):
|
||
"""解析真实的 ODS BD_manual 文档（如果存在）。"""
|
||
doc = Path("apps/etl/connectors/feiqiu/docs/database/ODS/main/BD_manual_assistant_accounts_master.md")
|
||
if doc.exists():
|
||
result = parse_bd_manual_fields(doc)
|
||
assert len(result) > 0
|
||
# 应包含 id 字段
|
||
assert "id" in result
|
||
|
||
def test_parse_real_dwd_doc(self):
|
||
"""解析真实的 DWD BD_manual 文档（如果存在）。"""
|
||
doc = Path("apps/etl/connectors/feiqiu/docs/database/DWD/main/BD_manual_dim_assistant.md")
|
||
if doc.exists():
|
||
result = parse_bd_manual_fields(doc)
|
||
assert len(result) > 0
|
||
assert "assistant_id" in result
|
||
|
||
|
||
class TestDumpBdDescriptions:
|
||
"""dump_bd_descriptions 输出 BD 描述 JSON。"""
|
||
|
||
def test_dump_creates_files(self, tmp_path):
|
||
"""为每张表生成 bd_descriptions JSON 文件。"""
|
||
results = [_make_result(table="test_table")]
|
||
dump_bd_descriptions(results, tmp_path)
|
||
|
||
bd_file = tmp_path / "bd_descriptions" / "test_table.json"
|
||
assert bd_file.exists()
|
||
data = _json.loads(bd_file.read_text(encoding="utf-8"))
|
||
assert data["ods_table"] == "test_table"
|
||
assert "ods_fields" in data
|
||
assert "dwd_fields" in data
|
||
|
||
def test_dump_empty_results(self, tmp_path):
|
||
"""空结果列表不报错。"""
|
||
dump_bd_descriptions([], tmp_path)
|
||
assert (tmp_path / "bd_descriptions").is_dir()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 新增测试：gen_dataflow_report.py 报告生成器
|
||
# ---------------------------------------------------------------------------
|
||
from gen_dataflow_report import generate_report, _esc, _format_samples, _is_enum_like
|
||
|
||
|
||
class TestReportHelpers:
|
||
"""报告生成器辅助函数测试。"""
|
||
|
||
def test_esc_pipe(self):
|
||
"""管道符被转义。"""
|
||
assert _esc("a|b") == "a\\|b"
|
||
|
||
def test_esc_newline(self):
|
||
"""换行符被替换为空格。"""
|
||
assert _esc("a\nb") == "a b"
|
||
|
||
def test_esc_none(self):
|
||
"""空字符串返回空。"""
|
||
assert _esc("") == ""
|
||
|
||
def test_format_samples_basic(self):
|
||
"""基本多示例格式化。"""
|
||
result = _format_samples(["a", "b", "c"])
|
||
assert "`a`" in result
|
||
assert "`b`" in result
|
||
assert "`c`" in result
|
||
|
||
def test_format_samples_truncate(self):
|
||
"""超长示例值被截断。"""
|
||
long_val = "x" * 50
|
||
result = _format_samples([long_val])
|
||
assert "..." in result
|
||
|
||
def test_format_samples_empty(self):
|
||
"""空列表返回空字符串。"""
|
||
assert _format_samples([]) == ""
|
||
|
||
def test_is_enum_like_true(self):
|
||
"""少量不同值 + 足够记录数 → 枚举。"""
|
||
assert _is_enum_like(["1", "2", "3"], 100) is True
|
||
|
||
def test_is_enum_like_false_too_many(self):
|
||
"""超过 8 个不同值 → 非枚举。"""
|
||
assert _is_enum_like([str(i) for i in range(10)], 100) is False
|
||
|
||
def test_is_enum_like_false_single(self):
|
||
"""只有 1 个值 → 非枚举。"""
|
||
assert _is_enum_like(["1"], 100) is False
|
||
|
||
def test_is_enum_like_false_few_records(self):
|
||
"""记录数太少 → 非枚举。"""
|
||
assert _is_enum_like(["1", "2"], 3) is False
|
||
|
||
|
||
class TestGenerateReport:
|
||
"""generate_report 集成测试。"""
|
||
|
||
def _setup_data_dir(self, tmp_path):
|
||
"""构造最小化的数据目录用于报告生成。"""
|
||
# collection_manifest.json
|
||
manifest = {
|
||
"timestamp": "2026-02-16T22:00:00+08:00",
|
||
"table_map": {"dwd.dwd_test": "ods.test_table"},
|
||
"tables": [{
|
||
"table": "test_table",
|
||
"task_code": "ODS_TEST",
|
||
"description": "测试表",
|
||
"record_count": 10,
|
||
"json_field_count": 2,
|
||
"ods_column_count": 2,
|
||
"dwd_tables": ["dwd_test"],
|
||
"dwd_column_count": 3,
|
||
"error": None,
|
||
}],
|
||
}
|
||
(tmp_path / "collection_manifest.json").write_text(
|
||
_json.dumps(manifest, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
|
||
# json_trees/test_table.json
|
||
jt_dir = tmp_path / "json_trees"
|
||
jt_dir.mkdir()
|
||
jt = {
|
||
"table": "test_table", "total_records": 10,
|
||
"fields": [
|
||
{"path": "id", "json_type": "integer", "sample": "1",
|
||
"depth": 0, "occurrence": 10, "total_records": 10,
|
||
"samples": ["1", "2", "3"]},
|
||
{"path": "status", "json_type": "integer", "sample": "1",
|
||
"depth": 0, "occurrence": 10, "total_records": 10,
|
||
"samples": ["0", "1", "2"]},
|
||
],
|
||
}
|
||
(jt_dir / "test_table.json").write_text(
|
||
_json.dumps(jt, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
|
||
# db_schemas/
|
||
db_dir = tmp_path / "db_schemas"
|
||
db_dir.mkdir()
|
||
ods = {"schema": "ods", "table": "test_table", "columns": [
|
||
{"name": "id", "data_type": "BIGINT", "is_nullable": False,
|
||
"column_default": None, "comment": "主键", "ordinal_position": 1},
|
||
{"name": "status", "data_type": "INTEGER", "is_nullable": True,
|
||
"column_default": None, "comment": None, "ordinal_position": 2},
|
||
]}
|
||
(db_dir / "ods_test_table.json").write_text(
|
||
_json.dumps(ods, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
dwd = {"schema": "dwd", "table": "dwd_test", "ods_source": "test_table", "columns": [
|
||
{"name": "test_id", "data_type": "BIGINT", "is_nullable": False,
|
||
"column_default": None, "comment": None, "ordinal_position": 1},
|
||
{"name": "status", "data_type": "INTEGER", "is_nullable": True,
|
||
"column_default": None, "comment": None, "ordinal_position": 2},
|
||
{"name": "scd2_start_time", "data_type": "TIMESTAMPTZ", "is_nullable": True,
|
||
"column_default": None, "comment": None, "ordinal_position": 3},
|
||
]}
|
||
(db_dir / "dwd_dwd_test.json").write_text(
|
||
_json.dumps(dwd, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
|
||
# field_mappings/test_table.json
|
||
fm_dir = tmp_path / "field_mappings"
|
||
fm_dir.mkdir()
|
||
fm = {
|
||
"ods_table": "test_table",
|
||
"anchors": {
|
||
"api": "api-test-table", "ods": "ods-test-table",
|
||
"dwd": {"dwd_test": "dwd-dwd-test"},
|
||
},
|
||
"json_to_ods": [
|
||
{"json_path": "id", "ods_col": "id", "match_type": "exact",
|
||
"json_type": "integer", "occurrence_pct": 100.0},
|
||
{"json_path": "status", "ods_col": "status", "match_type": "exact",
|
||
"json_type": "integer", "occurrence_pct": 100.0},
|
||
],
|
||
"ods_to_dwd": {
|
||
"id": [{"dwd_table": "dwd_test", "dwd_col": "test_id", "cast": None, "note": "字段重命名"}],
|
||
"status": [{"dwd_table": "dwd_test", "dwd_col": "status", "cast": None, "note": "同名直传"}],
|
||
},
|
||
"dwd_to_ods": {
|
||
"dwd_test": [
|
||
{"dwd_col": "test_id", "type": "BIGINT", "ods_source": "id", "mapping_type": "直接", "note": "字段重命名"},
|
||
{"dwd_col": "status", "type": "INTEGER", "ods_source": "status", "mapping_type": "直接", "note": "同名直传"},
|
||
{"dwd_col": "scd2_start_time", "type": "TIMESTAMPTZ", "ods_source": "—", "mapping_type": "SCD2", "note": "SCD2 元数据"},
|
||
],
|
||
},
|
||
}
|
||
(fm_dir / "test_table.json").write_text(
|
||
_json.dumps(fm, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
|
||
# bd_descriptions/test_table.json
|
||
bd_dir = tmp_path / "bd_descriptions"
|
||
bd_dir.mkdir()
|
||
bd = {
|
||
"ods_table": "test_table",
|
||
"ods_fields": {"id": "主键 ID", "status": "状态：0=禁用，1=启用"},
|
||
"dwd_fields": {"dwd_test": {"test_id": "唯一标识", "status": "状态"}},
|
||
}
|
||
(bd_dir / "test_table.json").write_text(
|
||
_json.dumps(bd, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
|
||
return tmp_path
|
||
|
||
def test_report_contains_json_field_count(self, tmp_path):
|
||
"""总览表包含 API JSON 字段数列。"""
|
||
data_dir = self._setup_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
assert "API JSON 字段数" in report
|
||
|
||
def test_report_contains_field_diff(self, tmp_path):
|
||
"""报告包含字段对比差异章节。"""
|
||
data_dir = self._setup_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
assert "API↔ODS↔DWD 字段对比差异" in report
|
||
|
||
def test_report_contains_biz_desc_in_coverage(self, tmp_path):
|
||
"""覆盖率表包含业务描述列。"""
|
||
data_dir = self._setup_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
# 2.3 覆盖率表应有业务描述列头
|
||
lines = report.split("\n")
|
||
coverage_header = [l for l in lines if "业务描述" in l and "JSON 字段数" in l]
|
||
assert len(coverage_header) > 0
|
||
|
||
def test_report_contains_biz_desc_in_api_section(self, tmp_path):
|
||
"""API 源字段表包含业务描述列。"""
|
||
data_dir = self._setup_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
assert "主键 ID" in report # 来自 bd_descriptions
|
||
|
||
def test_report_contains_biz_desc_in_ods_section(self, tmp_path):
|
||
"""ODS 表结构包含业务描述列。"""
|
||
data_dir = self._setup_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
# ODS 表头应有业务描述
|
||
assert "| # | ODS 列名 | 类型 | ← JSON 源 | → DWD 目标 | 业务描述 |" in report
|
||
|
||
def test_report_contains_biz_desc_in_dwd_section(self, tmp_path):
|
||
"""DWD 表结构包含业务描述列。"""
|
||
data_dir = self._setup_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
assert "| # | DWD 列名 | 类型 | ← ODS 来源 | 转换 | 业务描述 |" in report
|
||
|
||
def test_report_contains_enum_samples(self, tmp_path):
|
||
"""枚举字段展示枚举值列表。"""
|
||
data_dir = self._setup_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
# status 有 3 个不同值，应被识别为枚举
|
||
assert "枚举值" in report
|
||
|
||
def test_report_anchors_present(self, tmp_path):
|
||
"""报告包含锚点标签。"""
|
||
data_dir = self._setup_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
assert '<a id="api-test-table"></a>' in report
|
||
assert '<a id="ods-test-table"></a>' in report
|
||
assert '<a id="dwd-dwd-test"></a>' in report
|
||
|
||
|
||
class TestFieldDiffSubTables:
|
||
"""测试 1.1 字段差异分表的生成逻辑。"""
|
||
|
||
def _setup_diff_data_dir(self, tmp_path):
|
||
"""构造包含多种差异类型的测试数据。"""
|
||
import json as _json
|
||
|
||
manifest = {
|
||
"timestamp": "2026-02-17T22:00:00+08:00",
|
||
"table_map": {"dwd.dwd_alpha": "ods.alpha_table"},
|
||
"tables": [{
|
||
"table": "alpha_table",
|
||
"task_code": "ODS_ALPHA",
|
||
"description": "差异测试表",
|
||
"record_count": 50,
|
||
"json_field_count": 6,
|
||
"ods_column_count": 5,
|
||
"dwd_tables": ["dwd_alpha"],
|
||
"dwd_column_count": 5,
|
||
"error": None,
|
||
}],
|
||
}
|
||
(tmp_path / "collection_manifest.json").write_text(
|
||
_json.dumps(manifest, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
|
||
# json_trees
|
||
jt_dir = tmp_path / "json_trees"
|
||
jt_dir.mkdir()
|
||
jt = {
|
||
"table": "alpha_table", "total_records": 50,
|
||
"fields": [
|
||
{"path": "id", "json_type": "integer", "sample": "1",
|
||
"depth": 0, "occurrence": 50, "total_records": 50, "samples": ["1"]},
|
||
{"path": "name", "json_type": "string", "sample": "foo",
|
||
"depth": 0, "occurrence": 50, "total_records": 50, "samples": ["foo"]},
|
||
{"path": "extra_flat", "json_type": "string", "sample": "x",
|
||
"depth": 0, "occurrence": 50, "total_records": 50, "samples": ["x"]},
|
||
{"path": "nested.a", "json_type": "string", "sample": "na",
|
||
"depth": 1, "occurrence": 50, "total_records": 50, "samples": ["na"]},
|
||
{"path": "nested.b", "json_type": "string", "sample": "nb",
|
||
"depth": 1, "occurrence": 50, "total_records": 50, "samples": ["nb"]},
|
||
{"path": "status", "json_type": "integer", "sample": "1",
|
||
"depth": 0, "occurrence": 50, "total_records": 50, "samples": ["0", "1"]},
|
||
],
|
||
}
|
||
(jt_dir / "alpha_table.json").write_text(
|
||
_json.dumps(jt, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
|
||
# db_schemas
|
||
db_dir = tmp_path / "db_schemas"
|
||
db_dir.mkdir()
|
||
ods = {"schema": "ods", "table": "alpha_table", "columns": [
|
||
{"name": "id", "data_type": "BIGINT", "is_nullable": False,
|
||
"column_default": None, "comment": None, "ordinal_position": 1},
|
||
{"name": "name", "data_type": "TEXT", "is_nullable": True,
|
||
"column_default": None, "comment": None, "ordinal_position": 2},
|
||
{"name": "status", "data_type": "INTEGER", "is_nullable": True,
|
||
"column_default": None, "comment": None, "ordinal_position": 3},
|
||
{"name": "ods_only_col", "data_type": "TEXT", "is_nullable": True,
|
||
"column_default": None, "comment": None, "ordinal_position": 4},
|
||
{"name": "no_dwd_col", "data_type": "TEXT", "is_nullable": True,
|
||
"column_default": None, "comment": None, "ordinal_position": 5},
|
||
]}
|
||
(db_dir / "ods_alpha_table.json").write_text(
|
||
_json.dumps(ods, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
dwd = {"schema": "dwd", "table": "dwd_alpha", "ods_source": "alpha_table", "columns": [
|
||
{"name": "alpha_id", "data_type": "BIGINT", "is_nullable": False,
|
||
"column_default": None, "comment": None, "ordinal_position": 1},
|
||
{"name": "name", "data_type": "TEXT", "is_nullable": True,
|
||
"column_default": None, "comment": None, "ordinal_position": 2},
|
||
{"name": "status", "data_type": "INTEGER", "is_nullable": True,
|
||
"column_default": None, "comment": None, "ordinal_position": 3},
|
||
{"name": "scd2_ver", "data_type": "INTEGER", "is_nullable": True,
|
||
"column_default": None, "comment": None, "ordinal_position": 4},
|
||
{"name": "derived_flag", "data_type": "BOOLEAN", "is_nullable": True,
|
||
"column_default": None, "comment": None, "ordinal_position": 5},
|
||
]}
|
||
(db_dir / "dwd_dwd_alpha.json").write_text(
|
||
_json.dumps(dwd, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
|
||
# field_mappings — 构造各种差异
|
||
fm_dir = tmp_path / "field_mappings"
|
||
fm_dir.mkdir()
|
||
fm = {
|
||
"ods_table": "alpha_table",
|
||
"anchors": {
|
||
"api": "api-alpha-table", "ods": "ods-alpha-table",
|
||
"dwd": {"dwd_alpha": "dwd-dwd-alpha"},
|
||
},
|
||
"json_to_ods": [
|
||
{"json_path": "id", "ods_col": "id", "match_type": "exact",
|
||
"json_type": "integer", "occurrence_pct": 100.0},
|
||
{"json_path": "name", "ods_col": "name", "match_type": "exact",
|
||
"json_type": "string", "occurrence_pct": 100.0},
|
||
{"json_path": "status", "ods_col": "status", "match_type": "exact",
|
||
"json_type": "integer", "occurrence_pct": 100.0},
|
||
# 平层未映射
|
||
{"json_path": "extra_flat", "ods_col": None, "match_type": "unmapped",
|
||
"json_type": "string", "occurrence_pct": 100.0},
|
||
# 嵌套未映射
|
||
{"json_path": "nested.a", "ods_col": None, "match_type": "unmapped",
|
||
"json_type": "string", "occurrence_pct": 100.0},
|
||
{"json_path": "nested.b", "ods_col": None, "match_type": "unmapped",
|
||
"json_type": "string", "occurrence_pct": 100.0},
|
||
],
|
||
"ods_to_dwd": {
|
||
"id": [{"dwd_table": "dwd_alpha", "dwd_col": "alpha_id", "cast": None, "note": "重命名"}],
|
||
"name": [{"dwd_table": "dwd_alpha", "dwd_col": "name", "cast": None, "note": ""}],
|
||
"status": [{"dwd_table": "dwd_alpha", "dwd_col": "status", "cast": None, "note": ""}],
|
||
# ods_only_col 和 no_dwd_col 不在此 → ODS→DWD 未映射
|
||
},
|
||
"dwd_to_ods": {
|
||
"dwd_alpha": [
|
||
{"dwd_col": "alpha_id", "type": "BIGINT", "ods_source": "id",
|
||
"mapping_type": "直接", "note": "重命名"},
|
||
{"dwd_col": "name", "type": "TEXT", "ods_source": "name",
|
||
"mapping_type": "直接", "note": ""},
|
||
{"dwd_col": "status", "type": "INTEGER", "ods_source": "status",
|
||
"mapping_type": "直接", "note": ""},
|
||
# DWD 无 ODS 源
|
||
{"dwd_col": "scd2_ver", "type": "INTEGER", "ods_source": "—",
|
||
"mapping_type": "SCD2", "note": "SCD2 元数据"},
|
||
{"dwd_col": "derived_flag", "type": "BOOLEAN", "ods_source": "—",
|
||
"mapping_type": "派生", "note": "派生列"},
|
||
],
|
||
},
|
||
}
|
||
(fm_dir / "alpha_table.json").write_text(
|
||
_json.dumps(fm, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
|
||
# bd_descriptions
|
||
bd_dir = tmp_path / "bd_descriptions"
|
||
bd_dir.mkdir()
|
||
bd = {
|
||
"ods_table": "alpha_table",
|
||
"ods_fields": {"id": "主键", "name": "名称"},
|
||
"dwd_fields": {"dwd_alpha": {"alpha_id": "标识", "name": "名称"}},
|
||
}
|
||
(bd_dir / "alpha_table.json").write_text(
|
||
_json.dumps(bd, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
|
||
return tmp_path
|
||
|
||
def test_summary_table_has_links(self, tmp_path):
|
||
"""汇总表中非零差异数应包含跳转链接。"""
|
||
data_dir = self._setup_diff_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
# API→ODS 未映射 = 3（1 平层 + 2 嵌套），应有链接
|
||
assert "[3](#diff-alpha-table)" in report
|
||
|
||
def test_summary_table_zero_no_link(self, tmp_path):
|
||
"""汇总表中差异数为 0 时不应有链接。"""
|
||
import json as _json
|
||
data_dir = self._setup_diff_data_dir(tmp_path)
|
||
# 添加一个全部映射的表，使其所有差异列为 0
|
||
manifest = _json.loads((data_dir / "collection_manifest.json").read_text(encoding="utf-8"))
|
||
manifest["tables"].append({
|
||
"table": "zero_table", "task_code": "ODS_ZERO", "description": "零差异",
|
||
"record_count": 5, "json_field_count": 1, "ods_column_count": 1,
|
||
"dwd_tables": ["dwd_zero"], "dwd_column_count": 1, "error": None,
|
||
})
|
||
(data_dir / "collection_manifest.json").write_text(
|
||
_json.dumps(manifest, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
fm_z = {
|
||
"ods_table": "zero_table",
|
||
"anchors": {"api": "api-zero-table", "ods": "ods-zero-table",
|
||
"dwd": {"dwd_zero": "dwd-dwd-zero"}},
|
||
"json_to_ods": [{"json_path": "id", "ods_col": "id", "match_type": "exact",
|
||
"json_type": "integer", "occurrence_pct": 100.0}],
|
||
"ods_to_dwd": {"id": [{"dwd_table": "dwd_zero", "dwd_col": "id", "cast": None, "note": ""}]},
|
||
"dwd_to_ods": {"dwd_zero": [{"dwd_col": "id", "type": "BIGINT", "ods_source": "id",
|
||
"mapping_type": "直接", "note": ""}]},
|
||
}
|
||
(data_dir / "field_mappings" / "zero_table.json").write_text(
|
||
_json.dumps(fm_z, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
ods_z = {"schema": "ods", "table": "zero_table", "columns": [
|
||
{"name": "id", "data_type": "BIGINT", "is_nullable": False,
|
||
"column_default": None, "comment": None, "ordinal_position": 1}]}
|
||
(data_dir / "db_schemas" / "ods_zero_table.json").write_text(
|
||
_json.dumps(ods_z, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
dwd_z = {"schema": "dwd", "table": "dwd_zero", "ods_source": "zero_table", "columns": [
|
||
{"name": "id", "data_type": "BIGINT", "is_nullable": False,
|
||
"column_default": None, "comment": None, "ordinal_position": 1}]}
|
||
(data_dir / "db_schemas" / "dwd_dwd_zero.json").write_text(
|
||
_json.dumps(dwd_z, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
jt_z = {"table": "zero_table", "total_records": 5, "fields": [
|
||
{"path": "id", "json_type": "integer", "sample": "1",
|
||
"depth": 0, "occurrence": 5, "total_records": 5, "samples": ["1"]}]}
|
||
(data_dir / "json_trees" / "zero_table.json").write_text(
|
||
_json.dumps(jt_z, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
bd_z = {"ods_table": "zero_table", "ods_fields": {}, "dwd_fields": {}}
|
||
(data_dir / "bd_descriptions" / "zero_table.json").write_text(
|
||
_json.dumps(bd_z, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
report = generate_report(data_dir)
|
||
# zero_table 行应有 "| 0 |" 且不含链接
|
||
zero_line = [l for l in report.split("\n") if "zero_table" in l and "| 0 |" in l]
|
||
assert len(zero_line) > 0
|
||
assert "#diff-zero-table" not in report
|
||
|
||
def test_diff_anchor_present(self, tmp_path):
|
||
"""分表应包含锚点标签。"""
|
||
data_dir = self._setup_diff_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
assert '<a id="diff-alpha-table"></a>' in report
|
||
|
||
def test_diff_subtable_title(self, tmp_path):
|
||
"""分表应有递增编号标题。"""
|
||
data_dir = self._setup_diff_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
assert "#### 1.1.1 alpha_table 字段差异明细" in report
|
||
# 不应出现 1.1.x
|
||
assert "1.1.x" not in report
|
||
|
||
def test_api_unmapped_flat_bold(self, tmp_path):
|
||
"""平层未映射字段行应加粗。"""
|
||
data_dir = self._setup_diff_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
assert "**[`extra_flat`](#api-alpha-table)**" in report
|
||
assert "**⚠️ 未映射**" in report
|
||
|
||
def test_api_unmapped_nested_in_details(self, tmp_path):
|
||
"""嵌套未映射字段应在 <details> 折叠块中。"""
|
||
data_dir = self._setup_diff_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
assert "<details><summary>API→ODS 未映射（嵌套对象）" in report
|
||
assert "nested.a" in report
|
||
assert "nested.b" in report
|
||
assert "</details>" in report
|
||
|
||
def test_ods_no_dwd_fields_bold(self, tmp_path):
|
||
"""ODS→DWD 未映射字段行应加粗。"""
|
||
data_dir = self._setup_diff_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
# ods_only_col 和 no_dwd_col 不在 ods_to_dwd 中
|
||
assert "**[`ods_only_col`](#ods-alpha-table)**" in report
|
||
assert "**[`no_dwd_col`](#ods-alpha-table)**" in report
|
||
|
||
def test_dwd_no_ods_fields_bold(self, tmp_path):
|
||
"""DWD 无 ODS 源字段行应加粗。"""
|
||
data_dir = self._setup_diff_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
assert "**[`scd2_ver`](#dwd-dwd-alpha)**" in report
|
||
assert "**[`derived_flag`](#dwd-dwd-alpha)**" in report
|
||
assert "**⚠️ 无 ODS 源**" in report
|
||
|
||
def test_no_diff_table_skipped(self, tmp_path):
|
||
"""无差异的表不应生成分表。"""
|
||
import json as _json
|
||
data_dir = self._setup_diff_data_dir(tmp_path)
|
||
# 添加一个无差异的表
|
||
manifest = _json.loads((data_dir / "collection_manifest.json").read_text(encoding="utf-8"))
|
||
manifest["tables"].append({
|
||
"table": "clean_table",
|
||
"task_code": "ODS_CLEAN",
|
||
"description": "无差异表",
|
||
"record_count": 10,
|
||
"json_field_count": 1,
|
||
"ods_column_count": 1,
|
||
"dwd_tables": ["dwd_clean"],
|
||
"dwd_column_count": 1,
|
||
"error": None,
|
||
})
|
||
(data_dir / "collection_manifest.json").write_text(
|
||
_json.dumps(manifest, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
# 构造完全匹配的 field_mappings
|
||
fm_clean = {
|
||
"ods_table": "clean_table",
|
||
"anchors": {"api": "api-clean-table", "ods": "ods-clean-table",
|
||
"dwd": {"dwd_clean": "dwd-dwd-clean"}},
|
||
"json_to_ods": [
|
||
{"json_path": "id", "ods_col": "id", "match_type": "exact",
|
||
"json_type": "integer", "occurrence_pct": 100.0},
|
||
],
|
||
"ods_to_dwd": {"id": [{"dwd_table": "dwd_clean", "dwd_col": "id",
|
||
"cast": None, "note": ""}]},
|
||
"dwd_to_ods": {"dwd_clean": [
|
||
{"dwd_col": "id", "type": "BIGINT", "ods_source": "id",
|
||
"mapping_type": "直接", "note": ""},
|
||
]},
|
||
}
|
||
(data_dir / "field_mappings" / "clean_table.json").write_text(
|
||
_json.dumps(fm_clean, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
ods_clean = {"schema": "ods", "table": "clean_table", "columns": [
|
||
{"name": "id", "data_type": "BIGINT", "is_nullable": False,
|
||
"column_default": None, "comment": None, "ordinal_position": 1},
|
||
]}
|
||
(data_dir / "db_schemas" / "ods_clean_table.json").write_text(
|
||
_json.dumps(ods_clean, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
dwd_clean = {"schema": "dwd", "table": "dwd_clean", "ods_source": "clean_table", "columns": [
|
||
{"name": "id", "data_type": "BIGINT", "is_nullable": False,
|
||
"column_default": None, "comment": None, "ordinal_position": 1},
|
||
]}
|
||
(data_dir / "db_schemas" / "dwd_dwd_clean.json").write_text(
|
||
_json.dumps(dwd_clean, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
jt_clean = {
|
||
"table": "clean_table", "total_records": 10,
|
||
"fields": [{"path": "id", "json_type": "integer", "sample": "1",
|
||
"depth": 0, "occurrence": 10, "total_records": 10, "samples": ["1"]}],
|
||
}
|
||
(data_dir / "json_trees" / "clean_table.json").write_text(
|
||
_json.dumps(jt_clean, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
bd_clean = {"ods_table": "clean_table", "ods_fields": {}, "dwd_fields": {}}
|
||
(data_dir / "bd_descriptions" / "clean_table.json").write_text(
|
||
_json.dumps(bd_clean, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
|
||
report = generate_report(data_dir)
|
||
# clean_table 不应有差异分表
|
||
assert "clean_table 字段差异明细" not in report
|
||
# 但 alpha_table 应有
|
||
assert "alpha_table 字段差异明细" in report
|
||
|
||
def test_reason_includes_nested_and_scd2(self, tmp_path):
|
||
"""差异原因应包含嵌套对象和 SCD2 信息。"""
|
||
data_dir = self._setup_diff_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
assert "嵌套对象 2 个" in report
|
||
assert "SCD2/派生列 2 个" in report
|
||
|
||
|
||
|
||
|
||
class TestDiffSubTableColumns:
|
||
"""测试差异分表中列的输出格式（推测用途/置信度已移除，改为人工处理）。"""
|
||
|
||
def test_flat_unmapped_header(self, tmp_path):
|
||
"""平层未映射分表应包含示例值、说明列（无推测用途/置信度）。"""
|
||
from test_dataflow_analyzer import TestFieldDiffSubTables
|
||
inst = TestFieldDiffSubTables()
|
||
data_dir = inst._setup_diff_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
assert "| # | JSON 字段 | 示例值 | 说明 | 状态 |" in report
|
||
|
||
def test_dwd_no_ods_header(self, tmp_path):
|
||
"""DWD 无 ODS 源子表应包含说明列（无推测用途/置信度）。"""
|
||
from test_dataflow_analyzer import TestFieldDiffSubTables
|
||
inst = TestFieldDiffSubTables()
|
||
data_dir = inst._setup_diff_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
assert "| # | DWD 表 | DWD 列 | 说明 | 状态 |" in report
|
||
|
||
def test_section_numbering_incremental(self, tmp_path):
|
||
"""多个差异分表应有递增编号 1.1.1, 1.1.2, ...。"""
|
||
import json as _json
|
||
from test_dataflow_analyzer import TestFieldDiffSubTables
|
||
inst = TestFieldDiffSubTables()
|
||
data_dir = inst._setup_diff_data_dir(tmp_path)
|
||
manifest = _json.loads((data_dir / "collection_manifest.json").read_text(encoding="utf-8"))
|
||
manifest["tables"].append({
|
||
"table": "beta_table", "task_code": "ODS_BETA", "description": "第二表",
|
||
"record_count": 10, "json_field_count": 1, "ods_column_count": 1,
|
||
"dwd_tables": ["dwd_beta"], "dwd_column_count": 2, "error": None,
|
||
})
|
||
(data_dir / "collection_manifest.json").write_text(
|
||
_json.dumps(manifest, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
fm_b = {
|
||
"ods_table": "beta_table",
|
||
"anchors": {"api": "api-beta-table", "ods": "ods-beta-table",
|
||
"dwd": {"dwd_beta": "dwd-dwd-beta"}},
|
||
"json_to_ods": [{"json_path": "id", "ods_col": "id", "match_type": "exact",
|
||
"json_type": "integer", "occurrence_pct": 100.0}],
|
||
"ods_to_dwd": {"id": [{"dwd_table": "dwd_beta", "dwd_col": "id", "cast": None, "note": ""}]},
|
||
"dwd_to_ods": {"dwd_beta": [
|
||
{"dwd_col": "id", "type": "BIGINT", "ods_source": "id", "mapping_type": "直接", "note": ""},
|
||
{"dwd_col": "scd2_flag", "type": "INTEGER", "ods_source": "—", "mapping_type": "SCD2", "note": ""},
|
||
]},
|
||
}
|
||
(data_dir / "field_mappings" / "beta_table.json").write_text(
|
||
_json.dumps(fm_b, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
ods_b = {"schema": "ods", "table": "beta_table", "columns": [
|
||
{"name": "id", "data_type": "BIGINT", "is_nullable": False,
|
||
"column_default": None, "comment": None, "ordinal_position": 1}]}
|
||
(data_dir / "db_schemas" / "ods_beta_table.json").write_text(
|
||
_json.dumps(ods_b, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
dwd_b = {"schema": "dwd", "table": "dwd_beta", "ods_source": "beta_table", "columns": [
|
||
{"name": "id", "data_type": "BIGINT", "is_nullable": False,
|
||
"column_default": None, "comment": None, "ordinal_position": 1},
|
||
{"name": "scd2_flag", "data_type": "INTEGER", "is_nullable": True,
|
||
"column_default": None, "comment": None, "ordinal_position": 2},
|
||
]}
|
||
(data_dir / "db_schemas" / "dwd_dwd_beta.json").write_text(
|
||
_json.dumps(dwd_b, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
jt_b = {"table": "beta_table", "total_records": 10, "fields": [
|
||
{"path": "id", "json_type": "integer", "sample": "1",
|
||
"depth": 0, "occurrence": 10, "total_records": 10, "samples": ["1"]}]}
|
||
(data_dir / "json_trees" / "beta_table.json").write_text(
|
||
_json.dumps(jt_b, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
bd_b = {"ods_table": "beta_table", "ods_fields": {}, "dwd_fields": {}}
|
||
(data_dir / "bd_descriptions" / "beta_table.json").write_text(
|
||
_json.dumps(bd_b, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
|
||
report = generate_report(data_dir)
|
||
assert "#### 1.1.1 alpha_table 字段差异明细" in report
|
||
assert "#### 1.1.2 beta_table 字段差异明细" in report
|
||
|
||
def test_sample_value_in_flat_unmapped(self, tmp_path):
|
||
"""平层未映射字段应显示来自 json_trees 的示例值。"""
|
||
from test_dataflow_analyzer import TestFieldDiffSubTables
|
||
inst = TestFieldDiffSubTables()
|
||
data_dir = inst._setup_diff_data_dir(tmp_path)
|
||
report = generate_report(data_dir)
|
||
lines = report.split("\n")
|
||
flat_rows = [l for l in lines if "extra_flat" in l and "未映射" in l]
|
||
assert len(flat_rows) >= 1
|
||
assert "`x`" in flat_rows[0]
|
||
|
||
def test_bd_desc_in_ods_no_json(self, tmp_path):
|
||
"""ODS 无 JSON 源子表应显示来自 bd_descriptions 的说明。"""
|
||
import json as _json
|
||
from test_dataflow_analyzer import TestFieldDiffSubTables
|
||
inst = TestFieldDiffSubTables()
|
||
data_dir = inst._setup_diff_data_dir(tmp_path)
|
||
bd = {"ods_table": "alpha_table",
|
||
"ods_fields": {"ods_only_col": "仅ODS存在的测试列"},
|
||
"dwd_fields": {}}
|
||
(data_dir / "bd_descriptions" / "alpha_table.json").write_text(
|
||
_json.dumps(bd, ensure_ascii=False), encoding="utf-8"
|
||
)
|
||
report = generate_report(data_dir)
|
||
lines = report.split("\n")
|
||
ods_only_rows = [l for l in lines if "ods_only_col" in l and "无 JSON 源" in l]
|
||
assert len(ods_only_rows) >= 1
|
||
assert "仅ODS存在的测试列" in ods_only_rows[0]
|
||
|