Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/browse/views/test_origin.py
# Copyright (C) 2017-2019 The Software Heritage developers | # Copyright (C) 2017-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 | ||||
import re | |||||
from unittest.mock import patch | |||||
from django.utils.html import escape | from django.utils.html import escape | ||||
from hypothesis import given | from hypothesis import given | ||||
from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
from swh.web.browse.utils import process_snapshot_branches | from swh.web.browse.utils import process_snapshot_branches | ||||
from swh.web.common.exc import NotFoundExc | from swh.web.common.exc import NotFoundExc | ||||
from swh.web.common.utils import ( | from swh.web.common.utils import ( | ||||
reverse, gen_path_info, format_utc_iso_date, | reverse, gen_path_info, format_utc_iso_date, | ||||
parse_timestamp, get_swh_persistent_id | parse_timestamp, get_swh_persistent_id | ||||
) | ) | ||||
from swh.web.tests.data import get_content | from swh.web.tests.data import get_content | ||||
from swh.web.tests.django_asserts import assert_contains, assert_template_used | |||||
from swh.web.tests.strategies import ( | from swh.web.tests.strategies import ( | ||||
origin, origin_with_multiple_visits, new_origin, | origin, origin_with_multiple_visits, new_origin, | ||||
new_snapshot, visit_dates, revisions | new_snapshot, visit_dates, revisions | ||||
) | ) | ||||
from swh.web.tests.testcase import WebTestCase | |||||
class SwhBrowseOriginTest(WebTestCase): | |||||
@given(origin_with_multiple_visits()) | @given(origin_with_multiple_visits()) | ||||
def test_origin_visits_browse(self, origin): | def test_origin_visits_browse(client, archive_data, origin): | ||||
url = reverse('browse-origin-visits', | url = reverse('browse-origin-visits', | ||||
url_args={'origin_url': origin['url']}) | url_args={'origin_url': origin['url']}) | ||||
resp = self.client.get(url) | resp = client.get(url) | ||||
self.assertEqual(resp.status_code, 200) | assert resp.status_code == 200 | ||||
self.assertTemplateUsed('origin-visits.html') | assert_template_used('origin-visits.html') | ||||
url = reverse('browse-origin-visits', | url = reverse('browse-origin-visits', | ||||
url_args={'origin_url': origin['url']}) | url_args={'origin_url': origin['url']}) | ||||
resp = self.client.get(url) | resp = client.get(url) | ||||
self.assertEqual(resp.status_code, 200) | assert resp.status_code == 200 | ||||
self.assertTemplateUsed('origin-visits.html') | assert_template_used('origin-visits.html') | ||||
visits = self.origin_visit_get(origin['url']) | visits = archive_data.origin_visit_get(origin['url']) | ||||
for v in visits: | for v in visits: | ||||
vdate = format_utc_iso_date(v['date'], '%Y-%m-%dT%H:%M:%SZ') | vdate = format_utc_iso_date(v['date'], '%Y-%m-%dT%H:%M:%SZ') | ||||
browse_dir_url = reverse('browse-origin-directory', | browse_dir_url = reverse('browse-origin-directory', | ||||
url_args={'origin_url': origin['url'], | url_args={'origin_url': origin['url'], | ||||
'timestamp': vdate}) | 'timestamp': vdate}) | ||||
self.assertContains(resp, browse_dir_url) | assert_contains(resp, browse_dir_url) | ||||
@given(origin_with_multiple_visits()) | |||||
def test_origin_content_view(client, archive_data, origin): | |||||
origin_visits = archive_data.origin_visit_get(origin['url']) | |||||
def _get_archive_data(visit_idx): | |||||
snapshot = archive_data.snapshot_get( | |||||
origin_visits[visit_idx]['snapshot']) | |||||
head_rev_id = archive_data.snapshot_get_head(snapshot) | |||||
head_rev = archive_data.revision_get(head_rev_id) | |||||
dir_content = archive_data.directory_ls(head_rev['directory']) | |||||
dir_files = [e for e in dir_content if e['type'] == 'file'] | |||||
dir_file = random.choice(dir_files) | |||||
branches, releases = process_snapshot_branches(snapshot) | |||||
return { | |||||
'branches': branches, | |||||
'releases': releases, | |||||
'root_dir_sha1': head_rev['directory'], | |||||
'content': get_content(dir_file['checksums']['sha1']), | |||||
'visit': origin_visits[visit_idx] | |||||
} | |||||
tdata = _get_archive_data(-1) | |||||
_origin_content_view_test_helper(client, origin, origin_visits, | |||||
tdata['branches'], | |||||
tdata['releases'], | |||||
tdata['root_dir_sha1'], | |||||
tdata['content']) | |||||
_origin_content_view_test_helper(client, origin, origin_visits, | |||||
tdata['branches'], | |||||
tdata['releases'], | |||||
tdata['root_dir_sha1'], | |||||
tdata['content'], | |||||
timestamp=tdata['visit']['date']) | |||||
visit_unix_ts = parse_timestamp(tdata['visit']['date']).timestamp() | |||||
visit_unix_ts = int(visit_unix_ts) | |||||
_origin_content_view_test_helper(client, origin, origin_visits, | |||||
tdata['branches'], | |||||
tdata['releases'], | |||||
tdata['root_dir_sha1'], | |||||
tdata['content'], | |||||
timestamp=visit_unix_ts) | |||||
tdata = _get_archive_data(0) | |||||
_origin_content_view_test_helper(client, origin, origin_visits, | |||||
tdata['branches'], | |||||
tdata['releases'], | |||||
tdata['root_dir_sha1'], | |||||
tdata['content'], | |||||
visit_id=tdata['visit']['visit']) | |||||
@given(origin()) | |||||
def test_origin_root_directory_view(client, archive_data, origin): | |||||
origin_visits = archive_data.origin_visit_get(origin['url']) | |||||
visit = origin_visits[-1] | |||||
snapshot = archive_data.snapshot_get(visit['snapshot']) | |||||
head_rev_id = archive_data.snapshot_get_head(snapshot) | |||||
head_rev = archive_data.revision_get(head_rev_id) | |||||
root_dir_sha1 = head_rev['directory'] | |||||
dir_content = archive_data.directory_ls(root_dir_sha1) | |||||
branches, releases = process_snapshot_branches(snapshot) | |||||
visit_unix_ts = parse_timestamp(visit['date']).timestamp() | |||||
visit_unix_ts = int(visit_unix_ts) | |||||
_origin_directory_view_test_helper(client, origin, origin_visits, branches, | |||||
releases, root_dir_sha1, dir_content) | |||||
_origin_directory_view_test_helper(client, origin, origin_visits, branches, | |||||
releases, root_dir_sha1, dir_content, | |||||
visit_id=visit['visit']) | |||||
_origin_directory_view_test_helper(client, origin, origin_visits, branches, | |||||
releases, root_dir_sha1, dir_content, | |||||
timestamp=visit_unix_ts) | |||||
_origin_directory_view_test_helper(client, origin, origin_visits, branches, | |||||
releases, root_dir_sha1, dir_content, | |||||
timestamp=visit['date']) | |||||
origin = dict(origin) | |||||
del origin['type'] | |||||
_origin_directory_view_test_helper(client, origin, origin_visits, branches, | |||||
releases, root_dir_sha1, dir_content) | |||||
_origin_directory_view_test_helper(client, origin, origin_visits, branches, | |||||
releases, root_dir_sha1, dir_content, | |||||
visit_id=visit['visit']) | |||||
_origin_directory_view_test_helper(client, origin, origin_visits, branches, | |||||
releases, root_dir_sha1, dir_content, | |||||
timestamp=visit_unix_ts) | |||||
_origin_directory_view_test_helper(client, origin, origin_visits, branches, | |||||
releases, root_dir_sha1, dir_content, | |||||
timestamp=visit['date']) | |||||
@given(origin()) | |||||
def test_origin_sub_directory_view(client, archive_data, origin): | |||||
origin_visits = archive_data.origin_visit_get(origin['url']) | |||||
visit = origin_visits[-1] | |||||
snapshot = archive_data.snapshot_get(visit['snapshot']) | |||||
head_rev_id = archive_data.snapshot_get_head(snapshot) | |||||
head_rev = archive_data.revision_get(head_rev_id) | |||||
root_dir_sha1 = head_rev['directory'] | |||||
subdirs = [e for e in archive_data.directory_ls(root_dir_sha1) | |||||
if e['type'] == 'dir'] | |||||
branches, releases = process_snapshot_branches(snapshot) | |||||
visit_unix_ts = parse_timestamp(visit['date']).timestamp() | |||||
visit_unix_ts = int(visit_unix_ts) | |||||
if len(subdirs) == 0: | |||||
return | |||||
subdir = random.choice(subdirs) | |||||
subdir_content = archive_data.directory_ls(subdir['target']) | |||||
subdir_path = subdir['name'] | |||||
_origin_directory_view_test_helper(client, origin, origin_visits, branches, | |||||
releases, root_dir_sha1, subdir_content, | |||||
path=subdir_path) | |||||
_origin_directory_view_test_helper(client, origin, origin_visits, branches, | |||||
releases, root_dir_sha1, subdir_content, | |||||
path=subdir_path, | |||||
visit_id=visit['visit']) | |||||
_origin_directory_view_test_helper(client, origin, origin_visits, branches, | |||||
releases, root_dir_sha1, subdir_content, | |||||
path=subdir_path, | |||||
timestamp=visit_unix_ts) | |||||
_origin_directory_view_test_helper(client, origin, origin_visits, branches, | |||||
releases, root_dir_sha1, subdir_content, | |||||
path=subdir_path, | |||||
timestamp=visit['date']) | |||||
origin = dict(origin) | |||||
del origin['type'] | |||||
_origin_directory_view_test_helper(client, origin, origin_visits, branches, | |||||
releases, root_dir_sha1, subdir_content, | |||||
path=subdir_path) | |||||
_origin_directory_view_test_helper(client, origin, origin_visits, branches, | |||||
releases, root_dir_sha1, subdir_content, | |||||
path=subdir_path, | |||||
visit_id=visit['visit']) | |||||
_origin_directory_view_test_helper(client, origin, origin_visits, branches, | |||||
releases, root_dir_sha1, subdir_content, | |||||
path=subdir_path, | |||||
timestamp=visit_unix_ts) | |||||
_origin_directory_view_test_helper(client, origin, origin_visits, branches, | |||||
releases, root_dir_sha1, subdir_content, | |||||
path=subdir_path, | |||||
timestamp=visit['date']) | |||||
@given(origin()) | |||||
def test_origin_branches(client, archive_data, origin): | |||||
origin_visits = archive_data.origin_visit_get(origin['url']) | |||||
visit = origin_visits[-1] | |||||
snapshot = archive_data.snapshot_get(visit['snapshot']) | |||||
snapshot_content = process_snapshot_branches(snapshot) | |||||
_origin_branches_test_helper(client, origin, snapshot_content) | |||||
origin = dict(origin) | |||||
origin['type'] = None | |||||
_origin_branches_test_helper(client, origin, snapshot_content) | |||||
@given(origin()) | |||||
def test_origin_releases(client, archive_data, origin): | |||||
origin_visits = archive_data.origin_visit_get(origin['url']) | |||||
visit = origin_visits[-1] | |||||
snapshot = archive_data.snapshot_get(visit['snapshot']) | |||||
snapshot_content = process_snapshot_branches(snapshot) | |||||
_origin_releases_test_helper(client, origin, snapshot_content) | |||||
origin = dict(origin) | |||||
origin['type'] = None | |||||
_origin_releases_test_helper(client, origin, snapshot_content) | |||||
@given(new_origin(), new_snapshot(min_size=4, max_size=4), visit_dates(), | |||||
revisions(min_size=3, max_size=3)) | |||||
def test_origin_snapshot_null_branch(client, archive_data, new_origin, | |||||
new_snapshot, visit_dates, revisions): | |||||
snp_dict = new_snapshot.to_dict() | |||||
new_origin = archive_data.origin_add([new_origin])[0] | |||||
for i, branch in enumerate(snp_dict['branches'].keys()): | |||||
if i == 0: | |||||
snp_dict['branches'][branch] = None | |||||
else: | |||||
snp_dict['branches'][branch] = { | |||||
'target_type': 'revision', | |||||
'target': hash_to_bytes(revisions[i-1]), | |||||
} | |||||
archive_data.snapshot_add([snp_dict]) | |||||
visit = archive_data.origin_visit_add( | |||||
new_origin['url'], visit_dates[0], type='git') | |||||
archive_data.origin_visit_update(new_origin['url'], visit['visit'], | |||||
status='partial', | |||||
snapshot=snp_dict['id']) | |||||
def origin_content_view_helper(self, origin_info, origin_visits, | url = reverse('browse-origin-directory', | ||||
url_args={'origin_url': new_origin['url']}) | |||||
rv = client.get(url) | |||||
assert rv.status_code == 200 | |||||
@given(new_origin(), new_snapshot(min_size=4, max_size=4), visit_dates(), | |||||
revisions(min_size=3, max_size=3)) | |||||
def test_origin_snapshot_invalid_branch(client, archive_data, new_origin, | |||||
new_snapshot, visit_dates, revisions): | |||||
snp_dict = new_snapshot.to_dict() | |||||
new_origin = archive_data.origin_add([new_origin])[0] | |||||
for i, branch in enumerate(snp_dict['branches'].keys()): | |||||
if i == 0: | |||||
invalid_branch = branch | |||||
else: | |||||
snp_dict['branches'][branch] = { | |||||
'target_type': 'revision', | |||||
'target': hash_to_bytes(revisions[i-1]), | |||||
} | |||||
del snp_dict['branches'][invalid_branch] | |||||
archive_data.snapshot_add([snp_dict]) | |||||
visit = archive_data.origin_visit_add( | |||||
new_origin['url'], visit_dates[0], type='git') | |||||
archive_data.origin_visit_update(new_origin['url'], visit['visit'], | |||||
status='full', | |||||
snapshot=snp_dict['id']) | |||||
url = reverse('browse-origin-directory', | |||||
url_args={'origin_url': new_origin['url']}, | |||||
query_params={'branch': invalid_branch}) | |||||
rv = client.get(url) | |||||
assert rv.status_code == 404 | |||||
def test_origin_request_errors(client, archive_data, mocker): | |||||
mock_snapshot_service = mocker.patch( | |||||
'swh.web.browse.views.utils.snapshot_context.service') | |||||
mock_origin_service = mocker.patch('swh.web.browse.views.origin.service') | |||||
mock_utils_service = mocker.patch('swh.web.browse.utils.service') | |||||
mock_get_origin_visit_snapshot = mocker.patch( | |||||
'swh.web.browse.utils.get_origin_visit_snapshot') | |||||
mock_get_origin_visits = mocker.patch( | |||||
'swh.web.common.origin_visits.get_origin_visits') | |||||
mock_request_content = mocker.patch( | |||||
'swh.web.browse.views.utils.snapshot_context.request_content') | |||||
mock_origin_service.lookup_origin.side_effect = NotFoundExc( | |||||
'origin not found') | |||||
url = reverse('browse-origin-visits', | |||||
url_args={'origin_url': 'bar'}) | |||||
resp = client.get(url) | |||||
assert resp.status_code == 404 | |||||
assert_template_used('error.html') | |||||
assert_contains(resp, 'origin not found', status_code=404) | |||||
mock_origin_service.lookup_origin.side_effect = None | |||||
mock_origin_service.lookup_origin.return_value = {'type': 'foo', | |||||
'url': 'bar', | |||||
'id': 457} | |||||
mock_get_origin_visits.return_value = [] | |||||
url = reverse('browse-origin-directory', | |||||
url_args={'origin_url': 'bar'}) | |||||
resp = client.get(url) | |||||
assert resp.status_code == 404 | |||||
assert_template_used('error.html') | |||||
assert_contains(resp, "No visit", status_code=404) | |||||
mock_get_origin_visits.return_value = [{'visit': 1}] | |||||
mock_get_origin_visit_snapshot.side_effect = NotFoundExc('visit not found') | |||||
url = reverse('browse-origin-directory', | |||||
url_args={'origin_url': 'bar'}, | |||||
query_params={'visit_id': 2}) | |||||
resp = client.get(url) | |||||
assert resp.status_code == 404 | |||||
assert_template_used('error.html') | |||||
assert re.search('Visit.*not found', resp.content.decode('utf-8')) | |||||
mock_get_origin_visits.return_value = [{ | |||||
'date': '2015-09-26T09:30:52.373449+00:00', | |||||
'metadata': {}, | |||||
'origin': 457, | |||||
'snapshot': 'bdaf9ac436488a8c6cda927a0f44e172934d3f65', | |||||
'status': 'full', | |||||
'visit': 1 | |||||
}] | |||||
mock_get_origin_visit_snapshot.side_effect = None | |||||
mock_get_origin_visit_snapshot.return_value = ( | |||||
[{'directory': 'ae59ceecf46367e8e4ad800e231fc76adc3afffb', | |||||
'name': 'HEAD', | |||||
'revision': '7bc08e1aa0b08cb23e18715a32aa38517ad34672', | |||||
'date': '04 May 2017, 13:27 UTC', | |||||
'message': ''}], | |||||
[] | |||||
) | |||||
mock_utils_service.lookup_snapshot_size.return_value = { | |||||
'revision': 1, | |||||
'release': 0 | |||||
} | |||||
mock_lookup_directory = mock_utils_service.lookup_directory | |||||
mock_lookup_directory.side_effect = NotFoundExc('Directory not found') | |||||
url = reverse('browse-origin-directory', | |||||
url_args={'origin_url': 'bar'}) | |||||
resp = client.get(url) | |||||
assert resp.status_code == 404 | |||||
assert_template_used('error.html') | |||||
assert_contains(resp, 'Directory not found', status_code=404) | |||||
mock_origin_service.lookup_origin.side_effect = None | |||||
mock_origin_service.lookup_origin.return_value = {'type': 'foo', | |||||
'url': 'bar', | |||||
'id': 457} | |||||
mock_get_origin_visits.return_value = [] | |||||
url = reverse('browse-origin-content', | |||||
url_args={'origin_url': 'bar', | |||||
'path': 'foo'}) | |||||
resp = client.get(url) | |||||
assert resp.status_code == 404 | |||||
assert_template_used('error.html') | |||||
assert_contains(resp, "No visit", status_code=404) | |||||
mock_get_origin_visits.return_value = [{'visit': 1}] | |||||
mock_get_origin_visit_snapshot.side_effect = NotFoundExc('visit not found') | |||||
url = reverse('browse-origin-content', | |||||
url_args={'origin_url': 'bar', | |||||
'path': 'foo'}, | |||||
query_params={'visit_id': 2}) | |||||
resp = client.get(url) | |||||
assert resp.status_code == 404 | |||||
assert_template_used('error.html') | |||||
assert re.search('Visit.*not found', resp.content.decode('utf-8')) | |||||
mock_get_origin_visits.return_value = [{ | |||||
'date': '2015-09-26T09:30:52.373449+00:00', | |||||
'metadata': {}, | |||||
'origin': 457, | |||||
'snapshot': 'bdaf9ac436488a8c6cda927a0f44e172934d3f65', | |||||
'status': 'full', | |||||
'visit': 1 | |||||
}] | |||||
mock_get_origin_visit_snapshot.side_effect = None | |||||
mock_get_origin_visit_snapshot.return_value = ([], []) | |||||
url = reverse('browse-origin-content', | |||||
url_args={'origin_url': 'bar', | |||||
'path': 'baz'}) | |||||
resp = client.get(url) | |||||
assert resp.status_code == 404 | |||||
assert_template_used('error.html') | |||||
assert re.search('Origin.*has an empty list of branches', | |||||
resp.content.decode('utf-8')) | |||||
mock_get_origin_visit_snapshot.return_value = ( | |||||
[{'directory': 'ae59ceecf46367e8e4ad800e231fc76adc3afffb', | |||||
'name': 'HEAD', | |||||
'revision': '7bc08e1aa0b08cb23e18715a32aa38517ad34672', | |||||
'date': '04 May 2017, 13:27 UTC', | |||||
'message': ''}], | |||||
[] | |||||
) | |||||
mock_snapshot_service.lookup_directory_with_path.return_value = { | |||||
'target': '5ecd9f37b7a2d2e9980d201acd6286116f2ba1f1' | |||||
} | |||||
mock_request_content.side_effect = NotFoundExc('Content not found') | |||||
url = reverse('browse-origin-content', | |||||
url_args={'origin_url': 'bar', | |||||
'path': 'baz'}) | |||||
resp = client.get(url) | |||||
assert resp.status_code == 404 | |||||
assert_template_used('error.html') | |||||
assert_contains(resp, 'Content not found', status_code=404) | |||||
mock_get_snapshot_context = mocker.patch( | |||||
'swh.web.browse.views.utils.snapshot_context.get_snapshot_context') | |||||
mock_get_snapshot_context.side_effect = NotFoundExc('Snapshot not found') | |||||
url = reverse('browse-origin-directory', | |||||
url_args={'origin_url': 'bar'}) | |||||
resp = client.get(url) | |||||
assert resp.status_code == 404 | |||||
assert_template_used('error.html') | |||||
assert_contains(resp, 'Snapshot not found', status_code=404) | |||||
def test_origin_empty_snapshot(client, mocker): | |||||
mock_utils_service = mocker.patch('swh.web.browse.utils.service') | |||||
mock_get_origin_visit_snapshot = mocker.patch( | |||||
'swh.web.browse.utils.get_origin_visit_snapshot') | |||||
mock_get_origin_visits = mocker.patch( | |||||
'swh.web.common.origin_visits.get_origin_visits') | |||||
mock_get_origin_visits.return_value = [{ | |||||
'date': '2015-09-26T09:30:52.373449+00:00', | |||||
'metadata': {}, | |||||
'origin': 457, | |||||
'snapshot': 'bdaf9ac436488a8c6cda927a0f44e172934d3f65', | |||||
'status': 'full', | |||||
'type': 'git', | |||||
'visit': 1 | |||||
}] | |||||
mock_get_origin_visit_snapshot.return_value = ([], []) | |||||
mock_utils_service.lookup_snapshot_size.return_value = { | |||||
'revision': 0, | |||||
'release': 0 | |||||
} | |||||
mock_utils_service.lookup_origin.return_value = { | |||||
'id': 457, | |||||
'url': 'https://github.com/foo/bar' | |||||
} | |||||
url = reverse('browse-origin-directory', | |||||
url_args={'origin_url': 'bar'}) | |||||
resp = client.get(url) | |||||
assert resp.status_code == 200 | |||||
assert_template_used('content.html') | |||||
assert re.search('snapshot.*is empty', resp.content.decode('utf-8')) | |||||
def _origin_content_view_test_helper(client, origin_info, origin_visits, | |||||
origin_branches, origin_releases, | origin_branches, origin_releases, | ||||
root_dir_sha1, content, | root_dir_sha1, content, | ||||
visit_id=None, timestamp=None): | visit_id=None, timestamp=None): | ||||
content_path = '/'.join(content['path'].split('/')[1:]) | content_path = '/'.join(content['path'].split('/')[1:]) | ||||
url_args = {'origin_url': origin_info['url'], | url_args = {'origin_url': origin_info['url'], | ||||
'path': content_path} | 'path': content_path} | ||||
if not visit_id: | if not visit_id: | ||||
visit_id = origin_visits[-1]['visit'] | visit_id = origin_visits[-1]['visit'] | ||||
query_params = {} | query_params = {} | ||||
if timestamp: | if timestamp: | ||||
url_args['timestamp'] = timestamp | url_args['timestamp'] = timestamp | ||||
if visit_id: | if visit_id: | ||||
query_params['visit_id'] = visit_id | query_params['visit_id'] = visit_id | ||||
url = reverse('browse-origin-content', | url = reverse('browse-origin-content', | ||||
url_args=url_args, | url_args=url_args, | ||||
query_params=query_params) | query_params=query_params) | ||||
resp = self.client.get(url) | resp = client.get(url) | ||||
self.assertEqual(resp.status_code, 200) | assert resp.status_code == 200 | ||||
self.assertTemplateUsed('content.html') | assert_template_used('content.html') | ||||
self.assertContains(resp, '<code class="%s">' % | assert_contains(resp, '<code class="%s">' % | ||||
content['hljs_language']) | content['hljs_language']) | ||||
self.assertContains(resp, escape(content['data'])) | assert_contains(resp, escape(content['data'])) | ||||
split_path = content_path.split('/') | split_path = content_path.split('/') | ||||
filename = split_path[-1] | filename = split_path[-1] | ||||
path = content_path.replace(filename, '')[:-1] | path = content_path.replace(filename, '')[:-1] | ||||
path_info = gen_path_info(path) | path_info = gen_path_info(path) | ||||
del url_args['path'] | del url_args['path'] | ||||
if timestamp: | if timestamp: | ||||
url_args['timestamp'] = \ | url_args['timestamp'] = format_utc_iso_date( | ||||
format_utc_iso_date(parse_timestamp(timestamp).isoformat(), | parse_timestamp(timestamp).isoformat(), '%Y-%m-%dT%H:%M:%S') | ||||
'%Y-%m-%dT%H:%M:%S') | |||||
root_dir_url = reverse('browse-origin-directory', | root_dir_url = reverse('browse-origin-directory', | ||||
url_args=url_args, | url_args=url_args, | ||||
query_params=query_params) | query_params=query_params) | ||||
self.assertContains(resp, '<li class="swh-path">', | assert_contains(resp, '<li class="swh-path">', | ||||
count=len(path_info)+1) | count=len(path_info)+1) | ||||
self.assertContains(resp, '<a href="%s">%s</a>' % | assert_contains(resp, '<a href="%s">%s</a>' % | ||||
(root_dir_url, root_dir_sha1[:7])) | (root_dir_url, root_dir_sha1[:7])) | ||||
for p in path_info: | for p in path_info: | ||||
url_args['path'] = p['path'] | url_args['path'] = p['path'] | ||||
dir_url = reverse('browse-origin-directory', | dir_url = reverse('browse-origin-directory', | ||||
url_args=url_args, | url_args=url_args, | ||||
query_params=query_params) | query_params=query_params) | ||||
self.assertContains(resp, '<a href="%s">%s</a>' % | assert_contains(resp, '<a href="%s">%s</a>' % | ||||
(dir_url, p['name'])) | (dir_url, p['name'])) | ||||
self.assertContains(resp, '<li>%s</li>' % filename) | assert_contains(resp, '<li>%s</li>' % filename) | ||||
query_string = 'sha1_git:' + content['sha1_git'] | query_string = 'sha1_git:' + content['sha1_git'] | ||||
url_raw = reverse('browse-content-raw', | url_raw = reverse('browse-content-raw', | ||||
url_args={'query_string': query_string}, | url_args={'query_string': query_string}, | ||||
query_params={'filename': filename}) | query_params={'filename': filename}) | ||||
self.assertContains(resp, url_raw) | assert_contains(resp, url_raw) | ||||
if 'args' in url_args: | if 'args' in url_args: | ||||
del url_args['path'] | del url_args['path'] | ||||
origin_branches_url = reverse('browse-origin-branches', | origin_branches_url = reverse('browse-origin-branches', | ||||
url_args=url_args, | url_args=url_args, | ||||
query_params=query_params) | query_params=query_params) | ||||
self.assertContains(resp, '<a href="%s">Branches (%s)</a>' % | assert_contains(resp, '<a href="%s">Branches (%s)</a>' % | ||||
(origin_branches_url, len(origin_branches))) | (origin_branches_url, len(origin_branches))) | ||||
origin_releases_url = reverse('browse-origin-releases', | origin_releases_url = reverse('browse-origin-releases', | ||||
url_args=url_args, | url_args=url_args, | ||||
query_params=query_params) | query_params=query_params) | ||||
self.assertContains(resp, '<a href="%s">Releases (%s)</a>' % | assert_contains(resp, '<a href="%s">Releases (%s)</a>' % | ||||
(origin_releases_url, len(origin_releases))) | (origin_releases_url, len(origin_releases))) | ||||
self.assertContains(resp, '<li class="swh-branch">', | assert_contains(resp, '<li class="swh-branch">', | ||||
count=len(origin_branches)) | count=len(origin_branches)) | ||||
url_args['path'] = content_path | url_args['path'] = content_path | ||||
for branch in origin_branches: | for branch in origin_branches: | ||||
query_params['branch'] = branch['name'] | query_params['branch'] = branch['name'] | ||||
root_dir_branch_url = reverse('browse-origin-content', | root_dir_branch_url = reverse('browse-origin-content', | ||||
url_args=url_args, | url_args=url_args, | ||||
query_params=query_params) | query_params=query_params) | ||||
self.assertContains(resp, '<a href="%s">' % root_dir_branch_url) | assert_contains(resp, '<a href="%s">' % root_dir_branch_url) | ||||
self.assertContains(resp, '<li class="swh-release">', | assert_contains(resp, '<li class="swh-release">', | ||||
count=len(origin_releases)) | count=len(origin_releases)) | ||||
query_params['branch'] = None | query_params['branch'] = None | ||||
for release in origin_releases: | for release in origin_releases: | ||||
query_params['release'] = release['name'] | query_params['release'] = release['name'] | ||||
root_dir_release_url = reverse('browse-origin-content', | root_dir_release_url = reverse('browse-origin-content', | ||||
url_args=url_args, | url_args=url_args, | ||||
query_params=query_params) | query_params=query_params) | ||||
self.assertContains(resp, '<a href="%s">' % root_dir_release_url) | assert_contains(resp, '<a href="%s">' % root_dir_release_url) | ||||
url = reverse('browse-origin-content', | url = reverse('browse-origin-content', | ||||
url_args=url_args, | url_args=url_args, | ||||
query_params=query_params) | query_params=query_params) | ||||
resp = self.client.get(url) | resp = client.get(url) | ||||
self.assertEqual(resp.status_code, 200) | assert resp.status_code == 200 | ||||
self.assertTemplateUsed('content.html') | assert_template_used('content.html') | ||||
swh_cnt_id = get_swh_persistent_id('content', content['sha1_git']) | swh_cnt_id = get_swh_persistent_id('content', content['sha1_git']) | ||||
swh_cnt_id_url = reverse('browse-swh-id', | swh_cnt_id_url = reverse('browse-swh-id', | ||||
url_args={'swh_id': swh_cnt_id}) | url_args={'swh_id': swh_cnt_id}) | ||||
self.assertContains(resp, swh_cnt_id) | assert_contains(resp, swh_cnt_id) | ||||
self.assertContains(resp, swh_cnt_id_url) | assert_contains(resp, swh_cnt_id_url) | ||||
self.assertContains(resp, 'swh-take-new-snapshot') | assert_contains(resp, 'swh-take-new-snapshot') | ||||
@given(origin_with_multiple_visits()) | |||||
def test_origin_content_view(self, origin): | |||||
origin_visits = self.origin_visit_get(origin['url']) | |||||
def _get_test_data(visit_idx): | def _origin_directory_view_test_helper(client, origin_info, origin_visits, | ||||
snapshot = self.snapshot_get(origin_visits[visit_idx]['snapshot']) | |||||
head_rev_id = self.snapshot_get_head(snapshot) | |||||
head_rev = self.revision_get(head_rev_id) | |||||
dir_content = self.directory_ls(head_rev['directory']) | |||||
dir_files = [e for e in dir_content if e['type'] == 'file'] | |||||
dir_file = random.choice(dir_files) | |||||
branches, releases = process_snapshot_branches(snapshot) | |||||
return { | |||||
'branches': branches, | |||||
'releases': releases, | |||||
'root_dir_sha1': head_rev['directory'], | |||||
'content': get_content(dir_file['checksums']['sha1']), | |||||
'visit': origin_visits[visit_idx] | |||||
} | |||||
test_data = _get_test_data(-1) | |||||
self.origin_content_view_helper(origin, | |||||
origin_visits, | |||||
test_data['branches'], | |||||
test_data['releases'], | |||||
test_data['root_dir_sha1'], | |||||
test_data['content']) | |||||
self.origin_content_view_helper(origin, | |||||
origin_visits, | |||||
test_data['branches'], | |||||
test_data['releases'], | |||||
test_data['root_dir_sha1'], | |||||
test_data['content'], | |||||
timestamp=test_data['visit']['date']) | |||||
visit_unix_ts = parse_timestamp(test_data['visit']['date']).timestamp() | |||||
visit_unix_ts = int(visit_unix_ts) | |||||
self.origin_content_view_helper(origin, | |||||
origin_visits, | |||||
test_data['branches'], | |||||
test_data['releases'], | |||||
test_data['root_dir_sha1'], | |||||
test_data['content'], | |||||
timestamp=visit_unix_ts) | |||||
test_data = _get_test_data(0) | |||||
self.origin_content_view_helper(origin, | |||||
origin_visits, | |||||
test_data['branches'], | |||||
test_data['releases'], | |||||
test_data['root_dir_sha1'], | |||||
test_data['content'], | |||||
visit_id=test_data['visit']['visit']) | |||||
def origin_directory_view_helper(self, origin_info, origin_visits, | |||||
origin_branches, origin_releases, | origin_branches, origin_releases, | ||||
root_directory_sha1, directory_entries, | root_directory_sha1, directory_entries, | ||||
visit_id=None, timestamp=None, path=None): | visit_id=None, timestamp=None, | ||||
path=None): | |||||
dirs = [e for e in directory_entries | dirs = [e for e in directory_entries | ||||
if e['type'] in ('dir', 'rev')] | if e['type'] in ('dir', 'rev')] | ||||
files = [e for e in directory_entries | files = [e for e in directory_entries | ||||
if e['type'] == 'file'] | if e['type'] == 'file'] | ||||
if not visit_id: | if not visit_id: | ||||
visit_id = origin_visits[-1]['visit'] | visit_id = origin_visits[-1]['visit'] | ||||
url_args = {'origin_url': origin_info['url']} | url_args = {'origin_url': origin_info['url']} | ||||
query_params = {} | query_params = {} | ||||
if timestamp: | if timestamp: | ||||
url_args['timestamp'] = timestamp | url_args['timestamp'] = timestamp | ||||
else: | else: | ||||
query_params['visit_id'] = visit_id | query_params['visit_id'] = visit_id | ||||
if path: | if path: | ||||
url_args['path'] = path | url_args['path'] = path | ||||
url = reverse('browse-origin-directory', | url = reverse('browse-origin-directory', | ||||
url_args=url_args, | url_args=url_args, | ||||
query_params=query_params) | query_params=query_params) | ||||
resp = self.client.get(url) | resp = client.get(url) | ||||
self.assertEqual(resp.status_code, 200) | assert resp.status_code == 200 | ||||
self.assertTemplateUsed('directory.html') | assert_template_used('directory.html') | ||||
self.assertEqual(resp.status_code, 200) | assert resp.status_code == 200 | ||||
self.assertTemplateUsed('directory.html') | assert_template_used('directory.html') | ||||
self.assertContains(resp, '<td class="swh-directory">', | assert_contains(resp, '<td class="swh-directory">', | ||||
count=len(dirs)) | count=len(dirs)) | ||||
self.assertContains(resp, '<td class="swh-content">', | assert_contains(resp, '<td class="swh-content">', | ||||
count=len(files)) | count=len(files)) | ||||
if timestamp: | if timestamp: | ||||
url_args['timestamp'] = \ | url_args['timestamp'] = format_utc_iso_date( | ||||
format_utc_iso_date(parse_timestamp(timestamp).isoformat(), | parse_timestamp(timestamp).isoformat(), '%Y-%m-%dT%H:%M:%S') | ||||
'%Y-%m-%dT%H:%M:%S') | |||||
for d in dirs: | for d in dirs: | ||||
if d['type'] == 'rev': | if d['type'] == 'rev': | ||||
dir_url = reverse('browse-revision', | dir_url = reverse('browse-revision', | ||||
url_args={'sha1_git': d['target']}) | url_args={'sha1_git': d['target']}) | ||||
else: | else: | ||||
dir_path = d['name'] | dir_path = d['name'] | ||||
if path: | if path: | ||||
dir_path = "%s/%s" % (path, d['name']) | dir_path = "%s/%s" % (path, d['name']) | ||||
dir_url_args = dict(url_args) | dir_url_args = dict(url_args) | ||||
dir_url_args['path'] = dir_path | dir_url_args['path'] = dir_path | ||||
dir_url = reverse('browse-origin-directory', | dir_url = reverse('browse-origin-directory', | ||||
url_args=dir_url_args, | url_args=dir_url_args, | ||||
query_params=query_params) | query_params=query_params) | ||||
self.assertContains(resp, dir_url) | assert_contains(resp, dir_url) | ||||
for f in files: | for f in files: | ||||
file_path = f['name'] | file_path = f['name'] | ||||
if path: | if path: | ||||
file_path = "%s/%s" % (path, f['name']) | file_path = "%s/%s" % (path, f['name']) | ||||
file_url_args = dict(url_args) | file_url_args = dict(url_args) | ||||
file_url_args['path'] = file_path | file_url_args['path'] = file_path | ||||
file_url = reverse('browse-origin-content', | file_url = reverse('browse-origin-content', | ||||
url_args=file_url_args, | url_args=file_url_args, | ||||
query_params=query_params) | query_params=query_params) | ||||
self.assertContains(resp, file_url) | assert_contains(resp, file_url) | ||||
if 'path' in url_args: | if 'path' in url_args: | ||||
del url_args['path'] | del url_args['path'] | ||||
root_dir_branch_url = \ | root_dir_branch_url = reverse('browse-origin-directory', | ||||
reverse('browse-origin-directory', | |||||
url_args=url_args, | url_args=url_args, | ||||
query_params=query_params) | query_params=query_params) | ||||
nb_bc_paths = 1 | nb_bc_paths = 1 | ||||
if path: | if path: | ||||
nb_bc_paths = len(path.split('/')) + 1 | nb_bc_paths = len(path.split('/')) + 1 | ||||
self.assertContains(resp, '<li class="swh-path">', count=nb_bc_paths) | assert_contains(resp, '<li class="swh-path">', count=nb_bc_paths) | ||||
self.assertContains(resp, '<a href="%s">%s</a>' % | assert_contains(resp, '<a href="%s">%s</a>' % | ||||
(root_dir_branch_url, | (root_dir_branch_url, | ||||
root_directory_sha1[:7])) | root_directory_sha1[:7])) | ||||
origin_branches_url = reverse('browse-origin-branches', | origin_branches_url = reverse('browse-origin-branches', | ||||
url_args=url_args, | url_args=url_args, | ||||
query_params=query_params) | query_params=query_params) | ||||
self.assertContains(resp, '<a href="%s">Branches (%s)</a>' % | assert_contains(resp, '<a href="%s">Branches (%s)</a>' % | ||||
(origin_branches_url, len(origin_branches))) | (origin_branches_url, len(origin_branches))) | ||||
origin_releases_url = reverse('browse-origin-releases', | origin_releases_url = reverse('browse-origin-releases', | ||||
url_args=url_args, | url_args=url_args, | ||||
query_params=query_params) | query_params=query_params) | ||||
nb_releases = len(origin_releases) | nb_releases = len(origin_releases) | ||||
if nb_releases > 0: | if nb_releases > 0: | ||||
self.assertContains(resp, '<a href="%s">Releases (%s)</a>' % | assert_contains(resp, '<a href="%s">Releases (%s)</a>' % | ||||
(origin_releases_url, nb_releases)) | (origin_releases_url, nb_releases)) | ||||
if path: | if path: | ||||
url_args['path'] = path | url_args['path'] = path | ||||
self.assertContains(resp, '<li class="swh-branch">', | assert_contains(resp, '<li class="swh-branch">', | ||||
count=len(origin_branches)) | count=len(origin_branches)) | ||||
for branch in origin_branches: | for branch in origin_branches: | ||||
query_params['branch'] = branch['name'] | query_params['branch'] = branch['name'] | ||||
root_dir_branch_url = \ | root_dir_branch_url = reverse('browse-origin-directory', | ||||
reverse('browse-origin-directory', | |||||
url_args=url_args, | url_args=url_args, | ||||
query_params=query_params) | query_params=query_params) | ||||
self.assertContains(resp, '<a href="%s">' % root_dir_branch_url) | assert_contains(resp, '<a href="%s">' % root_dir_branch_url) | ||||
self.assertContains(resp, '<li class="swh-release">', | assert_contains(resp, '<li class="swh-release">', | ||||
count=len(origin_releases)) | count=len(origin_releases)) | ||||
query_params['branch'] = None | query_params['branch'] = None | ||||
for release in origin_releases: | for release in origin_releases: | ||||
query_params['release'] = release['name'] | query_params['release'] = release['name'] | ||||
root_dir_release_url = \ | root_dir_release_url = reverse('browse-origin-directory', | ||||
reverse('browse-origin-directory', | |||||
url_args=url_args, | url_args=url_args, | ||||
query_params=query_params) | query_params=query_params) | ||||
self.assertContains(resp, '<a href="%s">' % root_dir_release_url) | assert_contains(resp, '<a href="%s">' % root_dir_release_url) | ||||
self.assertContains(resp, 'vault-cook-directory') | assert_contains(resp, 'vault-cook-directory') | ||||
self.assertContains(resp, 'vault-cook-revision') | assert_contains(resp, 'vault-cook-revision') | ||||
swh_dir_id = get_swh_persistent_id('directory', directory_entries[0]['dir_id']) # noqa | swh_dir_id = get_swh_persistent_id('directory', directory_entries[0]['dir_id']) # noqa | ||||
swh_dir_id_url = reverse('browse-swh-id', | swh_dir_id_url = reverse('browse-swh-id', | ||||
url_args={'swh_id': swh_dir_id}) | url_args={'swh_id': swh_dir_id}) | ||||
self.assertContains(resp, swh_dir_id) | assert_contains(resp, swh_dir_id) | ||||
self.assertContains(resp, swh_dir_id_url) | assert_contains(resp, swh_dir_id_url) | ||||
self.assertContains(resp, 'swh-take-new-snapshot') | |||||
@given(origin()) | |||||
def test_origin_root_directory_view(self, origin): | |||||
origin_visits = self.origin_visit_get(origin['url']) | |||||
visit = origin_visits[-1] | assert_contains(resp, 'swh-take-new-snapshot') | ||||
snapshot = self.snapshot_get(visit['snapshot']) | |||||
head_rev_id = self.snapshot_get_head(snapshot) | |||||
head_rev = self.revision_get(head_rev_id) | |||||
root_dir_sha1 = head_rev['directory'] | |||||
dir_content = self.directory_ls(root_dir_sha1) | |||||
branches, releases = process_snapshot_branches(snapshot) | |||||
visit_unix_ts = parse_timestamp(visit['date']).timestamp() | |||||
visit_unix_ts = int(visit_unix_ts) | |||||
self.origin_directory_view_helper(origin, origin_visits, | |||||
branches, | |||||
releases, | |||||
root_dir_sha1, | |||||
dir_content) | |||||
self.origin_directory_view_helper(origin, origin_visits, | |||||
branches, | |||||
releases, | |||||
root_dir_sha1, | |||||
dir_content, | |||||
visit_id=visit['visit']) | |||||
self.origin_directory_view_helper(origin, origin_visits, | |||||
branches, | |||||
releases, | |||||
root_dir_sha1, | |||||
dir_content, | |||||
timestamp=visit_unix_ts) | |||||
self.origin_directory_view_helper(origin, origin_visits, | |||||
branches, | |||||
releases, | |||||
root_dir_sha1, | |||||
dir_content, | |||||
timestamp=visit['date']) | |||||
origin = dict(origin) | |||||
del origin['type'] | |||||
self.origin_directory_view_helper(origin, origin_visits, | |||||
branches, | |||||
releases, | |||||
root_dir_sha1, | |||||
dir_content) | |||||
self.origin_directory_view_helper(origin, origin_visits, | |||||
branches, | |||||
releases, | |||||
root_dir_sha1, | |||||
dir_content, | |||||
visit_id=visit['visit']) | |||||
self.origin_directory_view_helper(origin, origin_visits, | |||||
branches, | |||||
releases, | |||||
root_dir_sha1, | |||||
dir_content, | |||||
timestamp=visit_unix_ts) | |||||
self.origin_directory_view_helper(origin, origin_visits, | def _origin_branches_test_helper(client, origin_info, origin_snapshot): | ||||
branches, | |||||
releases, | |||||
root_dir_sha1, | |||||
dir_content, | |||||
timestamp=visit['date']) | |||||
@given(origin()) | |||||
def test_origin_sub_directory_view(self, origin): | |||||
origin_visits = self.origin_visit_get(origin['url']) | |||||
visit = origin_visits[-1] | |||||
snapshot = self.snapshot_get(visit['snapshot']) | |||||
head_rev_id = self.snapshot_get_head(snapshot) | |||||
head_rev = self.revision_get(head_rev_id) | |||||
root_dir_sha1 = head_rev['directory'] | |||||
subdirs = [e for e in self.directory_ls(root_dir_sha1) | |||||
if e['type'] == 'dir'] | |||||
branches, releases = process_snapshot_branches(snapshot) | |||||
visit_unix_ts = parse_timestamp(visit['date']).timestamp() | |||||
visit_unix_ts = int(visit_unix_ts) | |||||
if len(subdirs) == 0: | |||||
return | |||||
subdir = random.choice(subdirs) | |||||
subdir_content = self.directory_ls(subdir['target']) | |||||
subdir_path = subdir['name'] | |||||
self.origin_directory_view_helper(origin, origin_visits, | |||||
branches, | |||||
releases, | |||||
root_dir_sha1, | |||||
subdir_content, | |||||
path=subdir_path) | |||||
self.origin_directory_view_helper(origin, origin_visits, | |||||
branches, | |||||
releases, | |||||
root_dir_sha1, | |||||
subdir_content, | |||||
path=subdir_path, | |||||
visit_id=visit['visit']) | |||||
self.origin_directory_view_helper(origin, origin_visits, | |||||
branches, | |||||
releases, | |||||
root_dir_sha1, | |||||
subdir_content, | |||||
path=subdir_path, | |||||
timestamp=visit_unix_ts) | |||||
self.origin_directory_view_helper(origin, origin_visits, | |||||
branches, | |||||
releases, | |||||
root_dir_sha1, | |||||
subdir_content, | |||||
path=subdir_path, | |||||
timestamp=visit['date']) | |||||
origin = dict(origin) | |||||
del origin['type'] | |||||
self.origin_directory_view_helper(origin, origin_visits, | |||||
branches, | |||||
releases, | |||||
root_dir_sha1, | |||||
subdir_content, | |||||
path=subdir_path) | |||||
self.origin_directory_view_helper(origin, origin_visits, | |||||
branches, | |||||
releases, | |||||
root_dir_sha1, | |||||
subdir_content, | |||||
path=subdir_path, | |||||
visit_id=visit['visit']) | |||||
self.origin_directory_view_helper(origin, origin_visits, | |||||
branches, | |||||
releases, | |||||
root_dir_sha1, | |||||
subdir_content, | |||||
path=subdir_path, | |||||
timestamp=visit_unix_ts) | |||||
self.origin_directory_view_helper(origin, origin_visits, | |||||
branches, | |||||
releases, | |||||
root_dir_sha1, | |||||
subdir_content, | |||||
path=subdir_path, | |||||
timestamp=visit['date']) | |||||
def origin_branches_helper(self, origin_info, origin_snapshot): | |||||
url_args = {'origin_url': origin_info['url']} | url_args = {'origin_url': origin_info['url']} | ||||
url = reverse('browse-origin-branches', | url = reverse('browse-origin-branches', | ||||
url_args=url_args) | url_args=url_args) | ||||
resp = self.client.get(url) | resp = client.get(url) | ||||
self.assertEqual(resp.status_code, 200) | assert resp.status_code == 200 | ||||
self.assertTemplateUsed('branches.html') | assert_template_used('branches.html') | ||||
origin_branches = origin_snapshot[0] | origin_branches = origin_snapshot[0] | ||||
origin_releases = origin_snapshot[1] | origin_releases = origin_snapshot[1] | ||||
origin_branches_url = reverse('browse-origin-branches', | origin_branches_url = reverse('browse-origin-branches', | ||||
url_args=url_args) | url_args=url_args) | ||||
self.assertContains(resp, '<a href="%s">Branches (%s)</a>' % | assert_contains(resp, '<a href="%s">Branches (%s)</a>' % | ||||
(origin_branches_url, len(origin_branches))) | (origin_branches_url, len(origin_branches))) | ||||
origin_releases_url = reverse('browse-origin-releases', | origin_releases_url = reverse('browse-origin-releases', | ||||
url_args=url_args) | url_args=url_args) | ||||
nb_releases = len(origin_releases) | nb_releases = len(origin_releases) | ||||
if nb_releases > 0: | if nb_releases > 0: | ||||
self.assertContains(resp, '<a href="%s">Releases (%s)</a>' % | assert_contains(resp, '<a href="%s">Releases (%s)</a>' % | ||||
(origin_releases_url, nb_releases)) | (origin_releases_url, nb_releases)) | ||||
self.assertContains(resp, '<tr class="swh-branch-entry', | assert_contains(resp, '<tr class="swh-branch-entry', | ||||
count=len(origin_branches)) | count=len(origin_branches)) | ||||
for branch in origin_branches: | for branch in origin_branches: | ||||
browse_branch_url = reverse( | browse_branch_url = reverse( | ||||
'browse-origin-directory', | 'browse-origin-directory', | ||||
url_args={'origin_url': origin_info['url']}, | url_args={'origin_url': origin_info['url']}, | ||||
query_params={'branch': branch['name']}) | query_params={'branch': branch['name']}) | ||||
self.assertContains(resp, '<a href="%s">' % | assert_contains(resp, '<a href="%s">' % | ||||
escape(browse_branch_url)) | escape(browse_branch_url)) | ||||
browse_revision_url = reverse( | browse_revision_url = reverse( | ||||
'browse-revision', | 'browse-revision', | ||||
url_args={'sha1_git': branch['revision']}, | url_args={'sha1_git': branch['revision']}, | ||||
query_params={'origin': origin_info['url']}) | query_params={'origin': origin_info['url']}) | ||||
self.assertContains(resp, '<a href="%s">' % | assert_contains(resp, '<a href="%s">' % | ||||
escape(browse_revision_url)) | escape(browse_revision_url)) | ||||
@given(origin()) | |||||
def test_origin_branches(self, origin): | |||||
origin_visits = self.origin_visit_get(origin['url']) | |||||
visit = origin_visits[-1] | |||||
snapshot = self.snapshot_get(visit['snapshot']) | |||||
snapshot_content = process_snapshot_branches(snapshot) | |||||
self.origin_branches_helper(origin, snapshot_content) | |||||
origin = dict(origin) | def _origin_releases_test_helper(client, origin_info, origin_snapshot): | ||||
origin['type'] = None | |||||
self.origin_branches_helper(origin, snapshot_content) | |||||
def origin_releases_helper(self, origin_info, origin_snapshot): | |||||
url_args = {'origin_url': origin_info['url']} | url_args = {'origin_url': origin_info['url']} | ||||
url = reverse('browse-origin-releases', | url = reverse('browse-origin-releases', | ||||
url_args=url_args) | url_args=url_args) | ||||
resp = self.client.get(url) | resp = client.get(url) | ||||
self.assertEqual(resp.status_code, 200) | assert resp.status_code == 200 | ||||
self.assertTemplateUsed('releases.html') | assert_template_used('releases.html') | ||||
origin_branches = origin_snapshot[0] | origin_branches = origin_snapshot[0] | ||||
origin_releases = origin_snapshot[1] | origin_releases = origin_snapshot[1] | ||||
origin_branches_url = reverse('browse-origin-branches', | origin_branches_url = reverse('browse-origin-branches', | ||||
url_args=url_args) | url_args=url_args) | ||||
self.assertContains(resp, '<a href="%s">Branches (%s)</a>' % | assert_contains(resp, '<a href="%s">Branches (%s)</a>' % | ||||
(origin_branches_url, len(origin_branches))) | (origin_branches_url, len(origin_branches))) | ||||
origin_releases_url = reverse('browse-origin-releases', | origin_releases_url = reverse('browse-origin-releases', | ||||
url_args=url_args) | url_args=url_args) | ||||
nb_releases = len(origin_releases) | nb_releases = len(origin_releases) | ||||
if nb_releases > 0: | if nb_releases > 0: | ||||
self.assertContains(resp, '<a href="%s">Releases (%s)</a>' % | assert_contains(resp, '<a href="%s">Releases (%s)</a>' % | ||||
(origin_releases_url, nb_releases)) | (origin_releases_url, nb_releases)) | ||||
self.assertContains(resp, '<tr class="swh-release-entry', | assert_contains(resp, '<tr class="swh-release-entry', | ||||
count=nb_releases) | count=nb_releases) | ||||
for release in origin_releases: | for release in origin_releases: | ||||
browse_release_url = reverse( | browse_release_url = reverse( | ||||
'browse-release', | 'browse-release', | ||||
url_args={'sha1_git': release['id']}, | url_args={'sha1_git': release['id']}, | ||||
query_params={'origin': origin_info['url']}) | query_params={'origin': origin_info['url']}) | ||||
browse_revision_url = reverse( | browse_revision_url = reverse( | ||||
'browse-revision', | 'browse-revision', | ||||
url_args={'sha1_git': release['target']}, | url_args={'sha1_git': release['target']}, | ||||
query_params={'origin': origin_info['url']}) | query_params={'origin': origin_info['url']}) | ||||
self.assertContains(resp, '<a href="%s">' % | assert_contains(resp, '<a href="%s">' % | ||||
escape(browse_release_url)) | escape(browse_release_url)) | ||||
self.assertContains(resp, '<a href="%s">' % | assert_contains(resp, '<a href="%s">' % | ||||
escape(browse_revision_url)) | escape(browse_revision_url)) | ||||
@given(origin()) | |||||
def test_origin_releases(self, origin): | |||||
origin_visits = self.origin_visit_get(origin['url']) | |||||
visit = origin_visits[-1] | |||||
snapshot = self.snapshot_get(visit['snapshot']) | |||||
snapshot_content = process_snapshot_branches(snapshot) | |||||
self.origin_releases_helper(origin, snapshot_content) | |||||
origin = dict(origin) | |||||
origin['type'] = None | |||||
self.origin_releases_helper(origin, snapshot_content) | |||||
@given(new_origin(), new_snapshot(min_size=4, max_size=4), visit_dates(), | |||||
revisions(min_size=3, max_size=3)) | |||||
def test_origin_snapshot_null_branch(self, new_origin, new_snapshot, | |||||
visit_dates, revisions): | |||||
snp_dict = new_snapshot.to_dict() | |||||
new_origin = self.storage.origin_add([new_origin])[0] | |||||
for i, branch in enumerate(snp_dict['branches'].keys()): | |||||
if i == 0: | |||||
snp_dict['branches'][branch] = None | |||||
else: | |||||
snp_dict['branches'][branch] = { | |||||
'target_type': 'revision', | |||||
'target': hash_to_bytes(revisions[i-1]), | |||||
} | |||||
self.storage.snapshot_add([snp_dict]) | |||||
visit = self.storage.origin_visit_add( | |||||
new_origin['url'], visit_dates[0], type='git') | |||||
self.storage.origin_visit_update(new_origin['url'], visit['visit'], | |||||
status='partial', | |||||
snapshot=snp_dict['id']) | |||||
url = reverse('browse-origin-directory', | |||||
url_args={'origin_url': new_origin['url']}) | |||||
rv = self.client.get(url) | |||||
self.assertEqual(rv.status_code, 200) | |||||
@given(new_origin(), new_snapshot(min_size=4, max_size=4), visit_dates(), | |||||
revisions(min_size=3, max_size=3)) | |||||
def test_origin_snapshot_invalid_branch(self, new_origin, new_snapshot, | |||||
visit_dates, revisions): | |||||
snp_dict = new_snapshot.to_dict() | |||||
new_origin = self.storage.origin_add([new_origin])[0] | |||||
for i, branch in enumerate(snp_dict['branches'].keys()): | |||||
if i == 0: | |||||
invalid_branch = branch | |||||
else: | |||||
snp_dict['branches'][branch] = { | |||||
'target_type': 'revision', | |||||
'target': hash_to_bytes(revisions[i-1]), | |||||
} | |||||
del snp_dict['branches'][invalid_branch] | |||||
self.storage.snapshot_add([snp_dict]) | |||||
visit = self.storage.origin_visit_add( | |||||
new_origin['url'], visit_dates[0], type='git') | |||||
self.storage.origin_visit_update(new_origin['url'], visit['visit'], | |||||
status='full', | |||||
snapshot=snp_dict['id']) | |||||
url = reverse('browse-origin-directory', | |||||
url_args={'origin_url': new_origin['url']}, | |||||
query_params={'branch': invalid_branch}) | |||||
rv = self.client.get(url) | |||||
self.assertEqual(rv.status_code, 404) | |||||
@patch('swh.web.browse.views.utils.snapshot_context.request_content') | |||||
@patch('swh.web.common.origin_visits.get_origin_visits') | |||||
@patch('swh.web.browse.utils.get_origin_visit_snapshot') | |||||
@patch('swh.web.browse.utils.service') | |||||
@patch('swh.web.browse.views.origin.service') | |||||
@patch('swh.web.browse.views.utils.snapshot_context.service') | |||||
def test_origin_request_errors(self, | |||||
mock_snapshot_service, | |||||
mock_origin_service, | |||||
mock_utils_service, | |||||
mock_get_origin_visit_snapshot, | |||||
mock_get_origin_visits, | |||||
mock_request_content): | |||||
mock_origin_service.lookup_origin.side_effect = NotFoundExc( | |||||
'origin not found') | |||||
url = reverse('browse-origin-visits', | |||||
url_args={'origin_url': 'bar'}) | |||||
resp = self.client.get(url) | |||||
self.assertEqual(resp.status_code, 404) | |||||
self.assertTemplateUsed('error.html') | |||||
self.assertContains(resp, 'origin not found', status_code=404) | |||||
mock_origin_service.lookup_origin.side_effect = None | |||||
mock_origin_service.lookup_origin.return_value = {'type': 'foo', | |||||
'url': 'bar', | |||||
'id': 457} | |||||
mock_get_origin_visits.return_value = [] | |||||
url = reverse('browse-origin-directory', | |||||
url_args={'origin_url': 'bar'}) | |||||
resp = self.client.get(url) | |||||
self.assertEqual(resp.status_code, 404) | |||||
self.assertTemplateUsed('error.html') | |||||
self.assertContains(resp, "No visit", status_code=404) | |||||
mock_get_origin_visits.return_value = [{'visit': 1}] | |||||
mock_get_origin_visit_snapshot.side_effect = \ | |||||
NotFoundExc('visit not found') | |||||
url = reverse('browse-origin-directory', | |||||
url_args={'origin_url': 'bar'}, | |||||
query_params={'visit_id': 2}) | |||||
resp = self.client.get(url) | |||||
self.assertEqual(resp.status_code, 404) | |||||
self.assertTemplateUsed('error.html') | |||||
self.assertRegex(resp.content.decode('utf-8'), 'Visit.*not found') | |||||
mock_get_origin_visits.return_value = [{ | |||||
'date': '2015-09-26T09:30:52.373449+00:00', | |||||
'metadata': {}, | |||||
'origin': 457, | |||||
'snapshot': 'bdaf9ac436488a8c6cda927a0f44e172934d3f65', | |||||
'status': 'full', | |||||
'visit': 1 | |||||
}] | |||||
mock_get_origin_visit_snapshot.side_effect = None | |||||
mock_get_origin_visit_snapshot.return_value = ( | |||||
[{'directory': 'ae59ceecf46367e8e4ad800e231fc76adc3afffb', | |||||
'name': 'HEAD', | |||||
'revision': '7bc08e1aa0b08cb23e18715a32aa38517ad34672', | |||||
'date': '04 May 2017, 13:27 UTC', | |||||
'message': ''}], | |||||
[] | |||||
) | |||||
mock_utils_service.lookup_snapshot_size.return_value = { | |||||
'revision': 1, | |||||
'release': 0 | |||||
} | |||||
mock_utils_service.lookup_directory.side_effect = \ | |||||
NotFoundExc('Directory not found') | |||||
url = reverse('browse-origin-directory', | |||||
url_args={'origin_url': 'bar'}) | |||||
resp = self.client.get(url) | |||||
self.assertEqual(resp.status_code, 404) | |||||
self.assertTemplateUsed('error.html') | |||||
self.assertContains(resp, 'Directory not found', status_code=404) | |||||
with patch('swh.web.browse.views.utils.snapshot_context.' | |||||
'get_snapshot_context') as mock_get_snapshot_context: | |||||
mock_get_snapshot_context.side_effect = \ | |||||
NotFoundExc('Snapshot not found') | |||||
url = reverse('browse-origin-directory', | |||||
url_args={'origin_url': 'bar'}) | |||||
resp = self.client.get(url) | |||||
self.assertEqual(resp.status_code, 404) | |||||
self.assertTemplateUsed('error.html') | |||||
self.assertContains(resp, 'Snapshot not found', status_code=404) | |||||
mock_origin_service.lookup_origin.side_effect = None | |||||
mock_origin_service.lookup_origin.return_value = {'type': 'foo', | |||||
'url': 'bar', | |||||
'id': 457} | |||||
mock_get_origin_visits.return_value = [] | |||||
url = reverse('browse-origin-content', | |||||
url_args={'origin_url': 'bar', | |||||
'path': 'foo'}) | |||||
resp = self.client.get(url) | |||||
self.assertEqual(resp.status_code, 404) | |||||
self.assertTemplateUsed('error.html') | |||||
self.assertContains(resp, "No visit", status_code=404) | |||||
mock_get_origin_visits.return_value = [{'visit': 1}] | |||||
mock_get_origin_visit_snapshot.side_effect = \ | |||||
NotFoundExc('visit not found') | |||||
url = reverse('browse-origin-content', | |||||
url_args={'origin_url': 'bar', | |||||
'path': 'foo'}, | |||||
query_params={'visit_id': 2}) | |||||
resp = self.client.get(url) | |||||
self.assertEqual(resp.status_code, 404) | |||||
self.assertTemplateUsed('error.html') | |||||
self.assertRegex(resp.content.decode('utf-8'), 'Visit.*not found') | |||||
mock_get_origin_visits.return_value = [{ | |||||
'date': '2015-09-26T09:30:52.373449+00:00', | |||||
'metadata': {}, | |||||
'origin': 457, | |||||
'snapshot': 'bdaf9ac436488a8c6cda927a0f44e172934d3f65', | |||||
'status': 'full', | |||||
'visit': 1 | |||||
}] | |||||
mock_get_origin_visit_snapshot.side_effect = None | |||||
mock_get_origin_visit_snapshot.return_value = ([], []) | |||||
url = reverse('browse-origin-content', | |||||
url_args={'origin_url': 'bar', | |||||
'path': 'baz'}) | |||||
resp = self.client.get(url) | |||||
self.assertEqual(resp.status_code, 404) | |||||
self.assertTemplateUsed('error.html') | |||||
self.assertRegex(resp.content.decode('utf-8'), | |||||
'Origin.*has an empty list of branches') | |||||
mock_get_origin_visit_snapshot.return_value = ( | |||||
[{'directory': 'ae59ceecf46367e8e4ad800e231fc76adc3afffb', | |||||
'name': 'HEAD', | |||||
'revision': '7bc08e1aa0b08cb23e18715a32aa38517ad34672', | |||||
'date': '04 May 2017, 13:27 UTC', | |||||
'message': ''}], | |||||
[] | |||||
) | |||||
mock_snapshot_service.lookup_directory_with_path.return_value = \ | |||||
{'target': '5ecd9f37b7a2d2e9980d201acd6286116f2ba1f1'} | |||||
mock_request_content.side_effect = \ | |||||
NotFoundExc('Content not found') | |||||
url = reverse('browse-origin-content', | |||||
url_args={'origin_url': 'bar', | |||||
'path': 'baz'}) | |||||
resp = self.client.get(url) | |||||
self.assertEqual(resp.status_code, 404) | |||||
self.assertTemplateUsed('error.html') | |||||
self.assertContains(resp, 'Content not found', status_code=404) | |||||
@patch('swh.web.common.origin_visits.get_origin_visits') | |||||
@patch('swh.web.browse.utils.get_origin_visit_snapshot') | |||||
@patch('swh.web.browse.utils.service') | |||||
def test_origin_empty_snapshot(self, mock_utils_service, | |||||
mock_get_origin_visit_snapshot, | |||||
mock_get_origin_visits): | |||||
mock_get_origin_visits.return_value = [{ | |||||
'date': '2015-09-26T09:30:52.373449+00:00', | |||||
'metadata': {}, | |||||
'origin': 457, | |||||
'snapshot': 'bdaf9ac436488a8c6cda927a0f44e172934d3f65', | |||||
'status': 'full', | |||||
'type': 'git', | |||||
'visit': 1 | |||||
}] | |||||
mock_get_origin_visit_snapshot.return_value = ([], []) | |||||
mock_utils_service.lookup_snapshot_size.return_value = { | |||||
'revision': 0, | |||||
'release': 0 | |||||
} | |||||
mock_utils_service.lookup_origin.return_value = { | |||||
'id': 457, | |||||
'url': 'https://github.com/foo/bar' | |||||
} | |||||
url = reverse('browse-origin-directory', | |||||
url_args={'origin_url': 'bar'}) | |||||
resp = self.client.get(url) | |||||
self.assertEqual(resp.status_code, 200) | |||||
self.assertTemplateUsed('content.html') | |||||
self.assertRegex(resp.content.decode('utf-8'), 'snapshot.*is empty') |