如何在Pytest中Mock多次Socket Error异常以适配重试逻辑？

针对你的场景：父类parentjob新增了3次重试逻辑，原测试用例仅mock单次socket.error无法覆盖完整重试流程，这里提供两种可行的Mock方式：

方法一：使用异常列表匹配重试次数 #

直接通过requests-mock的exc参数传入异常列表，列表长度对应重试次数，让每次请求都抛出socket.error：

def test_socketerror(monkeypatch, requests_mock):
    monkeypatch.setenv('ENV_ID', 'Local')
    # 模拟3次socket.error，与重试次数对应
    requests_mock.post(
        "http://localhost:8080/samplejob",
        exc=[socket.error] * 3
    )

    with pytest.raises(CommunicationError) as ce:
        job_id = 'JOB_ID'
        executor = parentjob()
        executor.samplejob(job_id, None, None, None, None)
    assert str(ce.value.args[0]) == "error message on parent method"

逻辑说明 #

• [socket.error] * 3会生成包含3个socket.error实例的列表，requests-mock会按顺序每次请求抛出一个异常。
• 每次异常触发后，parentjob中的response保持为None，执行sleep后进入下一次循环；3次循环结束后，因response仍为None，最终抛出CommunicationError，与断言的错误信息匹配。

方法二：使用side_effect实现动态异常抛出 #

如果后续重试次数可能调整，用side_effect传入一个抛出异常的函数，每次请求都会触发该函数抛出socket.error，不需要固定列表长度：

def test_socketerror(monkeypatch, requests_mock):
    monkeypatch.setenv('ENV_ID', 'Local')
    
    def mock_socketerror():
        raise socket.error
    
    requests_mock.post(
        "http://localhost:8080/samplejob",
        side_effect=mock_socketerror
    )

    with pytest.raises(CommunicationError) as ce:
        job_id = 'JOB_ID'
        executor = parentjob()
        executor.samplejob(job_id, None, None, None, None)
    assert str(ce.value.args[0]) == "error message on parent method"

逻辑说明 #

• side_effect绑定的函数会在每次请求时执行，每次都抛出socket.error，完美适配任意次数的重试逻辑。
• 循环执行完所有重试次数后，parentjob会因response为None抛出CommunicationError，满足测试断言要求。

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