第三方库调用封装函数的异常单元测试必要性及精准异常捕获验证方案咨询

问题背景 #

我封装了一个调用pandas read_sql_query的get_table函数，主要是给不同异常场景添加了针对性的日志输出，然后重新抛出异常。现在有两个核心疑问：

1. 这些针对不同异常的测试是否是浪费时间？毕竟不同异常分支的差异只是日志内容；
2. 当前的测试只能验证异常被抛出，但没法确认是走了对应的except分支，还是被最后的通用Exception分支兜底，有没有办法精准验证？

另外我也纠结，这个函数本质就是对read_sql_query的简单封装，除了日志没别的逻辑，到底需要做多少测试？参数相关的问题read_sql_query本身会抛异常，是不是不用额外测试参数了？

get_table 函数实现 #

def get_table(sql, engine):
    try:
        df = pd.read_sql_query(sql, engine)
        log.info("Imported table from product database using query: " + sql)
    except exc.ObjectNotExecutableError as e:
        log.critical('Invalid \'sql\' parameter - string object not passed: ' + str(e))
        raise
    except exc.ArgumentError as e:
        log.critical('Invalid \'engine\' parameter - engine object not passed: ' + str(e))
        raise
    except exc.ProgrammingError as e:
        log.critical("Could not import table from product database. Invalid query text using read_sql_query: " + str(e))
        raise
    except exc.OperationalError as e:
        log.critical(
            "Could not import table from product database. Failed to connect to database: "
            + str(e)
        )
        raise
    except Exception as e:
        log.critical("Could import table from product database due to unexpected exception: " + str(e))
        raise

    return df

测试代码 #

class TestGetTable:
    def test_success(self, mock_df, monkeypatch):
        monkeypatch.setattr(pd, 'read_sql_query', lambda x, y: mock_df)

        result = udd.get_table('mock_sql', 'mock_engine')
        expected = mock_df

        assert pd.testing.assert_frame_equal(left=result, right=expected, check_dtype=False) is None

    def test_failure_object_not_executable_error(self, mock_read_sql_query_one_error):
        with pytest.raises(exc.ObjectNotExecutableError):
            udd.get_table('mock_sql', 'mock_engine')

    def test_failure_argument_error(self, mock_read_sql_query_argument_error):
        with pytest.raises(exc.ArgumentError):
            udd.get_table('mock_sql', 'mock_engine')

    def test_failure_programming_error(self, mock_read_sql_query_programming_error):
        with pytest.raises(exc.ProgrammingError):
            udd.get_table('mock_sql', 'mock_engine')

    def test_failure_operational_error(self, mock_read_sql_query_operational_error):
        with pytest.raises(exc.OperationalError):
            udd.get_table('mock_sql', 'mock_engine')

    def test_failure_generic_exception(self, mock_read_sql_query_generic_exception):
        with pytest.raises(Exception):
            udd.get_table('mock_sql', 'mock_engine')

conftest.py 夹具 #

@pytest.fixture
def mock_df():
    return pd.DataFrame({'col_a': [1, 2, 4], 'col_b': ['red', 'blue', 'green']})


@pytest.fixture
def mock_read_sql_query(monkeypatch, mock_df):
    monkeypatch.setattr('read_sql_query', mock_df)


# one_error means ObjectNotExecutableError
@pytest.fixture
def mock_read_sql_query_one_error(monkeypatch):
    def fail_read_sql_query(sql, engine):
        raise exc.ObjectNotExecutableError(sql)

    monkeypatch.setattr(pd, 'read_sql_query', fail_read_sql_query)


@pytest.fixture
def mock_read_sql_query_argument_error(monkeypatch):
    def fail_read_sql_query(sql, engine):
        raise exc.ArgumentError()

    monkeypatch.setattr(pd, 'read_sql_query', fail_read_sql_query)


@pytest.fixture
def mock_read_sql_query_programming_error(monkeypatch):
    def fail_read_sql_query(sql, engine):
        raise exc.ProgrammingError('mock_statement', None, Exception())

    monkeypatch.setattr(pd, 'read_sql_query', fail_read_sql_query)


@pytest.fixture
def mock_read_sql_query_operational_error(monkeypatch):
    def fail_read_sql_query(sql, engine):
        raise exc.OperationalError('mock_statement', None, Exception())

    monkeypatch.setattr(pd, 'read_sql_query', fail_read_sql_query)


@pytest.fixture
def mock_read_sql_query_generic_exception(monkeypatch):
    def fail_read_sql_query(sql, engine):
        raise Exception()

    monkeypatch.setattr(pd, 'read_sql_query', fail_read_sql_query)

我的解答 #

一、这些异常测试绝对不是浪费时间！ #

你封装这个函数的核心价值之一就是给不同错误场景添加了精准的日志，这对线上问题排查至关重要，而测试就是要保障这个核心逻辑不失效：

• 日志是排查问题的关键线索：如果ObjectNotExecutableError被错误地打到通用Exception的日志里，排查时会误以为是未知错误，浪费大量时间；
• 防止分支退化：后续维护时，如果有人不小心调整了except分支的顺序（比如把通用Exception提前），这些测试能立刻发现问题；
• 测试本身就是文档：其他开发者看测试用例，就能清楚知道这个函数在哪些错误场景下会输出什么日志、抛出什么异常，比看代码更直观。

至于参数测试，你的异常测试已经间接覆盖了核心场景——比如ObjectNotExecutableError对应sql参数问题，ArgumentError对应engine参数问题，这些测试已经验证了参数异常时函数的行为符合预期。如果想更直接，也可以加几个参数非法的测试用例（比如传非字符串sql、非engine对象），但现有测试已经足够覆盖核心需求。

二、如何验证特定异常被对应分支捕获？ #

当前测试只验证了异常被抛出，但没法确认走的是哪个分支。解决办法很简单：用pytest的caplog夹具捕获日志，验证日志内容是否匹配对应分支的文案。

修改后的测试用例示例：

def test_failure_object_not_executable_error(self, mock_read_sql_query_one_error, caplog):
    caplog.set_level(logging.CRITICAL)  # 设置捕获日志级别
    with pytest.raises(exc.ObjectNotExecutableError):
        udd.get_table('mock_sql', 'mock_engine')
    # 验证日志内容是否符合该分支的文案
    assert "Invalid 'sql' parameter - string object not passed" in caplog.text

同理，其他异常测试也可以添加对应的日志断言：

• ArgumentError测试：检查Invalid 'engine' parameter - engine object not passed是否在日志中；
• ProgrammingError测试：检查Invalid query text using read_sql_query是否在日志中；
• OperationalError测试：检查Failed to connect to database是否在日志中；
• 通用Exception测试：检查unexpected exception是否在日志中。

这样就能确保每个异常都走了对应的处理分支，而不是被通用Exception兜底。

额外小提醒 #

你的代码里有个小笔误：通用Exception分支的日志写的是Could import table，应该是Could not import table，记得修正这个拼写错误哦~

备注：内容来源于stack exchange，提问作者dabo_tusev