我有一个使用 django ninja 编写的单一端点。该端点执行一个执行多个请求的爬网程序。请求序列有 3 个不同的例程,具体取决于爬网程序在其会话中是否已具有有效的登录凭据。
我想为这 3 个例程以及可能引发的每个可能的异常编写 e2e 测试。问题是,在一个序列中,有多达 6 个不同的请求发送到几个具有不同负载的不同页面。
我能想到的就是为每个测试单独修补每个请求,但这会变得很难看。我的另一个想法是,我应该对爬虫控制器的各个组件的足够的单元测试覆盖率感到满意。
这是说明我的问题的简化代码
api.py
# This is what I want to test
@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)
# I have 5 of these right now and more to come
@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)
爬虫.py
class Crawler(BaseAPI):
def __init__(self, session_cookies=cookies):
# Inherits requests session and methods
super().__init__()
if self.session_cookies:
self.session.cookies = cookiejar_from_dict(session_cookies)
else:
self.login()
# Each page visit in the following two methods are a unique
# combination of payload and request method
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
# Each patch would load a sample response that has been stored locally in a pickle file
class TestDataSearchEndpoint(unittest.Testcase)
# 6 different patches
@mock.patch("visit_splash_page", mock_visit_splash_page)
...
@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)
...
@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)
这感觉很脏。我最终将拥有大约 12 个不同的补丁来覆盖我的所有情况,并且每次测试都必须调用 3-6 个补丁。即使我的单元测试覆盖率很高,我仍然觉得每次推送时都需要运行服务器并手动测试此端点。
我尝试了你的测试方法:
class Crawler(BaseAPI):
def __init__(self, session_cookies=cookies):
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
# Adding individual methods to simplify testing
def _visit_splash_page(self):
# logic to visit splash page
pass
def _visit_login_page(self, payload):
# logic to visit login page
pass
def _perform_login(self, payload):
# logic to perform login
pass
def _visit_service_table_page(self):
# logic to visit service table page
pass
def _search_for_data(self, payload):
# logic to search for data
pass
def _visit_data_details(self, payload):
# logic to visit data details
pass
修复和分组测试:
import unittest
from unittest import mock
# Define fixtures for common responses
def mock_splash_page():
return {"status": "ok", "page": "splash"}
def mock_login_page():
return {"status": "ok", "page": "login"}
def mock_start_page():
return {"status": "ok", "page": "start"}
def mock_service_table_page():
return {"status": "ok", "page": "service_table"}
def mock_search_results_page():
return {"status": "ok", "page": "search_results"}
def mock_search_details_page():
return {"status": "ok", "page": "search_details"}
class TestDataSearchEndpoint(unittest.TestCase):
def setUp(self):
# Setup common patches
self.patcher1 = mock.patch('crawler.Crawler._visit_splash_page', mock_splash_page)
self.patcher2 = mock.patch('crawler.Crawler._visit_login_page', mock_login_page)
self.patcher3 = mock.patch('crawler.Crawler._perform_login', mock_start_page)
self.patcher4 = mock.patch('crawler.Crawler._visit_service_table_page', mock_service_table_page)
self.patcher5 = mock.patch('crawler.Crawler._search_for_data', mock_search_results_page)
self.patcher6 = mock.patch('crawler.Crawler._visit_data_details', mock_search_details_page)
# Start patches
self.mock_splash_page = self.patcher1.start()
self.mock_login_page = self.patcher2.start()
self.mock_start_page = self.patcher3.start()
self.mock_service_table_page = self.patcher4.start()
self.mock_search_results_page = self.patcher5.start()
self.mock_search_details_page = self.patcher6.start()
def tearDown(self):
# Stop patches
self.patcher1.stop()
self.patcher2.stop()
self.patcher3.stop()
self.patcher4.stop()
self.patcher5.stop()
self.patcher6.stop()
def test_successful_search_with_login(self):
response = self.client.get('/data/123')
self.assertEqual(response.status_code, 200)
self.assertIn('page', response.json())
self.assertEqual(response.json()['page'], 'search_details')
@mock.patch('crawler.Crawler._perform_login', side_effect=ChangePasswordError)
def test_failed_login_change_password(self, mock_login):
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.",
})
这样,您的测试将得到清理,减少冗余,并确保您全面覆盖 E2E 场景。