Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/storage_tests.py
Show All 24 Lines | from swh.model.model import ( | ||||
Content, | Content, | ||||
Directory, | Directory, | ||||
MetadataTargetType, | MetadataTargetType, | ||||
Origin, | Origin, | ||||
OriginVisit, | OriginVisit, | ||||
OriginVisitStatus, | OriginVisitStatus, | ||||
Person, | Person, | ||||
Revision, | Revision, | ||||
SkippedContent, | |||||
Snapshot, | Snapshot, | ||||
TargetType, | 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.common import origin_url_to_sha1 as sha1 | from swh.storage.common 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 | ||||
▲ Show 20 Lines • Show All 720 Lines • ▼ Show 20 Lines | def test_directory_ls_missing_content(self, swh_storage, sample_data): | ||||
"sha1_git": None, | "sha1_git": None, | ||||
"sha256": None, | "sha256": None, | ||||
"status": None, | "status": None, | ||||
"target": sample_data.directory2.entries[0].target, | "target": sample_data.directory2.entries[0].target, | ||||
"type": "file", | "type": "file", | ||||
}, | }, | ||||
] | ] | ||||
def test_directory_ls_skipped_content(self, swh_storage, sample_data): | |||||
swh_storage.directory_add([sample_data.directory2]) | |||||
cont = SkippedContent( | |||||
sha1_git=sample_data.directory2.entries[0].target, | |||||
sha1=b"c" * 20, | |||||
sha256=None, | |||||
blake2s256=None, | |||||
length=42, | |||||
status="absent", | |||||
reason="You need a premium subscription to access this content", | |||||
ardumont: lol | |||||
) | |||||
swh_storage.skipped_content_add([cont]) | |||||
ardumontUnsubmitted Not Done Inline Actionsthere may exist skipped-content already in the sample-data fixture. ardumont: there may exist skipped-content already in the sample-data fixture. | |||||
assert list(swh_storage.directory_ls(sample_data.directory2.id)) == [ | |||||
{ | |||||
"dir_id": sample_data.directory2.id, | |||||
"length": 42, | |||||
"name": b"oof", | |||||
"perms": 33188, | |||||
"sha1": b"c" * 20, | |||||
"sha1_git": sample_data.directory2.entries[0].target, | |||||
"sha256": None, | |||||
"status": "absent", | |||||
"target": sample_data.directory2.entries[0].target, | |||||
"type": "file", | |||||
}, | |||||
] | |||||
def test_directory_entry_get_by_path(self, swh_storage, sample_data): | def test_directory_entry_get_by_path(self, swh_storage, sample_data): | ||||
cont, content2 = sample_data.contents[:2] | cont, content2 = sample_data.contents[:2] | ||||
dir1, dir2, dir3, dir4, dir5 = sample_data.directories[:5] | dir1, dir2, dir3, dir4, dir5 = sample_data.directories[:5] | ||||
# given | # given | ||||
dir_ids = [d.id for d in [dir1, dir2, dir3, dir4, dir5]] | dir_ids = [d.id for d in [dir1, dir2, dir3, dir4, dir5]] | ||||
init_missing = list(swh_storage.directory_missing(dir_ids)) | init_missing = list(swh_storage.directory_missing(dir_ids)) | ||||
assert init_missing == dir_ids | assert init_missing == dir_ids | ||||
▲ Show 20 Lines • Show All 3,111 Lines • Show Last 20 Lines |
lol