如何优雅编写多请求序列的Django Ninja端点E2E测试？

问题背景 #

我用Django Ninja写了一个端点，内部会执行爬虫发起多个请求。根据会话中是否有有效登录凭证，请求序列分3种不同流程。我要为这3种流程和所有可能的异常写E2E测试，但单个请求序列最多包含6个不同页面、不同负载的请求。

目前的方案要么是逐个Mock这些请求，导致代码冗余混乱；要么只依赖单元测试，但每次推送代码都得手动启动服务器测端点，很麻烦。

简化代码示例 #

api.py

# 要测试的端点
@api.get("/data/{data_id}")
def get_data(request, data_id):
    controller = SearchController(crawler=Crawler, scraper=Scraper)
    response_data = controller.get_data(data_id)
    return api.create_response(request=request, data=response_data, status=200)

# 已有5个这类异常处理器，后续还会增加
@api.exception_handler(ChangePasswordError)
def password_change_error_handler(request):
    data = {
        "error": "Failed to login",
        "message": "Provider is demanding a new password be set.",
    }
    return api.create_response(request=request, data=data, status=500)

crawler.py

class Crawler(BaseAPI):
    def __init__(self, session_cookies=cookies):
        # 继承requests会话及方法
        super().__init__()
        if self.session_cookies:
            self.session.cookies = cookiejar_from_dict(session_cookies)
        else:
            self.login()

    # 以下两个方法中的每个页面访问都是负载和请求方法的独特组合
    def login(self):
        self.session.cookies.clear()
        splash_page = self._visit_splash_page()
        login_page = self._visit_login_page(login_page_payload)
        start_page = self._perform_login(login_payload)
        self.save_cookies()
        return start_page

    def get_service_note_page(self, data_id: str) -> dict:
        table_page = self._visit_service_table_page()
        search_results_page = self._search_for_data(search_payload)
        search_details_page = self._visit_data_details(search_details_payload)
        return search_details_page

test_endpoint.py（当前冗余的测试写法）

# 每个patch都会加载本地pickle文件中存储的示例响应
class TestDataSearchEndpoint(unittest.Testcase):
    # 6个不同的patch装饰器
    @mock.patch("visit_splash_page", mock_visit_splash_page)
    # ... 省略其他4个patch
    @mock.patch("visit_data_details", mock_visit_splash_page)
    def test_successful_search_with_login(self):
        # 断言成功逻辑等
        assert success etc ...

    @mock.patch("visit_splash_page", mock_visit_splash_page)
    # ... 省略其他patch
    @mock.patch("perform_login", mock_perform_login_change_password_res, payload)
    def test_failed_login_change_password(self):
        self.assertRaises(call_endpoint, ChangePasswordError)
        self.assertDictEquals(response.json(), sample_error)

解决方案 #

1. 封装Mock组合为自定义装饰器 #

把重复的Mock组合打包成装饰器，每个业务流程对应一个装饰器，避免每个测试方法都写一堆@mock.patch：

def mock_full_login_flow():
    def decorator(func):
        @mock.patch("path.to._visit_splash_page", return_value=splash_response)
        @mock.patch("path.to._visit_login_page", return_value=login_page_response)
        @mock.patch("path.to._perform_login", return_value=start_page_response)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)
        return wrapper
    return decorator

def mock_search_flow_with_existing_cookies():
    def decorator(func):
        @mock.patch("path.to._visit_service_table_page", return_value=table_response)
        @mock.patch("path.to._search_for_data", return_value=search_results_response)
        @mock.patch("path.to._visit_data_details", return_value=details_response)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)
        return wrapper
    return decorator

测试时直接用装饰器简化代码：

class TestDataSearchEndpoint(unittest.Testcase):
    @mock_full_login_flow()
    @mock_search_flow_with_existing_cookies()
    def test_successful_search_with_login(self):
        response = self.client.get("/data/123")
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json(), expected_data)

2. 直接Mock核心业务方法而非私有请求方法 #

跳过对爬虫内部私有请求方法的逐个Mock，直接MockCrawler类的核心业务方法，聚焦端点的行为而非爬虫内部实现：

def test_successful_search_with_login(self):
    with mock.patch("path.to.Crawler") as mock_crawler_cls:
        mock_crawler_instance = mock_crawler_cls.return_value
        mock_crawler_instance.get_service_note_page.return_value = mock_details_response
        
        response = self.client.get("/data/123")
        self.assertEqual(response.status_code, 200)
        # 验证Crawler的初始化和方法调用逻辑
        mock_crawler_cls.assert_called_once()
        mock_crawler_instance.get_service_note_page.assert_called_once_with("123")

3. 测试数据与Mock逻辑分离 #

把本地pickle中的响应数据按流程分类存储（比如test_data/login_success.pkl、test_data/search_success.pkl），再写工具函数统一加载配置：

import pickle

def load_mock_data(filename):
    with open(f"test_data/{filename}.pkl", "rb") as f:
        return pickle.load(f)

def configure_crawler_mocks(mock_crawler, flow_type):
    if flow_type == "login_success":
        mock_crawler._visit_splash_page.return_value = load_mock_data("splash_page")
        mock_crawler._visit_login_page.return_value = load_mock_data("login_page")
        mock_crawler._perform_login.return_value = load_mock_data("login_success")
    elif flow_type == "search_success":
        mock_crawler._visit_service_table_page.return_value = load_mock_data("service_table")
        mock_crawler._search_for_data.return_value = load_mock_data("search_results")
        mock_crawler._visit_data_details.return_value = load_mock_data("data_details")

测试时调用工具函数快速配置Mock：

def test_failed_login_change_password(self):
    with mock.patch("path.to.Crawler") as mock_crawler_cls:
        mock_crawler = mock_crawler_cls.return_value
        configure_crawler_mocks(mock_crawler, "login_change_password")
        # 单独指定异常逻辑
        mock_crawler._perform_login.side_effect = ChangePasswordError()
        
        response = self.client.get("/data/123")
        self.assertEqual(response.status_code, 500)
        self.assertEqual(response.json(), {"error": "Failed to login", "message": "Provider is demanding a new password be set."})

4. 用setUp/tearDown复用公共Mock配置 #

如果多个测试共享基础Mock配置，在测试类的setUp中初始化公共Mock，tearDown中清理，单个测试只调整差异部分：

class TestDataSearchEndpoint(unittest.Testcase):
    def setUp(self):
        self.mock_crawler_patch = mock.patch("path.to.Crawler")
        self.mock_crawler_cls = self.mock_crawler_patch.start()
        self.mock_crawler = self.mock_crawler_cls.return_value
        # 配置公共默认Mock
        self.mock_crawler._visit_splash_page.return_value = load_mock_data("splash_page")
    
    def tearDown(self):
        self.mock_crawler_patch.stop()
    
    def test_successful_search_with_login(self):
        configure_crawler_mocks(self.mock_crawler, "login_success")
        configure_crawler_mocks(self.mock_crawler, "search_success")
        
        response = self.client.get("/data/123")
        self.assertEqual(response.status_code, 200)
    
    def test_failed_login_change_password(self):
        configure_crawler_mocks(self.mock_crawler, "login_change_password")
        self.mock_crawler._perform_login.side_effect = ChangePasswordError()
        
        response = self.client.get("/data/123")
        self.assertEqual(response.status_code, 500)

总结 #

优先Mock核心业务方法而非内部私有请求，减少Mock数量；通过装饰器、工具函数、setUp复用配置，降低代码冗余；同时让测试聚焦端点的输入输出，而非爬虫的内部执行细节。

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