Changeset View
Changeset View
Standalone View
Standalone View
swh/lister/phabricator/tests/test_lister.py
| # Copyright (C) 2019 the Software Heritage developers | # Copyright (C) 2019 the Software Heritage developers | ||||
| # 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 json | import json | ||||
| import unittest | import unittest | ||||
| import requests_mock | |||||
| from swh.lister.core.tests.test_lister import HttpListerTester | from swh.lister.core.tests.test_lister import HttpListerTester | ||||
| from swh.lister.phabricator.lister import PhabricatorLister | from swh.lister.phabricator.lister import PhabricatorLister | ||||
| from swh.lister.phabricator.lister import get_repo_url | from swh.lister.phabricator.lister import get_repo_url | ||||
| class PhabricatorListerTester(HttpListerTester, unittest.TestCase): | class PhabricatorListerTester(HttpListerTester, unittest.TestCase): | ||||
| Lister = PhabricatorLister | Lister = PhabricatorLister | ||||
| test_re = re.compile(r'\&after=([^?&]+)') | # first request will have the after parameter empty | ||||
| test_re = re.compile(r'\&after=([^?&]*)') | |||||
| lister_subdir = 'phabricator' | lister_subdir = 'phabricator' | ||||
| good_api_response_file = 'api_response.json' | good_api_response_file = 'api_response.json' | ||||
| good_api_response_undefined_protocol = 'api_response_undefined_'\ | good_api_response_undefined_protocol = 'api_response_undefined_'\ | ||||
| 'protocol.json' | 'protocol.json' | ||||
| bad_api_response_file = 'api_empty_response.json' | bad_api_response_file = 'api_empty_response.json' | ||||
| first_index = 1 | # first_index must be retrieved through a bootstrap process for Phabricator | ||||
| first_index = None | |||||
| last_index = 12 | last_index = 12 | ||||
| entries_per_page = 10 | entries_per_page = 10 | ||||
| convert_type = int | convert_type = int | ||||
| def request_index(self, request): | |||||
| """(Override) This is needed to emulate the listing bootstrap | |||||
| when no min_bound is provided to run | |||||
| """ | |||||
| m = self.test_re.search(request.path_url) | |||||
| idx = m.group(1) | |||||
| if idx == str(self.last_index): | |||||
| return int(idx) | |||||
| def get_fl(self, override_config=None): | def get_fl(self, override_config=None): | ||||
| """(Override) Retrieve an instance of fake lister (fl). | """(Override) Retrieve an instance of fake lister (fl). | ||||
| """ | """ | ||||
| if override_config or self.fl is None: | if override_config or self.fl is None: | ||||
| credentials = {'phabricator': {'fake': [ | credentials = {'phabricator': {'fake': [ | ||||
| {'password': 'toto'} | {'password': 'toto'} | ||||
| ]}} | ]}} | ||||
| Show All 19 Lines | def test_get_repo_url(self): | ||||
| f = open('swh/lister/%s/tests/%s' % | f = open('swh/lister/%s/tests/%s' % | ||||
| (self.lister_subdir, | (self.lister_subdir, | ||||
| self.good_api_response_undefined_protocol)) | self.good_api_response_undefined_protocol)) | ||||
| repo = json.load(f) | repo = json.load(f) | ||||
| self.assertEqual( | self.assertEqual( | ||||
| 'https://svn.blender.org/svnroot/bf-blender/', | 'https://svn.blender.org/svnroot/bf-blender/', | ||||
| get_repo_url(repo['attachments']['uris']['uris'])) | get_repo_url(repo['attachments']['uris']['uris'])) | ||||
| @requests_mock.Mocker() | |||||
| def test_full_listing(self, http_mocker): | |||||
| fl = self.create_fl_with_db(http_mocker) | |||||
| fl.run() | |||||
| self.assertEqual(fl.db_last_index(), self.last_index) | |||||
| ingested_repos = list(fl.db_query_range(self.first_index, | |||||
| self.last_index)) | |||||
| self.assertEqual(len(ingested_repos), self.entries_per_page) | |||||
| @requests_mock.Mocker() | |||||
| def test_range_listing(self, http_mocker): | |||||
| fl = self.create_fl_with_db(http_mocker) | |||||
| fl.run(max_bound=self.last_index - 1) | |||||
| self.assertEqual(fl.db_last_index(), self.last_index - 1) | |||||
| ingested_repos = list(fl.db_query_range(self.first_index, | |||||
| self.last_index)) | |||||
| self.assertEqual(len(ingested_repos), self.entries_per_page - 1) | |||||