Changeset View
Changeset View
Standalone View
Standalone View
swh/lister/github/tests/test_lister.py
| # Copyright (C) 2017-2019 The Software Heritage developers | # Copyright (C) 2017-2020 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import re | import re | ||||
| import unittest | import unittest | ||||
| import requests_mock | import requests_mock | ||||
| from datetime import datetime, timedelta | |||||
| from swh.lister.core.tests.test_lister import HttpListerTester | from swh.lister.core.tests.test_lister import HttpListerTester | ||||
| from swh.lister.github.lister import GitHubLister | from swh.lister.github.lister import GitHubLister | ||||
| class GitHubListerTester(HttpListerTester, unittest.TestCase): | class GitHubListerTester(HttpListerTester, unittest.TestCase): | ||||
| Lister = GitHubLister | Lister = GitHubLister | ||||
| test_re = re.compile(r'/repositories\?since=([^?&]+)') | test_re = re.compile(r'/repositories\?since=([^?&]+)') | ||||
| lister_subdir = 'github' | lister_subdir = 'github' | ||||
| Show All 19 Lines | def response_headers(self, request): | ||||
| ' rel="first"' | ' rel="first"' | ||||
| }) | }) | ||||
| return headers | return headers | ||||
| def mock_rate_quota(self, n, request, context): | def mock_rate_quota(self, n, request, context): | ||||
| self.rate_limit += 1 | self.rate_limit += 1 | ||||
| context.status_code = 403 | context.status_code = 403 | ||||
| context.headers['X-RateLimit-Remaining'] = '0' | context.headers['X-RateLimit-Remaining'] = '0' | ||||
| one_second = int((datetime.now() + timedelta(seconds=1.5)).timestamp()) | context.headers['Retry-After'] = '1' # 1 second | ||||
| context.headers['X-RateLimit-Reset'] = str(one_second) | |||||
| return '{"error":"dummy"}' | return '{"error":"dummy"}' | ||||
| @requests_mock.Mocker() | @requests_mock.Mocker() | ||||
| def test_scheduled_tasks(self, http_mocker): | def test_scheduled_tasks(self, http_mocker): | ||||
| self.scheduled_tasks_test( | self.scheduled_tasks_test( | ||||
| 'data/https_api.github.com/next_response.json', 876, http_mocker) | 'data/https_api.github.com/next_response.json', 876, http_mocker) | ||||
| Show All 24 Lines | |||||