Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/storage_tests.py
Show First 20 Lines • Show All 1,105 Lines • ▼ Show 20 Lines | def test_extid_add_git(self, swh_storage, sample_data): | ||||
assert summary == {"extid:add": len(gitids)} | assert summary == {"extid:add": len(gitids)} | ||||
assert swh_storage.extid_get_from_extid("git", gitids) == extids | assert swh_storage.extid_get_from_extid("git", gitids) == extids | ||||
assert swh_storage.extid_get_from_target(ObjectType.REVISION, gitids) == extids | assert swh_storage.extid_get_from_target(ObjectType.REVISION, gitids) == extids | ||||
assert swh_storage.extid_get_from_extid("hg", gitids) == [] | assert swh_storage.extid_get_from_extid("hg", gitids) == [] | ||||
assert swh_storage.extid_get_from_target(ObjectType.RELEASE, gitids) == [] | assert swh_storage.extid_get_from_target(ObjectType.RELEASE, gitids) == [] | ||||
# check ExtIDs have been added to the journal | |||||
extids_in_journal = [ | |||||
obj | |||||
for (obj_type, obj) in swh_storage.journal_writer.journal.objects | |||||
if obj_type == "extid" | |||||
] | |||||
assert extids == extids_in_journal | |||||
def test_extid_add_hg(self, swh_storage, sample_data): | def test_extid_add_hg(self, swh_storage, sample_data): | ||||
def get_node(revision): | def get_node(revision): | ||||
node = None | node = None | ||||
if revision.extra_headers: | if revision.extra_headers: | ||||
node = dict(revision.extra_headers).get(b"node") | node = dict(revision.extra_headers).get(b"node") | ||||
if node is None and revision.metadata: | if node is None and revision.metadata: | ||||
node = hash_to_bytes(revision.metadata.get("node")) | node = hash_to_bytes(revision.metadata.get("node")) | ||||
return node | return node | ||||
Show All 26 Lines | def test_extid_add_hg(self, swh_storage, sample_data): | ||||
assert swh_storage.extid_get_from_extid("hg", extids) == extid_objs | assert swh_storage.extid_get_from_extid("hg", extids) == extid_objs | ||||
assert ( | assert ( | ||||
swh_storage.extid_get_from_target(ObjectType.REVISION, swhids) == extid_objs | swh_storage.extid_get_from_target(ObjectType.REVISION, swhids) == extid_objs | ||||
) | ) | ||||
assert swh_storage.extid_get_from_extid("git", extids) == [] | assert swh_storage.extid_get_from_extid("git", extids) == [] | ||||
assert swh_storage.extid_get_from_target(ObjectType.RELEASE, swhids) == [] | assert swh_storage.extid_get_from_target(ObjectType.RELEASE, swhids) == [] | ||||
# check ExtIDs have been added to the journal | |||||
extids_in_journal = [ | |||||
obj | |||||
for (obj_type, obj) in swh_storage.journal_writer.journal.objects | |||||
if obj_type == "extid" | |||||
] | |||||
assert extid_objs == extids_in_journal | |||||
def test_extid_add_twice(self, swh_storage, sample_data): | def test_extid_add_twice(self, swh_storage, sample_data): | ||||
gitids = [ | gitids = [ | ||||
revision.id | revision.id | ||||
for revision in sample_data.revisions | for revision in sample_data.revisions | ||||
if revision.type.value == "git" | if revision.type.value == "git" | ||||
] | ] | ||||
▲ Show 20 Lines • Show All 3,184 Lines • Show Last 20 Lines |