Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/browse/views/test_directory.py
# Copyright (C) 2017-2021 The Software Heritage developers | # Copyright (C) 2017-2021 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 hypothesis import given | from hypothesis import given | ||||
import pytest | import pytest | ||||
from django.utils.html import escape | from django.utils.html import escape | ||||
from swh.model.from_disk import DentryPerms | from swh.model.from_disk import DentryPerms | ||||
from swh.model.hashutil import hash_to_bytes, hash_to_hex | from swh.model.hashutil import hash_to_bytes, hash_to_hex | ||||
Show All 12 Lines | |||||
from swh.model.model import Directory, DirectoryEntry | from swh.model.model import Directory, DirectoryEntry | ||||
from swh.model.model import ObjectType as ModelObjectType | from swh.model.model import ObjectType as ModelObjectType | ||||
from swh.model.swhids import ObjectType | from swh.model.swhids import ObjectType | ||||
from swh.storage.utils import now | from swh.storage.utils import now | ||||
from swh.web.browse.snapshot_context import process_snapshot_branches | from swh.web.browse.snapshot_context import process_snapshot_branches | ||||
from swh.web.common.identifiers import gen_swhid | from swh.web.common.identifiers import gen_swhid | ||||
from swh.web.common.utils import gen_path_info, reverse | from swh.web.common.utils import gen_path_info, reverse | ||||
from swh.web.tests.django_asserts import assert_contains, assert_not_contains | from swh.web.tests.django_asserts import assert_contains, assert_not_contains | ||||
from swh.web.tests.strategies import new_person, new_swh_date | from swh.web.tests.strategies import ( | ||||
new_origin, | |||||
new_person, | |||||
new_snapshot, | |||||
new_swh_date, | |||||
visit_dates, | |||||
) | |||||
from swh.web.tests.utils import check_html_get_response | from swh.web.tests.utils import check_html_get_response | ||||
def test_root_directory_view(client, archive_data, directory): | def test_root_directory_view(client, archive_data, directory): | ||||
_directory_view_checks(client, directory, archive_data.directory_ls(directory)) | _directory_view_checks(client, directory, archive_data.directory_ls(directory)) | ||||
def test_sub_directory_view(client, archive_data, directory_with_subdirs): | def test_sub_directory_view(client, archive_data, directory_with_subdirs): | ||||
▲ Show 20 Lines • Show All 498 Lines • ▼ Show 20 Lines | browse_url = reverse( | ||||
"release": release_name, | "release": release_name, | ||||
"snapshot": snapshot.id.hex(), | "snapshot": snapshot.id.hex(), | ||||
}, | }, | ||||
) | ) | ||||
check_html_get_response( | check_html_get_response( | ||||
client, browse_url, status_code=200, template_used="browse/directory.html" | client, browse_url, status_code=200, template_used="browse/directory.html" | ||||
) | ) | ||||
# starts | |||||
# def test_pull_request_branches_filtering( | |||||
# client, origin_with_pull_request_branches, archive_data | |||||
# ): | |||||
# origin_url = origin_with_pull_request_branches.url | |||||
# # check no pull request branches are displayed in the Branches / Releases dropdown | |||||
# url = reverse("browse-origin-directory", query_params={"origin_url": origin_url}) | |||||
# resp = check_html_get_response( | |||||
# client, url, status_code=200, template_used="browse/directory.html" | |||||
# ) | |||||
# assert_not_contains(resp, "refs/pull/") | |||||
# snapshot = archive_data.snapshot_get_latest(origin_url) | |||||
# # check no pull request branches are displayed in the branches view | |||||
# url = reverse( | |||||
# "browse-snapshot-branches", | |||||
# url_args={"snapshot_id": snapshot["id"]}, | |||||
# query_params={"origin_url": origin_url}, | |||||
# ) | |||||
# resp = check_html_get_response( | |||||
# client, url, status_code=200, template_used="browse/branches.html" | |||||
# ) | |||||
# assert_not_contains(resp, "refs/pull/") | |||||
# @pytest.mark.django_db | |||||
# @pytest.mark.parametrize("object_type", ["directory"]) | |||||
# @given(new_origin()) | |||||
# def test_browse_origin_content_directory_empty_snapshot( | |||||
# client, staff_user, archive_data, object_type, new_origin | |||||
# ): | |||||
# _add_empty_snapshot_origin(new_origin, archive_data) | |||||
# # to check proper generation of raw extrinsic metadata api links | |||||
# client.force_login(staff_user) | |||||
# url = reverse( | |||||
# f"browse-origin-{object_type}", | |||||
# query_params={"origin_url": new_origin.url, "path": "baz"}, | |||||
# ) | |||||
# resp = check_html_get_response( | |||||
# client, url, status_code=200, template_used=f"browse/{object_type}.html" | |||||
# ) | |||||
# assert re.search("snapshot.*is empty", resp.content.decode("utf-8")) | |||||
# def test_browse_directory_snapshot_not_found(client, mocker, origin): | |||||
# mock_get_snapshot_context = mocker.patch( | |||||
# "swh.web.browse.snapshot_context.get_snapshot_context" | |||||
# ) | |||||
# mock_get_snapshot_context.side_effect = NotFoundExc("Snapshot not found") | |||||
# url = reverse("browse-origin-directory", | |||||
# query_params={"origin_url": origin["url"]}) | |||||
# resp = check_html_get_response( | |||||
# client, url, status_code=404, template_used="error.html" | |||||
# ) | |||||
# assert_contains(resp, "Snapshot not found", status_code=404) | |||||
# assert mock_get_snapshot_context.called | |||||
# @given(new_origin()) | |||||
# def test_origin_empty_snapshot(client, archive_data, new_origin): | |||||
# _add_empty_snapshot_origin(new_origin, archive_data) | |||||
# url = reverse( | |||||
# "browse-origin-directory", query_params={"origin_url": new_origin.url} | |||||
# ) | |||||
# resp = check_html_get_response( | |||||
# client, url, status_code=200, template_used="browse/directory.html" | |||||
# ) | |||||
# resp_content = resp.content.decode("utf-8") | |||||
# assert re.search("snapshot.*is empty", resp_content) | |||||
# assert not re.search("swh-tr-link", resp_content) | |||||
# def test_origin_sub_directory_view(client, archive_data, swh_scheduler, origin): | |||||
# origin_visits = archive_data.origin_visit_get(origin["url"]) | |||||
# visit = origin_visits[-1] | |||||
# snapshot = archive_data.snapshot_get(visit["snapshot"]) | |||||
# snapshot_sizes = archive_data.snapshot_count_branches(snapshot["id"]) | |||||
# 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) | |||||
# 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, | |||||
# archive_data, | |||||
# origin, | |||||
# visit, | |||||
# snapshot_sizes, | |||||
# branches, | |||||
# releases, | |||||
# root_dir_sha1, | |||||
# subdir_content, | |||||
# path=subdir_path, | |||||
# ) | |||||
# _origin_directory_view_test_helper( | |||||
# client, | |||||
# archive_data, | |||||
# origin, | |||||
# visit, | |||||
# snapshot_sizes, | |||||
# branches, | |||||
# releases, | |||||
# root_dir_sha1, | |||||
# subdir_content, | |||||
# path=subdir_path, | |||||
# visit_id=visit["visit"], | |||||
# ) | |||||
# _origin_directory_view_test_helper( | |||||
# client, | |||||
# archive_data, | |||||
# origin, | |||||
# visit, | |||||
# snapshot_sizes, | |||||
# branches, | |||||
# releases, | |||||
# root_dir_sha1, | |||||
# subdir_content, | |||||
# path=subdir_path, | |||||
# timestamp=visit["date"], | |||||
# ) | |||||
# _origin_directory_view_test_helper( | |||||
# client, | |||||
# archive_data, | |||||
# origin, | |||||
# visit, | |||||
# snapshot_sizes, | |||||
# branches, | |||||
# releases, | |||||
# root_dir_sha1, | |||||
# subdir_content, | |||||
# path=subdir_path, | |||||
# snapshot_id=visit["snapshot"], | |||||
# ) | |||||
# _origin_directory_view_test_helper( | |||||
# client, | |||||
# archive_data, | |||||
# origin, | |||||
# visit, | |||||
# snapshot_sizes, | |||||
# branches, | |||||
# releases, | |||||
# root_dir_sha1, | |||||
# subdir_content, | |||||
# path=subdir_path, | |||||
# ) | |||||
# _origin_directory_view_test_helper( | |||||
# client, | |||||
# archive_data, | |||||
# origin, | |||||
# visit, | |||||
# snapshot_sizes, | |||||
# branches, | |||||
# releases, | |||||
# root_dir_sha1, | |||||
# subdir_content, | |||||
# path=subdir_path, | |||||
# visit_id=visit["visit"], | |||||
# ) | |||||
# _origin_directory_view_test_helper( | |||||
# client, | |||||
# archive_data, | |||||
# origin, | |||||
# visit, | |||||
# snapshot_sizes, | |||||
# branches, | |||||
# releases, | |||||
# root_dir_sha1, | |||||
# subdir_content, | |||||
# path=subdir_path, | |||||
# timestamp=visit["date"], | |||||
# ) | |||||
# _origin_directory_view_test_helper( | |||||
# client, | |||||
# archive_data, | |||||
# origin, | |||||
# visit, | |||||
# snapshot_sizes, | |||||
# branches, | |||||
# releases, | |||||
# root_dir_sha1, | |||||
# subdir_content, | |||||
# path=subdir_path, | |||||
# snapshot_id=visit["snapshot"], | |||||
# ) | |||||
# def test_origin_root_directory_view(client, archive_data, swh_scheduler, origin): | |||||
# origin_visits = archive_data.origin_visit_get(origin["url"]) | |||||
# visit = origin_visits[-1] | |||||
# snapshot = archive_data.snapshot_get(visit["snapshot"]) | |||||
# snapshot_sizes = archive_data.snapshot_count_branches(snapshot["id"]) | |||||
# 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) | |||||
# _origin_directory_view_test_helper( | |||||
# client, | |||||
# archive_data, | |||||
# origin, | |||||
# visit, | |||||
# snapshot_sizes, | |||||
# branches, | |||||
# releases, | |||||
# root_dir_sha1, | |||||
# dir_content, | |||||
# ) | |||||
# _origin_directory_view_test_helper( | |||||
# client, | |||||
# archive_data, | |||||
# origin, | |||||
# visit, | |||||
# snapshot_sizes, | |||||
# branches, | |||||
# releases, | |||||
# root_dir_sha1, | |||||
# dir_content, | |||||
# visit_id=visit["visit"], | |||||
# ) | |||||
# _origin_directory_view_test_helper( | |||||
# client, | |||||
# archive_data, | |||||
# origin, | |||||
# visit, | |||||
# snapshot_sizes, | |||||
# branches, | |||||
# releases, | |||||
# root_dir_sha1, | |||||
# dir_content, | |||||
# timestamp=visit["date"], | |||||
# ) | |||||
# _origin_directory_view_test_helper( | |||||
# client, | |||||
# archive_data, | |||||
# origin, | |||||
# visit, | |||||
# snapshot_sizes, | |||||
# branches, | |||||
# releases, | |||||
# root_dir_sha1, | |||||
# dir_content, | |||||
# snapshot_id=visit["snapshot"], | |||||
# ) | |||||
# _origin_directory_view_test_helper( | |||||
# client, | |||||
# archive_data, | |||||
# origin, | |||||
# visit, | |||||
# snapshot_sizes, | |||||
# branches, | |||||
# releases, | |||||
# root_dir_sha1, | |||||
# dir_content, | |||||
# ) | |||||
# _origin_directory_view_test_helper( | |||||
# client, | |||||
# archive_data, | |||||
# origin, | |||||
# visit, | |||||
# snapshot_sizes, | |||||
# branches, | |||||
# releases, | |||||
# root_dir_sha1, | |||||
# dir_content, | |||||
# visit_id=visit["visit"], | |||||
# ) | |||||
# _origin_directory_view_test_helper( | |||||
# client, | |||||
# archive_data, | |||||
# origin, | |||||
# visit, | |||||
# snapshot_sizes, | |||||
# branches, | |||||
# releases, | |||||
# root_dir_sha1, | |||||
# dir_content, | |||||
# timestamp=visit["date"], | |||||
# ) | |||||
# _origin_directory_view_test_helper( | |||||
# client, | |||||
# archive_data, | |||||
# origin, | |||||
# visit, | |||||
# snapshot_sizes, | |||||
# branches, | |||||
# releases, | |||||
# root_dir_sha1, | |||||
# dir_content, | |||||
# snapshot_id=visit["snapshot"], | |||||
# ) | |||||
# @given( | |||||
# new_origin(), new_snapshot(min_size=4, max_size=4), visit_dates(), | |||||
# ) | |||||
# def test_browse_directory_snapshot_null_branch( | |||||
# client, archive_data, revisions_list, new_origin, new_snapshot, visit_dates, | |||||
# ): | |||||
# revisions = revisions_list(size=4) | |||||
# snp_dict = new_snapshot.to_dict() | |||||
# archive_data.origin_add([new_origin]) | |||||
# 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([Snapshot.from_dict(snp_dict)]) | |||||
# visit = archive_data.origin_visit_add( | |||||
# [OriginVisit(origin=new_origin.url, date=visit_dates[0], type="git",)] | |||||
# )[0] | |||||
# visit_status = OriginVisitStatus( | |||||
# origin=new_origin.url, | |||||
# visit=visit.visit, | |||||
# date=now(), | |||||
# status="partial", | |||||
# snapshot=snp_dict["id"], | |||||
# ) | |||||
# archive_data.origin_visit_status_add([visit_status]) | |||||
# url = reverse( | |||||
# "browse-directory", query_params={"origin_url": new_origin.url} | |||||
# ) | |||||
# check_html_get_response( | |||||
# client, url, status_code=200, template_used="error.html" | |||||
# ) | |||||
@given( | |||||
new_origin(), new_snapshot(min_size=4, max_size=4), visit_dates(), | |||||
) | |||||
def test_browse_directory_snapshot_invalid_branch( | |||||
client, archive_data, revisions_list, new_origin, new_snapshot, visit_dates, | |||||
): | |||||
revisions = revisions_list(size=4) | |||||
snp_dict = new_snapshot.to_dict() | |||||
archive_data.origin_add([new_origin]) | |||||
for i, branch in enumerate(snp_dict["branches"].keys()): | |||||
snp_dict["branches"][branch] = { | |||||
"target_type": "revision", | |||||
"target": hash_to_bytes(revisions[i]), | |||||
} | |||||
archive_data.snapshot_add([Snapshot.from_dict(snp_dict)]) | |||||
visit = archive_data.origin_visit_add( | |||||
[OriginVisit(origin=new_origin.url, date=visit_dates[0], type="git",)] | |||||
)[0] | |||||
visit_status = OriginVisitStatus( | |||||
origin=new_origin.url, | |||||
visit=visit.visit, | |||||
date=now(), | |||||
status="full", | |||||
snapshot=snp_dict["id"], | |||||
) | |||||
archive_data.origin_visit_status_add([visit_status]) | |||||
url = reverse( | |||||
"browse-directory", | |||||
query_params={"origin_url": new_origin.url, "branch": "invalid_branch"}, | |||||
) | |||||
check_html_get_response(client, url, status_code=404, template_used="error.html") | |||||
def test_browse_directory_no_visit(client, mocker, origin): | |||||
mock_get_origin_visits = mocker.patch( | |||||
"swh.web.common.origin_visits.get_origin_visits" | |||||
) | |||||
mock_get_origin_visits.return_value = [] | |||||
mock_archive = mocker.patch("swh.web.common.origin_visits.archive") | |||||
mock_archive.lookup_origin_visit_latest.return_value = None | |||||
url = reverse("browse-directory", query_params={"origin_url": origin["url"]}) | |||||
resp = check_html_get_response( | |||||
client, url, status_code=404, template_used="error.html" | |||||
) | |||||
assert_contains(resp, "No valid visit", status_code=404) | |||||
assert not mock_get_origin_visits.called | |||||
def test_browse_directory_unknown_visit(client, mocker, origin): | |||||
mock_get_origin_visits = mocker.patch( | |||||
"swh.web.common.origin_visits.get_origin_visits" | |||||
) | |||||
mock_get_origin_visits.return_value = [{"visit": 1}] | |||||
url = reverse( | |||||
"browse-directory", query_params={"origin_url": origin["url"], "visit_id": 2}, | |||||
) | |||||
resp = check_html_get_response( | |||||
client, url, status_code=404, template_used="error.html" | |||||
) | |||||
assert re.search("Visit.*not found", resp.content.decode("utf-8")) | |||||
assert mock_get_origin_visits.called | |||||
def test_browse_origin_directory_not_found(client, origin): | |||||
url = reverse( | |||||
"browse-directory", | |||||
query_params={"origin_url": origin["url"], "path": "/invalid/dir/path/"}, | |||||
) | |||||
resp = check_html_get_response( | |||||
client, url, status_code=404, template_used="error.html" | |||||
) | |||||
assert re.search( | |||||
"Directory entry with path.*from root directory", resp.content.decode("utf-8") | |||||
) | |||||
@given(new_origin()) | |||||
def test_directory_browse_empty_snapshot_null_revision( | |||||
client, archive_data, new_origin | |||||
): | |||||
snapshot = Snapshot( | |||||
branches={ | |||||
b"HEAD": SnapshotBranch( | |||||
target="refs/head/master".encode(), target_type=TargetType.ALIAS, | |||||
), | |||||
b"refs/head/master": None, | |||||
} | |||||
) | |||||
archive_data.origin_add([new_origin]) | |||||
archive_data.snapshot_add([snapshot]) | |||||
visit = archive_data.origin_visit_add( | |||||
[OriginVisit(origin=new_origin.url, date=now(), type="git",)] | |||||
)[0] | |||||
visit_status = OriginVisitStatus( | |||||
origin=new_origin.url, | |||||
visit=visit.visit, | |||||
date=now(), | |||||
status="partial", | |||||
snapshot=snapshot.id, | |||||
) | |||||
archive_data.origin_visit_status_add([visit_status]) | |||||
url = reverse("browse-directory", query_params={"origin_url": new_origin.url},) | |||||
resp = check_html_get_response( | |||||
client, url, status_code=404, template_used="error.html" | |||||
) | |||||
resp_content = resp.content.decode("utf-8") | |||||
assert re.search("The requested directory is not available", resp_content) | |||||
def test_content_browse_empty_sha1_missing_identifier(client): | |||||
url = reverse("browse-directory", query_params={},) | |||||
resp = check_html_get_response(client, url, status_code=400) | |||||
assert_contains( | |||||
resp, | |||||
"The origin_url or snapshot query parameter must be provided.", | |||||
status_code=400, | |||||
) | |||||
def test_content_browse_empty_sha1(client, directory): | |||||
url = reverse("browse-directory", query_params={"sha1_git": directory},) | |||||
resp = check_html_get_response(client, url, status_code=302) | |||||
redict_url = reverse( | |||||
"browse-directory", | |||||
url_args={"sha1_git": directory}, | |||||
query_params={"sha1_git": directory}, | |||||
) | |||||
assert resp.url == redict_url | |||||
def test_content_browse_empty_sha1_with_snapshot(client, snapshot, archive_data): | |||||
head_branch = archive_data.snapshot_get_branches(snapshot)["branches"][ | |||||
"refs/heads/master" | |||||
]["target"] | |||||
directory = archive_data.revision_get(head_branch)["directory"] | |||||
url = reverse("browse-directory", query_params={"snapshot": snapshot},) | |||||
resp = check_html_get_response(client, url, status_code=302) | |||||
redict_url = reverse( | |||||
"browse-directory", | |||||
url_args={"sha1_git": directory}, | |||||
query_params={"snapshot": snapshot}, | |||||
) | |||||
assert resp.url == redict_url | |||||
def test_content_browse_empty_sha1_with_origin(client, origin, archive_data): | |||||
snapshot = archive_data.snapshot_get_latest(origin["url"]) | |||||
head_branch = archive_data.snapshot_get_branches(snapshot["id"])["branches"][ | |||||
"refs/heads/master" | |||||
]["target"] | |||||
directory = archive_data.revision_get(head_branch)["directory"] | |||||
url = reverse("browse-directory", query_params={"origin_url": origin["url"]},) | |||||
resp = check_html_get_response(client, url, status_code=302) | |||||
redict_url = reverse( | |||||
"browse-directory", | |||||
url_args={"sha1_git": directory}, | |||||
query_params={"origin_url": origin["url"]}, | |||||
) | |||||
assert resp.url == redict_url |