Changeset View
Changeset View
Standalone View
Standalone View
swh/web/tests/common/test_identifiers.py
| Show All 10 Lines | |||||
| from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
| from swh.model.identifiers import ( | from swh.model.identifiers import ( | ||||
| CONTENT, | CONTENT, | ||||
| DIRECTORY, | DIRECTORY, | ||||
| RELEASE, | RELEASE, | ||||
| REVISION, | REVISION, | ||||
| SNAPSHOT, | SNAPSHOT, | ||||
| PersistentId, | SWHID, | ||||
| ) | ) | ||||
| from swh.web.common.exc import BadInputExc | from swh.web.common.exc import BadInputExc | ||||
| from swh.web.common.identifiers import ( | from swh.web.common.identifiers import ( | ||||
| get_swh_persistent_id, | gen_swhid, | ||||
| resolve_swh_persistent_id, | resolve_swhid, | ||||
| get_persistent_identifier, | get_swhid, | ||||
| group_swh_persistent_identifiers, | group_swhids, | ||||
| get_swhids_info, | get_swhids_info, | ||||
| ) | ) | ||||
| from swh.web.browse.snapshot_context import get_snapshot_context | from swh.web.browse.snapshot_context import get_snapshot_context | ||||
| from swh.web.common.utils import reverse | from swh.web.common.utils import reverse | ||||
| from swh.web.common.typing import SWHObjectInfo | from swh.web.common.typing import SWHObjectInfo | ||||
| 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 ( | ||||
| content, | content, | ||||
| directory, | directory, | ||||
| release, | release, | ||||
| revision, | revision, | ||||
| snapshot, | snapshot, | ||||
| origin, | origin, | ||||
| origin_with_multiple_visits, | origin_with_multiple_visits, | ||||
| directory_with_subdirs, | directory_with_subdirs, | ||||
| ) | ) | ||||
| @given(content()) | @given(content()) | ||||
| def test_get_swh_persistent_id(content): | def test_gen_swhid(content): | ||||
| swh_object_type = CONTENT | swh_object_type = CONTENT | ||||
| sha1_git = content["sha1_git"] | sha1_git = content["sha1_git"] | ||||
| expected_swh_id = "swh:1:cnt:" + sha1_git | expected_swhid = "swh:1:cnt:" + sha1_git | ||||
| assert get_swh_persistent_id(swh_object_type, sha1_git) == expected_swh_id | assert gen_swhid(swh_object_type, sha1_git) == expected_swhid | ||||
| with pytest.raises(BadInputExc) as e: | with pytest.raises(BadInputExc) as e: | ||||
| get_swh_persistent_id("foo", sha1_git) | gen_swhid("foo", sha1_git) | ||||
| assert e.match("Invalid object") | assert e.match("Invalid object") | ||||
| with pytest.raises(BadInputExc) as e: | with pytest.raises(BadInputExc) as e: | ||||
| get_swh_persistent_id(swh_object_type, "not a valid id") | gen_swhid(swh_object_type, "not a valid id") | ||||
| assert e.match("Invalid object") | assert e.match("Invalid object") | ||||
| @given(content(), directory(), release(), revision(), snapshot()) | @given(content(), directory(), release(), revision(), snapshot()) | ||||
| def test_resolve_swh_persistent_id_legacy( | def test_resolve_swhid_legacy(content, directory, release, revision, snapshot): | ||||
| content, directory, release, revision, snapshot | |||||
| ): | |||||
| for obj_type, obj_id in ( | for obj_type, obj_id in ( | ||||
| (CONTENT, content["sha1_git"]), | (CONTENT, content["sha1_git"]), | ||||
| (DIRECTORY, directory), | (DIRECTORY, directory), | ||||
| (RELEASE, release), | (RELEASE, release), | ||||
| (REVISION, revision), | (REVISION, revision), | ||||
| (SNAPSHOT, snapshot), | (SNAPSHOT, snapshot), | ||||
| ): | ): | ||||
| swh_pid = get_swh_persistent_id(obj_type, obj_id) | swhid = gen_swhid(obj_type, obj_id) | ||||
| url_args = {} | url_args = {} | ||||
| if obj_type == CONTENT: | if obj_type == CONTENT: | ||||
| url_args["query_string"] = f"sha1_git:{obj_id}" | url_args["query_string"] = f"sha1_git:{obj_id}" | ||||
| elif obj_type == SNAPSHOT: | elif obj_type == SNAPSHOT: | ||||
| url_args["snapshot_id"] = obj_id | url_args["snapshot_id"] = obj_id | ||||
| else: | else: | ||||
| url_args["sha1_git"] = obj_id | url_args["sha1_git"] = obj_id | ||||
| query_params = {"origin_url": "some-origin"} | query_params = {"origin_url": "some-origin"} | ||||
| browse_url = reverse( | browse_url = reverse( | ||||
| f"browse-{obj_type}", url_args=url_args, query_params=query_params | f"browse-{obj_type}", url_args=url_args, query_params=query_params | ||||
| ) | ) | ||||
| resolved_pid = resolve_swh_persistent_id(swh_pid, query_params) | resolved_swhid = resolve_swhid(swhid, query_params) | ||||
| assert isinstance(resolved_pid["swh_id_parsed"], PersistentId) | assert isinstance(resolved_swhid["swhid_parsed"], SWHID) | ||||
| assert str(resolved_pid["swh_id_parsed"]) == swh_pid | assert str(resolved_swhid["swhid_parsed"]) == swhid | ||||
| assert resolved_pid["browse_url"] == browse_url | assert resolved_swhid["browse_url"] == browse_url | ||||
| with pytest.raises(BadInputExc, match="Origin PIDs"): | with pytest.raises(BadInputExc, match="Origin SWHIDs"): | ||||
| resolve_swh_persistent_id(f"swh:1:ori:{random_sha1()}") | resolve_swhid(f"swh:1:ori:{random_sha1()}") | ||||
| @given(content(), directory(), release(), revision(), snapshot()) | @given(content(), directory(), release(), revision(), snapshot()) | ||||
| def test_get_persistent_identifier(content, directory, release, revision, snapshot): | def test_get_swhid(content, directory, release, revision, snapshot): | ||||
| for obj_type, obj_id in ( | for obj_type, obj_id in ( | ||||
| (CONTENT, content["sha1_git"]), | (CONTENT, content["sha1_git"]), | ||||
| (DIRECTORY, directory), | (DIRECTORY, directory), | ||||
| (RELEASE, release), | (RELEASE, release), | ||||
| (REVISION, revision), | (REVISION, revision), | ||||
| (SNAPSHOT, snapshot), | (SNAPSHOT, snapshot), | ||||
| ): | ): | ||||
| swh_pid = get_swh_persistent_id(obj_type, obj_id) | swhid = gen_swhid(obj_type, obj_id) | ||||
| swh_parsed_pid = get_persistent_identifier(swh_pid) | swh_parsed_swhid = get_swhid(swhid) | ||||
| assert isinstance(swh_parsed_pid, PersistentId) | assert isinstance(swh_parsed_swhid, SWHID) | ||||
| assert str(swh_parsed_pid) == swh_pid | assert str(swh_parsed_swhid) == swhid | ||||
| with pytest.raises(BadInputExc, match="Error when parsing identifier"): | with pytest.raises(BadInputExc, match="Error when parsing identifier"): | ||||
| get_persistent_identifier("foo") | get_swhid("foo") | ||||
| @given(content(), directory(), release(), revision(), snapshot()) | @given(content(), directory(), release(), revision(), snapshot()) | ||||
| def test_group_persistent_identifiers(content, directory, release, revision, snapshot): | def test_group_swhids(content, directory, release, revision, snapshot): | ||||
| swh_pids = [] | swhids = [] | ||||
| expected = {} | expected = {} | ||||
| for obj_type, obj_id in ( | for obj_type, obj_id in ( | ||||
| (CONTENT, content["sha1_git"]), | (CONTENT, content["sha1_git"]), | ||||
| (DIRECTORY, directory), | (DIRECTORY, directory), | ||||
| (RELEASE, release), | (RELEASE, release), | ||||
| (REVISION, revision), | (REVISION, revision), | ||||
| (SNAPSHOT, snapshot), | (SNAPSHOT, snapshot), | ||||
| ): | ): | ||||
| swh_pid = get_swh_persistent_id(obj_type, obj_id) | swhid = gen_swhid(obj_type, obj_id) | ||||
| swh_pid = get_persistent_identifier(swh_pid) | swhid = get_swhid(swhid) | ||||
| swh_pids.append(swh_pid) | swhids.append(swhid) | ||||
| expected[obj_type] = [hash_to_bytes(obj_id)] | expected[obj_type] = [hash_to_bytes(obj_id)] | ||||
| pid_groups = group_swh_persistent_identifiers(swh_pids) | swhid_groups = group_swhids(swhids) | ||||
| assert pid_groups == expected | assert swhid_groups == expected | ||||
| @given(directory_with_subdirs()) | @given(directory_with_subdirs()) | ||||
| def test_get_swhids_info_directory_context(archive_data, directory): | def test_get_swhids_info_directory_context(archive_data, directory): | ||||
| extra_context = {"path": "/"} | extra_context = {"path": "/"} | ||||
| swhid = get_swhids_info( | swhid = get_swhids_info( | ||||
| [SWHObjectInfo(object_type=DIRECTORY, object_id=directory)], | [SWHObjectInfo(object_type=DIRECTORY, object_id=directory)], | ||||
| snapshot_context=None, | snapshot_context=None, | ||||
| extra_context=extra_context, | extra_context=extra_context, | ||||
| )[0] | )[0] | ||||
| swhid_dir_parsed = get_persistent_identifier(swhid["swhid_with_context"]) | swhid_dir_parsed = get_swhid(swhid["swhid_with_context"]) | ||||
| assert swhid_dir_parsed.metadata == extra_context | assert swhid_dir_parsed.metadata == extra_context | ||||
| dir_content = archive_data.directory_ls(directory) | dir_content = archive_data.directory_ls(directory) | ||||
| dir_subdirs = [e for e in dir_content if e["type"] == "dir"] | dir_subdirs = [e for e in dir_content if e["type"] == "dir"] | ||||
| dir_subdir = random.choice(dir_subdirs) | dir_subdir = random.choice(dir_subdirs) | ||||
| dir_subdir_path = f'/{dir_subdir["name"]}/' | dir_subdir_path = f'/{dir_subdir["name"]}/' | ||||
| Show All 17 Lines | if dir_subdir_files: | ||||
| object_type=CONTENT, object_id=dir_subdir_file["checksums"]["sha1_git"] | object_type=CONTENT, object_id=dir_subdir_file["checksums"]["sha1_git"] | ||||
| ) | ) | ||||
| ) | ) | ||||
| swhids = get_swhids_info( | swhids = get_swhids_info( | ||||
| swh_objects_info, snapshot_context=None, extra_context=extra_context, | swh_objects_info, snapshot_context=None, extra_context=extra_context, | ||||
| ) | ) | ||||
| swhid_dir_parsed = get_persistent_identifier(swhids[0]["swhid_with_context"]) | swhid_dir_parsed = get_swhid(swhids[0]["swhid_with_context"]) | ||||
| anchor = get_swh_persistent_id(DIRECTORY, directory) | anchor = gen_swhid(DIRECTORY, directory) | ||||
| assert swhid_dir_parsed.metadata == { | assert swhid_dir_parsed.metadata == { | ||||
| "anchor": anchor, | "anchor": anchor, | ||||
| "path": dir_subdir_path, | "path": dir_subdir_path, | ||||
| } | } | ||||
| if dir_subdir_files: | if dir_subdir_files: | ||||
| swhid_cnt_parsed = get_persistent_identifier(swhids[1]["swhid_with_context"]) | swhid_cnt_parsed = get_swhid(swhids[1]["swhid_with_context"]) | ||||
| assert swhid_cnt_parsed.metadata == { | assert swhid_cnt_parsed.metadata == { | ||||
| "anchor": anchor, | "anchor": anchor, | ||||
| "path": f'{dir_subdir_path}{dir_subdir_file["name"]}', | "path": f'{dir_subdir_path}{dir_subdir_file["name"]}', | ||||
| } | } | ||||
| @given(revision()) | @given(revision()) | ||||
| Show All 17 Lines | if dir_entry["type"] == "file": | ||||
| ) | ) | ||||
| extra_context["filename"] = dir_entry["name"] | extra_context["filename"] = dir_entry["name"] | ||||
| swhids = get_swhids_info( | swhids = get_swhids_info( | ||||
| swh_objects, snapshot_context=None, extra_context=extra_context, | swh_objects, snapshot_context=None, extra_context=extra_context, | ||||
| ) | ) | ||||
| assert swhids[0]["context"] == {} | assert swhids[0]["context"] == {} | ||||
| swhid_dir_parsed = get_persistent_identifier(swhids[1]["swhid_with_context"]) | swhid_dir_parsed = get_swhid(swhids[1]["swhid_with_context"]) | ||||
| anchor = get_swh_persistent_id(REVISION, revision) | anchor = gen_swhid(REVISION, revision) | ||||
| assert swhid_dir_parsed.metadata == { | assert swhid_dir_parsed.metadata == { | ||||
| "anchor": anchor, | "anchor": anchor, | ||||
| "path": "/", | "path": "/", | ||||
| } | } | ||||
| if dir_entry["type"] == "file": | if dir_entry["type"] == "file": | ||||
| swhid_cnt_parsed = get_persistent_identifier(swhids[2]["swhid_with_context"]) | swhid_cnt_parsed = get_swhid(swhids[2]["swhid_with_context"]) | ||||
| assert swhid_cnt_parsed.metadata == { | assert swhid_cnt_parsed.metadata == { | ||||
| "anchor": anchor, | "anchor": anchor, | ||||
| "path": f'/{dir_entry["name"]}', | "path": f'/{dir_entry["name"]}', | ||||
| } | } | ||||
| @given(origin_with_multiple_visits()) | @given(origin_with_multiple_visits()) | ||||
| def test_get_swhids_info_origin_snapshot_context(archive_data, origin): | def test_get_swhids_info_origin_snapshot_context(archive_data, origin): | ||||
| ▲ Show 20 Lines • Show All 103 Lines • ▼ Show 20 Lines | for visit in visits: | ||||
| ) | ) | ||||
| swhids = get_swhids_info( | swhids = get_swhids_info( | ||||
| swh_objects, | swh_objects, | ||||
| snapshot_context, | snapshot_context, | ||||
| extra_context={"path": "/", "filename": dir_file["name"]}, | extra_context={"path": "/", "filename": dir_file["name"]}, | ||||
| ) | ) | ||||
| swhid_cnt_parsed = get_persistent_identifier( | swhid_cnt_parsed = get_swhid(swhids[0]["swhid_with_context"]) | ||||
| swhids[0]["swhid_with_context"] | swhid_dir_parsed = get_swhid(swhids[1]["swhid_with_context"]) | ||||
| ) | swhid_rev_parsed = get_swhid(swhids[2]["swhid_with_context"]) | ||||
| swhid_dir_parsed = get_persistent_identifier( | |||||
| swhids[1]["swhid_with_context"] | |||||
| ) | |||||
| swhid_rev_parsed = get_persistent_identifier( | |||||
| swhids[2]["swhid_with_context"] | |||||
| ) | |||||
| swhid_snp_parsed = get_persistent_identifier( | swhid_snp_parsed = get_swhid( | ||||
| swhids[3]["swhid_with_context"] or swhids[3]["swhid"] | swhids[3]["swhid_with_context"] or swhids[3]["swhid"] | ||||
| ) | ) | ||||
| swhid_rel_parsed = None | swhid_rel_parsed = None | ||||
| if "release_name" in snp_ctx_params: | if "release_name" in snp_ctx_params: | ||||
| swhid_rel_parsed = get_persistent_identifier( | swhid_rel_parsed = get_swhid(swhids[4]["swhid_with_context"]) | ||||
| swhids[4]["swhid_with_context"] | |||||
| ) | |||||
| anchor = get_swh_persistent_id( | anchor = gen_swhid( | ||||
| object_type=anchor_info["anchor_type"], | object_type=anchor_info["anchor_type"], | ||||
| object_id=anchor_info["anchor_id"], | object_id=anchor_info["anchor_id"], | ||||
| ) | ) | ||||
| snapshot_swhid = get_swh_persistent_id( | snapshot_swhid = gen_swhid(object_type=SNAPSHOT, object_id=snapshot_id) | ||||
| object_type=SNAPSHOT, object_id=snapshot_id | |||||
| ) | |||||
| expected_cnt_context = { | expected_cnt_context = { | ||||
| "visit": snapshot_swhid, | "visit": snapshot_swhid, | ||||
| "anchor": anchor, | "anchor": anchor, | ||||
| "path": f'/{dir_file["name"]}', | "path": f'/{dir_file["name"]}', | ||||
| } | } | ||||
| expected_dir_context = { | expected_dir_context = { | ||||
| ▲ Show 20 Lines • Show All 134 Lines • ▼ Show 20 Lines | ): | ||||
| obj_context = {} | obj_context = {} | ||||
| query_params = {} | query_params = {} | ||||
| if origin_url: | if origin_url: | ||||
| obj_context["origin"] = origin_url | obj_context["origin"] = origin_url | ||||
| query_params["origin_url"] = origin_url | query_params["origin_url"] = origin_url | ||||
| obj_context["visit"] = get_swh_persistent_id(SNAPSHOT, snapshot_id) | obj_context["visit"] = gen_swhid(SNAPSHOT, snapshot_id) | ||||
| query_params["snapshot"] = snapshot_id | query_params["snapshot"] = snapshot_id | ||||
| if object_type in (CONTENT, DIRECTORY, REVISION): | if object_type in (CONTENT, DIRECTORY, REVISION): | ||||
| if snapshot_context["release"]: | if snapshot_context["release"]: | ||||
| obj_context["anchor"] = get_swh_persistent_id( | obj_context["anchor"] = gen_swhid(RELEASE, snapshot_context["release_id"]) | ||||
| RELEASE, snapshot_context["release_id"] | |||||
| ) | |||||
| query_params["release"] = snapshot_context["release"] | query_params["release"] = snapshot_context["release"] | ||||
| else: | else: | ||||
| obj_context["anchor"] = get_swh_persistent_id( | obj_context["anchor"] = gen_swhid(REVISION, snapshot_context["revision_id"]) | ||||
| REVISION, snapshot_context["revision_id"] | |||||
| ) | |||||
| if ( | if ( | ||||
| snapshot_context["branch"] | snapshot_context["branch"] | ||||
| and snapshot_context["branch"] != snapshot_context["revision_id"] | and snapshot_context["branch"] != snapshot_context["revision_id"] | ||||
| ): | ): | ||||
| branch = snapshot_context["branch"] | branch = snapshot_context["branch"] | ||||
| if branch == "HEAD": | if branch == "HEAD": | ||||
| for b in snapshot_context["branches"]: | for b in snapshot_context["branches"]: | ||||
| if ( | if ( | ||||
| Show All 16 Lines | if path: | ||||
| query_params["path"] = path[1:-1] | query_params["path"] = path[1:-1] | ||||
| if object_type == DIRECTORY: | if object_type == DIRECTORY: | ||||
| object_id = snapshot_context["root_directory"] | object_id = snapshot_context["root_directory"] | ||||
| if lines: | if lines: | ||||
| obj_context["lines"] = lines | obj_context["lines"] = lines | ||||
| obj_swhid = get_swh_persistent_id(object_type, object_id, metadata=obj_context) | obj_swhid = gen_swhid(object_type, object_id, metadata=obj_context) | ||||
| obj_swhid_resolved = resolve_swh_persistent_id(obj_swhid) | obj_swhid_resolved = resolve_swhid(obj_swhid) | ||||
| url_args = {"sha1_git": object_id} | url_args = {"sha1_git": object_id} | ||||
| if object_type == CONTENT: | if object_type == CONTENT: | ||||
| url_args = {"query_string": f"sha1_git:{object_id}"} | url_args = {"query_string": f"sha1_git:{object_id}"} | ||||
| elif object_type == SNAPSHOT: | elif object_type == SNAPSHOT: | ||||
| url_args = {"snapshot_id": object_id} | url_args = {"snapshot_id": object_id} | ||||
| expected_url = reverse( | expected_url = reverse( | ||||
| Show All 9 Lines | |||||