Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show All 31 Lines | from swh.model.model import ( | ||||
| MetadataTargetType, | MetadataTargetType, | ||||
| Origin, | Origin, | ||||
| OriginVisit, | OriginVisit, | ||||
| OriginVisitStatus, | OriginVisitStatus, | ||||
| Person, | Person, | ||||
| Release, | Release, | ||||
| Revision, | Revision, | ||||
| Snapshot, | Snapshot, | ||||
| TargetType, | |||||
| ) | ) | ||||
| from swh.model.hypothesis_strategies import objects | from swh.model.hypothesis_strategies import objects | ||||
| from swh.storage import get_storage | from swh.storage import get_storage | ||||
| from swh.storage.converters import origin_url_to_sha1 as sha1 | from swh.storage.converters import origin_url_to_sha1 as sha1 | ||||
| from swh.storage.exc import HashCollision, StorageArgumentException | from swh.storage.exc import HashCollision, StorageArgumentException | ||||
| from swh.storage.interface import ListOrder, PagedResult, StorageInterface | from swh.storage.interface import ListOrder, PagedResult, StorageInterface | ||||
| from swh.storage.utils import content_hex_hashes, now | from swh.storage.utils import content_hex_hashes, now | ||||
| ▲ Show 20 Lines • Show All 2,565 Lines • ▼ Show 20 Lines | def test_snapshot_add_count_branches(self, swh_storage, sample_data): | ||||
| assert snp_size == expected_snp_size | assert snp_size == expected_snp_size | ||||
| def test_snapshot_add_get_paginated(self, swh_storage, sample_data): | def test_snapshot_add_get_paginated(self, swh_storage, sample_data): | ||||
| complete_snapshot = sample_data.snapshots[2] | complete_snapshot = sample_data.snapshots[2] | ||||
| swh_storage.snapshot_add([complete_snapshot]) | swh_storage.snapshot_add([complete_snapshot]) | ||||
| snp_id = complete_snapshot.id | snp_id = complete_snapshot.id | ||||
| branches = complete_snapshot.to_dict()["branches"] | branches = complete_snapshot.branches | ||||
| branch_names = list(sorted(branches)) | branch_names = list(sorted(branches)) | ||||
| # Test branch_from | # Test branch_from | ||||
| snapshot = swh_storage.snapshot_get_branches(snp_id, branches_from=b"release") | snapshot = swh_storage.snapshot_get_branches(snp_id, branches_from=b"release") | ||||
| rel_idx = branch_names.index(b"release") | rel_idx = branch_names.index(b"release") | ||||
| expected_snapshot = { | expected_snapshot = { | ||||
| "id": snp_id, | "id": snp_id, | ||||
| ▲ Show 20 Lines • Show All 51 Lines • ▼ Show 20 Lines | def test_snapshot_add_get_filtered(self, swh_storage, sample_data): | ||||
| date=now(), | date=now(), | ||||
| status="ongoing", | status="ongoing", | ||||
| snapshot=complete_snapshot.id, | snapshot=complete_snapshot.id, | ||||
| ) | ) | ||||
| ] | ] | ||||
| ) | ) | ||||
| snp_id = complete_snapshot.id | snp_id = complete_snapshot.id | ||||
| branches = complete_snapshot.to_dict()["branches"] | branches = complete_snapshot.branches | ||||
| snapshot = swh_storage.snapshot_get_branches( | snapshot = swh_storage.snapshot_get_branches( | ||||
| snp_id, target_types=["release", "revision"] | snp_id, target_types=["release", "revision"] | ||||
| ) | ) | ||||
| expected_snapshot = { | expected_snapshot = { | ||||
| "id": snp_id, | "id": snp_id, | ||||
| "branches": { | "branches": { | ||||
| name: tgt | name: tgt | ||||
| for name, tgt in branches.items() | for name, tgt in branches.items() | ||||
| if tgt and tgt["target_type"] in ["release", "revision"] | if tgt and tgt.target_type in [TargetType.RELEASE, TargetType.REVISION] | ||||
| }, | }, | ||||
| "next_branch": None, | "next_branch": None, | ||||
| } | } | ||||
| assert snapshot == expected_snapshot | assert snapshot == expected_snapshot | ||||
| snapshot = swh_storage.snapshot_get_branches(snp_id, target_types=["alias"]) | snapshot = swh_storage.snapshot_get_branches(snp_id, target_types=["alias"]) | ||||
| expected_snapshot = { | expected_snapshot = { | ||||
| "id": snp_id, | "id": snp_id, | ||||
| "branches": { | "branches": { | ||||
| name: tgt | name: tgt | ||||
| for name, tgt in branches.items() | for name, tgt in branches.items() | ||||
| if tgt and tgt["target_type"] == "alias" | if tgt and tgt.target_type == TargetType.ALIAS | ||||
| }, | }, | ||||
| "next_branch": None, | "next_branch": None, | ||||
| } | } | ||||
| assert snapshot == expected_snapshot | assert snapshot == expected_snapshot | ||||
| def test_snapshot_add_get_filtered_and_paginated(self, swh_storage, sample_data): | def test_snapshot_add_get_filtered_and_paginated(self, swh_storage, sample_data): | ||||
| complete_snapshot = sample_data.snapshots[2] | complete_snapshot = sample_data.snapshots[2] | ||||
| swh_storage.snapshot_add([complete_snapshot]) | swh_storage.snapshot_add([complete_snapshot]) | ||||
| snp_id = complete_snapshot.id | snp_id = complete_snapshot.id | ||||
| branches = complete_snapshot.to_dict()["branches"] | branches = complete_snapshot.branches | ||||
| branch_names = list(sorted(branches)) | branch_names = list(sorted(branches)) | ||||
| # Test branch_from | # Test branch_from | ||||
| snapshot = swh_storage.snapshot_get_branches( | snapshot = swh_storage.snapshot_get_branches( | ||||
| snp_id, target_types=["directory", "release"], branches_from=b"directory2" | snp_id, target_types=["directory", "release"], branches_from=b"directory2" | ||||
| ) | ) | ||||
| ▲ Show 20 Lines • Show All 1,454 Lines • Show Last 20 Lines | |||||