Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/storage_tests.py
Show First 20 Lines • Show All 1,689 Lines • ▼ Show 20 Lines | def test_extid_version_behavior_failure(self, swh_storage, sample_data): | ||||||||||
with pytest.raises( | with pytest.raises( | ||||||||||
(ValueError, RemoteException), match="both extid_type and extid_version" | (ValueError, RemoteException), match="both extid_type and extid_version" | ||||||||||
): | ): | ||||||||||
swh_storage.extid_get_from_target( | swh_storage.extid_get_from_target( | ||||||||||
ObjectType.REVISION, [ids[0]], extid_type="git" | ObjectType.REVISION, [ids[0]], extid_type="git" | ||||||||||
) | ) | ||||||||||
def test_extid_payload(self, swh_storage, sample_data): | |||||||||||
target = sample_data.directory2.swhid() | |||||||||||
extids = [ | |||||||||||
ExtID( | |||||||||||
extid=b"abc123", | |||||||||||
extid_type="test", | |||||||||||
target=target, | |||||||||||
payload_type="test_payload", | |||||||||||
vlorentzUnsubmitted Not Done Inline Actions
vlorentz: | |||||||||||
payload=sample_data.content.sha1_git, | |||||||||||
), | |||||||||||
] | |||||||||||
assert swh_storage.extid_get_from_extid("test", [b"abc123"]) == [] | |||||||||||
assert ( | |||||||||||
swh_storage.extid_get_from_target(target.object_type, [target.object_id]) | |||||||||||
== [] | |||||||||||
) | |||||||||||
summary = swh_storage.extid_add(extids) | |||||||||||
assert summary == {"extid:add": 1} | |||||||||||
assert swh_storage.extid_get_from_extid("test", [b"abc123"]) == extids | |||||||||||
assert ( | |||||||||||
swh_storage.extid_get_from_target(target.object_type, [target.object_id]) | |||||||||||
== extids | |||||||||||
) | |||||||||||
# 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_release_add(self, swh_storage, sample_data): | def test_release_add(self, swh_storage, sample_data): | ||||||||||
release, release2 = sample_data.releases[:2] | release, release2 = sample_data.releases[:2] | ||||||||||
init_missing = swh_storage.release_missing([release.id, release2.id]) | init_missing = swh_storage.release_missing([release.id, release2.id]) | ||||||||||
assert list(init_missing) == [release.id, release2.id] | assert list(init_missing) == [release.id, release2.id] | ||||||||||
actual_result = swh_storage.release_add([release, release2]) | actual_result = swh_storage.release_add([release, release2]) | ||||||||||
assert actual_result == {"release:add": 2} | assert actual_result == {"release:add": 2} | ||||||||||
▲ Show 20 Lines • Show All 3,972 Lines • Show Last 20 Lines |