70 lines
2.0 KiB
Python
70 lines
2.0 KiB
Python
"""io_csv 模块测试。"""
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
import pytest
|
|
from ts_anomaly_td.io_csv import read_csv
|
|
|
|
|
|
def test_read_csv_basic():
|
|
"""基本读取:三列完整。"""
|
|
content = "timestamp,value,label\n2020-01-01 00:00:00,1.0,0\n2020-01-01 01:00:00,2.0,1\n"
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f:
|
|
f.write(content)
|
|
tmp = f.name
|
|
|
|
try:
|
|
ts, vals, labels = read_csv(tmp)
|
|
assert len(ts) == 2
|
|
assert len(vals) == 2
|
|
assert len(labels) == 2
|
|
assert vals[0] == 1.0
|
|
assert labels[1] == 1
|
|
finally:
|
|
Path(tmp).unlink()
|
|
|
|
|
|
def test_read_csv_no_label_column():
|
|
"""label 列缺失时填 0。"""
|
|
content = "timestamp,value\n2020-01-01 00:00:00,5.0\n"
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f:
|
|
f.write(content)
|
|
tmp = f.name
|
|
|
|
try:
|
|
ts, vals, labels = read_csv(tmp)
|
|
assert labels == [0]
|
|
finally:
|
|
Path(tmp).unlink()
|
|
|
|
|
|
def test_read_csv_missing_required_column():
|
|
"""缺少 timestamp 或 value 列时抛出 ValueError。"""
|
|
content = "time,val\n2020-01-01,1.0\n"
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f:
|
|
f.write(content)
|
|
tmp = f.name
|
|
|
|
try:
|
|
with pytest.raises(ValueError):
|
|
read_csv(tmp)
|
|
finally:
|
|
Path(tmp).unlink()
|
|
|
|
|
|
def test_read_csv_file_not_found():
|
|
"""文件不存在抛出 FileNotFoundError。"""
|
|
with pytest.raises(FileNotFoundError):
|
|
read_csv("/nonexistent/file.csv")
|
|
|
|
|
|
def test_read_csv_finance_001():
|
|
"""验证真实 finance_001.csv 可读。"""
|
|
path = Path(__file__).parent.parent / "data" / "finance_001.csv"
|
|
if not path.exists():
|
|
pytest.skip("finance_001.csv not found")
|
|
ts, vals, labels = read_csv(str(path))
|
|
assert len(ts) == len(vals) == len(labels)
|
|
assert len(ts) > 0
|
|
assert all(isinstance(v, float) for v in vals)
|