Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/api/views/test_snapshot.py
# Copyright (C) 2018-2019 The Software Heritage developers | # Copyright (C) 2018-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 random | import random | ||||
from hypothesis import given | from hypothesis import given | ||||
from swh.model.hashutil import hash_to_hex | from swh.model.hashutil import hash_to_hex | ||||
from swh.web.api.utils import enrich_snapshot | |||||
from swh.web.common.utils import reverse | from swh.web.common.utils import reverse | ||||
from swh.web.tests.data import random_sha1 | from swh.web.tests.data import random_sha1 | ||||
from swh.web.tests.strategies import ( | from swh.web.tests.strategies import ( | ||||
snapshot, new_snapshot | snapshot, new_snapshot | ||||
) | ) | ||||
@given(snapshot()) | @given(snapshot()) | ||||
def test_api_snapshot(api_client, archive_data, snapshot): | def test_api_snapshot(api_client, archive_data, snapshot): | ||||
url = reverse('api-1-snapshot', | url = reverse('api-1-snapshot', | ||||
url_args={'snapshot_id': snapshot}) | url_args={'snapshot_id': snapshot}) | ||||
rv = api_client.get(url) | rv = api_client.get(url) | ||||
assert rv.status_code == 200, rv.data | assert rv.status_code == 200, rv.data | ||||
assert rv['Content-Type'] == 'application/json' | assert rv['Content-Type'] == 'application/json' | ||||
expected_data = archive_data.snapshot_get(snapshot) | expected_data = archive_data.snapshot_get(snapshot) | ||||
expected_data = _enrich_snapshot(archive_data, expected_data) | expected_data = enrich_snapshot(expected_data, rv.wsgi_request) | ||||
assert rv.data == expected_data | assert rv.data == expected_data | ||||
@given(snapshot()) | @given(snapshot()) | ||||
def test_api_snapshot_paginated(api_client, archive_data, snapshot): | def test_api_snapshot_paginated(api_client, archive_data, snapshot): | ||||
branches_offset = 0 | branches_offset = 0 | ||||
branches_count = 2 | branches_count = 2 | ||||
Show All 17 Lines | while branches_offset < len(snapshot_branches): | ||||
query_params={'branches_from': branches_from, | query_params={'branches_from': branches_from, | ||||
'branches_count': branches_count}) | 'branches_count': branches_count}) | ||||
rv = api_client.get(url) | rv = api_client.get(url) | ||||
assert rv.status_code == 200, rv.data | assert rv.status_code == 200, rv.data | ||||
assert rv['Content-Type'] == 'application/json' | assert rv['Content-Type'] == 'application/json' | ||||
expected_data = archive_data.snapshot_get_branches( | expected_data = archive_data.snapshot_get_branches( | ||||
snapshot, branches_from, branches_count) | snapshot, branches_from, branches_count) | ||||
expected_data = _enrich_snapshot(archive_data, expected_data) | expected_data = enrich_snapshot(expected_data, rv.wsgi_request) | ||||
branches_offset += branches_count | branches_offset += branches_count | ||||
if branches_offset < len(snapshot_branches): | if branches_offset < len(snapshot_branches): | ||||
next_branch = snapshot_branches[branches_offset]['name'] | next_branch = snapshot_branches[branches_offset]['name'] | ||||
expected_data['next_branch'] = next_branch | expected_data['next_branch'] = next_branch | ||||
else: | else: | ||||
expected_data['next_branch'] = None | expected_data['next_branch'] = None | ||||
Show All 36 Lines | def test_api_snapshot_filtered(api_client, archive_data, snapshot): | ||||
url = reverse('api-1-snapshot', | url = reverse('api-1-snapshot', | ||||
url_args={'snapshot_id': snapshot}, | url_args={'snapshot_id': snapshot}, | ||||
query_params={'target_types': target_type}) | query_params={'target_types': target_type}) | ||||
rv = api_client.get(url) | rv = api_client.get(url) | ||||
expected_data = archive_data.snapshot_get_branches( | expected_data = archive_data.snapshot_get_branches( | ||||
snapshot, target_types=target_type) | snapshot, target_types=target_type) | ||||
expected_data = _enrich_snapshot(archive_data, expected_data) | expected_data = enrich_snapshot(expected_data, rv.wsgi_request) | ||||
assert rv.status_code == 200, rv.data | assert rv.status_code == 200, rv.data | ||||
assert rv['Content-Type'] == 'application/json' | assert rv['Content-Type'] == 'application/json' | ||||
assert rv.data == expected_data | assert rv.data == expected_data | ||||
def test_api_snapshot_errors(api_client): | def test_api_snapshot_errors(api_client): | ||||
unknown_snapshot_ = random_sha1() | unknown_snapshot_ = random_sha1() | ||||
Show All 30 Lines | def test_api_snapshot_null_branch(api_client, archive_data, new_snapshot): | ||||
for branch in snp_dict['branches'].keys(): | for branch in snp_dict['branches'].keys(): | ||||
snp_dict['branches'][branch] = None | snp_dict['branches'][branch] = None | ||||
break | break | ||||
archive_data.snapshot_add([snp_dict]) | archive_data.snapshot_add([snp_dict]) | ||||
url = reverse('api-1-snapshot', | url = reverse('api-1-snapshot', | ||||
url_args={'snapshot_id': snp_id}) | url_args={'snapshot_id': snp_id}) | ||||
rv = api_client.get(url) | rv = api_client.get(url) | ||||
assert rv.status_code == 200, rv.data | assert rv.status_code == 200, rv.data | ||||
def _enrich_snapshot(archive_data, snapshot): | |||||
def _get_branch_url(target_type, target): | |||||
url = None | |||||
if target_type == 'revision': | |||||
url = reverse('api-1-revision', url_args={'sha1_git': target}) | |||||
if target_type == 'release': | |||||
url = reverse('api-1-release', url_args={'sha1_git': target}) | |||||
return url | |||||
for branch in snapshot['branches'].keys(): | |||||
target = snapshot['branches'][branch]['target'] | |||||
target_type = snapshot['branches'][branch]['target_type'] | |||||
snapshot['branches'][branch]['target_url'] = \ | |||||
_get_branch_url(target_type, target) | |||||
for branch in snapshot['branches'].keys(): | |||||
target = snapshot['branches'][branch]['target'] | |||||
target_type = snapshot['branches'][branch]['target_type'] | |||||
if target_type == 'alias': | |||||
if target in snapshot['branches']: | |||||
snapshot['branches'][branch]['target_url'] = \ | |||||
snapshot['branches'][target]['target_url'] | |||||
else: | |||||
snp = archive_data.snapshot_get_branches(snapshot['id'], | |||||
branches_from=target, | |||||
branches_count=1) | |||||
alias_target = snp['branches'][target]['target'] | |||||
alias_target_type = snp['branches'][target]['target_type'] | |||||
snapshot['branches'][branch]['target_url'] = \ | |||||
_get_branch_url(alias_target_type, alias_target) | |||||
return snapshot |