Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/common/test_service.py
# Copyright (C) 2015-2019 The Software Heritage developers | # Copyright (C) 2015-2019 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 Affero General Public License version 3, or any later version | # License: GNU Affero 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 itertools | import itertools | ||||
import pytest | import pytest | ||||
import random | import random | ||||
from collections import defaultdict | from collections import defaultdict | ||||
from hypothesis import given | from hypothesis import given | ||||
from hypothesis.strategies import integers | |||||
from swh.model.hashutil import hash_to_bytes, hash_to_hex | from swh.model.hashutil import hash_to_bytes, hash_to_hex | ||||
from swh.model.from_disk import DentryPerms | from swh.model.from_disk import DentryPerms | ||||
from swh.web.common import service | from swh.web.common import service | ||||
from swh.web.common.exc import BadInputExc, NotFoundExc | from swh.web.common.exc import BadInputExc, NotFoundExc | ||||
from swh.web.tests.strategies import ( | from swh.web.tests.strategies import ( | ||||
content, contents, unknown_content, unknown_contents, | content, contents, unknown_content, unknown_contents, | ||||
▲ Show 20 Lines • Show All 167 Lines • ▼ Show 20 Lines | class ServiceTestCase(WebTestCase): | ||||
@given(new_origin(), visit_dates()) | @given(new_origin(), visit_dates()) | ||||
def test_lookup_origin_visits(self, new_origin, visit_dates): | def test_lookup_origin_visits(self, new_origin, visit_dates): | ||||
origin_id = self.storage.origin_add_one(new_origin) | origin_id = self.storage.origin_add_one(new_origin) | ||||
for ts in visit_dates: | for ts in visit_dates: | ||||
self.storage.origin_visit_add(origin_id, ts) | self.storage.origin_visit_add(origin_id, ts) | ||||
actual_origin_visits = list( | actual_origin_visits = list( | ||||
service.lookup_origin_visits(origin_id, per_page=100)) | service.lookup_origin_visits( | ||||
hash_to_hex(origin_id), per_page=100)) | |||||
expected_visits = self.origin_visit_get(origin_id) | expected_visits = self.origin_visit_get(origin_id) | ||||
self.assertEqual(actual_origin_visits, expected_visits) | self.assertEqual(actual_origin_visits, expected_visits) | ||||
@given(new_origin(), visit_dates()) | @given(new_origin(), visit_dates()) | ||||
def test_lookup_origin_visit(self, new_origin, visit_dates): | def test_lookup_origin_visit(self, new_origin, visit_dates): | ||||
origin_id = self.storage.origin_add_one(new_origin) | origin_id = self.storage.origin_add_one(new_origin) | ||||
visits = [] | visits = [] | ||||
for ts in visit_dates: | for ts in visit_dates: | ||||
visits.append(self.storage.origin_visit_add(origin_id, ts)) | visits.append(self.storage.origin_visit_add(origin_id, ts)) | ||||
visit = random.choice(visits)['visit'] | visit = random.choice(visits)['visit'] | ||||
actual_origin_visit = service.lookup_origin_visit(origin_id, visit) | actual_origin_visit = service.lookup_origin_visit( | ||||
hash_to_hex(origin_id), visit) | |||||
expected_visit = dict(self.storage.origin_visit_get_by(origin_id, | expected_visit = dict(self.storage.origin_visit_get_by( | ||||
visit)) | origin_id, visit)) | ||||
expected_visit['date'] = expected_visit['date'].isoformat() | expected_visit['date'] = expected_visit['date'].isoformat() | ||||
expected_visit['metadata'] = {} | expected_visit['metadata'] = {} | ||||
expected_visit['origin'] = hash_to_hex(expected_visit['origin']) | |||||
self.assertEqual(actual_origin_visit, expected_visit) | self.assertEqual(actual_origin_visit, expected_visit) | ||||
@given(new_origin()) | @given(new_origin()) | ||||
def test_lookup_origin(self, new_origin): | def test_lookup_origin(self, new_origin): | ||||
origin_id = self.storage.origin_add_one(new_origin) | origin_id = self.storage.origin_add_one(new_origin) | ||||
actual_origin = service.lookup_origin({'id': origin_id}) | actual_origin = service.lookup_origin( | ||||
{'id': hash_to_hex(origin_id)}) | |||||
expected_origin = self.storage.origin_get({'id': origin_id}) | expected_origin = self.storage.origin_get({'id': origin_id}) | ||||
expected_origin['id'] = hash_to_hex(expected_origin['id']) | |||||
self.assertEqual(actual_origin, expected_origin) | self.assertEqual(actual_origin, expected_origin) | ||||
actual_origin = service.lookup_origin({'type': new_origin['type'], | actual_origin = service.lookup_origin({'type': new_origin['type'], | ||||
'url': new_origin['url']}) | 'url': new_origin['url']}) | ||||
expected_origin = self.storage.origin_get({'type': new_origin['type'], | expected_origin = self.storage.origin_get({'type': new_origin['type'], | ||||
'url': new_origin['url']}) | 'url': new_origin['url']}) | ||||
expected_origin['id'] = hash_to_hex(expected_origin['id']) | |||||
self.assertEqual(actual_origin, expected_origin) | self.assertEqual(actual_origin, expected_origin) | ||||
@given(invalid_sha1()) | @given(invalid_sha1()) | ||||
def test_lookup_release_ko_id_checksum_not_a_sha1(self, invalid_sha1): | def test_lookup_release_ko_id_checksum_not_a_sha1(self, invalid_sha1): | ||||
with self.assertRaises(BadInputExc) as cm: | with self.assertRaises(BadInputExc) as cm: | ||||
service.lookup_release(invalid_sha1) | service.lookup_release(invalid_sha1) | ||||
self.assertIn('invalid checksum', cm.exception.args[0].lower()) | self.assertIn('invalid checksum', cm.exception.args[0].lower()) | ||||
▲ Show 20 Lines • Show All 525 Lines • ▼ Show 20 Lines | def test_lookup_revision_through_with_context_by(self, origin): | ||||
branches = self._get_origin_branches(origin) | branches = self._get_origin_branches(origin) | ||||
branch_name = random.choice(list(branches.keys())) | branch_name = random.choice(list(branches.keys())) | ||||
root_rev = branches[branch_name]['target'] | root_rev = branches[branch_name]['target'] | ||||
root_rev_log = self.revision_log(root_rev) | root_rev_log = self.revision_log(root_rev) | ||||
rev = root_rev_log[-1]['id'] | rev = root_rev_log[-1]['id'] | ||||
self.assertEqual(service.lookup_revision_through({ | self.assertEqual(service.lookup_revision_through({ | ||||
'origin_id': origin['id'], | 'origin_id': hash_to_hex(origin['id']), | ||||
'branch_name': branch_name, | 'branch_name': branch_name, | ||||
'ts': None, | 'ts': None, | ||||
'sha1_git': rev | 'sha1_git': rev | ||||
}), | }), | ||||
service.lookup_revision_with_context_by( | service.lookup_revision_with_context_by( | ||||
origin['id'], branch_name, None, rev) | origin['id'], branch_name, None, rev) | ||||
) | ) | ||||
@given(origin()) | @given(origin()) | ||||
def test_lookup_revision_through_with_revision_by(self, origin): | def test_lookup_revision_through_with_revision_by(self, origin): | ||||
branches = self._get_origin_branches(origin) | branches = self._get_origin_branches(origin) | ||||
branch_name = random.choice(list(branches.keys())) | branch_name = random.choice(list(branches.keys())) | ||||
self.assertEqual(service.lookup_revision_through({ | self.assertEqual(service.lookup_revision_through({ | ||||
'origin_id': origin['id'], | 'origin_id': hash_to_hex(origin['id']), | ||||
'branch_name': branch_name, | 'branch_name': branch_name, | ||||
'ts': None, | 'ts': None, | ||||
}), | }), | ||||
service.lookup_revision_by( | service.lookup_revision_by( | ||||
origin['id'], branch_name, None) | origin['id'], branch_name, None) | ||||
) | ) | ||||
@given(ancestor_revisions()) | @given(ancestor_revisions()) | ||||
▲ Show 20 Lines • Show All 55 Lines • ▼ Show 20 Lines | def test_lookup_directory_through_revision_ok_with_data(self, revision): | ||||
service.lookup_directory_through_revision({'sha1_git': revision}, | service.lookup_directory_through_revision({'sha1_git': revision}, | ||||
dir_entry['name'], | dir_entry['name'], | ||||
with_data=True), | with_data=True), | ||||
(revision, | (revision, | ||||
service.lookup_directory_with_revision( | service.lookup_directory_with_revision( | ||||
revision, dir_entry['name'], with_data=True)) | revision, dir_entry['name'], with_data=True)) | ||||
) | ) | ||||
@given(new_origins(20)) | @given(new_origins(10), integers(min_value=1, max_value=9)) | ||||
def test_lookup_origins(self, new_origins): | def test_lookup_origins(self, new_origins, origin_count): | ||||
nb_origins = len(new_origins) | nb_origins = len(new_origins) | ||||
expected_origins = self.storage.origin_add(new_origins) | self.storage.origin_add(new_origins) | ||||
origins = list(self.storage.origin_get_range(origin_count=10000)) | |||||
origins.sort(key=lambda orig: orig['id']) | |||||
origin_from_idx = random.randint(1, nb_origins-1) - 1 | origin_from_idx = random.randint(1, nb_origins-1) - 1 | ||||
origin_from = expected_origins[origin_from_idx]['id'] | origin_from = origins[origin_from_idx]['id'] | ||||
max_origin_idx = expected_origins[-1]['id'] | if origin_from_idx + origin_count >= len(origins): | ||||
origin_count = random.randint(1, max_origin_idx - origin_from) | origin_count = len(origins) - origin_from_idx - 1 | ||||
actual_origins = list(service.lookup_origins(origin_from, | actual_origins = list(service.lookup_origins(origin_from, | ||||
origin_count)) | origin_count)) | ||||
expected_origins = list(self.storage.origin_get_range(origin_from, | expected_origins = list(self.storage.origin_get_range(origin_from, | ||||
origin_count)) | origin_count)) | ||||
for orig in expected_origins: | |||||
orig['id'] = hash_to_hex(orig['id']) | |||||
self.assertEqual(actual_origins, expected_origins) | self.assertEqual(actual_origins, expected_origins) |