如何使用Python pytest对两个DataFrame进行全量数据断言对比

你原有代码直接遍历DataFrame对象时，默认返回的是列名，zip(df1, df2)等价于zip(df1.columns, df2.columns)，所以仅对比了列名完全没有校验行级数据，自然所有用例都会通过。

推荐两种实现方式，可根据需求选择：

方案1：直接用pandas内置断言（推荐，报错信息最清晰） #

pandas自带的pd.testing.assert_frame_equal会自动对比两个DataFrame的索引、列、所有行数据，不一致时会直接输出差异的位置和内容，不需要自己写遍历逻辑。
修改后代码如下：

import pytest
import pandas as pd
from sqlalchemy import create_engine
import pymysql

# 数据读取部分保持不变
df1 = pd.read_csv(r'Filename.csv')
sqlEngine = create_engine('mysql+pymysql://root:root@localhost', pool_recycle=3600)
dbConnection = sqlEngine.connect()
df2 = pd.read_sql("SELECT * FROM tablename", dbConnection)
dbConnection.close()

def test_compare_src_trg_data():
    # 可选操作：统一两个DataFrame的列顺序，避免列排序不同导致报错
    df2_sorted = df2[df1.columns]
    # 核心断言，可根据需求调整参数
    pd.testing.assert_frame_equal(
        df1,
        df2_sorted,
        check_dtype=False,  # 如果不需要严格校验数据类型（比如csv读的int和数据库的str）可以设为False
        check_like=False # 如果允许列顺序不同可以设为True
    )

运行后如果数据不一致，会直接报错提示：Name列第0行预期是James，实际是Linda。

方案2：逐行参数化断言（适合需要每行作为独立用例的场景） #

如果需要把每一行都作为独立测试用例，出现错误时不影响其他行的校验，可以按行遍历两个DataFrame生成参数：

import pytest
import pandas as pd
from sqlalchemy import create_engine
import pymysql

# 数据读取部分保持不变
df1 = pd.read_csv(r'Filename.csv')
sqlEngine = create_engine('mysql+pymysql://root:root@localhost', pool_recycle=3600)
dbConnection = sqlEngine.connect()
df2 = pd.read_sql("SELECT * FROM tablename", dbConnection)
dbConnection.close()

# 生成逐行对比的参数
def get_compare_params():
    # 先统一列顺序
    df2_sorted = df2[df1.columns]
    # 按行遍历，返回(行号, 源数据行, 目标数据行)
    for idx, (row1, row2) in enumerate(zip(df1.itertuples(index=False), df2_sorted.itertuples(index=False))):
        yield idx, row1, row2

@pytest.mark.parametrize('idx, row1, row2', get_compare_params())
def test_compare_src_trg_data(idx, row1, row2):
    assert row1 == row2, f"第{idx}行数据不一致"

运行后对应行号的用例会直接失败，提示该行数据不一致。

内容的提问来源于stack exchange，提问作者EverydayLearner