Python API请求Mock测试失败排查：寻求技术协助

我编写的数据提取器功能正常，但对应的单元测试编写遇到阻碍，执行一直失败，请求协助排查问题。

测试代码 #

from unittest.mock import MagicMock, patch
import pandas as pd
from my_project.task import extractor
from my_project.tests import utils
from prefect.logging import disable_run_logger

CONTACT_RECORD = utils.TEST_CONTACT_RECORD
PAGED_CONTACT_RECORD = utils.TEST_PAGED_CONTACT_RECORD
EXPECTED_CONTACT_RECORD = utils.EXPECTED_CONTACT_RECORD


@patch("requests.get")
def test_contact_extractor(get: MagicMock):
    """
    调用一次requests.get并返回包含联系人数据的json
    """
    get.return_value.json.return_value = CONTACT_RECORD

    with disable_run_logger():
        result = extractor.get_contacts()
        assert get.call_count == 1
        assert result == pd.DataFrame(EXPECTED_CONTACT_RECORD)


@patch("my_project.extractor.get_contacts")
def test_get_paged_contacts(get_contacts: MagicMock):
    """
    循环调用requests.get直到['has-more']为False且无offset值
    """
    get_contacts.return_value.json.side_effect = [
        PAGED_CONTACT_RECORD,
        PAGED_CONTACT_RECORD,
        PAGED_CONTACT_RECORD,
        CONTACT_RECORD,
    ]
    with disable_run_logger():
        data = extractor.get_paged_contacts(
            endpoint=MagicMock, query_string=MagicMock, df=MagicMock
        )
    assert get_contacts.call_count == 4
    assert data == pd.DataFrame(EXPECTED_CONTACT_RECORD)

错误历程 #

• 最初错误：
  • requests已导入但未使用
  • callable[[Union[str,bytes],....], Response]没有属性"return_value"
• 修正拼写错误（btyes→bytes）后，第二个错误消失，出现新错误：

  AttributeError: 'NoneType' object has no attribute 'client'
  

get_paged_contacts函数实现 #

import typing
import requests
import time
import json
from pandas import json_normalize
import pandas as pd
from prefect import logger

def get_paged_contacts(
    endpoint: str, query_string: typing.Dict[str, typing.Any], df: pd.DataFrame
) -> pd.DataFrame:
    """
    返回GET请求的结果。
    循环处理API分页响应，将所有页结果合并到传入的DataFrame中。
    """
    url = endpoint
    contacts = []
    # 注意：此处header变量需提前定义，否则会引发NoneType错误
    response = requests.request("GET", url, headers=header, params=query_string).json()
    has_more = response["has-more"]
    offset = response["vid-offset"]

    while has_more is True:
        querystring = {"limit": "100", "archived": "false", "offset": offset}
        try:
            response = requests.request(
                "GET", url, headers=header, params=querystring
            ).json()
            # 原代码未更新has_more和offset，会导致无限循环
            has_more = response["has-more"]
            offset = response["vid-offset"]
            # 原代码未收集中间页数据，仅保留最后一页
            contacts.extend(response["contacts"])
            time.sleep(10)
        except (requests.exceptions.ConnectionError, json.decoder.JSONDecodeError) as j:
            logger.error(f"Error occurred: {j}.")
            break
    # 补充收集最后一页数据
    contacts.extend(response["contacts"])

    contacts = json_normalize(contacts)
    merged = pd.concat([df, contacts])
    return merged

问题排查与修复方案 #

1. 测试代码问题修复 #

针对test_contact_extractor

• 删除未使用的import requests语句
• 用pd.testing.assert_frame_equal替代直接==比较DataFrame（直接比较会返回布尔矩阵，导致断言失败）

  pd.testing.assert_frame_equal(result, pd.DataFrame(EXPECTED_CONTACT_RECORD))
  

针对test_get_paged_contacts

• 修正Patch目标：函数实际调用的是requests.request而非get_contacts，需改为@patch("requests.request")
• 替换MagicMock参数：传入真实的字符串、字典和空DataFrame，避免类型错误
• 模拟正确的返回结构：给requests.request的返回值添加json()方法，模拟API响应
• 调整断言逻辑：断言请求调用次数，并验证最终DataFrame的行数是否符合分页总数

修复后的测试代码：

@patch("requests.request")
def test_get_paged_contacts(request_mock: MagicMock):
    """
    循环调用requests.request直到has-more为False，合并所有分页数据
    """
    # 模拟分页响应：前3页有更多数据，最后一页无更多数据
    mock_responses = [
        MagicMock(json=lambda: PAGED_CONTACT_RECORD),
        MagicMock(json=lambda: PAGED_CONTACT_RECORD),
        MagicMock(json=lambda: PAGED_CONTACT_RECORD),
        MagicMock(json=lambda: CONTACT_RECORD),
    ]
    request_mock.side_effect = mock_responses

    # 传入真实参数
    test_endpoint = "https://api.example.com/contacts"
    test_query_string = {"limit": "100", "archived": "false"}
    test_df = pd.DataFrame()

    with disable_run_logger():
        data = extractor.get_paged_contacts(
            endpoint=test_endpoint, query_string=test_query_string, df=test_df
        )
    
    assert request_mock.call_count == 4
    # 验证总数据量：3页分页数据 + 1页最终数据
    expected_total = len(PAGED_CONTACT_RECORD["contacts"])*3 + len(CONTACT_RECORD["contacts"])
    assert len(data) == expected_total

2. 函数实现问题修复 #

• 定义header变量：将header作为函数参数传入，或者从配置文件读取，避免出现NoneType错误
• 修复分页循环逻辑：每次请求后更新has_more和offset，并收集当前页的contacts，避免无限循环和数据丢失
• 补充缺失的导入：确保time、json、json_normalize、logger等依赖已正确导入

3. 解决AttributeError: 'NoneType' object has no attribute 'client' #

该错误是因为header变量未定义，导致requests.request传入了None作为headers。解决方式：

• 将header改为函数参数：

  def get_paged_contacts(
      endpoint: str, query_string: typing.Dict[str, typing.Any], df: pd.DataFrame, headers: typing.Dict[str, str]
  ) -> pd.DataFrame:
      response = requests.request("GET", url, headers=headers, params=query_string).json()
      # ...后续代码使用headers替代header
  

• 测试时传入模拟的headers：

  test_headers = {"Authorization": "Bearer test_token"}
  data = extractor.get_paged_contacts(
      endpoint=test_endpoint, query_string=test_query_string, df=test_df, headers=test_headers
  )
  

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