如何Mock requests.models.Response类型以完成Python response_check函数的单元测试

别发愁，Mock requests.models.Response 其实比你想的简单——我们只需要用Python标准库的unittest.mock来模拟这个对象的核心行为：也就是让它的json()方法返回我们预设的响应数据，同时可以控制多次请求的不同返回值。下面我会带你一步步写出覆盖所有场景的单元测试。

核心思路 #

requests.Response对我们的函数来说，关键就是它的json()方法返回的字典，以及我们调用requests.get时能返回这个模拟的响应对象。所以我们可以：

• 用MagicMock创建一个模拟的Response实例，给它的json.return_value赋值为我们需要的状态数据
• 用@patch装饰器替换函数里调用的requests.get方法，让它返回我们的模拟Response
• 同时Mocktime.sleep避免测试时真的等待10秒

完整测试示例 #

假设你的函数放在dataform_tasks.py文件里，下面是对应的测试代码：

import unittest
from unittest.mock import patch, MagicMock
from airflow.exceptions import AirflowException
import time
from dataform_tasks import response_check  # 替换成你的函数所在模块

class TestResponseCheck(unittest.TestCase):
    def setUp(self):
        # 预设基础参数，所有测试用例共用
        self.base_url = "https://api.dataform.example.com/runs"
        self.headers = {"Authorization": "Bearer test-token"}
        self.schedule_name = "daily_sales_load"
        # 初始的response（调用函数时传入的参数）
        self.initial_response = MagicMock()
        self.initial_response.json.return_value = {"id": "test-run-123"}

    @patch('requests.get')
    @patch('time.sleep')
    def test_job_running_then_succeeds(self, mock_sleep, mock_get):
        # 模拟两次requests.get的返回：第一次RUNNING，第二次SUCCESS
        mock_response_running = MagicMock()
        mock_response_running.json.return_value = {"status": "RUNNING"}
        
        mock_response_success = MagicMock()
        mock_response_success.json.return_value = {"status": "SUCCESS"}
        
        # 让mock_get第一次返回RUNNING，第二次返回SUCCESS
        mock_get.side_effect = [mock_response_running, mock_response_success]

        # 执行函数
        result = response_check(
            self.base_url, 
            self.headers, 
            self.schedule_name, 
            self.initial_response
        )

        # 断言结果正确
        self.assertEqual(result, "Dataform job finished")
        # 断言sleep被调用了一次
        mock_sleep.assert_called_once_with(10)
        # 断言requests.get被调用了两次，且URL正确
        expected_url = f"{self.base_url}/test-run-123"
        mock_get.assert_any_call(expected_url, headers=self.headers)
        self.assertEqual(mock_get.call_count, 2)

    @patch('requests.get')
    def test_job_fails_raises_exception(self, mock_get):
        # 模拟返回FAILED状态
        mock_response_failed = MagicMock()
        mock_response_failed.json.return_value = {
            "status": "FAILED",
            "runLogUrl": "https://logs.example.com/fail-123"
        }
        mock_get.return_value = mock_response_failed

        # 断言函数抛出AirflowException，且信息正确
        with self.assertRaises(AirflowException) as ctx:
            response_check(
                self.base_url, 
                self.headers, 
                self.schedule_name, 
                self.initial_response
            )
        
        expected_msg = f'Dataform task {self.schedule_name} has been FAILED for reason https://logs.example.com/fail-123'
        self.assertEqual(str(ctx.exception), expected_msg)
        # 断言requests.get只被调用了一次
        mock_get.assert_called_once()

    @patch('requests.get')
    @patch('time.sleep')
    def test_multiple_running_then_succeeds(self, mock_sleep, mock_get):
        # 模拟三次RUNNING，最后一次SUCCESS
        mock_running = MagicMock()
        mock_running.json.return_value = {"status": "RUNNING"}
        
        mock_success = MagicMock()
        mock_success.json.return_value = {"status": "SUCCESS"}
        
        # side_effect接受列表，按顺序返回
        mock_get.side_effect = [mock_running, mock_running, mock_running, mock_success]

        response_check(
            self.base_url, 
            self.headers, 
            self.schedule_name, 
            self.initial_response
        )

        # 断言sleep被调用了3次
        self.assertEqual(mock_sleep.call_count, 3)
        # 断言requests.get被调用了4次
        self.assertEqual(mock_get.call_count, 4)

if __name__ == '__main__':
    unittest.main()

关键细节说明 #

1. @patch的路径：确保你patch的是函数实际调用的requests模块。比如如果你的response_check函数在my_module.py里，那么你应该patch'my_module.requests.get'，而不是'requests.get'——这是Mock的常见坑，因为你要替换的是函数所在作用域里的requests引用。
2. side_effect的用法：当你需要模拟多次调用的不同返回值时，用side_effect传入一个列表，每次调用会按顺序取列表里的元素返回。如果是固定返回值，用return_value即可。
3. Mock time.sleep：测试循环逻辑时，我们不想真的等待10秒，所以用@patch('time.sleep')替换sleep方法，让它什么都不做。

这样你就能覆盖所有场景：任务正常完成、任务失败抛出异常、多次运行后完成，完美测试你的response_check函数啦！

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