Changeset View
Changeset View
Standalone View
Standalone View
swh/lister/core/tests/test_lister.py
Show First 20 Lines • Show All 320 Lines • ▼ Show 20 Lines | def test_api_request(self, http_mocker): | ||||
"""Test API request for rate limit handling | """Test API request for rate limit handling | ||||
""" | """ | ||||
http_mocker.get(self.test_re, text=self.mock_limit_twice_response) | http_mocker.get(self.test_re, text=self.mock_limit_twice_response) | ||||
with patch.object(time, 'sleep', wraps=time.sleep) as sleepmock: | with patch.object(time, 'sleep', wraps=time.sleep) as sleepmock: | ||||
self.get_api_response(self.first_index) | self.get_api_response(self.first_index) | ||||
self.assertEqual(sleepmock.call_count, 2) | self.assertEqual(sleepmock.call_count, 2) | ||||
@requests_mock.Mocker() | |||||
def test_request_headers(self, http_mocker): | |||||
fl = self.create_fl_with_db(http_mocker) | |||||
fl.run() | |||||
self.assertNotEqual(len(http_mocker.request_history), 0) | |||||
for request in http_mocker.request_history: | |||||
assert 'User-Agent' in request.headers | |||||
user_agent = request.headers['User-Agent'] | |||||
assert 'Software Heritage Lister' in user_agent | |||||
assert swh.lister.__version__ in user_agent | |||||
def scheduled_tasks_test(self, next_api_response_file, next_last_index, | def scheduled_tasks_test(self, next_api_response_file, next_last_index, | ||||
http_mocker): | http_mocker): | ||||
"""Check that no loading tasks get disabled when processing a new | """Check that no loading tasks get disabled when processing a new | ||||
page of repositories returned by a forge API | page of repositories returned by a forge API | ||||
""" | """ | ||||
fl = self.create_fl_with_db(http_mocker) | fl = self.create_fl_with_db(http_mocker) | ||||
# process first page of repositories listing | # process first page of repositories listing | ||||
▲ Show 20 Lines • Show All 97 Lines • Show Last 20 Lines |