Files
ts-anomaly-td/tests/test_e2e.py
T

51 lines
1.7 KiB
Python

"""E2E 测试(需要 TDengine 容器)。"""
import json
import subprocess
import pytest
@pytest.mark.e2e
def test_e2e_generates_png_and_json():
"""跑 e2e Python 子命令,验证 PNG + JSON 输出。"""
project_dir = __import__("pathlib").Path(__file__).parent.parent
# 检查 TDengine 容器是否在运行
result = subprocess.run(
["docker", "compose", "exec", "-T", "tdengine", "taos", "-s", "SELECT 1"],
capture_output=True, text=True, cwd=str(project_dir),
)
if result.returncode != 0:
pytest.skip("TDengine container not running (docker compose up -d first)")
# 清空旧输出
for d in ["render", "logs"]:
p = project_dir / d
if p.exists():
for f in p.iterdir():
f.unlink()
# 跑 e2e Python(不拉容器——假设已有)
r = subprocess.run(
["uv", "run", "python", "-m", "ts_anomaly_td", "e2e",
"--data-dir", "data", "--output-dir", "render", "--log-dir", "logs"],
capture_output=True, text=True, cwd=str(project_dir),
timeout=300,
)
print("STDOUT:", r.stdout[-2000:])
print("STDERR:", r.stderr[-2000:])
# 验证 PNG
pngs = list((project_dir / "render").glob("*.png"))
assert len(pngs) >= 1, f"no PNGs found, stdout: {r.stdout[-500:]}"
for png in pngs:
assert png.stat().st_size > 10240, f"{png.name} too small: {png.stat().st_size} bytes"
# 验证 JSON 日志
logs = list((project_dir / "logs").glob("e2e_*.json"))
assert len(logs) >= 1, "no JSON logs found"
log_data = json.loads(logs[0].read_text())
assert "datasets" in log_data
assert len(log_data["datasets"]) >= 1