guzhujushi
6758e81c9d
save_table 的两处收尾（用户授权代改）： - 后缀校验原先藏在 if/else 分支的 else 里，排在 path.parent.mkdir(...) 之后， 传 sub/x.txt 会先把 sub\ 建出来再抛 ValueError。改成守卫子句，集中在 mkdir 之前， 校验段结束后才产生副作用；此时 else 分支自然消失，用回 if/else 两路分派 - __all__ 与下一个 def 之间补回第二个空行（PEP 8 / E302） 测试： - 新增 test_save_table_validates_before_touching_disk， 断言参数非法时 tmp_path 下一个条目都不留（ValueError 与 TypeError 两条路径） - 40 passed / 44 skipped，两套环境一致；ruff All checks passed 文档： - docs/dev/personal-pitfalls.md 追加条目 5（参数校验排在副作用之后） 与条目 6（把 lint 的"没报错"当成"没问题"），索引表同步加行 - STATUS：io.py 收尾，L0 第 3 项完成；测试计数 40 - CHANGELOG：Fixed 段补两条
173 lines
5.7 KiB
Python
173 lines
5.7 KiB
Python
"""data 包测试。
|
|
|
|
L0 阶段的约定：**只测已实现的行为**。未实现的算法用逐条 skip 标出，
|
|
不要用模块级 pytestmark —— 那会把同文件里已实现的函数一起跳过，
|
|
让 load_table 这种已经写好的函数变成零覆盖。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
_NOT_IMPLEMENTED = "未实现：见 docs/03-roadmap.md L0/L1"
|
|
|
|
|
|
# ---------------------------------------------------------------- io: load_table
|
|
|
|
|
|
def test_load_table_detects_gbk(tmp_path):
|
|
"""国内赛题数据常见 GBK，编码探测必须能兜住。"""
|
|
from mcm_toolkit.data import load_table
|
|
|
|
p = tmp_path / "gbk.csv"
|
|
p.write_bytes("城市,人口\n北京,2189\n".encode("gbk"))
|
|
df = load_table(p)
|
|
assert list(df.columns) == ["城市", "人口"]
|
|
|
|
|
|
def test_load_table_keeps_utf8_readable(tmp_path):
|
|
"""UTF-8 文件必须按 UTF-8 读出，不能先试 GBK 变成乱码。"""
|
|
from mcm_toolkit.data import load_table
|
|
|
|
p = tmp_path / "utf8.csv"
|
|
p.write_text("省份,GDP增速\n广东,5.8\n", encoding="utf-8")
|
|
df = load_table(p)
|
|
assert list(df.columns) == ["省份", "GDP增速"]
|
|
assert df.loc[0, "省份"] == "广东"
|
|
|
|
|
|
def test_load_table_maps_na_like_tokens_to_nan(tmp_path):
|
|
"""赛题里的缺失值长得五花八门，默认 na_values 要全部吃成 NaN。"""
|
|
from mcm_toolkit.data import load_table
|
|
|
|
p = tmp_path / "na.csv"
|
|
p.write_text("a,b,c\n1,NA,\n2,-,#N/A\n", encoding="utf-8")
|
|
df = load_table(p)
|
|
assert df["b"].isna().all()
|
|
assert df["c"].isna().all()
|
|
assert df["a"].notna().all()
|
|
|
|
|
|
def test_load_table_respects_explicit_encoding(tmp_path):
|
|
"""显式给了 encoding 就照办，不再做探测。"""
|
|
from mcm_toolkit.data import load_table
|
|
|
|
p = tmp_path / "gbk.csv"
|
|
p.write_bytes("城市,人口\n上海,2487\n".encode("gbk"))
|
|
df = load_table(p, encoding="gbk")
|
|
assert list(df.columns) == ["城市", "人口"]
|
|
|
|
|
|
def test_load_table_rejects_unknown_suffix(tmp_path):
|
|
"""不支持的扩展名要抛 ValueError，而且要带上后缀名。"""
|
|
from mcm_toolkit.data import load_table
|
|
|
|
p = tmp_path / "data.txt"
|
|
p.write_text("a,b\n1,2\n", encoding="utf-8")
|
|
with pytest.raises(ValueError, match=r"\.txt"):
|
|
load_table(p)
|
|
|
|
|
|
def test_load_table_missing_file_raises(tmp_path):
|
|
"""文件不存在时要把路径原样带出来，方便定位。"""
|
|
from mcm_toolkit.data import load_table
|
|
|
|
missing = tmp_path / "nope.csv"
|
|
with pytest.raises(FileNotFoundError, match=r"nope\.csv"):
|
|
load_table(missing)
|
|
|
|
|
|
# ---------------------------------------------------------------- io: save_table
|
|
|
|
|
|
def test_save_table_round_trip(tmp_path):
|
|
"""存进去再读出来，内容要一致，并且返回实际写入的 Path。"""
|
|
from mcm_toolkit.data import load_table, save_table
|
|
|
|
df = pd.DataFrame({"城市": ["北京", "上海"], "人口": [2189.0, 2487.0]})
|
|
out = save_table(df, tmp_path / "out.csv")
|
|
assert isinstance(out, Path)
|
|
assert out.exists()
|
|
back = load_table(out)
|
|
assert list(back.columns) == ["城市", "人口"]
|
|
assert back["人口"].tolist() == [2189.0, 2487.0]
|
|
|
|
|
|
def test_save_table_rejects_non_dataframe(tmp_path):
|
|
"""传错类型要当场报错，不要写出一个奇怪的文件。"""
|
|
from mcm_toolkit.data import save_table
|
|
|
|
with pytest.raises(TypeError):
|
|
save_table([1, 2, 3], tmp_path / "bad.csv")
|
|
|
|
|
|
def test_save_table_validates_before_touching_disk(tmp_path):
|
|
"""校验必须早于副作用：参数不合法时，一个目录、一个文件都不能留下。"""
|
|
from mcm_toolkit.data import save_table
|
|
|
|
with pytest.raises(ValueError, match=r"\.txt"):
|
|
save_table(pd.DataFrame({"a": [1]}), tmp_path / "d1" / "x.txt")
|
|
with pytest.raises(TypeError):
|
|
save_table([1, 2, 3], tmp_path / "d2" / "x.csv")
|
|
|
|
assert sorted(p.name for p in tmp_path.iterdir()) == []
|
|
|
|
|
|
# ------------------------------------------------- 以下为未实现算法（L0/L1 待写）
|
|
|
|
|
|
@pytest.mark.skip(reason=_NOT_IMPLEMENTED)
|
|
def test_detect_outliers_finds_planted_points():
|
|
"""人工埋 3 个离群点，IQR 法应正好命中 3 个。"""
|
|
from mcm_toolkit.data import detect_outliers
|
|
|
|
x = np.concatenate([np.arange(100.0), [1e6, -1e6, 1e7]])
|
|
mask = np.asarray(detect_outliers(x, method="iqr"))
|
|
assert mask.sum() == 3
|
|
assert mask[-3:].all()
|
|
|
|
|
|
@pytest.mark.skip(reason=_NOT_IMPLEMENTED)
|
|
def test_detect_outliers_returns_mask_not_filtered_data():
|
|
"""约定：只标不删——返回与原数据同形的布尔掩码。"""
|
|
from mcm_toolkit.data import detect_outliers
|
|
|
|
x = np.arange(50.0)
|
|
mask = np.asarray(detect_outliers(x, method="zscore"))
|
|
assert mask.shape == x.shape
|
|
assert mask.dtype == bool
|
|
|
|
|
|
@pytest.mark.skip(reason=_NOT_IMPLEMENTED)
|
|
def test_normalize_minmax_range(rng):
|
|
from mcm_toolkit.data import normalize
|
|
|
|
x = rng.normal(size=(50, 3))
|
|
out = np.asarray(normalize(x, method="minmax"))
|
|
assert np.allclose(out.min(axis=0), 0.0)
|
|
assert np.allclose(out.max(axis=0), 1.0)
|
|
|
|
|
|
@pytest.mark.skip(reason=_NOT_IMPLEMENTED)
|
|
def test_standardize_has_zero_mean_unit_std(rng):
|
|
from mcm_toolkit.data import standardize
|
|
|
|
x = rng.normal(5, 3, size=1000)
|
|
out = np.asarray(standardize(x))
|
|
assert abs(out.mean()) < 1e-9
|
|
assert abs(out.std() - 1.0) < 1e-9
|
|
|
|
|
|
@pytest.mark.skip(reason=_NOT_IMPLEMENTED)
|
|
def test_train_test_split_sizes_and_no_overlap(rng):
|
|
from mcm_toolkit.data import train_test_split
|
|
|
|
X = rng.normal(size=(100, 2))
|
|
Xtr, Xte = train_test_split(X, test_size=0.2, random_state=0)
|
|
assert len(Xtr) == 80 and len(Xte) == 20
|
|
assert len({tuple(r) for r in Xtr} & {tuple(r) for r in Xte}) == 0
|