12 种测试实践 —— fixtures、schemas、property-based tests、snapshots、performance guards —— 每周能省不少排查问题的时间
Pandas 的 bug 有个特点，就是不会在控制台里大喊大叫，而是悄悄藏在 dtype 转换、索引操作、时区处理的某个角落，或者那种跑十万次才能复现一次的边界条件。所以如果你想找到和定位这种隐藏的BUG就需要一套相对简洁的测试手段能把大部分坑提前暴露出来。
下面这 12 个策略是实际项目里反复使用的测试方法，能让数据处理代码变得比较靠谱。
1) 用 Pytest Fixtures 做 DataFrame 工厂弄几个小巧的 fixture "工厂"来生成样例数据，这样setup 代码会少写很多，测试逻辑反而能写更充分。
# conftest.py import pandas as pd import numpy as np import pytest @pytest.fixture def sales_df(): return pd.DataFrame({ "order_id": [1, 2, 3], "country": ["IN", "US", "IN"], "amount": [99.0, 149.5, np.nan], "ts": pd.to_datetime(["2025-09-01", "2025-09-02", "2025-09-02"]) })
然后在哪里都能直接用：
def test_revenue_total(sales_df): assert sales_df["amount"].sum(skipna=True) == 248.5
一个标准样本反复使用，减少重复代码，测试的样例也更容易读懂。
2) Schema 层来约束数据Dtype 会漂，列也可能会丢。所以加个 schema 检查，这样违规的数据在边界就会被暴露出来。用 pandera 这类工具也行，或者自己写个轻量检查：
def assert_schema(df, expected): # expected: dict[column] -> dtype string, e.g. {"order_id": "int64", ...} assert set(df.columns) == set(expected), "Columns mismatch" for c, dt in expected.items(): assert str(df[c].dtype) == dt, f"{c} dtype mismatch: {df[c].dtype} != {dt}" def test_schema(sales_df): assert_schema(sales_df, { "order_id": "int64", "country": "object", "amount": "float64", "ts": "datetime64[ns]" })
数据结构变化能第一时间发现，不会传到转换逻辑深处才暴露。
3) Property-Based Testing 检查不变量有些规则应该对任意输入都成立，比如归一化之后总和还是 1。所以可以用 Hypothesis 自动生成各种输入来验证：
from hypothesis import given, strategies as st import pandas as pd import numpy as np @given(st.lists(st.floats(allow_nan=False, width=32), min_size=1, max_size=50)) def test_normalize_preserves_sum(xs): s = pd.Series(xs, dtype="float32") total = float(s.sum()) if total == 0: # define behavior on zero-sum return normalized = s / total assert np.isclose(float(normalized.sum()), 1.0, atol=1e-6)
一个测试用例能自动覆盖几十种形状、数值范围和边界情况。
4) 参数化测试把边缘 case 都列出来有一些经典的麻烦场景：空 DataFrame、单行数据、重复索引、全 null 列、混时区。
import pytest import pandas as pd import numpy as np @pytest.mark.parametrize("df", [ pd.DataFrame(columns=["a","b"]), pd.DataFrame({"a":[1], "b":[np.nan]}), pd.DataFrame({"a":[1,1], "b":[2,2]}).set_index("a"), ]) def test_transform_handles_edges(df): out = df.assign(b=df.get("b", pd.Series(dtype=float)).fillna(0.0)) assert "b" in out
一次性把健壮性锁定，以后就不用反复调同样的边界问题了。
5) Golden Snapshot + 校验和来固定输出复杂输出可以存个"黄金样本"，CI 里对比校验和。如果输出变了会直接报错。
import pandas as pd import hashlib def df_digest(df: pd.DataFrame) -> str: b = df.sort_index(axis=1).to_csv(index=False).encode() return hashlib.md5(b).hexdigest() def test_output_snapshot(tmp_path, sales_df): out = (sales_df .assign(day=sales_df["ts"].dt.date) .groupby(["country","day"], as_index=False)["amount"].sum()) expected = pd.read_parquet("tests/golden/agg.parquet") assert df_digest(out) == df_digest(expected)
6) 固定随机数和时间如果转换依赖随机或者当前时间，得把种子钉死。
import numpy as np import pandas as pd from datetime import datetime def stratified_sample(df, frac, rng): return df.groupby("country", group_keys=False).apply(lambda g: g.sample(frac=frac, random_state=rng)) def test_sample_is_deterministic(sales_df): rng = np.random.default_rng(42) a = stratified_sample(sales_df, 0.5, rng) rng = np.random.default_rng(42) b = stratified_sample(sales_df, 0.5, rng) pd.testing.assert_frame_equal(a.sort_index(), b.sort_index())
这个没什么说的，模型训练的时候也要固定随机种子
7) 明确测试 NA 的语义NaN、None、pd.NA 在不同操作下行为差异挺大的，这时候需要把预期行为显式写出来：
import pandas as pd import numpy as np def test_na_logic(): s = pd.Series([1, np.nan, 3]) s2 = s.fillna(0) assert s2.isna().sum() == 0 assert s2.iloc[1] == 0
NA 相关的 bug 经常藏在 groupby、merge 和数学运算里，得当成一等公民来测。
8) 索引、排序、唯一性约束函数如果保证"索引稳定"就测索引，依赖排序就断言排序状态。
def is_monotonic(df, column): return df[column].is_monotonic_increasing def test_index_and_sort(sales_df): out = sales_df.sort_values(["ts","order_id"]).set_index("order_id") assert out.index.is_unique assert is_monotonic(out.reset_index(), "ts")
很多逻辑错误其实就是顺序错了或者不小心有重复。
9) 双实现交叉验证聪明的向量化逻辑可以用"慢但一看就懂"的实现来验证：
import pandas as pd def vectorized_net(df): return df.assign(net=df["amount"] - df["amount"].mean()) def slow_net(df): mean = df["amount"].mean() df = df.copy() df["net"] = df["amount"].apply(lambda x: x - mean) return df def test_equivalence(sales_df): a = vectorized_net(sales_df) b = slow_net(sales_df) pd.testing.assert_series_equal(a["net"], b["net"], check_names=False)
防止向量化实现出现细微错误，同时保持性能优势。
10) 性能预算作为冒烟测试不需要精确的 benchmark，设个大概的护栏就够了。用小规模代表性数据跑一下，给个时间上限：
import time import pandas as pd def test_runs_fast_enough(sales_df): small = pd.concat([sales_df]*2000, ignore_index=True) # ~6k rows t0 = time.perf_counter() _ = small.groupby("country", as_index=False)["amount"].sum() dt = time.perf_counter() - t0 assert dt < 0.25 # budget for CI environment
11) I/O 往返保证CSV、Parquet、Arrow 格式往返可能会改类型，测一下关心的部分：
import pandas as pd import numpy as np def test_parquet_round_trip(tmp_path, sales_df): p = tmp_path / "sales.parquet" sales_df.to_parquet(p, index=False) back = pd.read_parquet(p) pd.testing.assert_frame_equal( sales_df.sort_index(axis=1), back.sort_index(axis=1), check_like=True )
"本地跑得好好的，生产环境因为 I/O 就挂了"这种谜之问题可能就出现在这里
12) Join 操作的基数和覆盖率检查Merge 是数据质量最容易出问题的地方，基数、重复、覆盖率都得显式断言。
import pandas as pd def test_merge_cardinality(): left = pd.DataFrame({"id":[1,2,3], "x":[10,20,30]}) right = pd.DataFrame({"id":[1,1,2], "y":[5,6,7]}) out = left.merge(right, on="id", how="left") # Expect duplicated rows for id=1 because right has two matches assert (out["id"] == 1).sum() == 2 # Coverage: every left id appears at least once assert set(left["id"]).issubset(set(out["id"]))
key 不唯一或者行数意外翻倍的时候能立刻发现。
小结好的 Pandas 代码不光要写得聪明，更重要的是可预测。这 12 个策略能让正确性变成默认状态：fixtures 快速启动、schemas 早期失败、property-based tests 探索各种古怪情况、简单的性能预算阻止慢代码偷偷溜进来。本周先试两三个，接到 CI 里，那些神秘的数据 bug 基本就消失了。
https://avoid.overfit.cn/post/6eca2c51cef244849e52ae02b932efa9
作者：Syntal