Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/common/test_archive.py
| Show All 25 Lines | |||||
| from swh.model.swhids import ObjectType | from swh.model.swhids import ObjectType | ||||
| from swh.web.common import archive | from swh.web.common import archive | ||||
| from swh.web.common.exc import BadInputExc, NotFoundExc | from swh.web.common.exc import BadInputExc, NotFoundExc | ||||
| from swh.web.common.typing import OriginInfo, PagedResult | from swh.web.common.typing import OriginInfo, PagedResult | ||||
| from swh.web.tests.conftest import ctags_json_missing, fossology_missing | from swh.web.tests.conftest import ctags_json_missing, fossology_missing | ||||
| from swh.web.tests.data import random_content, random_sha1 | from swh.web.tests.data import random_content, random_sha1 | ||||
| from swh.web.tests.strategies import ( | from swh.web.tests.strategies import ( | ||||
| ancestor_revisions, | ancestor_revisions, | ||||
| directory, | |||||
| empty_directory, | |||||
| invalid_sha1, | invalid_sha1, | ||||
| new_origin, | new_origin, | ||||
| new_revision, | new_revision, | ||||
| non_ancestor_revisions, | non_ancestor_revisions, | ||||
| origin, | origin, | ||||
| release, | release, | ||||
| releases, | releases, | ||||
| revision, | revision, | ||||
| ▲ Show 20 Lines • Show All 225 Lines • ▼ Show 20 Lines | |||||
| def test_lookup_release_multiple_none_found(): | def test_lookup_release_multiple_none_found(): | ||||
| unknown_releases_ = [random_sha1(), random_sha1(), random_sha1()] | unknown_releases_ = [random_sha1(), random_sha1(), random_sha1()] | ||||
| actual_releases = list(archive.lookup_release_multiple(unknown_releases_)) | actual_releases = list(archive.lookup_release_multiple(unknown_releases_)) | ||||
| assert actual_releases == [None] * len(unknown_releases_) | assert actual_releases == [None] * len(unknown_releases_) | ||||
| @given(directory()) | |||||
| def test_lookup_directory_with_path_not_found(directory): | def test_lookup_directory_with_path_not_found(directory): | ||||
| path = "some/invalid/path/here" | path = "some/invalid/path/here" | ||||
| with pytest.raises(NotFoundExc) as e: | with pytest.raises(NotFoundExc) as e: | ||||
| archive.lookup_directory_with_path(directory, path) | archive.lookup_directory_with_path(directory, path) | ||||
| assert e.match( | assert e.match( | ||||
| f"Directory entry with path {path} from root directory {directory} not found" | f"Directory entry with path {path} from root directory {directory} not found" | ||||
| ) | ) | ||||
| @given(directory()) | |||||
| def test_lookup_directory_with_path_found(archive_data, directory): | def test_lookup_directory_with_path_found(archive_data, directory): | ||||
| directory_content = archive_data.directory_ls(directory) | directory_content = archive_data.directory_ls(directory) | ||||
| directory_entry = random.choice(directory_content) | directory_entry = random.choice(directory_content) | ||||
| path = directory_entry["name"] | path = directory_entry["name"] | ||||
| actual_result = archive.lookup_directory_with_path(directory, path) | actual_result = archive.lookup_directory_with_path(directory, path) | ||||
| assert actual_result == directory_entry | assert actual_result == directory_entry | ||||
| ▲ Show 20 Lines • Show All 370 Lines • ▼ Show 20 Lines | def test_lookup_directory_not_found(): | ||||
| unknown_directory_ = random_sha1() | unknown_directory_ = random_sha1() | ||||
| with pytest.raises(NotFoundExc) as e: | with pytest.raises(NotFoundExc) as e: | ||||
| archive.lookup_directory(unknown_directory_) | archive.lookup_directory(unknown_directory_) | ||||
| assert e.match("Directory with sha1_git %s not found" % unknown_directory_) | assert e.match("Directory with sha1_git %s not found" % unknown_directory_) | ||||
| @given(directory()) | |||||
| def test_lookup_directory(archive_data, directory): | def test_lookup_directory(archive_data, directory): | ||||
| actual_directory_ls = list(archive.lookup_directory(directory)) | actual_directory_ls = list(archive.lookup_directory(directory)) | ||||
| expected_directory_ls = archive_data.directory_ls(directory) | expected_directory_ls = archive_data.directory_ls(directory) | ||||
| assert actual_directory_ls == expected_directory_ls | assert actual_directory_ls == expected_directory_ls | ||||
| @given(empty_directory()) | |||||
| def test_lookup_directory_empty(empty_directory): | def test_lookup_directory_empty(empty_directory): | ||||
| actual_directory_ls = list(archive.lookup_directory(empty_directory)) | actual_directory_ls = list(archive.lookup_directory(empty_directory)) | ||||
| assert actual_directory_ls == [] | assert actual_directory_ls == [] | ||||
| @given(origin()) | @given(origin()) | ||||
| def test_lookup_revision_by_nothing_found(origin): | def test_lookup_revision_by_nothing_found(origin): | ||||
| ▲ Show 20 Lines • Show All 138 Lines • ▼ Show 20 Lines | def test_lookup_directory_through_revision_ok_with_data(archive_data, revision): | ||||
| ) == ( | ) == ( | ||||
| revision, | revision, | ||||
| archive.lookup_directory_with_revision( | archive.lookup_directory_with_revision( | ||||
| revision, dir_entry["name"], with_data=True | revision, dir_entry["name"], with_data=True | ||||
| ), | ), | ||||
| ) | ) | ||||
| @given(directory(), release(), revision(), snapshot()) | @given(release(), revision(), snapshot()) | ||||
| def test_lookup_known_objects( | def test_lookup_known_objects( | ||||
| archive_data, content, directory, release, revision, snapshot | archive_data, content, directory, release, revision, snapshot | ||||
| ): | ): | ||||
| expected = archive_data.content_find(content) | expected = archive_data.content_find(content) | ||||
| assert archive.lookup_object(ObjectType.CONTENT, content["sha1_git"]) == expected | assert archive.lookup_object(ObjectType.CONTENT, content["sha1_git"]) == expected | ||||
| expected = archive_data.directory_get(directory) | expected = archive_data.directory_get(directory) | ||||
| assert archive.lookup_object(ObjectType.DIRECTORY, directory) == expected | assert archive.lookup_object(ObjectType.DIRECTORY, directory) == expected | ||||
| ▲ Show 20 Lines • Show All 88 Lines • ▼ Show 20 Lines | assert actual_result == { | ||||
| missing_cnt, | missing_cnt, | ||||
| missing_dir, | missing_dir, | ||||
| missing_rev, | missing_rev, | ||||
| missing_rel, | missing_rel, | ||||
| missing_snp, | missing_snp, | ||||
| } | } | ||||
| @given(directory()) | |||||
| def test_lookup_missing_hashes_some_present(content, directory): | def test_lookup_missing_hashes_some_present(content, directory): | ||||
| missing_rev = random_sha1() | missing_rev = random_sha1() | ||||
| missing_rel = random_sha1() | missing_rel = random_sha1() | ||||
| missing_snp = random_sha1() | missing_snp = random_sha1() | ||||
| grouped_swhids = { | grouped_swhids = { | ||||
| ObjectType.CONTENT: [hash_to_bytes(content["sha1_git"])], | ObjectType.CONTENT: [hash_to_bytes(content["sha1_git"])], | ||||
| ObjectType.DIRECTORY: [hash_to_bytes(directory)], | ObjectType.DIRECTORY: [hash_to_bytes(directory)], | ||||
| ▲ Show 20 Lines • Show All 199 Lines • ▼ Show 20 Lines | ): | ||||
| assert len(branches) == nb_results | assert len(branches) == nb_results | ||||
| for branch_name in branches: | for branch_name in branches: | ||||
| if include_pattern: | if include_pattern: | ||||
| assert include_pattern in branch_name | assert include_pattern in branch_name | ||||
| if exclude_prefix: | if exclude_prefix: | ||||
| assert not branch_name.startswith(exclude_prefix) | assert not branch_name.startswith(exclude_prefix) | ||||
| @given(directory(), revision()) | @given(revision()) | ||||
| def test_lookup_snapshot_branch_names_filtering_paginated( | def test_lookup_snapshot_branch_names_filtering_paginated( | ||||
| archive_data, directory, revision | archive_data, directory, revision | ||||
| ): | ): | ||||
| pattern = "foo" | pattern = "foo" | ||||
| nb_branches_by_target_type = 10 | nb_branches_by_target_type = 10 | ||||
| branches = {} | branches = {} | ||||
| for i in range(nb_branches_by_target_type): | for i in range(nb_branches_by_target_type): | ||||
| branches[f"branch/directory/bar{i}".encode()] = SnapshotBranch( | branches[f"branch/directory/bar{i}".encode()] = SnapshotBranch( | ||||
| ▲ Show 20 Lines • Show All 55 Lines • Show Last 20 Lines | |||||