feat: initialize DataTest ETL testing framework

This commit is contained in:
KISEKI 2026-08-23 00:52:32 +08:00
commit 6229c5f937
38 changed files with 7909 additions and 0 deletions

8
.gitignore vendored Normal file
View File

@ -0,0 +1,8 @@
.DS_Store
.datatest/
.datatest-backups/
.build/
__pycache__/
*.py[cod]
.pytest_cache/
.swiftpm/

163
README.md Normal file
View File

@ -0,0 +1,163 @@
# DataTest
DataTest 是一套以需求为中心的 ETL 数据测试框架。当前版本使用本地 SQLite 跑通:
```text
需求 → Metadata → 测试案例 → 确定性执行 → 断言 → 指标 → 报告
```
它同时提供:
- 原生 SwiftUI macOS 工作台
- `datatest` 命令行
- 可供 Codex 等 Agent 调用的本地 MCP Server
- 隔离调用本机 Codex CLI 的 AI 适配器
## 快速开始
当前版本不需要安装第三方 Python 包。
```bash
./bin/datatest init
./bin/datatest demo
./bin/datatest metadata REQ-CUSTOMER-001
./bin/datatest cases REQ-CUSTOMER-001
./bin/datatest run REQ-CUSTOMER-001 --batch-id 2026-08-22 --biz-date 2026-08-22
```
复杂大数据体验会创建 10 万客户、20 万账户、100 万笔增量交易、30 个业务日期分区、一个全量画像目标表和一个日增量指标目标表:
```bash
./bin/datatest complex-demo
```
该数据集故意保留一条风险评分/等级错误。运行 `REQ-RISK-002`预期只有复合风险指标一致性案例失败可继续使用界面的“Codex 调查根因”体验诊断。调试或自动化测试时可缩小规模,或通过 `--no-error` 生成完全正确的数据:
```bash
./bin/datatest complex-demo --customers 1000 --transactions 12000
./bin/datatest complex-demo --no-error
```
运行结果会输出 `run_id`。使用它查看结果、生成报告;如果某案例结果为 `FAIL``ERROR`,还可以明确调用 Codex 调查根因:
```bash
./bin/datatest result RUN-XXXXXXXXXXXX
./bin/datatest report RUN-XXXXXXXXXXXX
./bin/datatest analyze-failure RUN-XXXXXXXXXXXX CASE-003
```
本地数据默认保存在 `.datatest/`
```text
.datatest/
├── app.sqlite
├── source.sqlite
├── target.sqlite
└── artifacts/
```
## SwiftUI 客户端
构建:
```bash
swift build --package-path macos
```
运行:
```bash
swift run --package-path macos DataTestApp
```
也可以生成可双击启动的标准 macOS App
```bash
./bin/build-macos-app
open macos/.build/DataTest.app
```
客户端会调用同一套 Python 核心能力,并展示需求、案例、运行批次和历史指标。
## 导入真实需求
```bash
./bin/datatest requirement-import ./requirement.md \
--project-id PROJECT-001 \
--project-name 客户主题项目 \
--requirement-id REQ-001 \
--requirement-name 客户主题ETL需求
```
通过本机 Codex CLI 解析需求:
```bash
./bin/datatest ai-parse REQ-001
./bin/datatest requirement-confirm REQ-001
./bin/datatest ai-generate-cases REQ-001
```
AI 生成的案例状态为 `draft`,必须通过确定性校验并明确批准后才能运行:
```bash
./bin/datatest case-approve CASE-007
./bin/datatest case-approve-all REQ-001
./bin/datatest case-reject CASE-008 --comment "缺少分区条件"
```
完整状态流为:`导入需求 → Agent 只读探索数据库并解析需求 → 形成候选 Metadata →
人工确认范围 → 重新采集并锁定正式 Metadata → Codex 生成草稿 → 人工审核 → 确定性执行`。
探索阶段只读取真实表目录、DDL、字段、索引、行数和少量样例执行器始终只选择
`approved` 案例。
生成案例后可继续与 Codex 沟通调整或补充案例:
```bash
./bin/datatest case-chat REQ-001 "补充历史数据量波动和字段值分布案例"
```
沟通内容、Codex 回复和案例变更都会按需求保存。任何新增或修改案例都会重新置为
`draft`,包括原本已经批准的案例,必须再次人工审核后才能执行。
macOS 协作页面通过流式命令展示当前处理过程;也可以直接消费 JSONL 事件:
```bash
./bin/datatest case-chat-stream REQ-001 "补充历史数据量波动案例"
```
事件只包含上下文准备、Codex 活动摘要、确定性校验和草稿保存等运行阶段,不输出模型内部思维内容。
可使用 `DATATEST_CODEX_PATH` 指定 Codex CLI。默认依次检查 PATH 和:
```text
/Applications/ChatGPT.app/Contents/Resources/codex
```
内部 Codex 调用使用临时会话、`--ignore-user-config` 和只读沙箱,避免 DataTest MCP 递归调用自身。失败调查只生成结构化证据、根因判断、建议和只读验证 SQL不会修改测试数据结果及调用审计保存到 `app.sqlite`
## Codex MCP 接入
先初始化数据,然后将本地 STDIO Server 添加到 Codex。请将路径换成项目绝对路径
```bash
codex mcp add datatest -- \
/absolute/path/to/datatestool/bin/datatest \
--home /absolute/path/to/datatestool/.datatest mcp
```
当前提供的工具包括查看项目、需求、Metadata、案例调用 Codex 解析需求和生成案例,运行已审核案例,查询结果、生成报告,以及调查失败案例根因。
## 测试
```bash
PYTHONPATH=src python3 -m unittest discover -s tests -v
```
## 当前边界
- 当前只有 SQLite 数据源适配器。
- 测试 SQL 只允许 `SELECT``WITH`,数据库连接同时启用 `query_only`
- PDF、DOCX 文本提取和远程数据源尚未实现。
- Hive、Impala、Spark 的方言、分区和性能能力将在后续适配器中实现。
详细架构见 [docs/architecture.md](docs/architecture.md)。

28
bin/build-macos-app Executable file
View File

@ -0,0 +1,28 @@
#!/bin/sh
set -eu
SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
PROJECT_DIR=$(dirname "$SCRIPT_DIR")
APP_DIR="$PROJECT_DIR/macos/.build/DataTest.app"
PACKAGE_TEMP_DIR=$(mktemp -d /private/tmp/datatest-app.XXXXXX)
STAGED_APP_DIR="$PACKAGE_TEMP_DIR/DataTest.app"
trap 'rm -rf "$PACKAGE_TEMP_DIR"' EXIT
swift build --package-path "$PROJECT_DIR/macos" --disable-sandbox
"$PROJECT_DIR/bin/generate-macos-icons"
mkdir -p "$STAGED_APP_DIR/Contents/MacOS" "$STAGED_APP_DIR/Contents/Resources"
cp -X "$PROJECT_DIR/macos/.build/debug/DataTestApp" "$STAGED_APP_DIR/Contents/MacOS/DataTestApp"
chmod 755 "$STAGED_APP_DIR/Contents/MacOS/DataTestApp"
cp -X "$PROJECT_DIR/macos/App/Info.plist" "$STAGED_APP_DIR/Contents/Info.plist"
cp -X "$PROJECT_DIR/macos/.build/generated-icons/DataTestIcon.icns" "$STAGED_APP_DIR/Contents/Resources/DataTestIcon.icns"
cp -X "$PROJECT_DIR/macos/.build/generated-icons/DataTestIcon-Dark.icns" "$STAGED_APP_DIR/Contents/Resources/DataTestIcon-Dark.icns"
xattr -cr "$STAGED_APP_DIR"
codesign --force --deep --sign - "$STAGED_APP_DIR"
codesign --verify --deep --strict "$STAGED_APP_DIR"
if [ -d "$APP_DIR" ]; then
mv "$APP_DIR" "$PACKAGE_TEMP_DIR/previous-DataTest.app"
fi
ditto --noextattr --noqtn "$STAGED_APP_DIR" "$APP_DIR"
xattr -cr "$APP_DIR"
codesign --verify --deep --strict "$APP_DIR"
echo "$APP_DIR"

5
bin/datatest Executable file
View File

@ -0,0 +1,5 @@
#!/bin/sh
set -eu
SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
PROJECT_DIR=$(dirname "$SCRIPT_DIR")
PYTHONPATH="$PROJECT_DIR/src${PYTHONPATH:+:$PYTHONPATH}" exec python3 -m datatest.cli "$@"

35
bin/generate-macos-icons Executable file
View File

@ -0,0 +1,35 @@
#!/bin/sh
set -eu
SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
PROJECT_DIR=$(dirname "$SCRIPT_DIR")
ASSET_DIR="$PROJECT_DIR/macos/App/Assets"
OUTPUT_DIR="$PROJECT_DIR/macos/.build/generated-icons"
ICON_TEMP_DIR=$(mktemp -d /private/tmp/datatest-icons.XXXXXX)
trap 'rm -rf "$ICON_TEMP_DIR"' EXIT
generate_icon() {
source_png=$1
icon_name=$2
iconset_dir="$ICON_TEMP_DIR/$icon_name.iconset"
mkdir -p "$iconset_dir"
sips -z 16 16 "$source_png" --out "$iconset_dir/icon_16x16.png" >/dev/null
sips -z 32 32 "$source_png" --out "$iconset_dir/icon_16x16@2x.png" >/dev/null
sips -z 32 32 "$source_png" --out "$iconset_dir/icon_32x32.png" >/dev/null
sips -z 64 64 "$source_png" --out "$iconset_dir/icon_32x32@2x.png" >/dev/null
sips -z 128 128 "$source_png" --out "$iconset_dir/icon_128x128.png" >/dev/null
sips -z 256 256 "$source_png" --out "$iconset_dir/icon_128x128@2x.png" >/dev/null
sips -z 256 256 "$source_png" --out "$iconset_dir/icon_256x256.png" >/dev/null
sips -z 512 512 "$source_png" --out "$iconset_dir/icon_256x256@2x.png" >/dev/null
sips -z 512 512 "$source_png" --out "$iconset_dir/icon_512x512.png" >/dev/null
sips -z 1024 1024 "$source_png" --out "$iconset_dir/icon_512x512@2x.png" >/dev/null
if ! iconutil --convert icns "$iconset_dir" --output "$ICON_TEMP_DIR/$icon_name.icns" >/dev/null 2>&1; then
python3 "$SCRIPT_DIR/png-to-icns" "$iconset_dir" "$ICON_TEMP_DIR/$icon_name.icns"
fi
ditto --noextattr --noqtn "$ICON_TEMP_DIR/$icon_name.icns" "$OUTPUT_DIR/$icon_name.icns"
}
mkdir -p "$OUTPUT_DIR"
generate_icon "$ASSET_DIR/DataTestIcon-Light.png" "DataTestIcon"
generate_icon "$ASSET_DIR/DataTestIcon-Dark.png" "DataTestIcon-Dark"

33
bin/png-to-icns Executable file
View File

@ -0,0 +1,33 @@
#!/usr/bin/env python3
from __future__ import annotations
import struct
import sys
from pathlib import Path
def main() -> int:
if len(sys.argv) != 3:
raise SystemExit("usage: png-to-icns ICONSET_DIR OUTPUT.icns")
iconset = Path(sys.argv[1])
output = Path(sys.argv[2])
resources = [
(b"icp4", "icon_16x16.png"),
(b"icp5", "icon_32x32.png"),
(b"icp6", "icon_32x32@2x.png"),
(b"ic07", "icon_128x128.png"),
(b"ic08", "icon_256x256.png"),
(b"ic09", "icon_512x512.png"),
(b"ic10", "icon_512x512@2x.png"),
]
blocks: list[bytes] = []
for resource_type, filename in resources:
payload = (iconset / filename).read_bytes()
blocks.append(resource_type + struct.pack(">I", len(payload) + 8) + payload)
body = b"".join(blocks)
output.write_bytes(b"icns" + struct.pack(">I", len(body) + 8) + body)
return 0
if __name__ == "__main__":
raise SystemExit(main())

53
docs/architecture.md Normal file
View File

@ -0,0 +1,53 @@
# DataTest 架构
## 设计原则
1. 需求是聚合根文档、Metadata、案例、结果、指标和报告都绑定需求版本。
2. 必须先获取真实 Metadata再生成和校验测试案例。
3. AI 只生成结构化草稿,确定性引擎负责安全校验、执行和断言。
4. SwiftUI、CLI 和 MCP 共用同一个核心服务,避免行为分叉。
5. 数据源、SQL 方言、指标采集器和断言均采用可替换边界。
## 组件
```text
SwiftUI ─┐
CLI ─────┼→ DataTestService → Storage
MCP ─────┘ │ → SQLiteDataSource
├──────────→ DeterministicExecutor
└──────────→ CodexCLIAdapter
```
### Storage
`app.sqlite` 保存需求版本、ETL 任务、Metadata 快照、案例版本、运行结果、结构化失败分析、指标和 Agent 调用审计。原始文档以及报告作为 artifact 保存,数据库记录路径和哈希。
### Data source adapter
当前 `SQLiteDataSource``source.sqlite``target.sqlite` 分别附加为 `ods``dwd`。连接启用 `query_only`Metadata 查询封装在适配器内部。
### Case validation
执行前必须同时通过:案例名称、表字段存在性、只读 SQL、断言完整性以及审核状态校验。Agent 不能向运行工具提交临时 SQL。
### AI adapter
Codex CLI 使用 JSON Schema 输出结构化结果。内部调用使用临时会话、不加载用户配置,并启用只读沙箱,避免继承 DataTest MCP 后递归调用。失败分析绑定具体运行结果、需求版本和同版本 MetadataAgent 只提供调查结论,确定性执行器仍然独占测试状态判定。
### MCP
MCP Server 使用 STDIO JSON-RPC。只读工具带有只读标记运行、AI 生成和报告生成属于非破坏性写操作。
## 扩展路径
后续数据源适配器至少需要实现:
- Metadata 获取
- 方言和标识符规则
- 只读会话
- SQL 执行与取消
- 查询超时
- 查询计划与性能指标
- 分区、批次及数据新鲜度指标
Hive、Impala 和 Spark 专属能力不在 SQLite 中模拟,未实现的能力必须明确返回 `UNSUPPORTED`

View File

@ -0,0 +1,15 @@
# 客户主题 ETL 加工需求
任务名称:客户主题明细加工。
源表为 `ods.ods_customer`,目标表为 `dwd.dwd_customer_info`
字段映射:
- `customer_id` 映射到 `cust_id`,必须非空且唯一。
- `customer_name` 映射到 `cust_name`,去除首尾空格。
- `status` 映射到 `cust_status``1` 转换为 `ACTIVE``0` 转换为 `INACTIVE`
- `age` 映射到 `age`,合法范围为 0 到 120。
- `updated_at` 映射到 `updated_at`
仅装载 `is_deleted = 0` 的有效客户。目标表数据量应与有效源数据量一致。每天按业务日期执行,历史数据量环比波动不得超过 30%。

View File

@ -0,0 +1,114 @@
# 客户交易风险画像与日增量指标加工需求 V2
## 1. 需求目标
建设客户交易风险主题数据集,将客户、账户、风险标签、汇率和交易流水进行多表关联,产出:
1. 客户风险画像全量表 `dwd.dwd_customer_risk_profile_full`
2. 客户日交易风险增量表 `dwd.dws_customer_trade_risk_di`
测试批次日期为 `2026-08-22`,交易统计窗口覆盖 `2026-07-24``2026-08-22` 共 30 个业务日期。
## 2. 输入表及装载类型
| 库表 | 装载类型 | 主键/分区 | 用途 |
|---|---|---|---|
| `ods.ods_customer_master_full` | 每日全量 | `cust_id` | 客户主数据 |
| `ods.ods_account_full` | 每日全量 | `account_id` | 客户账户及余额 |
| `ods.ods_risk_tag_full` | 每日全量 | `risk_tag_id` | 风险标签、基础分和权重 |
| `ods.ods_fx_rate_full` | 每日全量 | `currency_code, rate_date` | 交易金额折算人民币 |
| `ods.ods_transaction_inc` | 日增量 | `biz_date`,流水主键 `txn_id` | 30 日交易流水 |
仅客户状态为 `ACTIVE`、账户状态为 `ACTIVE` 的数据参与风险指标加工。失败交易计入交易笔数和失败率,但不计入交易金额、均值、最大值、跨境笔数和大额笔数。
## 3. 全量客户风险画像
目标表:`dwd.dwd_customer_risk_profile_full`,每日按 `etl_batch_date` 全量覆盖。
基础字段:
- `cust_id``ods_customer_master_full.cust_id`,非空、唯一;
- `cust_name``customer_name`
- `customer_type``customer_type`
- `region_code``region_code`
- `total_account_count`:客户全部账户数;
- `active_account_count`:客户有效账户数;
- `total_balance`:有效账户余额求和,保留 2 位小数。
V2 新增字段:
- `risk_tag_code`:关联 `risk_tag_id` 获取;
- `txn_count_30d`30 日交易总笔数;
- `txn_amount_cny_30d`:成功交易按交易日汇率折算人民币后求和,保留 2 位;
- `avg_txn_amount_cny_30d`:人民币交易金额除以成功交易数,保留 2 位;
- `cross_border_ratio_30d`:成功跨境交易数除以成功交易数,保留 6 位;
- `large_txn_count_30d`:成功交易折算人民币后单笔金额大于等于 50,000 元的笔数;
- `failed_txn_ratio_30d`:失败交易数除以全部交易数,保留 6 位;
- `risk_score`:复合风险分,范围 0100保留 2 位;
- `risk_level`:风险等级;
- `data_quality_flag`:无有效账户时为 `NO_ACTIVE_ACCOUNT`,否则为 `OK`
- `profile_version`:固定为 `2`
- `etl_batch_date`:当前批次日期。
全量画像风险分公式:
```text
min(100,
base_score × score_weight
+ cross_border_ratio_30d × 25
+ large_txn_count_30d × 0.4
+ failed_txn_ratio_30d × 15
+ if txn_amount_cny_30d >= 1,000,000 then 5 else 0
)
```
## 4. 日增量交易风险指标
目标表:`dwd.dws_customer_trade_risk_di`,按 `biz_date` 日增量写入,以 `cust_id, biz_date` 为联合主键。重复调度同一业务日期时必须覆盖该分区,不允许追加重复记录。
指标字段:
- `txn_count`:当日交易总笔数;
- `successful_txn_count`:成功交易笔数;
- `failed_txn_count`:失败交易笔数;
- `txn_amount_cny`:成功交易人民币金额合计;
- `avg_txn_amount_cny`:成功交易人民币平均金额;
- `max_txn_amount_cny`:成功交易人民币最大金额;
- `cross_border_count`:成功跨境交易笔数;
- `cross_border_ratio`:成功跨境交易笔数除以成功交易笔数;
- `large_txn_count`:成功交易中人民币金额大于等于 50,000 元的笔数;
- `source_max_update_seq`:该客户当日参与加工的源流水最大更新序号;
- `etl_batch_time`:批次完成时间。
日增量风险分公式:
```text
min(100,
base_score × score_weight
+ cross_border_ratio × 25
+ large_txn_count × 2
+ failed_txn_count / txn_count × 20
+ if txn_amount_cny >= 500,000 then 10 else 0
)
```
风险等级统一按复合风险分映射:
- `risk_score >= 80``HIGH`
- `50 <= risk_score < 80``MEDIUM`
- `risk_score < 50``LOW`
## 5. 数据质量与核对要求
1. 全量画像记录数必须等于有效客户记录数;
2. 所有主键及联合主键必须唯一;
3. 新增字段必须存在且按规则非空;
4. 源交易汇率换算金额与目标汇总金额误差不得超过 0.01
5. `cross_border_ratio`、`failed_txn_ratio_30d` 必须在 01
6. 风险分和风险等级必须严格符合公式;
7. 增量目标必须覆盖 30 个业务日期,源目标有效交易总笔数必须一致;
8. Metadata、案例、运行结果、失败样例、Agent 调查结论和测试报告均须绑定本需求版本保存。
## 6. 演示验收说明
SQLite 演示数据会在日增量目标表中故意写入一条错误的 `risk_score/risk_level` 组合。预期只有“复合风险评分与等级计算一致性校验”失败,用于体验失败证据和 Codex 根因调查;框架不得把该错误自动修复为通过。

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.5 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.2 MiB

30
macos/App/Info.plist Normal file
View File

@ -0,0 +1,30 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
<key>CFBundleDevelopmentRegion</key>
<string>zh_CN</string>
<key>CFBundleDisplayName</key>
<string>DataTest</string>
<key>CFBundleExecutable</key>
<string>DataTestApp</string>
<key>CFBundleIdentifier</key>
<string>local.datatest.app</string>
<key>CFBundleInfoDictionaryVersion</key>
<string>6.0</string>
<key>CFBundleName</key>
<string>DataTest</string>
<key>CFBundleIconFile</key>
<string>DataTestIcon</string>
<key>CFBundlePackageType</key>
<string>APPL</string>
<key>CFBundleShortVersionString</key>
<string>0.1.0</string>
<key>CFBundleVersion</key>
<string>1</string>
<key>LSMinimumSystemVersion</key>
<string>14.0</string>
<key>NSHighResolutionCapable</key>
<true/>
</dict>
</plist>

13
macos/Package.swift Normal file
View File

@ -0,0 +1,13 @@
// swift-tools-version: 6.0
import PackageDescription
let package = Package(
name: "DataTestMac",
platforms: [.macOS(.v14)],
products: [
.executable(name: "DataTestApp", targets: ["DataTestApp"])
],
targets: [
.executableTarget(name: "DataTestApp")
]
)

View File

@ -0,0 +1,27 @@
import AppKit
@MainActor
final class AppDelegate: NSObject, NSApplicationDelegate {
private var appearanceObservation: NSKeyValueObservation?
func applicationDidFinishLaunching(_ notification: Notification) {
appearanceObservation = NSApp.observe(
\.effectiveAppearance,
options: [.initial, .new]
) { [weak self] _, _ in
Task { @MainActor in
self?.applyIconForCurrentAppearance()
}
}
}
private func applyIconForCurrentAppearance() {
let match = NSApp.effectiveAppearance.bestMatch(from: [.darkAqua, .aqua])
let resourceName = match == .darkAqua ? "DataTestIcon-Dark" : "DataTestIcon"
guard let url = Bundle.main.url(forResource: resourceName, withExtension: "icns"),
let image = NSImage(contentsOf: url) else {
return
}
NSApp.applicationIconImage = image
}
}

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,623 @@
import Foundation
enum CoreClientError: LocalizedError {
case projectNotFound
case commandFailed(String)
var errorDescription: String? {
switch self {
case .projectNotFound:
return "未找到 bin/datatest请从项目目录启动应用。"
case .commandFailed(let message):
return message
}
}
}
struct CoreClient: Sendable {
private func searchProjectRoot(startingAt start: URL) -> URL? {
var candidate = start
for _ in 0..<6 {
if FileManager.default.fileExists(atPath: candidate.appendingPathComponent("bin/datatest").path) {
return candidate
}
candidate.deleteLastPathComponent()
}
return nil
}
private func projectRoot() throws -> URL {
let workingDirectory = URL(fileURLWithPath: FileManager.default.currentDirectoryPath)
if let root = searchProjectRoot(startingAt: workingDirectory) {
return root
}
let executable = URL(fileURLWithPath: CommandLine.arguments[0]).standardizedFileURL
if let root = searchProjectRoot(startingAt: executable.deletingLastPathComponent()) {
return root
}
throw CoreClientError.projectNotFound
}
private func run(_ arguments: [String]) async throws -> Data {
try await Task.detached(priority: .userInitiated) {
let root = try projectRoot()
let process = Process()
let output = Pipe()
let errors = Pipe()
process.executableURL = root.appendingPathComponent("bin/datatest")
process.arguments = ["--home", root.appendingPathComponent(".datatest").path, "--compact"] + arguments
process.currentDirectoryURL = root
process.standardOutput = output
process.standardError = errors
try process.run()
// Drain both pipes while the child is running. Waiting for the process
// first can deadlock once dashboard JSON grows beyond the pipe buffer.
let outputReader = Task.detached(priority: .utility) {
output.fileHandleForReading.readDataToEndOfFile()
}
let errorReader = Task.detached(priority: .utility) {
errors.fileHandleForReading.readDataToEndOfFile()
}
process.waitUntilExit()
let data = await outputReader.value
let errorData = await errorReader.value
if process.terminationStatus != 0 {
let standardOutput = String(data: data, encoding: .utf8)?
.trimmingCharacters(in: .whitespacesAndNewlines) ?? ""
let standardError = String(data: errorData, encoding: .utf8)?
.trimmingCharacters(in: .whitespacesAndNewlines) ?? ""
let message = !standardOutput.isEmpty
? standardOutput
: (!standardError.isEmpty ? standardError : "DataTest 核心服务执行失败")
throw CoreClientError.commandFailed(message)
}
return data
}.value
}
func dashboard() async throws -> Dashboard {
let data = try await run(["dashboard"])
return try JSONDecoder().decode(Dashboard.self, from: data)
}
func initializeDemo() async throws {
_ = try await run(["demo"])
}
func initializeComplexDemo() async throws {
_ = try await run(["complex-demo"])
}
func importRequirement(fileURL: URL, requirementID: String, requirementName: String) async throws {
_ = try await run([
"requirement-import", fileURL.path,
"--project-id", "PROJECT-LOCAL",
"--project-name", "本地 ETL 测试项目",
"--requirement-id", requirementID,
"--requirement-name", requirementName,
])
}
func parseRequirement(requirementID: String, context: String? = nil) async throws {
var arguments = ["ai-parse", requirementID]
if let context, !context.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty {
arguments += ["--context", context]
}
_ = try await run(arguments)
}
func confirmRequirement(requirementID: String) async throws {
_ = try await run(["requirement-confirm", requirementID])
}
func refreshMetadata(requirementID: String) async throws {
_ = try await run(["metadata", requirementID, "--refresh"])
}
func generateCases(
requirementID: String
) -> AsyncThrowingStream<CaseChatProgressEvent, Error> {
streamProgress(
arguments: ["ai-generate-cases-stream", requirementID],
fallbackError: "Codex 测试案例生成失败"
)
}
func approveCase(caseID: String) async throws {
_ = try await run(["case-approve", caseID])
}
func approveAllCases(requirementID: String) async throws {
_ = try await run(["case-approve-all", requirementID])
}
func rejectCase(caseID: String) async throws {
_ = try await run(["case-reject", caseID])
}
func chatAboutCases(
requirementID: String,
message: String
) -> AsyncThrowingStream<CaseChatProgressEvent, Error> {
streamProgress(
arguments: ["case-chat-stream", requirementID, message],
fallbackError: "Codex 案例协作执行失败"
)
}
private func streamProgress(
arguments: [String],
fallbackError: String
) -> AsyncThrowingStream<CaseChatProgressEvent, Error> {
AsyncThrowingStream { continuation in
Task.detached(priority: .userInitiated) {
do {
let root = try projectRoot()
let process = Process()
let output = Pipe()
let errors = Pipe()
process.executableURL = root.appendingPathComponent("bin/datatest")
process.arguments = [
"--home", root.appendingPathComponent(".datatest").path,
"--compact",
] + arguments
process.currentDirectoryURL = root
process.standardOutput = output
process.standardError = errors
try process.run()
var buffer = ""
var streamedError: String?
func consumeLine(_ line: String) {
let trimmed = line.trimmingCharacters(in: .whitespacesAndNewlines)
guard !trimmed.isEmpty,
let data = trimmed.data(using: .utf8),
let event = try? JSONDecoder().decode(
CaseChatProgressEvent.self, from: data
) else { return }
if event.type == "error" { streamedError = event.detail }
continuation.yield(event)
}
while true {
let data = output.fileHandleForReading.availableData
if data.isEmpty { break }
buffer.append(String(decoding: data, as: UTF8.self))
while let newline = buffer.firstIndex(of: "\n") {
consumeLine(String(buffer[..<newline]))
buffer.removeSubrange(...newline)
}
}
if !buffer.isEmpty { consumeLine(buffer) }
process.waitUntilExit()
if process.terminationStatus != 0 {
let errorData = errors.fileHandleForReading.readDataToEndOfFile()
let standardError = String(data: errorData, encoding: .utf8)?
.trimmingCharacters(in: .whitespacesAndNewlines) ?? ""
throw CoreClientError.commandFailed(
streamedError ?? (!standardError.isEmpty
? standardError : fallbackError)
)
}
continuation.finish()
} catch {
continuation.finish(throwing: error)
}
}
}
}
func runApprovedCases(
requirementID: String
) -> AsyncThrowingStream<TestRunProgressEvent, Error> {
runCasesStream(requirementID: requirementID, caseID: nil)
}
func runSingleCase(
requirementID: String,
caseID: String
) -> AsyncThrowingStream<TestRunProgressEvent, Error> {
runCasesStream(requirementID: requirementID, caseID: caseID)
}
private func runCasesStream(
requirementID: String,
caseID: String?
) -> AsyncThrowingStream<TestRunProgressEvent, Error> {
AsyncThrowingStream { continuation in
Task.detached(priority: .userInitiated) {
do {
let root = try projectRoot()
let process = Process()
let output = Pipe()
let errors = Pipe()
process.executableURL = root.appendingPathComponent("bin/datatest")
var arguments = [
"--home", root.appendingPathComponent(".datatest").path,
"--compact", "run-stream", requirementID,
]
if let caseID { arguments += ["--case", caseID] }
process.arguments = arguments
process.currentDirectoryURL = root
process.standardOutput = output
process.standardError = errors
try process.run()
var buffer = ""
var streamedError: String?
func consumeLine(_ line: String) {
let trimmed = line.trimmingCharacters(in: .whitespacesAndNewlines)
guard !trimmed.isEmpty,
let data = trimmed.data(using: .utf8),
let event = try? JSONDecoder().decode(
TestRunProgressEvent.self, from: data
) else { return }
if event.type == "error" { streamedError = event.detail }
continuation.yield(event)
}
while true {
let data = output.fileHandleForReading.availableData
if data.isEmpty { break }
buffer.append(String(decoding: data, as: UTF8.self))
while let newline = buffer.firstIndex(of: "\n") {
consumeLine(String(buffer[..<newline]))
buffer.removeSubrange(...newline)
}
}
if !buffer.isEmpty { consumeLine(buffer) }
process.waitUntilExit()
if process.terminationStatus != 0 {
let errorData = errors.fileHandleForReading.readDataToEndOfFile()
let standardError = String(data: errorData, encoding: .utf8)?
.trimmingCharacters(in: .whitespacesAndNewlines) ?? ""
throw CoreClientError.commandFailed(
streamedError ?? (!standardError.isEmpty
? standardError : "测试案例执行失败")
)
}
continuation.finish()
} catch {
continuation.finish(throwing: error)
}
}
}
}
func generateReport(runID: String) async throws {
_ = try await run(["report", runID])
}
func analyzeFailure(runID: String, caseID: String) async throws {
_ = try await run(["analyze-failure", runID, caseID])
}
}
@MainActor
final class AppModel: ObservableObject {
@Published var dashboard = Dashboard.empty
@Published var isLoading = false
@Published var message: String?
@Published var runningCaseID: String?
@Published var lastImportedRequirementID: String?
@Published var generatingReportRunID: String?
@Published var analyzingCaseID: String?
@Published var isInitializingComplexDemo = false
@Published var workflowAction: String?
@Published var reviewingCaseID: String?
@Published var isGeneratingCases = false
@Published var caseGenerationRequirementID: String?
@Published var caseGenerationProgress: [CaseChatProgressEvent] = []
@Published var caseGenerationStartedAt: Date?
@Published var caseGenerationFinishedAt: Date?
@Published var caseGenerationError: String?
@Published var isChattingAboutCases = false
@Published var caseChatProgress: [CaseChatProgressEvent] = []
@Published var caseChatStartedAt: Date?
@Published var caseChatFinishedAt: Date?
@Published var caseChatError: String?
@Published var pendingCaseChatMessage: String?
@Published var isRunningTests = false
@Published var testRunRequirementID: String?
@Published var testRunProgress: [TestRunProgressEvent] = []
@Published var testRunStartedAt: Date?
@Published var testRunFinishedAt: Date?
@Published var testRunError: String?
private let client = CoreClient()
func refresh() {
isLoading = true
message = nil
Task {
do {
dashboard = try await client.dashboard()
} catch {
message = error.localizedDescription
}
isLoading = false
}
}
func initializeDemo() {
isLoading = true
message = nil
Task {
do {
try await client.initializeDemo()
dashboard = try await client.dashboard()
message = "演示需求、Metadata 和测试案例已初始化。"
} catch {
message = error.localizedDescription
}
isLoading = false
}
}
func initializeComplexDemo() {
isLoading = true
isInitializingComplexDemo = true
message = nil
Task {
do {
try await client.initializeComplexDemo()
dashboard = try await client.dashboard()
lastImportedRequirementID = "REQ-RISK-002"
message = "复杂大数据需求已准备10 万客户、20 万账户、100 万笔交易,并保留 1 条故意错误供失败诊断。"
} catch {
message = error.localizedDescription
}
isInitializingComplexDemo = false
isLoading = false
}
}
func importRequirement(from fileURL: URL) {
isLoading = true
message = nil
Task {
let hasAccess = fileURL.startAccessingSecurityScopedResource()
defer {
if hasAccess { fileURL.stopAccessingSecurityScopedResource() }
}
do {
let baseName = fileURL.deletingPathExtension().lastPathComponent
let timestamp = Int(Date().timeIntervalSince1970)
let requirementID = "REQ-LOCAL-\(timestamp)"
try await client.importRequirement(
fileURL: fileURL,
requirementID: requirementID,
requirementName: baseName
)
dashboard = try await client.dashboard()
lastImportedRequirementID = requirementID
message = "已导入需求“\(baseName)”,下一步请解析需求并获取 Metadata。"
} catch {
message = error.localizedDescription
}
isLoading = false
}
}
func parseRequirement(requirementID: String, context: String? = nil) {
performWorkflowAction("Agent 正在只读探索数据库并解析测试范围…") {
try await self.client.parseRequirement(requirementID: requirementID, context: context)
return "数据库探索和需求解析完成,候选 Metadata 已形成,请人工确认库、表、字段和规则。"
}
}
func confirmRequirement(requirementID: String) {
performWorkflowAction("正在确认范围并锁定正式 Metadata…") {
try await self.client.confirmRequirement(requirementID: requirementID)
return "需求范围已确认,正式 Metadata 已重新采集并锁定。"
}
}
func refreshMetadata(requirementID: String) {
performWorkflowAction("正在刷新需求 Metadata…") {
try await self.client.refreshMetadata(requirementID: requirementID)
return "Metadata 已刷新。"
}
}
func generateCases(requirementID: String) {
guard !isGeneratingCases else { return }
isLoading = true
isGeneratingCases = true
caseGenerationRequirementID = requirementID
caseGenerationProgress = []
caseGenerationStartedAt = Date()
caseGenerationFinishedAt = nil
caseGenerationError = nil
workflowAction = "Codex 正在根据需求和 Metadata 生成案例草稿…"
message = nil
Task {
do {
for try await event in client.generateCases(requirementID: requirementID) {
caseGenerationProgress.append(event)
if event.type == "error" { caseGenerationError = event.detail }
}
dashboard = try await client.dashboard()
message = "案例草稿已生成,请逐条审核;审核通过后才能执行。"
} catch {
caseGenerationError = error.localizedDescription
message = error.localizedDescription
if let refreshed = try? await client.dashboard() {
dashboard = refreshed
}
}
caseGenerationFinishedAt = Date()
workflowAction = nil
isGeneratingCases = false
isLoading = false
}
}
func approveCase(caseID: String) {
reviewingCaseID = caseID
performWorkflowAction("正在校验并批准 \(caseID)") {
defer { self.reviewingCaseID = nil }
try await self.client.approveCase(caseID: caseID)
return "案例 \(caseID) 已审核通过。"
}
}
func rejectCase(caseID: String) {
reviewingCaseID = caseID
performWorkflowAction("正在驳回 \(caseID)") {
defer { self.reviewingCaseID = nil }
try await self.client.rejectCase(caseID: caseID)
return "案例 \(caseID) 已驳回,可通过 Codex 沟通页面继续调整。"
}
}
func approveAllCases(requirementID: String) {
performWorkflowAction("正在校验并批量审核案例…") {
try await self.client.approveAllCases(requirementID: requirementID)
return "所有有效草稿案例已审核通过。"
}
}
func chatAboutCases(requirementID: String, message: String) {
guard !isChattingAboutCases else { return }
isChattingAboutCases = true
caseChatProgress = []
caseChatStartedAt = Date()
caseChatFinishedAt = nil
caseChatError = nil
pendingCaseChatMessage = message
self.message = nil
Task {
do {
for try await event in client.chatAboutCases(
requirementID: requirementID,
message: message
) {
caseChatProgress.append(event)
if event.type == "error" { caseChatError = event.detail }
}
dashboard = try await client.dashboard()
pendingCaseChatMessage = nil
self.message = "Codex 已回复;新增或修改案例已转为草稿,等待人工审核。"
} catch {
caseChatError = error.localizedDescription
self.message = error.localizedDescription
if let refreshed = try? await client.dashboard() {
dashboard = refreshed
pendingCaseChatMessage = nil
}
}
caseChatFinishedAt = Date()
isChattingAboutCases = false
}
}
private func performWorkflowAction(
_ progress: String,
operation: @escaping @MainActor () async throws -> String
) {
isLoading = true
workflowAction = progress
message = nil
Task {
do {
let successMessage = try await operation()
dashboard = try await client.dashboard()
message = successMessage
} catch {
message = error.localizedDescription
}
workflowAction = nil
isLoading = false
}
}
func runTests(requirementID: String) {
startTestRun(requirementID: requirementID, caseID: nil)
}
func runSingleCase(requirementID: String, caseID: String) {
startTestRun(requirementID: requirementID, caseID: caseID)
}
private func startTestRun(requirementID: String, caseID: String?) {
guard !isRunningTests else { return }
isLoading = true
isRunningTests = true
runningCaseID = caseID
testRunRequirementID = requirementID
testRunProgress = []
testRunStartedAt = Date()
testRunFinishedAt = nil
testRunError = nil
message = nil
Task {
do {
let stream = if let caseID {
client.runSingleCase(requirementID: requirementID, caseID: caseID)
} else {
client.runApprovedCases(requirementID: requirementID)
}
for try await event in stream {
testRunProgress.append(event)
if event.type == "error" { testRunError = event.detail }
}
dashboard = try await client.dashboard()
message = caseID == nil ? "已完成全部已审核案例。" : "案例 \(caseID!) 已完成单独重跑。"
} catch {
testRunError = error.localizedDescription
message = error.localizedDescription
if let refreshed = try? await client.dashboard() {
dashboard = refreshed
}
}
testRunFinishedAt = Date()
runningCaseID = nil
isRunningTests = false
isLoading = false
}
}
func generateReport(runID: String) {
isLoading = true
generatingReportRunID = runID
message = nil
Task {
do {
try await client.generateReport(runID: runID)
dashboard = try await client.dashboard()
message = "运行 \(runID) 的测试报告已生成。"
} catch {
message = error.localizedDescription
}
generatingReportRunID = nil
isLoading = false
}
}
func analyzeFailure(runID: String, caseID: String) {
isLoading = true
analyzingCaseID = caseID
message = nil
Task {
do {
try await client.analyzeFailure(runID: runID, caseID: caseID)
dashboard = try await client.dashboard()
message = "Codex 已完成 \(caseID) 的失败根因调查。"
} catch {
message = error.localizedDescription
}
analyzingCaseID = nil
isLoading = false
}
}
func clearMessage() {
message = nil
}
}

View File

@ -0,0 +1,16 @@
import SwiftUI
@main
struct DataTestApp: App {
@NSApplicationDelegateAdaptor(AppDelegate.self) private var appDelegate
@StateObject private var model = AppModel()
var body: some Scene {
WindowGroup("DataTest") {
ContentView()
.environmentObject(model)
.frame(minWidth: 1040, minHeight: 680)
}
.windowStyle(.titleBar)
}
}

View File

@ -0,0 +1,280 @@
import Foundation
struct ProjectItem: Codable, Identifiable, Sendable {
let id: String
let name: String
let created_at: String
}
struct RequirementItem: Codable, Identifiable, Sendable {
let id: String
let project_id: String
let name: String
let status: String
let current_version: Int
let created_at: String
let source_path: String
let extraction: RequirementExtraction?
let metadata_ready: Bool
let metadata_complete: Bool
let metadata_missing_tables: [String]
}
struct RequirementExtraction: Codable, Sendable {
let requirement_name: String
let tasks: [ExtractedTask]
let open_questions: [String]
}
struct ExtractedTask: Codable, Identifiable, Sendable {
var id: String { name + targets.joined(separator: "|") }
let name: String
let sources: [String]
let targets: [String]
let field_mappings: [JSONValue]
let rules: [String]
}
struct CaseCount: Codable, Identifiable, Sendable {
var id: String { status }
let status: String
let count: Int
}
struct TestCaseItem: Codable, Identifiable, Sendable {
let id: String
let requirement_id: String
let requirement_name: String
let name: String
let table_name: String
let category: String
let status: String
let version: Int
let fields: [String]
let sql: String
let assertions: [CaseAssertion]
let validation_errors: [String]
let created_at: String
}
struct CaseAssertion: Codable, Sendable {
let type: String
let actual: String?
let expected: JSONValue?
let minimum: Double?
let maximum: Double?
}
enum JSONValue: Codable, Sendable, CustomStringConvertible {
case string(String)
case number(Double)
case boolean(Bool)
case object([String: JSONValue])
case array([JSONValue])
case null
init(from decoder: Decoder) throws {
let container = try decoder.singleValueContainer()
if container.decodeNil() { self = .null }
else if let value = try? container.decode(Bool.self) { self = .boolean(value) }
else if let value = try? container.decode(Double.self) { self = .number(value) }
else if let value = try? container.decode(String.self) { self = .string(value) }
else if let value = try? container.decode([String: JSONValue].self) { self = .object(value) }
else { self = .array(try container.decode([JSONValue].self)) }
}
func encode(to encoder: Encoder) throws {
var container = encoder.singleValueContainer()
switch self {
case .string(let value): try container.encode(value)
case .number(let value): try container.encode(value)
case .boolean(let value): try container.encode(value)
case .object(let value): try container.encode(value)
case .array(let value): try container.encode(value)
case .null: try container.encodeNil()
}
}
var description: String {
switch self {
case .string(let value): value
case .number(let value): value.rounded() == value ? String(Int(value)) : String(value)
case .boolean(let value): String(value)
case .object(let value): value.map { "\($0.key)=\($0.value)" }.sorted().joined(separator: ", ")
case .array(let value): value.map(\.description).joined(separator: ", ")
case .null: "null"
}
}
}
struct MetadataColumn: Codable, Identifiable, Sendable {
var id: Int { cid }
let cid: Int
let name: String
let type: String
let notNull: Int
let defaultValue: String?
let pk: Int
var ordinalText: String { String(cid + 1) }
var nullableText: String { notNull == 0 ? "" : "" }
var defaultText: String { defaultValue ?? "" }
var primaryKeyText: String { pk > 0 ? "" : "" }
enum CodingKeys: String, CodingKey {
case cid, name, type, pk
case notNull = "notnull"
case defaultValue = "dflt_value"
}
}
struct MetadataTableItem: Codable, Identifiable, Sendable {
let id: String
let requirement_id: String
let requirement_name: String
let database_name: String
let name: String
let type: String
let row_count: Int
let columns: [MetadataColumn]
}
struct RunItem: Codable, Identifiable, Sendable {
let id: String
let requirement_id: String
let requirement_version: Int
let status: String
let batch_id: String
let biz_date: String?
let started_at: String
let finished_at: String?
}
struct MetricItem: Codable, Identifiable, Sendable {
let id: Int
let requirement_id: String
let table_name: String
let field_name: String?
let batch_id: String
let biz_date: String?
let metric_type: String
let metric_value: Double?
let metric_json: String?
let definition_version: Int
let collected_at: String
}
struct ReportItem: Codable, Identifiable, Sendable {
let id: String
let run_id: String
let requirement_id: String
let requirement_version: Int
let status: String
let format: String
let file_path: String
let content: String
let created_at: String
}
struct CaseResultItem: Codable, Identifiable, Sendable {
let id: Int
let run_id: String
let case_id: String
let requirement_id: String
let case_name: String
let status: String
let actual_json: String?
let assertion_json: String
let sample_json: String?
let error_message: String?
let duration_ms: Int
let created_at: String
}
struct FailureAnalysis: Codable, Sendable {
let summary: String
let suspected_layer: String
let root_cause: String
let evidence: [String]
let recommendations: [String]
let validation_sql: [String]
let confidence: String
}
struct FailureAnalysisItem: Codable, Identifiable, Sendable {
let id: Int
let result_id: Int
let run_id: String
let case_id: String
let requirement_id: String
let status: String
let created_at: String
let analysis: FailureAnalysis
}
struct CaseAgentMessageItem: Codable, Identifiable, Sendable {
let id: Int
let requirement_id: String
let role: String
let content: String
let operation_json: String?
let created_at: String
}
struct CaseChatProgressEvent: Codable, Identifiable, Sendable {
var id: Int { sequence }
let type: String
let sequence: Int
let phase: String
let status: String
let title: String
let detail: String
let timestamp: String
}
struct TestRunProgressEvent: Codable, Identifiable, Sendable {
var id: Int { sequence }
let type: String
let sequence: Int
let event: String
let run_id: String?
let case_id: String?
let case_name: String?
let status: String
let index: Int?
let total: Int?
let duration_ms: Int?
let error: String?
let detail: String?
let timestamp: String
}
struct CaseReviewEventItem: Codable, Identifiable, Sendable {
let id: Int
let case_id: String
let requirement_id: String
let case_version: Int
let decision: String
let comment: String?
let created_at: String
}
struct Dashboard: Codable, Sendable {
let projects: [ProjectItem]
let requirements: [RequirementItem]
let cases: [CaseCount]
let case_items: [TestCaseItem]
let metadata: [MetadataTableItem]
let runs: [RunItem]
let metrics: [MetricItem]
let reports: [ReportItem]
let result_items: [CaseResultItem]
let failure_analyses: [FailureAnalysisItem]
let case_agent_messages: [CaseAgentMessageItem]
let case_review_events: [CaseReviewEventItem]
static let empty = Dashboard(
projects: [], requirements: [], cases: [], case_items: [], metadata: [], runs: [], metrics: [], reports: [],
result_items: [], failure_analyses: [], case_agent_messages: [], case_review_events: []
)
}

View File

@ -0,0 +1,371 @@
import AppKit
import Foundation
import SwiftUI
struct SQLCodeView: View {
let sql: String
@Environment(\.colorScheme) private var colorScheme
@State private var copied = false
private var formattedSQL: String {
SQLPrettyPrinter.format(sql)
}
private var highlightedSQL: AttributedString {
SQLSyntaxHighlighter.highlight(formattedSQL, colorScheme: colorScheme)
}
private var lineNumbers: String {
let count = max(formattedSQL.split(separator: "\n", omittingEmptySubsequences: false).count, 1)
return (1...count).map(String.init).joined(separator: "\n")
}
var body: some View {
VStack(spacing: 0) {
HStack(spacing: 8) {
Label("格式化 SQL", systemImage: "chevron.left.forwardslash.chevron.right")
.font(.caption.bold())
.foregroundStyle(.secondary)
Spacer()
Button {
let pasteboard = NSPasteboard.general
pasteboard.clearContents()
pasteboard.setString(formattedSQL, forType: .string)
copied = true
Task { @MainActor in
try? await Task.sleep(for: .seconds(1.5))
copied = false
}
} label: {
Label(copied ? "已复制" : "复制", systemImage: copied ? "checkmark" : "doc.on.doc")
}
.buttonStyle(.borderless)
.help("复制格式化后的 SQL")
}
.padding(.horizontal, 12)
.padding(.vertical, 8)
.background(.quaternary.opacity(0.34))
Divider()
ScrollView(.horizontal) {
HStack(alignment: .top, spacing: 12) {
Text(lineNumbers)
.foregroundStyle(.tertiary)
.multilineTextAlignment(.trailing)
.textSelection(.disabled)
Divider()
Text(highlightedSQL)
.textSelection(.enabled)
}
.font(.system(size: 12.5, design: .monospaced))
.lineSpacing(3)
.padding(12)
.fixedSize(horizontal: true, vertical: true)
}
}
.background(Color(nsColor: .textBackgroundColor).opacity(colorScheme == .dark ? 0.72 : 0.88))
.clipShape(RoundedRectangle(cornerRadius: 9))
.overlay {
RoundedRectangle(cornerRadius: 9)
.stroke(Color(nsColor: .separatorColor).opacity(0.55), lineWidth: 1)
}
.onChange(of: sql) { _, _ in copied = false }
}
}
enum SQLPrettyPrinter {
private struct ParenthesisContext {
let isMultiline: Bool
let clause: String
let continuationIndent: Int
let closingIndent: Int
let parentIndent: Int
}
private static let tokenPattern = #"--[^\n]*|/\*[\s\S]*?\*/|'(?:''|[^'])*'|\"(?:\"\"|[^\"])*\"|`(?:``|[^`])*`|\[[^\]]*\]|[A-Za-z_][A-Za-z0-9_$]*|\d+(?:\.\d+)?|<>|!=|<=|>=|==|\|\||[-+*/%=<>,.;()]|[^\s]"#
private static let clausePhrases: Set<String> = [
"SELECT", "FROM", "WHERE", "GROUP BY", "HAVING", "ORDER BY",
"LIMIT", "OFFSET", "UNION", "UNION ALL", "EXCEPT", "INTERSECT",
"LEFT JOIN", "LEFT OUTER JOIN", "RIGHT JOIN", "RIGHT OUTER JOIN",
"INNER JOIN", "FULL JOIN", "FULL OUTER JOIN", "CROSS JOIN", "JOIN",
"VALUES", "SET"
]
static func format(_ sql: String) -> String {
let tokens = tokenize(sql)
guard !tokens.isEmpty else { return sql.trimmingCharacters(in: .whitespacesAndNewlines) }
var lines: [String] = []
var current = ""
var currentIndent = 0
var indent = 0
var continuationIndent = 0
var clause = ""
var parentheses: [ParenthesisContext] = []
var index = 0
func flush() {
let trimmed = current.trimmingCharacters(in: .whitespaces)
guard !trimmed.isEmpty else { return }
lines.append(String(repeating: " ", count: max(currentIndent, 0)) + trimmed)
current = ""
}
func newLine(_ requestedIndent: Int? = nil) {
flush()
currentIndent = max(requestedIndent ?? continuationIndent, 0)
}
func append(_ text: String, spaceBefore: Bool = true) {
if current.isEmpty {
current = text
} else if spaceBefore && !current.hasSuffix(" ") {
current += " " + text
} else {
current += text
}
}
while index < tokens.count {
let token = tokens[index]
let upper = token.uppercased()
let (phrase, consumed) = phrase(at: index, tokens: tokens)
let previous = index > 0 ? tokens[index - 1].uppercased() : ""
let next = index + consumed < tokens.count ? tokens[index + consumed].uppercased() : ""
if phrase == "WITH" {
newLine(indent)
append("WITH", spaceBefore: false)
continuationIndent = indent
clause = "WITH"
index += consumed
continue
}
if clausePhrases.contains(phrase) {
newLine(indent)
append(phrase, spaceBefore: false)
clause = phrase
continuationIndent = indent + 1
if phrase == "SELECT" {
newLine(continuationIndent)
}
index += consumed
continue
}
if phrase == "ON" {
newLine(indent + 1)
append("ON", spaceBefore: false)
clause = "ON"
continuationIndent = indent + 2
index += consumed
continue
}
if (phrase == "AND" || phrase == "OR")
&& ["WHERE", "HAVING", "ON"].contains(clause) {
newLine(indent + 1)
append(phrase, spaceBefore: false)
continuationIndent = indent + 2
index += consumed
continue
}
if phrase == "WHEN" || phrase == "ELSE" {
newLine(indent + 1)
append(phrase, spaceBefore: false)
continuationIndent = indent + 2
index += consumed
continue
}
if phrase == "END" {
newLine(indent)
append("END", spaceBefore: false)
continuationIndent = indent
index += consumed
continue
}
switch token {
case "(":
let multiline = next == "SELECT" || next == "WITH"
let needsSpace = multiline
? !current.isEmpty
: (!current.isEmpty
&& ![".", "("].contains(previous)
&& !isFunctionName(previous))
append("(", spaceBefore: needsSpace)
parentheses.append(ParenthesisContext(
isMultiline: multiline,
clause: clause,
continuationIndent: continuationIndent,
closingIndent: currentIndent,
parentIndent: indent
))
if multiline {
indent = currentIndent + 1
continuationIndent = indent
newLine(indent)
}
case ")":
let context = parentheses.popLast()
if context?.isMultiline == true {
indent = context?.parentIndent ?? max(indent - 1, 0)
newLine(context?.closingIndent ?? indent)
append(")", spaceBefore: false)
clause = context?.clause ?? clause
continuationIndent = context?.continuationIndent ?? indent
} else {
append(")", spaceBefore: false)
}
case ",":
append(",", spaceBefore: false)
let insideFunction = parentheses.last?.isMultiline == false
if !insideFunction && ["SELECT", "GROUP BY", "ORDER BY", "WITH"].contains(clause) {
newLine(continuationIndent)
}
case ".":
append(".", spaceBefore: false)
case ";":
append(";", spaceBefore: false)
newLine(indent)
case "=", "!=", "<>", "<", ">", "<=", ">=", "+", "-", "*", "/", "%", "||":
let compactWildcard = token == "*" && (previous == "(" || next == ")")
append(token, spaceBefore: !compactWildcard)
default:
if upper.hasPrefix("--") || upper.hasPrefix("/*") {
newLine(continuationIndent)
append(token, spaceBefore: false)
newLine(continuationIndent)
} else {
append(token, spaceBefore: previous != "." && previous != "(")
}
}
index += 1
}
flush()
return lines.joined(separator: "\n")
.trimmingCharacters(in: .whitespacesAndNewlines)
}
private static func tokenize(_ sql: String) -> [String] {
guard let expression = try? NSRegularExpression(pattern: tokenPattern) else { return [sql] }
let range = NSRange(sql.startIndex..., in: sql)
return expression.matches(in: sql, range: range).map {
(sql as NSString).substring(with: $0.range)
}
}
private static func phrase(at index: Int, tokens: [String]) -> (String, Int) {
let word = tokens[index].uppercased()
func matches(_ values: [String]) -> Bool {
guard index + values.count <= tokens.count else { return false }
return zip(tokens[index..<(index + values.count)], values).allSatisfy {
$0.0.uppercased() == $0.1
}
}
let phrases = [
["LEFT", "OUTER", "JOIN"], ["RIGHT", "OUTER", "JOIN"],
["FULL", "OUTER", "JOIN"], ["GROUP", "BY"], ["ORDER", "BY"],
["PARTITION", "BY"], ["UNION", "ALL"], ["LEFT", "JOIN"],
["RIGHT", "JOIN"], ["INNER", "JOIN"], ["FULL", "JOIN"],
["CROSS", "JOIN"]
]
for values in phrases where matches(values) {
return (values.joined(separator: " "), values.count)
}
return (word, 1)
}
private static func isFunctionName(_ token: String) -> Bool {
guard let first = token.first else { return false }
return first.isLetter || first == "_"
}
}
private enum SQLSyntaxHighlighter {
private static let tokenPattern = #"--[^\n]*|/\*[\s\S]*?\*/|'(?:''|[^'])*'|\"(?:\"\"|[^\"])*\"|`(?:``|[^`])*`|\[[^\]]*\]|\b\d+(?:\.\d+)?\b|\b[A-Za-z_][A-Za-z0-9_$]*\b"#
private static let keywords: Set<String> = [
"ALL", "ALTER", "AND", "AS", "ASC", "BETWEEN", "BY", "CASE", "CAST",
"CREATE", "CROSS", "CURRENT", "DELETE", "DESC", "DISTINCT", "DROP", "ELSE",
"END", "EXCEPT", "EXISTS", "FOLLOWING", "FROM", "FULL", "GROUP", "HAVING",
"IN", "INNER", "INSERT", "INTERSECT", "INTERVAL", "INTO", "IS", "JOIN",
"LEFT", "LIKE", "LIMIT", "NOT", "NULL", "OFFSET", "ON", "OR", "ORDER",
"OUTER", "OVER", "PARTITION", "PRECEDING", "RANGE", "RECURSIVE", "RIGHT",
"ROW", "ROWS", "SELECT", "SET", "TABLE", "THEN", "UNION", "UPDATE",
"VALUES", "VIEW", "WHEN", "WHERE", "WITH"
]
private static let functions: Set<String> = [
"ABS", "AVG", "CAST", "COALESCE", "COUNT", "DATE", "DENSE_RANK", "IFNULL",
"LAG", "LEAD", "LOWER", "MAX", "MIN", "NULLIF", "RANK", "ROUND",
"ROW_NUMBER", "SUBSTR", "SUBSTRING", "SUM", "TRIM", "UPPER"
]
static func highlight(_ source: String, colorScheme: ColorScheme) -> AttributedString {
var attributed = AttributedString(source)
guard let expression = try? NSRegularExpression(pattern: tokenPattern) else {
return attributed
}
let nsSource = source as NSString
let fullRange = NSRange(source.startIndex..., in: source)
for match in expression.matches(in: source, range: fullRange) {
let token = nsSource.substring(with: match.range)
let upper = token.uppercased()
let color: Color?
if token.hasPrefix("--") || token.hasPrefix("/*") {
color = .secondary
} else if token.hasPrefix("'") {
color = colorScheme == .dark ? Color(red: 0.56, green: 0.82, blue: 0.58) : Color(red: 0.12, green: 0.50, blue: 0.20)
} else if token.hasPrefix("\"") || token.hasPrefix("`") || token.hasPrefix("[") {
color = .teal
} else if token.first?.isNumber == true {
color = .orange
} else if keywords.contains(upper) {
color = colorScheme == .dark ? Color(red: 0.80, green: 0.58, blue: 0.98) : Color(red: 0.49, green: 0.18, blue: 0.72)
} else if functions.contains(upper) || isFollowedByOpeningParenthesis(match.range, in: nsSource) {
color = colorScheme == .dark ? Color(red: 0.42, green: 0.72, blue: 1.0) : Color(red: 0.05, green: 0.38, blue: 0.72)
} else {
color = nil
}
guard let color,
let stringRange = Range(match.range, in: source),
let lower = AttributedString.Index(stringRange.lowerBound, within: attributed),
let upperBound = AttributedString.Index(stringRange.upperBound, within: attributed) else {
continue
}
attributed[lower..<upperBound].foregroundColor = color
}
return attributed
}
private static func isFollowedByOpeningParenthesis(_ range: NSRange, in source: NSString) -> Bool {
var offset = NSMaxRange(range)
while offset < source.length {
let scalar = source.character(at: offset)
if scalar == 32 || scalar == 9 || scalar == 10 || scalar == 13 {
offset += 1
continue
}
return scalar == 40
}
return false
}
}

19
pyproject.toml Normal file
View File

@ -0,0 +1,19 @@
[project]
name = "datatest-tool"
version = "0.1.0"
description = "Requirement-centric deterministic ETL data testing framework"
requires-python = ">=3.9"
dependencies = []
[project.scripts]
datatest = "datatest.cli:main"
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[tool.setuptools]
package-dir = {"" = "src"}
[tool.setuptools.packages.find]
where = ["src"]

View File

@ -0,0 +1,51 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"required": ["assistant_message", "cases", "removed_case_ids"],
"properties": {
"assistant_message": {"type": "string"},
"cases": {
"type": "array",
"items": {
"type": "object",
"required": ["case_id", "name", "table_name", "database_name", "fields", "category", "sql", "sample_sql", "sample_limit", "assertions"],
"properties": {
"case_id": {"type": ["string", "null"]},
"name": {"type": "string"},
"table_name": {"type": "string"},
"database_name": {"type": "string"},
"fields": {"type": "array", "items": {"type": "string"}},
"category": {"type": "string"},
"sql": {"type": "string"},
"sample_sql": {"type": ["string", "null"]},
"sample_limit": {"type": "integer", "minimum": 1, "maximum": 1000},
"assertions": {
"type": "array",
"minItems": 1,
"items": {
"type": "object",
"required": ["type", "actual", "expected", "minimum", "maximum"],
"properties": {
"type": {
"type": "string",
"enum": ["equals", "not_equals", "greater_than", "less_than", "between", "change_rate_between", "result_is_empty"]
},
"actual": {"type": ["string", "null"]},
"expected": {"type": ["string", "number", "boolean", "null"]},
"minimum": {"type": ["number", "null"]},
"maximum": {"type": ["number", "null"]}
},
"additionalProperties": false
}
}
},
"additionalProperties": false
}
},
"removed_case_ids": {
"type": "array",
"items": {"type": "string"}
}
},
"additionalProperties": false
}

View File

@ -0,0 +1,18 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"required": ["summary", "suspected_layer", "root_cause", "evidence", "recommendations", "validation_sql", "confidence"],
"properties": {
"summary": {"type": "string"},
"suspected_layer": {
"type": "string",
"enum": ["source_data", "etl_logic", "target_data", "test_case", "environment", "requirement_ambiguity", "unknown"]
},
"root_cause": {"type": "string"},
"evidence": {"type": "array", "items": {"type": "string"}},
"recommendations": {"type": "array", "items": {"type": "string"}},
"validation_sql": {"type": "array", "items": {"type": "string"}},
"confidence": {"type": "string", "enum": ["low", "medium", "high"]}
},
"additionalProperties": false
}

View File

@ -0,0 +1,37 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"required": ["requirement_name", "tasks", "open_questions"],
"properties": {
"requirement_name": {"type": "string"},
"tasks": {
"type": "array",
"items": {
"type": "object",
"required": ["name", "sources", "targets", "field_mappings", "rules"],
"properties": {
"name": {"type": "string"},
"sources": {"type": "array", "items": {"type": "string"}},
"targets": {"type": "array", "items": {"type": "string"}},
"field_mappings": {
"type": "array",
"items": {
"type": "object",
"required": ["source_field", "target_field", "transformation"],
"properties": {
"source_field": {"type": "string"},
"target_field": {"type": "string"},
"transformation": {"type": ["string", "null"]}
},
"additionalProperties": false
}
},
"rules": {"type": "array", "items": {"type": "string"}}
},
"additionalProperties": false
}
},
"open_questions": {"type": "array", "items": {"type": "string"}}
},
"additionalProperties": false
}

View File

@ -0,0 +1,45 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"required": ["cases"],
"properties": {
"cases": {
"type": "array",
"items": {
"type": "object",
"required": ["name", "table_name", "database_name", "fields", "category", "sql", "sample_sql", "sample_limit", "assertions"],
"properties": {
"name": {"type": "string"},
"table_name": {"type": "string"},
"database_name": {"type": "string"},
"fields": {"type": "array", "items": {"type": "string"}},
"category": {"type": "string"},
"sql": {"type": "string"},
"sample_sql": {"type": ["string", "null"]},
"sample_limit": {"type": "integer", "minimum": 1, "maximum": 1000},
"assertions": {
"type": "array",
"minItems": 1,
"items": {
"type": "object",
"required": ["type", "actual", "expected", "minimum", "maximum"],
"properties": {
"type": {
"type": "string",
"enum": ["equals", "not_equals", "greater_than", "less_than", "between", "change_rate_between", "result_is_empty"]
},
"actual": {"type": ["string", "null"]},
"expected": {"type": ["string", "number", "boolean", "null"]},
"minimum": {"type": ["number", "null"]},
"maximum": {"type": ["number", "null"]}
},
"additionalProperties": false
}
}
},
"additionalProperties": false
}
}
},
"additionalProperties": false
}

3
src/datatest/__init__.py Normal file
View File

@ -0,0 +1,3 @@
"""DataTest: requirement-centric ETL testing."""
__version__ = "0.1.0"

136
src/datatest/ai.py Normal file
View File

@ -0,0 +1,136 @@
from __future__ import annotations
import hashlib
import json
import os
import shutil
import subprocess
import tempfile
import threading
from pathlib import Path
from typing import Any, Callable
class CodexCLIAdapter:
"""Isolated structured-output adapter. It never participates in pass/fail decisions."""
def __init__(self, executable: str | None = None):
self.executable = executable or os.environ.get("DATATEST_CODEX_PATH") or shutil.which("codex")
if not self.executable:
application_path = Path("/Applications/ChatGPT.app/Contents/Resources/codex")
if application_path.exists():
self.executable = str(application_path)
if not self.executable:
raise FileNotFoundError("未找到 Codex CLI请设置 DATATEST_CODEX_PATH")
def run_structured(self, instruction: str, input_payload: dict[str, Any],
output_schema: Path, timeout_seconds: int = 180,
reasoning_effort: str | None = None,
model: str | None = None) -> dict[str, Any]:
prompt = instruction + "\n\n输入数据:\n" + json.dumps(input_payload, ensure_ascii=False)
with tempfile.TemporaryDirectory(prefix="datatest-codex-") as temp_dir:
output_file = Path(temp_dir) / "last-message.json"
command = [
self.executable, "exec", "--ignore-user-config", "--ephemeral",
"--sandbox", "read-only", "--skip-git-repo-check", "--cd", temp_dir,
]
if model is not None:
command += ["--model", model]
if reasoning_effort is not None:
command += ["--config", f'model_reasoning_effort="{reasoning_effort}"']
command += [
"--output-schema", str(output_schema.resolve()),
"--output-last-message", str(output_file), "-",
]
completed = subprocess.run(
command, input=prompt, text=True, capture_output=True,
timeout=timeout_seconds, check=False,
)
if completed.returncode != 0:
raise RuntimeError(completed.stderr.strip() or "Codex CLI 调用失败")
return json.loads(output_file.read_text(encoding="utf-8"))
def run_structured_streaming(
self,
instruction: str,
input_payload: dict[str, Any],
output_schema: Path,
on_event: Callable[[dict[str, Any]], None],
timeout_seconds: int = 300,
reasoning_effort: str | None = None,
model: str | None = None,
) -> dict[str, Any]:
"""Run Codex with JSONL events while keeping the final response schema-bound."""
prompt = instruction + "\n\n输入数据:\n" + json.dumps(input_payload, ensure_ascii=False)
with tempfile.TemporaryDirectory(prefix="datatest-codex-") as temp_dir:
output_file = Path(temp_dir) / "last-message.json"
command = [
self.executable, "exec", "--ignore-user-config", "--ephemeral",
"--sandbox", "read-only", "--skip-git-repo-check", "--cd", temp_dir, "--json",
]
if model is not None:
command += ["--model", model]
if reasoning_effort is not None:
command += ["--config", f'model_reasoning_effort="{reasoning_effort}"']
command += [
"--output-schema", str(output_schema.resolve()),
"--output-last-message", str(output_file), "-",
]
process = subprocess.Popen(
command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
)
timed_out = threading.Event()
def terminate_on_timeout() -> None:
timed_out.set()
process.kill()
timer = threading.Timer(timeout_seconds, terminate_on_timeout)
timer.daemon = True
timer.start()
stderr_chunks: list[str] = []
def collect_stderr() -> None:
if process.stderr is not None:
stderr_chunks.append(process.stderr.read())
stderr_thread = threading.Thread(target=collect_stderr, daemon=True)
stderr_thread.start()
try:
if process.stdin is None or process.stdout is None:
raise RuntimeError("Codex CLI 管道初始化失败")
process.stdin.write(prompt)
process.stdin.close()
for line in process.stdout:
line = line.strip()
if not line:
continue
try:
event = json.loads(line)
except json.JSONDecodeError:
continue
if isinstance(event, dict):
on_event(event)
return_code = process.wait()
finally:
timer.cancel()
stderr_thread.join(timeout=1)
if timed_out.is_set():
raise TimeoutError(f"Codex CLI 超过 {timeout_seconds} 秒未完成")
if return_code != 0:
error_message = "".join(stderr_chunks).strip()
raise RuntimeError(error_message or "Codex CLI 调用失败")
if not output_file.exists():
raise RuntimeError("Codex CLI 未生成结构化最终响应")
return json.loads(output_file.read_text(encoding="utf-8"))
@staticmethod
def input_hash(payload: dict[str, Any]) -> str:
raw = json.dumps(payload, ensure_ascii=False, sort_keys=True).encode()
return hashlib.sha256(raw).hexdigest()

View File

@ -0,0 +1,268 @@
from __future__ import annotations
from .domain import AssertionSpec, TestCaseSpec
def customer_demo_cases(requirement_id: str, version: int) -> list[TestCaseSpec]:
base = {
"requirement_id": requirement_id,
"requirement_version": version,
"etl_task_id": "TASK-CUSTOMER-001",
"database_name": "dwd",
"table_name": "dwd_customer_info",
}
return [
TestCaseSpec(
**base,
name="dwd_customer_info_目标表数据非空校验",
fields=[], category="completeness",
sql="SELECT COUNT(*) AS row_count FROM dwd.dwd_customer_info",
assertions=[AssertionSpec("greater_than", "row_count", 0)],
),
TestCaseSpec(
**base,
name="dwd_customer_info_cust_id字段唯一性校验",
fields=["cust_id"], category="quality",
sql="SELECT COUNT(*) AS duplicate_count FROM (SELECT cust_id FROM dwd.dwd_customer_info GROUP BY cust_id HAVING COUNT(*) > 1)",
assertions=[AssertionSpec("equals", "duplicate_count", 0)],
sample_sql="SELECT cust_id, COUNT(*) AS duplicate_count FROM dwd.dwd_customer_info GROUP BY cust_id HAVING COUNT(*) > 1",
),
TestCaseSpec(
**base,
name="dwd_customer_info_cust_status字段值分布校验",
fields=["cust_status"], category="distribution",
sql="SELECT COUNT(*) AS invalid_count FROM dwd.dwd_customer_info WHERE cust_status NOT IN ('ACTIVE', 'INACTIVE') OR cust_status IS NULL",
assertions=[AssertionSpec("equals", "invalid_count", 0)],
sample_sql="SELECT * FROM dwd.dwd_customer_info WHERE cust_status NOT IN ('ACTIVE', 'INACTIVE') OR cust_status IS NULL",
),
TestCaseSpec(
**base,
name="dwd_customer_info_age字段范围校验",
fields=["age"], category="quality",
sql="SELECT COUNT(*) AS invalid_count FROM dwd.dwd_customer_info WHERE age NOT BETWEEN 0 AND 120",
assertions=[AssertionSpec("equals", "invalid_count", 0)],
sample_sql="SELECT * FROM dwd.dwd_customer_info WHERE age NOT BETWEEN 0 AND 120",
),
TestCaseSpec(
**base,
name="dwd_customer_info_源目标有效客户数量一致性校验",
fields=["cust_id"], category="reconciliation",
sql="SELECT (SELECT COUNT(*) FROM ods.ods_customer WHERE is_deleted = 0) - (SELECT COUNT(*) FROM dwd.dwd_customer_info) AS count_difference",
assertions=[AssertionSpec("equals", "count_difference", 0)],
),
TestCaseSpec(
**base,
name="dwd_customer_info_字段转换一致性校验",
fields=["cust_id", "cust_name", "cust_status", "age", "updated_at"],
category="transformation",
sql="""
SELECT COUNT(*) AS difference_count
FROM ods.ods_customer s
LEFT JOIN dwd.dwd_customer_info t ON s.customer_id = t.cust_id
WHERE s.is_deleted = 0 AND (
t.cust_id IS NULL OR t.cust_name <> TRIM(s.customer_name)
OR t.cust_status <> CASE s.status WHEN '1' THEN 'ACTIVE' WHEN '0' THEN 'INACTIVE' END
OR t.age IS NOT s.age OR t.updated_at <> s.updated_at
)
""",
assertions=[AssertionSpec("equals", "difference_count", 0)],
sample_sql="""
SELECT s.*, t.cust_name AS actual_name, t.cust_status AS actual_status
FROM ods.ods_customer s
LEFT JOIN dwd.dwd_customer_info t ON s.customer_id = t.cust_id
WHERE s.is_deleted = 0 AND (
t.cust_id IS NULL OR t.cust_name <> TRIM(s.customer_name)
OR t.cust_status <> CASE s.status WHEN '1' THEN 'ACTIVE' WHEN '0' THEN 'INACTIVE' END
OR t.age IS NOT s.age OR t.updated_at <> s.updated_at
)
""",
),
]
def complex_risk_demo_cases(requirement_id: str, version: int) -> list[TestCaseSpec]:
profile_base = {
"requirement_id": requirement_id,
"requirement_version": version,
"etl_task_id": "TASK-RISK-FULL-001",
"database_name": "dwd",
"table_name": "dwd_customer_risk_profile_full",
}
daily_base = {
"requirement_id": requirement_id,
"requirement_version": version,
"etl_task_id": "TASK-RISK-INC-002",
"database_name": "dwd",
"table_name": "dws_customer_trade_risk_di",
}
risk_formula = """
WITH expected AS (
SELECT t.cust_id, t.biz_date, t.risk_score AS actual_score,
t.risk_level AS actual_level,
ROUND(MIN(100.0,
r.base_score * r.score_weight
+ t.cross_border_ratio * 25.0
+ t.large_txn_count * 2.0
+ CASE WHEN t.txn_count = 0 THEN 0
ELSE t.failed_txn_count * 20.0 / t.txn_count END
+ CASE WHEN t.txn_amount_cny >= 500000 THEN 10 ELSE 0 END
), 2) AS expected_score
FROM dwd.dws_customer_trade_risk_di t
JOIN ods.ods_customer_master_full c ON c.cust_id = t.cust_id
JOIN ods.ods_risk_tag_full r ON r.risk_tag_id = c.risk_tag_id
), compared AS (
SELECT *, CASE WHEN expected_score >= 80 THEN 'HIGH'
WHEN expected_score >= 50 THEN 'MEDIUM' ELSE 'LOW' END AS expected_level
FROM expected
)
"""
return [
TestCaseSpec(
**profile_base,
name="dwd_customer_risk_profile_full_全量有效客户数量一致性校验",
fields=["cust_id"], category="reconciliation",
sql="""
SELECT
(SELECT COUNT(*) FROM ods.ods_customer_master_full WHERE status = 'ACTIVE')
- (SELECT COUNT(*) FROM dwd.dwd_customer_risk_profile_full) AS count_difference
""",
assertions=[AssertionSpec("equals", "count_difference", 0)],
),
TestCaseSpec(
**profile_base,
name="dwd_customer_risk_profile_full_cust_id字段唯一性校验",
fields=["cust_id"], category="quality",
sql="SELECT COUNT(*) AS duplicate_count FROM (SELECT cust_id FROM dwd.dwd_customer_risk_profile_full GROUP BY cust_id HAVING COUNT(*) > 1)",
assertions=[AssertionSpec("equals", "duplicate_count", 0)],
sample_sql="SELECT cust_id, COUNT(*) AS duplicate_count FROM dwd.dwd_customer_risk_profile_full GROUP BY cust_id HAVING COUNT(*) > 1",
),
TestCaseSpec(
**profile_base,
name="dwd_customer_risk_profile_full_新增风险指标字段完整性校验",
fields=[
"txn_count_30d", "txn_amount_cny_30d", "avg_txn_amount_cny_30d",
"cross_border_ratio_30d", "large_txn_count_30d", "failed_txn_ratio_30d",
"risk_score", "risk_level", "profile_version",
],
category="schema_evolution",
sql="""
SELECT COUNT(*) AS invalid_count
FROM dwd.dwd_customer_risk_profile_full
WHERE txn_count_30d IS NULL OR txn_amount_cny_30d IS NULL
OR avg_txn_amount_cny_30d IS NULL OR cross_border_ratio_30d IS NULL
OR large_txn_count_30d IS NULL OR failed_txn_ratio_30d IS NULL
OR risk_score IS NULL OR risk_level IS NULL OR profile_version <> 2
""",
assertions=[AssertionSpec("equals", "invalid_count", 0)],
sample_sql="SELECT * FROM dwd.dwd_customer_risk_profile_full WHERE risk_score IS NULL OR risk_level IS NULL OR profile_version <> 2",
),
TestCaseSpec(
**profile_base,
name="dwd_customer_risk_profile_full_账户汇总指标多表关联一致性校验",
fields=["cust_id", "total_account_count", "active_account_count", "total_balance"],
category="transformation",
sql="""
WITH expected AS (
SELECT c.cust_id, COUNT(a.account_id) AS total_count,
SUM(CASE WHEN a.status = 'ACTIVE' THEN 1 ELSE 0 END) AS active_count,
ROUND(SUM(CASE WHEN a.status = 'ACTIVE' THEN a.balance ELSE 0 END), 2) AS total_balance
FROM ods.ods_customer_master_full c
LEFT JOIN ods.ods_account_full a ON a.cust_id = c.cust_id
WHERE c.status = 'ACTIVE' GROUP BY c.cust_id
)
SELECT COUNT(*) AS difference_count
FROM expected e JOIN dwd.dwd_customer_risk_profile_full t ON t.cust_id = e.cust_id
WHERE t.total_account_count <> e.total_count
OR t.active_account_count <> e.active_count
OR ABS(t.total_balance - e.total_balance) > 0.01
""",
assertions=[AssertionSpec("equals", "difference_count", 0)],
),
TestCaseSpec(
**daily_base,
name="dws_customer_trade_risk_di_增量分区覆盖完整性校验",
fields=["biz_date"], category="incremental",
sql="SELECT COUNT(DISTINCT biz_date) AS partition_count FROM dwd.dws_customer_trade_risk_di",
assertions=[AssertionSpec("equals", "partition_count", 30)],
),
TestCaseSpec(
**daily_base,
name="dws_customer_trade_risk_di_客户业务日期联合主键唯一性校验",
fields=["cust_id", "biz_date"], category="quality",
sql="SELECT COUNT(*) AS duplicate_count FROM (SELECT cust_id, biz_date FROM dwd.dws_customer_trade_risk_di GROUP BY cust_id, biz_date HAVING COUNT(*) > 1)",
assertions=[AssertionSpec("equals", "duplicate_count", 0)],
sample_sql="SELECT cust_id, biz_date, COUNT(*) AS duplicate_count FROM dwd.dws_customer_trade_risk_di GROUP BY cust_id, biz_date HAVING COUNT(*) > 1",
),
TestCaseSpec(
**daily_base,
name="dws_customer_trade_risk_di_增量交易总量源目标一致性校验",
fields=["txn_count"], category="reconciliation",
sql="""
SELECT
(SELECT COUNT(*) FROM ods.ods_transaction_inc t
JOIN ods.ods_account_full a ON a.account_id = t.account_id AND a.status = 'ACTIVE'
JOIN ods.ods_customer_master_full c ON c.cust_id = a.cust_id AND c.status = 'ACTIVE')
- (SELECT COALESCE(SUM(txn_count), 0) FROM dwd.dws_customer_trade_risk_di)
AS count_difference
""",
assertions=[AssertionSpec("equals", "count_difference", 0)],
),
TestCaseSpec(
**daily_base,
name="dws_customer_trade_risk_di_人民币交易金额汇率换算一致性校验",
fields=["cust_id", "biz_date", "txn_amount_cny"], category="metric",
sql="""
WITH source_daily AS (
SELECT a.cust_id, t.biz_date,
ROUND(SUM(CASE WHEN t.status = 'SUCCESS' THEN t.amount * f.cny_rate ELSE 0 END), 2) AS expected_amount
FROM ods.ods_transaction_inc t
JOIN ods.ods_account_full a ON a.account_id = t.account_id AND a.status = 'ACTIVE'
JOIN ods.ods_customer_master_full c ON c.cust_id = a.cust_id AND c.status = 'ACTIVE'
JOIN ods.ods_fx_rate_full f ON f.currency_code = t.currency_code AND f.rate_date = t.biz_date
GROUP BY a.cust_id, t.biz_date
)
SELECT COUNT(*) AS difference_count
FROM source_daily s
LEFT JOIN dwd.dws_customer_trade_risk_di t
ON t.cust_id = s.cust_id AND t.biz_date = s.biz_date
WHERE t.cust_id IS NULL OR ABS(t.txn_amount_cny - s.expected_amount) > 0.01
""",
assertions=[AssertionSpec("equals", "difference_count", 0)],
),
TestCaseSpec(
**daily_base,
name="dws_customer_trade_risk_di_跨境交易占比分布校验",
fields=["cross_border_ratio"], category="distribution",
sql="SELECT COUNT(*) AS invalid_count FROM dwd.dws_customer_trade_risk_di WHERE cross_border_ratio < 0 OR cross_border_ratio > 1",
assertions=[AssertionSpec("equals", "invalid_count", 0)],
sample_sql="SELECT * FROM dwd.dws_customer_trade_risk_di WHERE cross_border_ratio < 0 OR cross_border_ratio > 1",
),
TestCaseSpec(
**daily_base,
name="dws_customer_trade_risk_di_复合风险评分与等级计算一致性校验",
fields=[
"cust_id", "biz_date", "txn_count", "failed_txn_count", "txn_amount_cny",
"cross_border_ratio", "large_txn_count", "risk_score", "risk_level",
],
category="complex_metric",
sql=risk_formula + """
SELECT COUNT(*) AS difference_count FROM compared
WHERE ABS(actual_score - expected_score) > 0.01 OR actual_level <> expected_level
""",
assertions=[AssertionSpec("equals", "difference_count", 0)],
sample_sql=risk_formula + """
SELECT cust_id, biz_date, actual_score, expected_score, actual_level, expected_level
FROM compared
WHERE ABS(actual_score - expected_score) > 0.01 OR actual_level <> expected_level
""",
sample_limit=20,
),
TestCaseSpec(
**daily_base,
name="dws_customer_trade_risk_di_风险等级字段值分布校验",
fields=["risk_level"], category="distribution",
sql="SELECT COUNT(*) AS invalid_count FROM dwd.dws_customer_trade_risk_di WHERE risk_level NOT IN ('LOW', 'MEDIUM', 'HIGH') OR risk_level IS NULL",
assertions=[AssertionSpec("equals", "invalid_count", 0)],
sample_sql="SELECT * FROM dwd.dws_customer_trade_risk_di WHERE risk_level NOT IN ('LOW', 'MEDIUM', 'HIGH') OR risk_level IS NULL",
),
]

271
src/datatest/cli.py Normal file
View File

@ -0,0 +1,271 @@
from __future__ import annotations
import argparse
import json
import os
import sys
from pathlib import Path
from typing import Any
from .service import DataTestService
from .storage import utc_now
def _print(value: Any, compact: bool = False, flush: bool = False) -> None:
print(
json.dumps(value, ensure_ascii=False, indent=None if compact else 2, default=str),
flush=flush,
)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(prog="datatest", description="ETL 数据测试框架")
parser.add_argument("--home", type=Path, default=Path(os.environ.get("DATATEST_HOME", ".datatest")))
parser.add_argument("--compact", action="store_true", help="输出单行 JSON")
sub = parser.add_subparsers(dest="command", required=True)
sub.add_parser("init", help="初始化本地存储")
demo = sub.add_parser("demo", help="初始化 SQLite 演示需求、数据和案例")
demo.add_argument("--requirement", type=Path, default=Path("examples/requirements/customer_etl.md"))
complex_demo = sub.add_parser("complex-demo", help="初始化大规模复杂 ETL 演示及故意错误")
complex_demo.add_argument(
"--requirement", type=Path,
default=Path("examples/requirements/customer_risk_complex.md"),
)
complex_demo.add_argument("--customers", type=int, default=100_000)
complex_demo.add_argument("--transactions", type=int, default=1_000_000)
complex_demo.add_argument("--no-error", action="store_true", help="不植入演示错误")
requirement_import = sub.add_parser("requirement-import", help="导入并版本化保存需求文档")
requirement_import.add_argument("path", type=Path)
requirement_import.add_argument("--project-id", required=True)
requirement_import.add_argument("--project-name", required=True)
requirement_import.add_argument("--requirement-id", required=True)
requirement_import.add_argument("--requirement-name", required=True)
workflow_reset = sub.add_parser(
"workflow-reset", help="保留需求文档和 SQLite 数据,将需求重置到刚导入状态"
)
workflow_reset.add_argument("requirement_id")
sub.add_parser("projects", help="列出项目")
sub.add_parser("requirements", help="列出需求")
metadata = sub.add_parser("metadata", help="查看或刷新 Metadata")
metadata.add_argument("requirement_id")
metadata.add_argument("--refresh", action="store_true")
cases = sub.add_parser("cases", help="列出案例")
cases.add_argument("requirement_id")
run = sub.add_parser("run", help="运行已审核案例")
run.add_argument("requirement_id")
run.add_argument("--case", action="append", dest="case_ids")
run.add_argument("--batch-id")
run.add_argument("--biz-date")
run_stream = sub.add_parser("run-stream", help="以 JSONL 事件流逐条执行已审核案例")
run_stream.add_argument("requirement_id")
run_stream.add_argument("--case", action="append", dest="case_ids")
run_stream.add_argument("--batch-id")
run_stream.add_argument("--biz-date")
result = sub.add_parser("result", help="查看运行结果")
result.add_argument("run_id")
parse = sub.add_parser("ai-parse", help="使用本机 Codex CLI 解析需求")
parse.add_argument("requirement_id")
parse.add_argument("--schema", type=Path, default=Path("schemas/requirement-extraction.schema.json"))
parse.add_argument("--context", help="补充业务说明,用于重新解析并消除歧义")
generate = sub.add_parser("ai-generate-cases", help="使用本机 Codex CLI 生成待审核案例")
generate.add_argument("requirement_id")
generate.add_argument("--schema", type=Path, default=Path("schemas/test-case.schema.json"))
generate_stream = sub.add_parser(
"ai-generate-cases-stream", help="以 JSONL 事件流生成待审核测试案例"
)
generate_stream.add_argument("requirement_id")
generate_stream.add_argument(
"--schema", type=Path, default=Path("schemas/test-case.schema.json")
)
confirm = sub.add_parser("requirement-confirm", help="确认已解析需求并建立 ETL 任务")
confirm.add_argument("requirement_id")
approve = sub.add_parser("case-approve", help="校验并批准一个候选案例")
approve.add_argument("case_id")
approve_all = sub.add_parser("case-approve-all", help="人工批量批准需求下全部有效草稿案例")
approve_all.add_argument("requirement_id")
reject = sub.add_parser("case-reject", help="驳回一个候选案例")
reject.add_argument("case_id")
reject.add_argument("--comment")
case_chat = sub.add_parser("case-chat", help="与 Codex 沟通调整或补充测试案例")
case_chat.add_argument("requirement_id")
case_chat.add_argument("message")
case_chat.add_argument(
"--schema", type=Path, default=Path("schemas/case-agent-response.schema.json")
)
case_chat_stream = sub.add_parser(
"case-chat-stream", help="以 JSONL 事件流与 Codex 沟通调整或补充测试案例"
)
case_chat_stream.add_argument("requirement_id")
case_chat_stream.add_argument("message")
case_chat_stream.add_argument(
"--schema", type=Path, default=Path("schemas/case-agent-response.schema.json")
)
report = sub.add_parser("report", help="生成一次运行的 Markdown 报告")
report.add_argument("run_id")
analyze = sub.add_parser("analyze-failure", help="使用本机 Codex CLI 调查失败案例根因")
analyze.add_argument("run_id")
analyze.add_argument("case_id")
analyze.add_argument("--schema", type=Path, default=Path("schemas/failure-analysis.schema.json"))
sub.add_parser("dashboard", help="输出 SwiftUI 仪表盘数据")
sub.add_parser("mcp", help="启动本地 STDIO MCP Server")
return parser
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
stream_sequence = 0
try:
service = DataTestService(args.home)
if args.command == "init":
output = {"status": "initialized", "home": str(service.store.home)}
elif args.command == "demo":
output = service.initialize_demo(args.requirement)
elif args.command == "complex-demo":
output = service.initialize_complex_demo(
args.requirement,
customer_count=args.customers,
transaction_count=args.transactions,
introduce_error=not args.no_error,
)
elif args.command == "requirement-import":
output = service.import_requirement(
args.project_id, args.project_name, args.requirement_id,
args.requirement_name, args.path,
)
elif args.command == "workflow-reset":
output = service.reset_requirement_workflow(args.requirement_id)
elif args.command == "projects":
output = service.list_projects()
elif args.command == "requirements":
output = service.list_requirements()
elif args.command == "metadata":
if args.refresh:
requirement = service.store.query(
"SELECT current_version FROM requirements WHERE id = ?", (args.requirement_id,)
)
if not requirement:
raise ValueError(f"需求不存在: {args.requirement_id}")
output = service.refresh_metadata(args.requirement_id, requirement[0]["current_version"])
else:
output = service.latest_metadata(args.requirement_id)
elif args.command == "cases":
output = service.list_cases(args.requirement_id)
elif args.command == "run":
output = service.run_cases(
args.requirement_id, args.case_ids, args.batch_id, args.biz_date
)
elif args.command == "run-stream":
def emit_run_progress(event: dict[str, Any]) -> None:
nonlocal stream_sequence
stream_sequence += 1
_print({
"type": "progress", "sequence": stream_sequence,
"timestamp": utc_now(), **event,
}, True, True)
result = service.run_cases(
args.requirement_id, args.case_ids, args.batch_id, args.biz_date,
emit_run_progress,
)
stream_sequence += 1
_print({
"type": "result", "sequence": stream_sequence,
"event": "result", "run_id": result["run_id"],
"status": result["status"], "total": len(result["results"]),
"detail": "测试批次已完成并保存。", "timestamp": utc_now(),
}, True, True)
return 0
elif args.command == "result":
output = service.get_run(args.run_id)
elif args.command == "ai-parse":
output = service.parse_requirement_with_ai(args.requirement_id, args.schema, args.context)
elif args.command == "ai-generate-cases":
output = service.generate_cases_with_ai(args.requirement_id, args.schema)
elif args.command == "ai-generate-cases-stream":
def emit_generation_progress(event: dict[str, Any]) -> None:
nonlocal stream_sequence
stream_sequence += 1
_print({
"type": "progress", "sequence": stream_sequence,
"timestamp": utc_now(), **event,
}, True, True)
result = service.generate_cases_with_ai(
args.requirement_id, args.schema, emit_generation_progress
)
stream_sequence += 1
_print({
"type": "result", "sequence": stream_sequence, "phase": "complete",
"status": "completed", "title": "测试案例生成完成",
"detail": f"已保存 {len(result)} 条待审核案例草稿。",
"timestamp": utc_now(), "result": result,
}, True, True)
return 0
elif args.command == "requirement-confirm":
output = service.confirm_requirement(args.requirement_id)
elif args.command == "case-approve":
output = service.approve_case(args.case_id)
elif args.command == "case-approve-all":
output = service.approve_all_cases(args.requirement_id)
elif args.command == "case-reject":
output = service.reject_case(args.case_id, args.comment)
elif args.command == "case-chat":
output = service.chat_about_cases_with_ai(
args.requirement_id, args.message, args.schema
)
elif args.command == "case-chat-stream":
def emit_progress(event: dict[str, Any]) -> None:
nonlocal stream_sequence
stream_sequence += 1
_print({
"type": "progress", "sequence": stream_sequence,
"timestamp": utc_now(), **event,
}, True, True)
result = service.chat_about_cases_with_ai(
args.requirement_id, args.message, args.schema, emit_progress
)
stream_sequence += 1
_print({
"type": "result", "sequence": stream_sequence, "phase": "complete",
"status": "completed", "title": "案例协作已完成",
"detail": "新增或修改案例已保存为待审核草稿。",
"timestamp": utc_now(), "result": result,
}, True, True)
return 0
elif args.command == "report":
output = service.generate_report(args.run_id)
elif args.command == "analyze-failure":
output = service.analyze_failure_with_ai(args.run_id, args.case_id, args.schema)
elif args.command == "dashboard":
output = service.dashboard()
elif args.command == "mcp":
from .mcp_server import serve
return serve(service)
else:
raise ValueError(f"未知命令: {args.command}")
_print(output, args.compact)
return 0
except Exception as error:
if args.command in {"case-chat-stream", "ai-generate-cases-stream", "run-stream"}:
stream_sequence += 1
if args.command == "case-chat-stream":
title = "案例协作失败"
elif args.command == "ai-generate-cases-stream":
title = "测试案例生成失败"
else:
title = "测试案例执行失败"
_print({
"type": "error", "sequence": stream_sequence, "phase": "failed",
"status": "failed",
"title": title, "event": "error",
"detail": str(error),
"timestamp": utc_now(),
}, True, True)
else:
_print({"error": str(error)}, True)
return 1
if __name__ == "__main__":
sys.exit(main())

50
src/datatest/domain.py Normal file
View File

@ -0,0 +1,50 @@
from __future__ import annotations
from dataclasses import asdict, dataclass, field
from typing import Any
@dataclass
class AssertionSpec:
type: str
actual: str | None = None
expected: Any = None
minimum: float | None = None
maximum: float | None = None
def to_dict(self) -> dict[str, Any]:
return asdict(self)
@dataclass
class TestCaseSpec:
name: str
requirement_id: str
requirement_version: int
etl_task_id: str
database_name: str
table_name: str
fields: list[str]
category: str
sql: str
assertions: list[AssertionSpec]
sample_sql: str | None = None
sample_limit: int = 100
parameters: dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> dict[str, Any]:
value = asdict(self)
value["assertions"] = [item.to_dict() for item in self.assertions]
return value
@dataclass
class AssertionOutcome:
assertion_type: str
status: str
actual: Any
expected: Any
message: str
def to_dict(self) -> dict[str, Any]:
return asdict(self)

77
src/datatest/executor.py Normal file
View File

@ -0,0 +1,77 @@
from __future__ import annotations
import json
import time
from typing import Any
from .domain import AssertionOutcome, AssertionSpec, TestCaseSpec
from .sqlite_source import SQLiteDataSource
from .validation import validate_read_only_sql
def _assert_value(spec: AssertionSpec, row: dict[str, Any], rows: list[dict[str, Any]]) -> AssertionOutcome:
if spec.type == "result_is_empty":
actual = len(rows)
passed = actual == 0
expected: Any = 0
else:
if not spec.actual:
raise ValueError(f"断言 {spec.type} 缺少 actual 字段")
if spec.actual not in row:
raise ValueError(f"查询结果中不存在断言字段 {spec.actual}")
actual = row[spec.actual]
expected = spec.expected
if spec.type == "equals":
passed = actual == expected
elif spec.type == "not_equals":
passed = actual != expected
elif spec.type == "greater_than":
passed = actual > expected
elif spec.type == "less_than":
passed = actual < expected
elif spec.type in {"between", "change_rate_between"}:
if spec.minimum is None or spec.maximum is None:
raise ValueError(f"断言 {spec.type} 缺少 minimum 或 maximum")
expected = {"minimum": spec.minimum, "maximum": spec.maximum}
passed = spec.minimum <= actual <= spec.maximum
else:
raise ValueError(f"暂不支持断言类型 {spec.type}")
status = "PASS" if passed else "FAIL"
return AssertionOutcome(spec.type, status, actual, expected, f"actual={actual}, expected={expected}")
class DeterministicExecutor:
def __init__(self, datasource: SQLiteDataSource):
self.datasource = datasource
def execute(self, spec: TestCaseSpec) -> dict[str, Any]:
validate_read_only_sql(spec.sql)
started = time.monotonic()
rows: list[dict[str, Any]] = []
samples: list[dict[str, Any]] = []
try:
with self.datasource.connect() as connection:
rows = [dict(row) for row in connection.execute(spec.sql, spec.parameters).fetchall()]
first = rows[0] if rows else {}
assertions = [_assert_value(item, first, rows) for item in spec.assertions]
if any(item.status == "FAIL" for item in assertions) and spec.sample_sql:
samples = [
dict(row)
for row in connection.execute(spec.sample_sql, spec.parameters).fetchmany(spec.sample_limit)
]
status = "FAIL" if any(item.status == "FAIL" for item in assertions) else "PASS"
error = None
except Exception as exc: # execution errors are data, not process crashes
assertions = []
status = "ERROR"
error = str(exc)
duration_ms = round((time.monotonic() - started) * 1000)
return {
"status": status,
"rows": rows,
"assertions": [item.to_dict() for item in assertions],
"samples": samples,
"error": error,
"duration_ms": duration_ms,
"result_json": json.dumps(rows, ensure_ascii=False),
}

229
src/datatest/mcp_server.py Normal file
View File

@ -0,0 +1,229 @@
from __future__ import annotations
import json
import sys
from pathlib import Path
from typing import Any
from .service import DataTestService
TOOLS: list[dict[str, Any]] = [
{
"name": "list_projects",
"description": "列出 DataTest 中的测试项目。只读。",
"inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
"annotations": {"readOnlyHint": True},
},
{
"name": "list_requirements",
"description": "列出按版本保存的 ETL 测试需求。只读。",
"inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
"annotations": {"readOnlyHint": True},
},
{
"name": "inspect_metadata",
"description": "读取某需求最近一次源表和目标表 Metadata 快照。只读。",
"inputSchema": {
"type": "object", "required": ["requirement_id"],
"properties": {"requirement_id": {"type": "string"}},
"additionalProperties": False,
},
"annotations": {"readOnlyHint": True},
},
{
"name": "adjust_test_cases",
"description": "根据明确的用户意见调用本机 Codex CLI 调整或补充案例。所有变更保存为草稿,仍需在 App 中人工审核。",
"inputSchema": {
"type": "object", "required": ["requirement_id", "message"],
"properties": {
"requirement_id": {"type": "string"},
"message": {"type": "string"}
},
"additionalProperties": False,
},
"annotations": {"readOnlyHint": False, "destructiveHint": False},
},
{
"name": "list_test_cases",
"description": "列出某需求的测试案例及审核状态。只读。",
"inputSchema": {
"type": "object", "required": ["requirement_id"],
"properties": {"requirement_id": {"type": "string"}},
"additionalProperties": False,
},
"annotations": {"readOnlyHint": True},
},
{
"name": "parse_requirement",
"description": "调用隔离的本机 Codex CLI 解析需求并保存结构化草稿。会写入新解析记录。",
"inputSchema": {
"type": "object", "required": ["requirement_id"],
"properties": {
"requirement_id": {"type": "string"},
"supplemental_context": {"type": "string"}
},
"additionalProperties": False,
},
"annotations": {"readOnlyHint": False, "destructiveHint": False},
},
{
"name": "generate_test_cases",
"description": "调用隔离的本机 Codex CLI根据需求和真实 Metadata 生成待审核案例。不会自动批准或执行。",
"inputSchema": {
"type": "object", "required": ["requirement_id"],
"properties": {"requirement_id": {"type": "string"}},
"additionalProperties": False,
},
"annotations": {"readOnlyHint": False, "destructiveHint": False},
},
{
"name": "run_test_cases",
"description": "运行某需求下已审核的案例。不能提交任意 SQL如提供 case_ids只运行指定案例。",
"inputSchema": {
"type": "object", "required": ["requirement_id"],
"properties": {
"requirement_id": {"type": "string"},
"case_ids": {"type": "array", "items": {"type": "string"}},
"batch_id": {"type": "string"},
"biz_date": {"type": "string"},
},
"additionalProperties": False,
},
"annotations": {"readOnlyHint": False, "destructiveHint": False},
},
{
"name": "get_test_result",
"description": "查询一次测试运行的断言和差异结果。只读。",
"inputSchema": {
"type": "object", "required": ["run_id"],
"properties": {"run_id": {"type": "string"}},
"additionalProperties": False,
},
"annotations": {"readOnlyHint": True},
},
{
"name": "generate_report",
"description": "根据已保存运行结果生成可追溯的 Markdown 测试报告。",
"inputSchema": {
"type": "object", "required": ["run_id"],
"properties": {"run_id": {"type": "string"}},
"additionalProperties": False,
},
"annotations": {"readOnlyHint": False, "destructiveHint": False},
},
{
"name": "analyze_failure",
"description": "调用隔离的本机 Codex CLI基于需求、Metadata、断言和失败样例调查 FAIL/ERROR 案例根因。",
"inputSchema": {
"type": "object", "required": ["run_id", "case_id"],
"properties": {
"run_id": {"type": "string"},
"case_id": {"type": "string"}
},
"additionalProperties": False
},
"annotations": {"readOnlyHint": False, "destructiveHint": False},
},
{
"name": "get_dashboard",
"description": "获取项目、需求、案例、运行和历史指标概览。只读。",
"inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
"annotations": {"readOnlyHint": True},
},
]
def _call(service: DataTestService, name: str, arguments: dict[str, Any]) -> Any:
if name == "list_projects":
return service.list_projects()
if name == "list_requirements":
return service.list_requirements()
if name == "inspect_metadata":
return service.latest_metadata(arguments["requirement_id"])
if name == "list_test_cases":
return service.list_cases(arguments["requirement_id"])
if name == "parse_requirement":
return service.parse_requirement_with_ai(
arguments["requirement_id"], Path("schemas/requirement-extraction.schema.json"),
arguments.get("supplemental_context"),
)
if name == "generate_test_cases":
return service.generate_cases_with_ai(
arguments["requirement_id"], Path("schemas/test-case.schema.json")
)
if name == "adjust_test_cases":
return service.chat_about_cases_with_ai(
arguments["requirement_id"], arguments["message"],
Path("schemas/case-agent-response.schema.json"),
)
if name == "run_test_cases":
return service.run_cases(
arguments["requirement_id"], arguments.get("case_ids"),
arguments.get("batch_id"), arguments.get("biz_date"),
)
if name == "get_test_result":
return service.get_run(arguments["run_id"])
if name == "generate_report":
return service.generate_report(arguments["run_id"])
if name == "analyze_failure":
return service.analyze_failure_with_ai(
arguments["run_id"], arguments["case_id"],
Path("schemas/failure-analysis.schema.json"),
)
if name == "get_dashboard":
return service.dashboard()
raise ValueError(f"未知 MCP 工具: {name}")
def _result(request_id: Any, value: Any) -> dict[str, Any]:
return {"jsonrpc": "2.0", "id": request_id, "result": value}
def serve(service: DataTestService) -> int:
"""Minimal MCP 2025-03-26 JSON-RPC server over newline-delimited STDIO."""
for raw_line in sys.stdin:
if not raw_line.strip():
continue
request: dict[str, Any] | None = None
try:
request = json.loads(raw_line)
method = request.get("method")
request_id = request.get("id")
if method == "initialize":
response = _result(
request_id,
{
"protocolVersion": "2025-03-26",
"capabilities": {"tools": {"listChanged": False}},
"serverInfo": {"name": "datatest", "version": "0.1.0"},
"instructions": (
"先查看需求和 Metadata再查看或运行案例。只能运行已审核案例"
"不要把缺少历史基线解释为通过。"
),
},
)
elif method == "tools/list":
response = _result(request_id, {"tools": TOOLS})
elif method == "tools/call":
params = request.get("params", {})
value = _call(service, params.get("name", ""), params.get("arguments", {}))
text = json.dumps(value, ensure_ascii=False, default=str)
response = _result(
request_id,
{"content": [{"type": "text", "text": text}], "structuredContent": value},
)
elif method and method.startswith("notifications/"):
continue
elif method == "ping":
response = _result(request_id, {})
else:
raise ValueError(f"不支持 MCP 方法: {method}")
except Exception as error:
response = {
"jsonrpc": "2.0", "id": request.get("id") if request else None,
"error": {"code": -32603, "message": str(error)},
}
sys.stdout.write(json.dumps(response, ensure_ascii=False) + "\n")
sys.stdout.flush()
return 0

1525
src/datatest/service.py Normal file

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,469 @@
from __future__ import annotations
import sqlite3
from contextlib import closing, contextmanager
from pathlib import Path
from typing import Any, Iterator
class SQLiteDataSource:
"""SQLite implementation of the future datasource adapter contract."""
def __init__(self, source_path: Path, target_path: Path):
self.source_path = source_path
self.target_path = target_path
@contextmanager
def connect(self) -> Iterator[sqlite3.Connection]:
connection = sqlite3.connect(":memory:")
connection.row_factory = sqlite3.Row
connection.execute("ATTACH DATABASE ? AS ods", (str(self.source_path),))
connection.execute("ATTACH DATABASE ? AS dwd", (str(self.target_path),))
connection.execute("PRAGMA query_only = ON")
try:
yield connection
finally:
connection.close()
def inspect(
self,
table_refs: set[str] | None = None,
include_samples: bool = False,
sample_limit: int = 3,
) -> dict[str, Any]:
result: dict[str, Any] = {"dialect": "sqlite", "databases": {}}
with self.connect() as connection:
for database in ("ods", "dwd"):
tables = connection.execute(
f"SELECT name, type, sql FROM {database}.sqlite_master "
"WHERE type IN ('table', 'view') AND name NOT LIKE 'sqlite_%' ORDER BY name"
).fetchall()
table_items: list[dict[str, Any]] = []
for table in tables:
if table_refs is not None and f"{database}.{table['name']}" not in table_refs:
continue
safe_name = str(table["name"]).replace("'", "''")
columns = [
dict(row)
for row in connection.execute(
f"PRAGMA {database}.table_info('{safe_name}')"
).fetchall()
]
indexes = [
dict(row)
for row in connection.execute(
f"PRAGMA {database}.index_list('{safe_name}')"
).fetchall()
]
row_count = connection.execute(
f'SELECT COUNT(*) AS count FROM {database}."{table["name"]}"'
).fetchone()["count"]
table_item = {
"name": table["name"],
"type": table["type"],
"ddl": table["sql"],
"columns": columns,
"indexes": indexes,
"row_count": row_count,
}
if include_samples:
safe_limit = max(0, min(int(sample_limit), 20))
rows = connection.execute(
f'SELECT * FROM {database}."{table["name"]}" LIMIT ?',
(safe_limit,),
).fetchall()
table_item["sample_rows"] = [
{
key: (
f"<BLOB {len(value)} bytes>"
if isinstance(value, bytes) else value
)
for key, value in dict(row).items()
}
for row in rows
]
table_items.append(table_item)
result["databases"][database] = table_items
return result
def seed_demo_databases(source_path: Path, target_path: Path) -> None:
source_path.parent.mkdir(parents=True, exist_ok=True)
with closing(sqlite3.connect(source_path)) as connection, connection:
connection.executescript(
"""
DROP TABLE IF EXISTS ods_customer;
CREATE TABLE ods_customer (
customer_id INTEGER NOT NULL,
customer_name TEXT,
status TEXT NOT NULL,
age INTEGER,
is_deleted INTEGER NOT NULL DEFAULT 0,
updated_at TEXT NOT NULL
);
INSERT INTO ods_customer VALUES
(1, ' Alice ', '1', 31, 0, '2026-08-20T08:00:00Z'),
(2, 'Bob', '1', 42, 0, '2026-08-20T08:01:00Z'),
(3, 'Carol', '0', 27, 0, '2026-08-20T08:02:00Z'),
(4, 'Deleted', '1', 50, 1, '2026-08-20T08:03:00Z');
"""
)
with closing(sqlite3.connect(target_path)) as connection, connection:
connection.executescript(
"""
DROP TABLE IF EXISTS dwd_customer_info;
CREATE TABLE dwd_customer_info (
cust_id INTEGER NOT NULL,
cust_name TEXT,
cust_status TEXT NOT NULL,
age INTEGER,
updated_at TEXT NOT NULL
);
CREATE UNIQUE INDEX idx_customer_id ON dwd_customer_info(cust_id);
INSERT INTO dwd_customer_info VALUES
(1, 'Alice', 'ACTIVE', 31, '2026-08-20T08:00:00Z'),
(2, 'Bob', 'ACTIVE', 42, '2026-08-20T08:01:00Z'),
(3, 'Carol', 'INACTIVE', 27, '2026-08-20T08:02:00Z');
"""
)
def seed_complex_databases(
source_path: Path,
target_path: Path,
customer_count: int = 100_000,
transaction_count: int = 1_000_000,
introduce_error: bool = True,
) -> dict[str, int | bool]:
"""Create a deterministic, sizeable multi-table ETL dataset.
The target contains one intentionally corrupted daily risk metric when
``introduce_error`` is true, so the failure-analysis workflow is testable.
"""
if customer_count < 100 or transaction_count < 1_000:
raise ValueError("复杂演示至少需要 100 个客户和 1,000 笔交易")
account_count = customer_count * 2
source_path.parent.mkdir(parents=True, exist_ok=True)
with closing(sqlite3.connect(source_path)) as connection, connection:
connection.execute("PRAGMA synchronous = OFF")
connection.execute("PRAGMA temp_store = MEMORY")
connection.executescript(
f"""
DROP TABLE IF EXISTS ods_transaction_inc;
DROP TABLE IF EXISTS ods_account_full;
DROP TABLE IF EXISTS ods_customer_master_full;
DROP TABLE IF EXISTS ods_risk_tag_full;
DROP TABLE IF EXISTS ods_fx_rate_full;
CREATE TABLE ods_customer_master_full (
cust_id INTEGER PRIMARY KEY,
customer_name TEXT NOT NULL,
customer_type TEXT NOT NULL,
region_code TEXT NOT NULL,
risk_tag_id INTEGER NOT NULL,
register_date TEXT NOT NULL,
status TEXT NOT NULL,
load_date TEXT NOT NULL
);
CREATE TABLE ods_account_full (
account_id INTEGER PRIMARY KEY,
cust_id INTEGER NOT NULL,
account_type TEXT NOT NULL,
balance REAL NOT NULL,
status TEXT NOT NULL,
open_date TEXT NOT NULL,
load_date TEXT NOT NULL
);
CREATE TABLE ods_risk_tag_full (
risk_tag_id INTEGER PRIMARY KEY,
risk_tag_code TEXT NOT NULL,
base_score REAL NOT NULL,
score_weight REAL NOT NULL,
effective_date TEXT NOT NULL,
expiry_date TEXT
);
CREATE TABLE ods_fx_rate_full (
currency_code TEXT NOT NULL,
rate_date TEXT NOT NULL,
cny_rate REAL NOT NULL,
PRIMARY KEY(currency_code, rate_date)
);
CREATE TABLE ods_transaction_inc (
txn_id INTEGER PRIMARY KEY,
account_id INTEGER NOT NULL,
txn_time TEXT NOT NULL,
biz_date TEXT NOT NULL,
txn_type TEXT NOT NULL,
currency_code TEXT NOT NULL,
amount REAL NOT NULL,
channel TEXT NOT NULL,
counterparty_country TEXT NOT NULL,
is_cross_border INTEGER NOT NULL,
status TEXT NOT NULL,
update_seq INTEGER NOT NULL
);
INSERT INTO ods_risk_tag_full VALUES
(1, 'NORMAL', 10.0, 1.00, '2026-01-01', NULL),
(2, 'WATCH', 35.0, 1.15, '2026-01-01', NULL),
(3, 'SENSITIVE', 55.0, 1.30, '2026-01-01', NULL),
(4, 'HIGH_RISK', 75.0, 1.50, '2026-01-01', NULL);
WITH RECURSIVE seq(n) AS (
SELECT 1 UNION ALL SELECT n + 1 FROM seq WHERE n < {customer_count}
)
INSERT INTO ods_customer_master_full
SELECT
n,
printf('Customer-%06d', n),
CASE WHEN n % 11 = 0 THEN 'CORPORATE' ELSE 'PERSONAL' END,
printf('R%02d', ((n - 1) % 20) + 1),
((n * 17) % 4) + 1,
date('2018-01-01', '+' || (n % 3000) || ' days'),
CASE WHEN n % 20 = 0 THEN 'INACTIVE' ELSE 'ACTIVE' END,
'2026-08-22'
FROM seq;
WITH RECURSIVE seq(n) AS (
SELECT 1 UNION ALL SELECT n + 1 FROM seq WHERE n < {account_count}
)
INSERT INTO ods_account_full
SELECT
n,
((n - 1) % {customer_count}) + 1,
CASE n % 3 WHEN 0 THEN 'CURRENT' WHEN 1 THEN 'SAVING' ELSE 'CREDIT' END,
ROUND(((n * 7919) % 50000000) / 100.0, 2),
CASE WHEN n % 25 = 0 THEN 'CLOSED' ELSE 'ACTIVE' END,
date('2019-01-01', '+' || (n % 2500) || ' days'),
'2026-08-22'
FROM seq;
WITH RECURSIVE days(n) AS (
SELECT 0 UNION ALL SELECT n + 1 FROM days WHERE n < 29
), currencies(code, base_rate) AS (
VALUES ('CNY', 1.0), ('USD', 7.18), ('EUR', 7.82), ('JPY', 0.049)
)
INSERT INTO ods_fx_rate_full
SELECT code, date('2026-07-24', '+' || n || ' days'),
ROUND(base_rate * (1.0 + ((n % 7) - 3) * 0.001), 6)
FROM days CROSS JOIN currencies;
WITH RECURSIVE seq(n) AS (
SELECT 1 UNION ALL SELECT n + 1 FROM seq WHERE n < {transaction_count}
)
INSERT INTO ods_transaction_inc
SELECT
n,
((n * 37 - 1) % {account_count}) + 1,
printf('%sT%02d:%02d:%02dZ',
date('2026-07-24', '+' || (n % 30) || ' days'),
n % 24, (n * 7) % 60, (n * 13) % 60),
date('2026-07-24', '+' || (n % 30) || ' days'),
CASE n % 4 WHEN 0 THEN 'TRANSFER' WHEN 1 THEN 'PAYMENT'
WHEN 2 THEN 'CASH' ELSE 'REFUND' END,
CASE n % 4 WHEN 0 THEN 'CNY' WHEN 1 THEN 'USD'
WHEN 2 THEN 'EUR' ELSE 'JPY' END,
ROUND(((n * 15485863) % 25000000) / 100.0 + 1.0, 2),
CASE n % 4 WHEN 0 THEN 'APP' WHEN 1 THEN 'WEB'
WHEN 2 THEN 'ATM' ELSE 'COUNTER' END,
CASE WHEN n % 7 = 0 THEN 'US' WHEN n % 11 = 0 THEN 'SG' ELSE 'CN' END,
CASE WHEN n % 7 = 0 OR n % 11 = 0 THEN 1 ELSE 0 END,
CASE WHEN n % 50 = 0 THEN 'FAILED' ELSE 'SUCCESS' END,
n * 10 + (n % 3)
FROM seq;
CREATE INDEX idx_complex_customer_status ON ods_customer_master_full(status);
CREATE INDEX idx_complex_account_customer ON ods_account_full(cust_id, status);
CREATE INDEX idx_complex_txn_account_date ON ods_transaction_inc(account_id, biz_date);
CREATE INDEX idx_complex_txn_date_status ON ods_transaction_inc(biz_date, status);
"""
)
with closing(sqlite3.connect(target_path)) as connection, connection:
connection.execute("PRAGMA synchronous = OFF")
connection.execute("PRAGMA temp_store = MEMORY")
connection.execute("ATTACH DATABASE ? AS ods", (str(source_path),))
connection.executescript(
"""
DROP TABLE IF EXISTS dwd_customer_risk_profile_full;
DROP TABLE IF EXISTS dws_customer_trade_risk_di;
CREATE TABLE dwd_customer_risk_profile_full (
cust_id INTEGER PRIMARY KEY,
cust_name TEXT NOT NULL,
customer_type TEXT NOT NULL,
region_code TEXT NOT NULL,
total_account_count INTEGER NOT NULL,
active_account_count INTEGER NOT NULL,
total_balance REAL NOT NULL,
risk_tag_code TEXT NOT NULL,
txn_count_30d INTEGER NOT NULL,
txn_amount_cny_30d REAL NOT NULL,
avg_txn_amount_cny_30d REAL NOT NULL,
cross_border_ratio_30d REAL NOT NULL,
large_txn_count_30d INTEGER NOT NULL,
failed_txn_ratio_30d REAL NOT NULL,
risk_score REAL NOT NULL,
risk_level TEXT NOT NULL,
data_quality_flag TEXT NOT NULL,
profile_version INTEGER NOT NULL,
etl_batch_date TEXT NOT NULL
);
CREATE TABLE dws_customer_trade_risk_di (
cust_id INTEGER NOT NULL,
biz_date TEXT NOT NULL,
txn_count INTEGER NOT NULL,
successful_txn_count INTEGER NOT NULL,
failed_txn_count INTEGER NOT NULL,
txn_amount_cny REAL NOT NULL,
avg_txn_amount_cny REAL NOT NULL,
max_txn_amount_cny REAL NOT NULL,
cross_border_count INTEGER NOT NULL,
cross_border_ratio REAL NOT NULL,
large_txn_count INTEGER NOT NULL,
risk_score REAL NOT NULL,
risk_level TEXT NOT NULL,
source_max_update_seq INTEGER NOT NULL,
etl_batch_time TEXT NOT NULL,
PRIMARY KEY(cust_id, biz_date)
);
WITH account_features AS (
SELECT cust_id,
COUNT(*) AS total_account_count,
SUM(CASE WHEN status = 'ACTIVE' THEN 1 ELSE 0 END) AS active_account_count,
ROUND(SUM(CASE WHEN status = 'ACTIVE' THEN balance ELSE 0 END), 2) AS total_balance
FROM ods.ods_account_full
GROUP BY cust_id
),
txn_features AS (
SELECT a.cust_id,
COUNT(*) AS txn_count,
SUM(CASE WHEN t.status = 'SUCCESS' THEN 1 ELSE 0 END) AS success_count,
SUM(CASE WHEN t.status = 'FAILED' THEN 1 ELSE 0 END) AS failed_count,
SUM(CASE WHEN t.status = 'SUCCESS' THEN t.amount * f.cny_rate ELSE 0 END) AS raw_amount_cny,
ROUND(SUM(CASE WHEN t.status = 'SUCCESS' THEN t.amount * f.cny_rate ELSE 0 END), 2) AS amount_cny,
SUM(CASE WHEN t.status = 'SUCCESS' AND t.is_cross_border = 1 THEN 1 ELSE 0 END) AS cross_border_count,
SUM(CASE WHEN t.status = 'SUCCESS' AND t.amount * f.cny_rate >= 50000 THEN 1 ELSE 0 END) AS large_count
FROM ods.ods_transaction_inc t
JOIN ods.ods_account_full a ON a.account_id = t.account_id AND a.status = 'ACTIVE'
JOIN ods.ods_fx_rate_full f
ON f.currency_code = t.currency_code AND f.rate_date = t.biz_date
GROUP BY a.cust_id
),
features AS (
SELECT c.cust_id, c.customer_name, c.customer_type, c.region_code,
COALESCE(a.total_account_count, 0) AS total_account_count,
COALESCE(a.active_account_count, 0) AS active_account_count,
COALESCE(a.total_balance, 0) AS total_balance,
r.risk_tag_code, r.base_score, r.score_weight,
COALESCE(t.txn_count, 0) AS txn_count,
COALESCE(t.success_count, 0) AS success_count,
COALESCE(t.failed_count, 0) AS failed_count,
COALESCE(t.raw_amount_cny, 0) AS raw_amount_cny,
COALESCE(t.amount_cny, 0) AS amount_cny,
COALESCE(t.cross_border_count, 0) AS cross_border_count,
COALESCE(t.large_count, 0) AS large_count
FROM ods.ods_customer_master_full c
JOIN ods.ods_risk_tag_full r ON r.risk_tag_id = c.risk_tag_id
LEFT JOIN account_features a ON a.cust_id = c.cust_id
LEFT JOIN txn_features t ON t.cust_id = c.cust_id
WHERE c.status = 'ACTIVE'
),
scored AS (
SELECT *,
ROUND(MIN(100.0,
base_score * score_weight
+ CASE WHEN success_count = 0 THEN 0 ELSE cross_border_count * 25.0 / success_count END
+ large_count * 0.4
+ CASE WHEN txn_count = 0 THEN 0 ELSE failed_count * 15.0 / txn_count END
+ CASE WHEN amount_cny >= 1000000 THEN 5 ELSE 0 END
), 2) AS calculated_score
FROM features
)
INSERT INTO dwd_customer_risk_profile_full
SELECT cust_id, customer_name, customer_type, region_code,
total_account_count, active_account_count, total_balance, risk_tag_code,
txn_count, amount_cny,
CASE WHEN success_count = 0 THEN 0 ELSE ROUND(raw_amount_cny / success_count, 2) END,
CASE WHEN success_count = 0 THEN 0 ELSE ROUND(cross_border_count * 1.0 / success_count, 6) END,
large_count,
CASE WHEN txn_count = 0 THEN 0 ELSE ROUND(failed_count * 1.0 / txn_count, 6) END,
calculated_score,
CASE WHEN calculated_score >= 80 THEN 'HIGH'
WHEN calculated_score >= 50 THEN 'MEDIUM' ELSE 'LOW' END,
CASE WHEN active_account_count = 0 THEN 'NO_ACTIVE_ACCOUNT' ELSE 'OK' END,
2,
'2026-08-22'
FROM scored;
WITH daily AS (
SELECT a.cust_id, t.biz_date, r.base_score, r.score_weight,
COUNT(*) AS txn_count,
SUM(CASE WHEN t.status = 'SUCCESS' THEN 1 ELSE 0 END) AS success_count,
SUM(CASE WHEN t.status = 'FAILED' THEN 1 ELSE 0 END) AS failed_count,
ROUND(SUM(CASE WHEN t.status = 'SUCCESS' THEN t.amount * f.cny_rate ELSE 0 END), 2) AS amount_cny,
ROUND(AVG(CASE WHEN t.status = 'SUCCESS' THEN t.amount * f.cny_rate END), 2) AS avg_amount_cny,
ROUND(MAX(CASE WHEN t.status = 'SUCCESS' THEN t.amount * f.cny_rate END), 2) AS max_amount_cny,
SUM(CASE WHEN t.status = 'SUCCESS' AND t.is_cross_border = 1 THEN 1 ELSE 0 END) AS cross_border_count,
SUM(CASE WHEN t.status = 'SUCCESS' AND t.amount * f.cny_rate >= 50000 THEN 1 ELSE 0 END) AS large_count,
MAX(t.update_seq) AS max_update_seq
FROM ods.ods_transaction_inc t
JOIN ods.ods_account_full a ON a.account_id = t.account_id AND a.status = 'ACTIVE'
JOIN ods.ods_customer_master_full c ON c.cust_id = a.cust_id AND c.status = 'ACTIVE'
JOIN ods.ods_risk_tag_full r ON r.risk_tag_id = c.risk_tag_id
JOIN ods.ods_fx_rate_full f
ON f.currency_code = t.currency_code AND f.rate_date = t.biz_date
GROUP BY a.cust_id, t.biz_date, r.base_score, r.score_weight
),
scored AS (
SELECT *,
CASE WHEN success_count = 0 THEN 0 ELSE ROUND(cross_border_count * 1.0 / success_count, 6) END AS cross_border_ratio,
ROUND(MIN(100.0,
base_score * score_weight
+ CASE WHEN success_count = 0 THEN 0 ELSE cross_border_count * 25.0 / success_count END
+ large_count * 2.0
+ failed_count * 20.0 / txn_count
+ CASE WHEN amount_cny >= 500000 THEN 10 ELSE 0 END
), 2) AS calculated_score
FROM daily
)
INSERT INTO dws_customer_trade_risk_di
SELECT cust_id, biz_date, txn_count, success_count, failed_count,
amount_cny, COALESCE(avg_amount_cny, 0), COALESCE(max_amount_cny, 0),
cross_border_count, cross_border_ratio, large_count, calculated_score,
CASE WHEN calculated_score >= 80 THEN 'HIGH'
WHEN calculated_score >= 50 THEN 'MEDIUM' ELSE 'LOW' END,
max_update_seq,
biz_date || 'T23:59:59Z'
FROM scored;
CREATE INDEX idx_complex_profile_region_risk
ON dwd_customer_risk_profile_full(region_code, risk_level);
CREATE INDEX idx_complex_daily_date_risk
ON dws_customer_trade_risk_di(biz_date, risk_level);
"""
)
if introduce_error:
connection.execute(
"""UPDATE dws_customer_trade_risk_di
SET risk_score = 99.99, risk_level = 'LOW'
WHERE rowid = (SELECT MIN(rowid) FROM dws_customer_trade_risk_di)"""
)
profile_rows = connection.execute(
"SELECT COUNT(*) FROM dwd_customer_risk_profile_full"
).fetchone()[0]
daily_rows = connection.execute(
"SELECT COUNT(*) FROM dws_customer_trade_risk_di"
).fetchone()[0]
return {
"customers": customer_count,
"accounts": account_count,
"transactions": transaction_count,
"profile_rows": profile_rows,
"daily_metric_rows": daily_rows,
"intentional_error": introduce_error,
}

228
src/datatest/storage.py Normal file
View File

@ -0,0 +1,228 @@
from __future__ import annotations
import hashlib
import json
import sqlite3
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterator
SCHEMA_VERSION = 2
def utc_now() -> str:
return datetime.now(timezone.utc).isoformat()
class DataStore:
def __init__(self, home: Path):
self.home = home.expanduser().resolve()
self.home.mkdir(parents=True, exist_ok=True)
self.artifacts_dir = self.home / "artifacts"
self.artifacts_dir.mkdir(parents=True, exist_ok=True)
self.db_path = self.home / "app.sqlite"
self.source_path = self.home / "source.sqlite"
self.target_path = self.home / "target.sqlite"
@contextmanager
def connect(self) -> Iterator[sqlite3.Connection]:
connection = sqlite3.connect(self.db_path)
connection.row_factory = sqlite3.Row
connection.execute("PRAGMA foreign_keys = ON")
try:
yield connection
connection.commit()
finally:
connection.close()
def initialize(self) -> None:
with self.connect() as connection:
connection.executescript(
"""
CREATE TABLE IF NOT EXISTS schema_info (
version INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS projects (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS requirements (
id TEXT PRIMARY KEY,
project_id TEXT NOT NULL REFERENCES projects(id),
name TEXT NOT NULL,
status TEXT NOT NULL,
current_version INTEGER NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS requirement_versions (
requirement_id TEXT NOT NULL REFERENCES requirements(id),
version INTEGER NOT NULL,
source_path TEXT NOT NULL,
content_hash TEXT NOT NULL,
content TEXT NOT NULL,
extracted_json TEXT,
created_at TEXT NOT NULL,
PRIMARY KEY(requirement_id, version)
);
CREATE TABLE IF NOT EXISTS etl_tasks (
id TEXT PRIMARY KEY,
requirement_id TEXT NOT NULL REFERENCES requirements(id),
requirement_version INTEGER NOT NULL,
name TEXT NOT NULL,
source_table TEXT,
target_table TEXT NOT NULL,
rule_json TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS metadata_snapshots (
id INTEGER PRIMARY KEY AUTOINCREMENT,
requirement_id TEXT NOT NULL REFERENCES requirements(id),
requirement_version INTEGER NOT NULL,
datasource TEXT NOT NULL,
snapshot_json TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS test_cases (
id TEXT PRIMARY KEY,
requirement_id TEXT NOT NULL REFERENCES requirements(id),
requirement_version INTEGER NOT NULL,
etl_task_id TEXT NOT NULL REFERENCES etl_tasks(id),
version INTEGER NOT NULL,
name TEXT NOT NULL,
table_name TEXT NOT NULL,
category TEXT NOT NULL,
status TEXT NOT NULL,
spec_json TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS test_runs (
id TEXT PRIMARY KEY,
requirement_id TEXT NOT NULL REFERENCES requirements(id),
requirement_version INTEGER NOT NULL,
status TEXT NOT NULL,
batch_id TEXT NOT NULL,
biz_date TEXT,
started_at TEXT NOT NULL,
finished_at TEXT
);
CREATE TABLE IF NOT EXISTS test_case_results (
id INTEGER PRIMARY KEY AUTOINCREMENT,
run_id TEXT NOT NULL REFERENCES test_runs(id),
case_id TEXT NOT NULL REFERENCES test_cases(id),
case_version INTEGER NOT NULL,
status TEXT NOT NULL,
actual_json TEXT,
assertion_json TEXT NOT NULL,
sample_json TEXT,
error_message TEXT,
duration_ms INTEGER NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS metric_snapshots (
id INTEGER PRIMARY KEY AUTOINCREMENT,
requirement_id TEXT NOT NULL REFERENCES requirements(id),
table_name TEXT NOT NULL,
field_name TEXT,
batch_id TEXT NOT NULL,
biz_date TEXT,
metric_type TEXT NOT NULL,
metric_value REAL,
metric_json TEXT,
definition_version INTEGER NOT NULL DEFAULT 1,
collected_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS test_reports (
id TEXT PRIMARY KEY,
run_id TEXT NOT NULL UNIQUE REFERENCES test_runs(id),
requirement_id TEXT NOT NULL REFERENCES requirements(id),
requirement_version INTEGER NOT NULL,
status TEXT NOT NULL,
format TEXT NOT NULL,
file_path TEXT NOT NULL,
content TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS failure_analyses (
id INTEGER PRIMARY KEY AUTOINCREMENT,
result_id INTEGER NOT NULL UNIQUE REFERENCES test_case_results(id),
run_id TEXT NOT NULL REFERENCES test_runs(id),
case_id TEXT NOT NULL REFERENCES test_cases(id),
requirement_id TEXT NOT NULL REFERENCES requirements(id),
status TEXT NOT NULL,
analysis_json TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS agent_invocations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
requirement_id TEXT,
operation TEXT NOT NULL,
input_hash TEXT NOT NULL,
status TEXT NOT NULL,
output_json TEXT,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS case_review_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
case_id TEXT NOT NULL REFERENCES test_cases(id),
requirement_id TEXT NOT NULL REFERENCES requirements(id),
case_version INTEGER NOT NULL,
decision TEXT NOT NULL,
comment TEXT,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS case_agent_messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
requirement_id TEXT NOT NULL REFERENCES requirements(id),
role TEXT NOT NULL,
content TEXT NOT NULL,
operation_json TEXT,
created_at TEXT NOT NULL
);
"""
)
row = connection.execute("SELECT version FROM schema_info LIMIT 1").fetchone()
if row is None:
connection.execute("INSERT INTO schema_info(version) VALUES (?)", (SCHEMA_VERSION,))
elif int(row["version"]) < SCHEMA_VERSION:
connection.execute("UPDATE schema_info SET version = ?", (SCHEMA_VERSION,))
def import_requirement(self, project_id: str, project_name: str, requirement_id: str,
requirement_name: str, source: Path) -> int:
content = source.read_text(encoding="utf-8")
digest = hashlib.sha256(content.encode()).hexdigest()
now = utc_now()
with self.connect() as connection:
connection.execute(
"INSERT OR IGNORE INTO projects(id, name, created_at) VALUES (?, ?, ?)",
(project_id, project_name, now),
)
existing = connection.execute(
"SELECT current_version FROM requirements WHERE id = ?", (requirement_id,)
).fetchone()
version = 1 if existing is None else int(existing["current_version"]) + 1
if existing is None:
connection.execute(
"INSERT INTO requirements VALUES (?, ?, ?, 'imported', ?, ?)",
(requirement_id, project_id, requirement_name, version, now),
)
else:
connection.execute(
"UPDATE requirements SET name = ?, current_version = ?, status = 'imported' WHERE id = ?",
(requirement_name, version, requirement_id),
)
connection.execute(
"INSERT INTO requirement_versions VALUES (?, ?, ?, ?, ?, NULL, ?)",
(requirement_id, version, str(source.resolve()), digest, content, now),
)
return version
def query(self, sql: str, parameters: tuple[Any, ...] = ()) -> list[dict[str, Any]]:
with self.connect() as connection:
return [dict(row) for row in connection.execute(sql, parameters).fetchall()]
@staticmethod
def json(value: Any) -> str:
return json.dumps(value, ensure_ascii=False, sort_keys=True)

View File

@ -0,0 +1,78 @@
from __future__ import annotations
import re
from typing import Any
from .domain import TestCaseSpec
READ_ONLY_START = re.compile(r"^\s*(SELECT|WITH)\b", re.IGNORECASE)
FORBIDDEN_SQL = re.compile(
r"\b(INSERT|UPDATE|DELETE|DROP|ALTER|CREATE|REPLACE|ATTACH|DETACH|VACUUM|REINDEX)\b",
re.IGNORECASE,
)
SUPPORTED_ASSERTION_TYPES = {
"equals", "not_equals", "greater_than", "less_than",
"between", "change_rate_between", "result_is_empty",
}
class ValidationError(ValueError):
pass
def validate_read_only_sql(sql: str) -> None:
normalized = sql.strip()
if not normalized:
raise ValidationError("SQL 不能为空")
if not READ_ONLY_START.search(normalized):
raise ValidationError("测试 SQL 只允许 SELECT 或 WITHMetadata PRAGMA 由适配器内部执行")
without_trailing = normalized.rstrip(";").strip()
if ";" in without_trailing:
raise ValidationError("单个案例只允许一条 SQL")
if FORBIDDEN_SQL.search(normalized):
raise ValidationError("测试 SQL 包含禁止的写入或结构变更语句")
def _metadata_index(metadata: dict[str, Any]) -> dict[str, set[str]]:
result: dict[str, set[str]] = {}
for database, tables in metadata.get("databases", {}).items():
for table in tables:
key = f"{database}.{table['name']}"
result[key] = {column["name"] for column in table.get("columns", [])}
return result
def validate_case(spec: TestCaseSpec, metadata: dict[str, Any]) -> list[str]:
errors: list[str] = []
expected_prefix = f"{spec.table_name}_"
if not spec.name.startswith(expected_prefix):
errors.append(f"案例名称必须以 {expected_prefix} 开头")
try:
validate_read_only_sql(spec.sql)
if spec.sample_sql:
validate_read_only_sql(spec.sample_sql)
except ValidationError as error:
errors.append(str(error))
index = _metadata_index(metadata)
table_key = f"{spec.database_name}.{spec.table_name}"
if table_key not in index:
errors.append(f"Metadata 中不存在表 {table_key}")
else:
missing = sorted(set(spec.fields) - index[table_key])
if missing:
errors.append(f"Metadata 中不存在字段: {', '.join(missing)}")
if not spec.assertions:
errors.append("案例至少需要一个确定性断言")
for index, assertion in enumerate(spec.assertions, start=1):
if assertion.type not in SUPPORTED_ASSERTION_TYPES:
errors.append(f"{index} 个断言类型不受支持: {assertion.type}")
continue
if assertion.type != "result_is_empty" and not assertion.actual:
errors.append(f"{index} 个断言缺少查询结果字段 actual")
if assertion.type in {"between", "change_rate_between"}:
if assertion.minimum is None or assertion.maximum is None:
errors.append(f"{index} 个断言缺少 minimum 或 maximum")
elif assertion.type != "result_is_empty" and assertion.expected is None:
errors.append(f"{index} 个断言缺少 expected")
return errors

102
tests/test_ai.py Normal file
View File

@ -0,0 +1,102 @@
from __future__ import annotations
import io
import json
import subprocess
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from datatest.ai import CodexCLIAdapter
class CodexAdapterTests(unittest.TestCase):
def test_uses_isolated_read_only_structured_invocation(self) -> None:
adapter = CodexCLIAdapter(executable="/usr/bin/true")
with tempfile.TemporaryDirectory() as directory:
schema = Path(directory) / "schema.json"
schema.write_text('{"type":"object"}', encoding="utf-8")
def fake_run(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
output_index = command.index("--output-last-message") + 1
Path(command[output_index]).write_text('{"ok":true}', encoding="utf-8")
self.assertIn("--ignore-user-config", command)
self.assertIn("--ephemeral", command)
self.assertEqual(command[command.index("--sandbox") + 1], "read-only")
self.assertEqual(command[command.index("--output-schema") + 1], str(schema.resolve()))
return subprocess.CompletedProcess(command, 0, "", "")
with patch("datatest.ai.subprocess.run", side_effect=fake_run):
result = adapter.run_structured("解析需求", {"content": "demo"}, schema)
self.assertEqual(result, {"ok": True})
def test_all_structured_output_objects_are_strict(self) -> None:
schemas = [
"requirement-extraction.schema.json",
"test-case.schema.json",
"case-agent-response.schema.json",
"failure-analysis.schema.json",
]
def validate_node(node: object, path: str) -> None:
if isinstance(node, dict):
if node.get("type") == "object":
self.assertIs(node.get("additionalProperties"), False, path)
properties = set(node.get("properties", {}))
self.assertEqual(set(node.get("required", [])), properties, path)
for key, value in node.items():
validate_node(value, f"{path}.{key}")
elif isinstance(node, list):
for index, value in enumerate(node):
validate_node(value, f"{path}[{index}]")
for name in schemas:
schema = json.loads((Path(__file__).resolve().parents[1] / "schemas" / name).read_text())
validate_node(schema, name)
def test_streams_jsonl_events_and_reads_schema_bound_final_message(self) -> None:
adapter = CodexCLIAdapter(executable="/usr/bin/codex")
received: list[dict[str, object]] = []
captured_command: list[str] = []
class FakeProcess:
def __init__(self) -> None:
self.stdin = io.StringIO()
self.stdout = io.StringIO(
'{"type":"thread.started","thread_id":"demo"}\n'
'{"type":"turn.started"}\n'
'{"type":"turn.completed","usage":{}}\n'
)
self.stderr = io.StringIO("")
def wait(self) -> int:
return 0
def kill(self) -> None:
pass
def fake_popen(command: list[str], **_: object) -> FakeProcess:
captured_command.extend(command)
output_index = command.index("--output-last-message") + 1
Path(command[output_index]).write_text('{"ok":true}', encoding="utf-8")
return FakeProcess()
with tempfile.TemporaryDirectory() as directory:
schema = Path(directory) / "schema.json"
schema.write_text('{"type":"object"}', encoding="utf-8")
with patch("datatest.ai.subprocess.Popen", side_effect=fake_popen):
result = adapter.run_structured_streaming(
"调整案例", {"message": "demo"}, schema, received.append
)
self.assertIn("--json", captured_command)
self.assertEqual([event["type"] for event in received], [
"thread.started", "turn.started", "turn.completed",
])
self.assertEqual(result, {"ok": True})
if __name__ == "__main__":
unittest.main()

383
tests/test_core.py Normal file
View File

@ -0,0 +1,383 @@
from __future__ import annotations
import json
import sqlite3
import tempfile
import unittest
from contextlib import closing
from pathlib import Path
from unittest.mock import patch
from datatest.case_factory import customer_demo_cases
from datatest.domain import AssertionSpec, TestCaseSpec
from datatest.service import DataTestService
from datatest.sqlite_source import seed_demo_databases
from datatest.validation import ValidationError, validate_case, validate_read_only_sql
ROOT = Path(__file__).resolve().parents[1]
class DataTestCoreTests(unittest.TestCase):
def setUp(self) -> None:
self.temporary = tempfile.TemporaryDirectory()
self.service = DataTestService(Path(self.temporary.name))
self.service.initialize_demo(ROOT / "examples/requirements/customer_etl.md")
def tearDown(self) -> None:
self.temporary.cleanup()
def test_metadata_is_collected_before_cases(self) -> None:
metadata = self.service.latest_metadata("REQ-CUSTOMER-001")
self.assertEqual([item["name"] for item in metadata["databases"]["ods"]], ["ods_customer"])
self.assertEqual([item["name"] for item in metadata["databases"]["dwd"]], ["dwd_customer_info"])
self.assertEqual(len(self.service.list_cases("REQ-CUSTOMER-001")), 6)
dashboard = self.service.dashboard()
self.assertEqual(len(dashboard["metadata"]), 2)
self.assertEqual(len(dashboard["case_items"]), 6)
self.assertEqual(dashboard["case_items"][0]["requirement_name"], "客户主题 ETL 加工需求")
self.assertEqual(
Path(dashboard["requirements"][0]["source_path"]).name,
"customer_etl.md",
)
target = next(item for item in dashboard["metadata"] if item["database_name"] == "dwd")
self.assertEqual(target["name"], "dwd_customer_info")
self.assertEqual(target["requirement_name"], "客户主题 ETL 加工需求")
self.assertEqual(len(target["columns"]), 5)
def test_all_demo_cases_follow_naming_rule(self) -> None:
cases = self.service.list_cases("REQ-CUSTOMER-001")
self.assertTrue(all(item["name"].startswith(f"{item['table_name']}_") for item in cases))
def test_workflow_reset_preserves_document_and_sqlite_data(self) -> None:
with closing(sqlite3.connect(self.service.store.source_path)) as connection:
source_count = connection.execute("SELECT COUNT(*) FROM ods_customer").fetchone()[0]
result = self.service.reset_requirement_workflow("REQ-CUSTOMER-001")
self.assertEqual(result["status"], "imported")
self.assertEqual(result["removed"]["cases"], 6)
requirement = self.service.list_requirements()[0]
self.assertEqual(requirement["status"], "imported")
self.assertIsNone(requirement["extraction"])
self.assertFalse(requirement["metadata_ready"])
self.assertEqual(self.service.list_cases("REQ-CUSTOMER-001"), [])
self.assertEqual(
self.service.store.query(
"SELECT * FROM etl_tasks WHERE requirement_id = 'REQ-CUSTOMER-001'"
),
[],
)
with self.assertRaisesRegex(ValueError, "尚未获取 Metadata"):
self.service.latest_metadata("REQ-CUSTOMER-001")
with closing(sqlite3.connect(self.service.store.source_path)) as connection:
self.assertEqual(
connection.execute("SELECT COUNT(*) FROM ods_customer").fetchone()[0],
source_count,
)
def test_valid_demo_run_passes_and_persists_metric(self) -> None:
progress_events: list[dict[str, object]] = []
result = self.service.run_cases(
"REQ-CUSTOMER-001", batch_id="2026-08-22", biz_date="2026-08-22",
progress_callback=progress_events.append,
)
self.assertEqual(result["status"], "PASS")
self.assertEqual(progress_events[0]["event"], "run_started")
self.assertEqual(
len([item for item in progress_events if item["event"] == "case_pending"]), 6
)
completed = [item for item in progress_events if item["event"] == "case_completed"]
self.assertEqual(len(completed), 6)
self.assertTrue(all(item["status"] == "PASS" for item in completed))
self.assertEqual(progress_events[-1]["event"], "run_completed")
persisted = self.service.get_run(result["run_id"])
self.assertEqual(len(persisted["results"]), 6)
report = self.service.generate_report(result["run_id"])
self.assertTrue(Path(report["report_path"]).exists())
reports = self.service.dashboard()["reports"]
self.assertEqual(len(reports), 1)
self.assertEqual(reports[0]["requirement_id"], "REQ-CUSTOMER-001")
self.assertIn("# ETL 测试报告", reports[0]["content"])
metrics = self.service.store.query("SELECT * FROM metric_snapshots")
self.assertEqual(metrics[0]["metric_value"], 3)
with self.assertRaisesRegex(ValueError, "只有 FAIL 或 ERROR"):
self.service.analyze_failure_with_ai(
result["run_id"], "CASE-001", ROOT / "schemas/failure-analysis.schema.json"
)
def test_bad_transformation_returns_failure_evidence(self) -> None:
with closing(sqlite3.connect(self.service.store.target_path)) as connection, connection:
connection.execute(
"UPDATE dwd_customer_info SET cust_status = 'BROKEN' WHERE cust_id = 2"
)
result = self.service.run_cases(
"REQ-CUSTOMER-001", case_ids=["CASE-003"], batch_id="bad-data"
)
self.assertEqual(result["status"], "FAIL")
self.assertEqual(result["results"][0]["samples"][0]["cust_id"], 2)
ai_output = {
"summary": "目标状态值不在允许枚举中",
"suspected_layer": "target_data",
"root_cause": "cust_id=2 的目标状态被写为 BROKEN",
"evidence": ["invalid_count=1", "失败样例 cust_id=2"],
"recommendations": ["检查状态映射逻辑"],
"validation_sql": ["SELECT * FROM dwd.dwd_customer_info WHERE cust_id = 2"],
"confidence": "high",
}
with patch("datatest.service.CodexCLIAdapter") as adapter_class:
adapter = adapter_class.return_value
adapter.run_structured.return_value = ai_output
adapter.input_hash.return_value = "test-hash"
analysis = self.service.analyze_failure_with_ai(
result["run_id"], "CASE-003", ROOT / "schemas/failure-analysis.schema.json"
)
self.assertEqual(analysis["analysis"]["confidence"], "high")
dashboard = self.service.dashboard()
self.assertEqual(dashboard["result_items"][0]["status"], "FAIL")
self.assertEqual(dashboard["failure_analyses"][0]["case_id"], "CASE-003")
def test_complex_demo_has_scoped_metadata_large_shapes_and_one_intentional_failure(self) -> None:
initialized = self.service.initialize_complex_demo(
ROOT / "examples/requirements/customer_risk_complex.md",
customer_count=1_000,
transaction_count=12_000,
)
self.assertEqual(initialized["data"]["customers"], 1_000)
self.assertEqual(initialized["data"]["transactions"], 12_000)
metadata = self.service.latest_metadata("REQ-RISK-002")
self.assertEqual(
{item["name"] for item in metadata["databases"]["ods"]},
{
"ods_customer_master_full", "ods_account_full", "ods_risk_tag_full",
"ods_fx_rate_full", "ods_transaction_inc",
},
)
self.assertEqual(
{item["name"] for item in metadata["databases"]["dwd"]},
{"dwd_customer_risk_profile_full", "dws_customer_trade_risk_di"},
)
profile = next(
item for item in metadata["databases"]["dwd"]
if item["name"] == "dwd_customer_risk_profile_full"
)
self.assertIn("risk_score", {item["name"] for item in profile["columns"]})
self.assertIn("profile_version", {item["name"] for item in profile["columns"]})
self.assertEqual(len(self.service.list_cases("REQ-RISK-002")), 11)
result = self.service.run_cases(
"REQ-RISK-002", batch_id="complex-bad-data", biz_date="2026-08-22"
)
failed = [item for item in result["results"] if item["status"] != "PASS"]
self.assertEqual(result["status"], "FAIL")
self.assertEqual(len(failed), 1)
self.assertIn("复合风险评分与等级计算一致性", failed[0]["name"])
self.assertEqual(len(failed[0]["samples"]), 1)
def test_write_sql_is_rejected(self) -> None:
with self.assertRaises(ValidationError):
validate_read_only_sql("DELETE FROM dwd.dwd_customer_info")
with self.assertRaises(ValidationError):
validate_read_only_sql("SELECT 1; DROP TABLE x")
with self.assertRaises(ValidationError):
validate_read_only_sql("PRAGMA writable_schema = ON")
def test_metadata_gate_rejects_unknown_field(self) -> None:
metadata = self.service.latest_metadata("REQ-CUSTOMER-001")
case = TestCaseSpec(
name="dwd_customer_info_未知字段校验",
requirement_id="REQ-CUSTOMER-001", requirement_version=1,
etl_task_id="TASK-CUSTOMER-001", database_name="dwd",
table_name="dwd_customer_info", fields=["not_exists"], category="quality",
sql="SELECT COUNT(*) AS count FROM dwd.dwd_customer_info",
assertions=[AssertionSpec("greater_than", "count", 0)],
)
errors = validate_case(case, metadata)
self.assertTrue(any("not_exists" in item for item in errors))
def test_metadata_gate_rejects_unsupported_assertion_type(self) -> None:
metadata = self.service.latest_metadata("REQ-CUSTOMER-001")
case = TestCaseSpec(
name="dwd_customer_info_错误断言类型校验",
requirement_id="REQ-CUSTOMER-001", requirement_version=1,
etl_task_id="TASK-CUSTOMER-001", database_name="dwd",
table_name="dwd_customer_info", fields=["cust_id"], category="quality",
sql="SELECT COUNT(*) AS count FROM dwd.dwd_customer_info",
assertions=[AssertionSpec("count_equals", "count", 3)],
)
errors = validate_case(case, metadata)
self.assertIn("第 1 个断言类型不受支持: count_equals", errors)
def test_import_to_metadata_generation_review_and_execution_workflow(self) -> None:
with tempfile.TemporaryDirectory() as directory:
service = DataTestService(Path(directory))
seed_demo_databases(service.store.source_path, service.store.target_path)
service.import_requirement(
"PROJECT-FLOW", "完整流程", "REQ-FLOW-001", "客户流程需求",
ROOT / "examples/requirements/customer_etl.md",
)
extraction = {
"requirement_name": "客户流程需求",
"tasks": [{
"name": "客户加工",
"sources": ["ods.ods_customer"],
"targets": ["dwd.dwd_customer_info"],
"field_mappings": [],
"rules": ["过滤 is_deleted = 0"],
}],
"open_questions": [],
}
generated = {
"cases": [{
"name": "dwd_customer_info_客户数量大于零校验",
"database_name": "dwd",
"table_name": "dwd_customer_info",
"fields": ["cust_id"],
"category": "reconciliation",
"sql": "SELECT COUNT(*) AS row_count FROM dwd.dwd_customer_info",
"assertions": [{"type": "greater_than", "actual": "row_count", "expected": 0}],
}]
}
with patch("datatest.service.CodexCLIAdapter") as adapter_class:
adapter = adapter_class.return_value
adapter.run_structured.side_effect = [extraction, generated]
adapter.input_hash.return_value = "flow-hash"
service.parse_requirement_with_ai(
"REQ-FLOW-001", ROOT / "schemas/requirement-extraction.schema.json"
)
candidate = service.latest_metadata("REQ-FLOW-001")
self.assertEqual(candidate["stage"], "candidate")
self.assertEqual(candidate["requirement_scope"], [
"dwd.dwd_customer_info", "ods.ods_customer",
])
self.assertEqual(
candidate["databases"]["ods"][0]["sample_rows"][0]["customer_id"],
1,
)
parse_payload = adapter.run_structured.call_args_list[0].args[1]
self.assertIn("database_catalog", parse_payload)
self.assertEqual(
parse_payload["read_only_database_access"]["command"],
"sqlite3 -readonly <database_path> <SQL>",
)
self.assertEqual(
parse_payload["database_catalog"]["databases"]["dwd"][0]["name"],
"dwd_customer_info",
)
pending = service.list_requirements()[0]
self.assertEqual(pending["status"], "pending_confirmation")
self.assertTrue(pending["metadata_ready"])
confirmed = service.confirm_requirement("REQ-FLOW-001")
self.assertEqual(service.latest_metadata("REQ-FLOW-001")["stage"], "confirmed")
created = service.generate_cases_with_ai(
"REQ-FLOW-001", ROOT / "schemas/test-case.schema.json"
)
self.assertEqual(confirmed["status"], "metadata_ready")
self.assertEqual(created[0]["status"], "draft")
with self.assertRaisesRegex(ValueError, "已审核"):
service.run_cases("REQ-FLOW-001")
service.approve_case(created[0]["id"])
result = service.run_cases("REQ-FLOW-001")
self.assertEqual(result["status"], "PASS")
self.assertEqual(service.list_requirements()[0]["status"], "ready")
def test_agent_exploration_blocks_unknown_database_objects_before_confirmation(self) -> None:
with tempfile.TemporaryDirectory() as directory:
service = DataTestService(Path(directory))
seed_demo_databases(service.store.source_path, service.store.target_path)
service.import_requirement(
"PROJECT-FLOW", "完整流程", "REQ-FLOW-UNKNOWN", "未知表需求",
ROOT / "examples/requirements/customer_etl.md",
)
extraction = {
"requirement_name": "未知表需求",
"tasks": [{
"name": "错误范围",
"sources": ["ods.ods_customer"],
"targets": ["dwd.not_existing"],
"field_mappings": [],
"rules": [],
}],
"open_questions": [],
}
with patch("datatest.service.CodexCLIAdapter") as adapter_class:
adapter = adapter_class.return_value
adapter.run_structured.return_value = extraction
adapter.input_hash.return_value = "unknown-hash"
service.parse_requirement_with_ai(
"REQ-FLOW-UNKNOWN", ROOT / "schemas/requirement-extraction.schema.json"
)
self.assertEqual(
service.latest_metadata("REQ-FLOW-UNKNOWN")["missing_tables"],
["dwd.not_existing"],
)
with self.assertRaisesRegex(ValueError, "数据库中不存在"):
service.confirm_requirement("REQ-FLOW-UNKNOWN")
def test_codex_case_conversation_resets_modified_case_to_draft(self) -> None:
original = self.service.store.query(
"SELECT version, spec_json FROM test_cases WHERE id = 'CASE-001'"
)[0]
spec = json.loads(original["spec_json"])
response = {
"assistant_message": "已将数量校验改为显式大于零,请重新审核。",
"cases": [{
"case_id": "CASE-001",
"name": "dwd_customer_info_目标表客户数量大于零校验",
"database_name": "dwd",
"table_name": "dwd_customer_info",
"fields": ["cust_id"],
"category": "reconciliation",
"sql": "SELECT COUNT(*) AS row_count FROM dwd.dwd_customer_info",
"assertions": [{"type": "greater_than", "actual": "row_count", "expected": 0}],
"sample_sql": spec.get("sample_sql"),
"sample_limit": 100,
}],
"removed_case_ids": [],
}
progress_events: list[dict[str, object]] = []
with patch("datatest.service.CodexCLIAdapter") as adapter_class:
adapter = adapter_class.return_value
def stream_response(
_instruction: str, _payload: dict[str, object], _schema: Path, on_event: object
) -> dict[str, object]:
on_event({"type": "thread.started", "thread_id": "demo"})
on_event({"type": "turn.started"})
on_event({"type": "item.completed", "item": {
"type": "reasoning", "text": "不应传到 UI 的内部内容",
}})
on_event({"type": "turn.completed", "usage": {}})
return response
adapter.run_structured_streaming.side_effect = stream_response
adapter.input_hash.return_value = "chat-hash"
result = self.service.chat_about_cases_with_ai(
"REQ-CUSTOMER-001", "把数量校验改成大于零",
ROOT / "schemas/case-agent-response.schema.json",
progress_events.append,
)
self.assertEqual(result["changed_cases"][0]["status"], "draft")
changed = self.service.store.query("SELECT * FROM test_cases WHERE id = 'CASE-001'")[0]
self.assertEqual(changed["version"], original["version"] + 1)
self.assertEqual(changed["status"], "draft")
with self.assertRaisesRegex(ValueError, "已审核"):
self.service.run_cases("REQ-CUSTOMER-001", case_ids=["CASE-001"])
messages = self.service.dashboard()["case_agent_messages"]
self.assertEqual([item["role"] for item in messages], ["user", "assistant"])
self.assertEqual(progress_events[0]["phase"], "context")
self.assertEqual(progress_events[-1]["phase"], "persistence")
self.assertNotIn(
"不应传到 UI 的内部内容",
" ".join(str(item) for item in progress_events),
)
self.service.approve_case("CASE-001")
self.assertEqual(
self.service.store.query("SELECT status FROM test_cases WHERE id = 'CASE-001'")[0]["status"],
"approved",
)
if __name__ == "__main__":
unittest.main()

41
tests/test_mcp.py Normal file
View File

@ -0,0 +1,41 @@
from __future__ import annotations
import json
import subprocess
import tempfile
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
class MCPServerTests(unittest.TestCase):
def test_initialize_and_list_tools(self) -> None:
with tempfile.TemporaryDirectory() as home:
subprocess.run(
[str(ROOT / "bin/datatest"), "--home", home, "demo"],
cwd=ROOT, check=True, capture_output=True, text=True,
)
messages = "\n".join(
[
json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}),
json.dumps({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}),
"",
]
)
completed = subprocess.run(
[str(ROOT / "bin/datatest"), "--home", home, "mcp"],
cwd=ROOT, input=messages, capture_output=True, text=True, check=True,
)
responses = [json.loads(line) for line in completed.stdout.splitlines()]
self.assertEqual(responses[0]["result"]["serverInfo"]["name"], "datatest")
names = {item["name"] for item in responses[1]["result"]["tools"]}
self.assertIn("inspect_metadata", names)
self.assertIn("run_test_cases", names)
self.assertIn("analyze_failure", names)
self.assertIn("adjust_test_cases", names)
if __name__ == "__main__":
unittest.main()