Page MenuHomeSoftware Heritage
Differential D3585 Diff 12634 swh/storage/tests/test_storage.py

Changeset View

Standalone View

View Options

swh/storage/tests/test_storage.py

Show All 22 Lines
from typing import ClassVar, Optional from typing import ClassVar, Optional
from swh.model import from_disk from swh.model import from_disk
from swh.model.hashutil import hash_to_bytes from swh.model.hashutil import hash_to_bytes
from swh.model.identifiers import SWHID from swh.model.identifiers import SWHID
from swh.model.model import ( from swh.model.model import (
Content, Content,
Directory,
MetadataTargetType, MetadataTargetType,
Origin, Origin,
OriginVisit, OriginVisit,
OriginVisitStatus, OriginVisitStatus,
Person, Person,
Release, Release,
Revision, Revision,
Snapshot, Snapshot,
▲ Show 20 Lines • Show All 623 Lines • ▼ Show 20 Lines def test_directory_add(self, swh_storage, sample_data_model):
init_missing = list(swh_storage.directory_missing([directory.id])) init_missing = list(swh_storage.directory_missing([directory.id]))
assert [directory.id] == init_missing assert [directory.id] == init_missing
actual_result = swh_storage.directory_add([directory]) actual_result = swh_storage.directory_add([directory])
assert actual_result == {"directory:add": 1} assert actual_result == {"directory:add": 1}
assert list(swh_storage.journal_writer.journal.objects) == [ assert list(swh_storage.journal_writer.journal.objects) == [
("directory", Directory.from_dict(data.dir)) ("directory", directory)
] ]
actual_data = list(swh_storage.directory_ls(directory.id)) actual_data = list(swh_storage.directory_ls(directory.id))
expected_data = list(transform_entries(directory)) expected_data = list(transform_entries(directory))
assert sorted(expected_data, key=cmpdir) == sorted(actual_data, key=cmpdir) assert sorted(expected_data, key=cmpdir) == sorted(actual_data, key=cmpdir)
after_missing = list(swh_storage.directory_missing([directory.id])) after_missing = list(swh_storage.directory_missing([directory.id]))
▲ Show 20 Lines • Show All 98 Lines • ▼ Show 20 Lines def test_directory_get_non_recursive(self, swh_storage, sample_data_model):
# List directory containing a known subdirectory, entries should # List directory containing a known subdirectory, entries should
# only be those of the parent directory, not of the subdir # only be those of the parent directory, not of the subdir
actual_data = list(swh_storage.directory_ls(dir3.id)) actual_data = list(swh_storage.directory_ls(dir3.id))
expected_data = list(transform_entries(dir3)) expected_data = list(transform_entries(dir3))
assert sorted(expected_data, key=cmpdir) == sorted(actual_data, key=cmpdir) assert sorted(expected_data, key=cmpdir) == sorted(actual_data, key=cmpdir)
def test_directory_entry_get_by_path(self, swh_storage, sample_data_model): def test_directory_entry_get_by_path(self, swh_storage, sample_data_model):
cont = sample_data_model["content"][0] cont = sample_data_model["content"][0]
dir1, dir2, dir3, dir4 = sample_data_model["directory"][:4] dir1, dir2, dir3, dir4, dir5 = sample_data_model["directory"][:5]
# given # given
init_missing = list(swh_storage.directory_missing([dir3.id])) init_missing = list(swh_storage.directory_missing([dir3.id]))
assert init_missing == [dir3.id] assert init_missing == [dir3.id]
actual_result = swh_storage.directory_add([dir3, dir4]) actual_result = swh_storage.directory_add([dir3, dir4])
assert actual_result == {"directory:add": 2} assert actual_result == {"directory:add": 2}
Show All 21 Lines def test_directory_entry_get_by_path(self, swh_storage, sample_data_model):
"status": None, "status": None,
"perms": from_disk.DentryPerms.directory, "perms": from_disk.DentryPerms.directory,
"length": None, "length": None,
}, },
{ {
"dir_id": dir3.id, "dir_id": dir3.id,
"name": b"hello", "name": b"hello",
"type": "file", "type": "file",
"target": b"12345678901234567890", "target": dir5.id,
"sha1": None, "sha1": None,
"sha1_git": None, "sha1_git": None,
"sha256": None, "sha256": None,
"status": None, "status": None,
"perms": from_disk.DentryPerms.content, "perms": from_disk.DentryPerms.content,
"length": None, "length": None,
}, },
] ]
Show All 9 Lines def test_directory_entry_get_by_path(self, swh_storage, sample_data_model):
for entry, expected_entry in zip(dir3.entries, expected_entries): for entry, expected_entry in zip(dir3.entries, expected_entries):
actual_entry = swh_storage.directory_entry_get_by_path( actual_entry = swh_storage.directory_entry_get_by_path(
dir4.id, [b"subdir1", entry.name] dir4.id, [b"subdir1", entry.name]
) )
expected_entry = expected_entry.copy() expected_entry = expected_entry.copy()
expected_entry["name"] = b"subdir1/" + expected_entry["name"] expected_entry["name"] = b"subdir1/" + expected_entry["name"]
assert actual_entry == expected_entry assert actual_entry == expected_entry
# when (nothing should be found here since data.dir is not persisted.) # when (nothing should be found here since `dir` is not persisted.)
for entry in dir2.entries: for entry in dir2.entries:
actual_entry = swh_storage.directory_entry_get_by_path( actual_entry = swh_storage.directory_entry_get_by_path(
dir2.id, [entry.name] dir2.id, [entry.name]
) )
assert actual_entry is None assert actual_entry is None
def test_directory_get_random(self, swh_storage, sample_data_model): def test_directory_get_random(self, swh_storage, sample_data_model):
dir1, dir2, dir3 = sample_data_model["directory"][:3] dir1, dir2, dir3 = sample_data_model["directory"][:3]
▲ Show 20 Lines • Show All 3,286 Lines • Show Last 20 Lines
About · Developer information · Legal