Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
Show First 20 Lines • Show All 181 Lines • ▼ Show 20 Lines | def test_content_add(self, swh_storage, sample_data): | ||||
for obj in contents: | for obj in contents: | ||||
assert insertion_start_time <= obj.ctime | assert insertion_start_time <= obj.ctime | ||||
assert obj.ctime <= insertion_end_time | assert obj.ctime <= insertion_end_time | ||||
assert obj == expected_cont | assert obj == expected_cont | ||||
swh_storage.refresh_stat_counters() | swh_storage.refresh_stat_counters() | ||||
assert swh_storage.stat_counters()["content"] == 1 | assert swh_storage.stat_counters()["content"] == 1 | ||||
def test_content_add_from_generator(self, swh_storage, sample_data): | |||||
cont = sample_data.content | |||||
def _cnt_gen(): | |||||
yield cont | |||||
actual_result = swh_storage.content_add(_cnt_gen()) | |||||
assert actual_result == { | |||||
"content:add": 1, | |||||
"content:add:bytes": cont.length, | |||||
} | |||||
swh_storage.refresh_stat_counters() | |||||
assert swh_storage.stat_counters()["content"] == 1 | |||||
def test_content_add_from_lazy_content(self, swh_storage, sample_data): | def test_content_add_from_lazy_content(self, swh_storage, sample_data): | ||||
cont = sample_data.content | cont = sample_data.content | ||||
lazy_content = LazyContent.from_dict(cont.to_dict()) | lazy_content = LazyContent.from_dict(cont.to_dict()) | ||||
insertion_start_time = now() | insertion_start_time = now() | ||||
actual_result = swh_storage.content_add([lazy_content]) | actual_result = swh_storage.content_add([lazy_content]) | ||||
▲ Show 20 Lines • Show All 458 Lines • ▼ Show 20 Lines | def test_directory_add(self, swh_storage, sample_data): | ||||
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])) | ||||
assert after_missing == [] | assert after_missing == [] | ||||
swh_storage.refresh_stat_counters() | swh_storage.refresh_stat_counters() | ||||
assert swh_storage.stat_counters()["directory"] == 1 | assert swh_storage.stat_counters()["directory"] == 1 | ||||
def test_directory_add_from_generator(self, swh_storage, sample_data): | |||||
directory = sample_data.directories[1] | |||||
def _dir_gen(): | |||||
yield directory | |||||
actual_result = swh_storage.directory_add(directories=_dir_gen()) | |||||
assert actual_result == {"directory:add": 1} | |||||
assert list(swh_storage.journal_writer.journal.objects) == [ | |||||
("directory", directory) | |||||
] | |||||
swh_storage.refresh_stat_counters() | |||||
assert swh_storage.stat_counters()["directory"] == 1 | |||||
def test_directory_add_twice(self, swh_storage, sample_data): | def test_directory_add_twice(self, swh_storage, sample_data): | ||||
directory = sample_data.directories[1] | directory = sample_data.directories[1] | ||||
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) | ("directory", directory) | ||||
▲ Show 20 Lines • Show All 172 Lines • ▼ Show 20 Lines | def test_revision_add(self, swh_storage, sample_data): | ||||
# already there so nothing added | # already there so nothing added | ||||
actual_result = swh_storage.revision_add([revision]) | actual_result = swh_storage.revision_add([revision]) | ||||
assert actual_result == {"revision:add": 0} | assert actual_result == {"revision:add": 0} | ||||
swh_storage.refresh_stat_counters() | swh_storage.refresh_stat_counters() | ||||
assert swh_storage.stat_counters()["revision"] == 1 | assert swh_storage.stat_counters()["revision"] == 1 | ||||
def test_revision_add_from_generator(self, swh_storage, sample_data): | |||||
revision = sample_data.revision | |||||
def _rev_gen(): | |||||
yield revision | |||||
actual_result = swh_storage.revision_add(_rev_gen()) | |||||
assert actual_result == {"revision:add": 1} | |||||
swh_storage.refresh_stat_counters() | |||||
assert swh_storage.stat_counters()["revision"] == 1 | |||||
def test_revision_add_twice(self, swh_storage, sample_data): | def test_revision_add_twice(self, swh_storage, sample_data): | ||||
revision, revision2 = sample_data.revisions[:2] | revision, revision2 = sample_data.revisions[:2] | ||||
actual_result = swh_storage.revision_add([revision]) | actual_result = swh_storage.revision_add([revision]) | ||||
assert actual_result == {"revision:add": 1} | assert actual_result == {"revision:add": 1} | ||||
assert list(swh_storage.journal_writer.journal.objects) == [ | assert list(swh_storage.journal_writer.journal.objects) == [ | ||||
("revision", revision) | ("revision", revision) | ||||
▲ Show 20 Lines • Show All 153 Lines • ▼ Show 20 Lines | def test_release_add(self, swh_storage, sample_data): | ||||
# already present so nothing added | # already present so nothing added | ||||
actual_result = swh_storage.release_add([release, release2]) | actual_result = swh_storage.release_add([release, release2]) | ||||
assert actual_result == {"release:add": 0} | assert actual_result == {"release:add": 0} | ||||
swh_storage.refresh_stat_counters() | swh_storage.refresh_stat_counters() | ||||
assert swh_storage.stat_counters()["release"] == 2 | assert swh_storage.stat_counters()["release"] == 2 | ||||
def test_release_add_from_generator(self, swh_storage, sample_data): | |||||
release, release2 = sample_data.releases[:2] | |||||
def _rel_gen(): | |||||
yield release | |||||
yield release2 | |||||
actual_result = swh_storage.release_add(_rel_gen()) | |||||
assert actual_result == {"release:add": 2} | |||||
assert list(swh_storage.journal_writer.journal.objects) == [ | |||||
("release", release), | |||||
("release", release2), | |||||
] | |||||
swh_storage.refresh_stat_counters() | |||||
assert swh_storage.stat_counters()["release"] == 2 | |||||
def test_release_add_no_author_date(self, swh_storage, sample_data): | def test_release_add_no_author_date(self, swh_storage, sample_data): | ||||
full_release = sample_data.release | full_release = sample_data.release | ||||
release = attr.evolve(full_release, author=None, date=None) | release = attr.evolve(full_release, author=None, date=None) | ||||
actual_result = swh_storage.release_add([release]) | actual_result = swh_storage.release_add([release]) | ||||
assert actual_result == {"release:add": 1} | assert actual_result == {"release:add": 1} | ||||
end_missing = swh_storage.release_missing([release.id]) | end_missing = swh_storage.release_missing([release.id]) | ||||
▲ Show 20 Lines • Show All 91 Lines • ▼ Show 20 Lines | def test_origin_add(self, swh_storage, sample_data): | ||||
assert set(swh_storage.journal_writer.journal.objects) == set( | assert set(swh_storage.journal_writer.journal.objects) == set( | ||||
[("origin", origins[0]), ("origin", origins[1]),] | [("origin", origins[0]), ("origin", origins[1]),] | ||||
) | ) | ||||
swh_storage.refresh_stat_counters() | swh_storage.refresh_stat_counters() | ||||
assert swh_storage.stat_counters()["origin"] == 2 | assert swh_storage.stat_counters()["origin"] == 2 | ||||
def test_origin_add_from_generator(self, swh_storage, sample_data): | |||||
origin, origin2 = sample_data.origins[:2] | |||||
def _ori_gen(): | |||||
yield origin | |||||
yield origin2 | |||||
stats = swh_storage.origin_add(_ori_gen()) | |||||
assert stats == {"origin:add": 2} | |||||
actual_origins = swh_storage.origin_get([origin.url, origin2.url]) | |||||
assert actual_origins == [origin, origin2] | |||||
assert set(swh_storage.journal_writer.journal.objects) == set( | |||||
[("origin", origin), ("origin", origin2),] | |||||
) | |||||
swh_storage.refresh_stat_counters() | |||||
assert swh_storage.stat_counters()["origin"] == 2 | |||||
def test_origin_add_twice(self, swh_storage, sample_data): | def test_origin_add_twice(self, swh_storage, sample_data): | ||||
origin, origin2 = sample_data.origins[:2] | origin, origin2 = sample_data.origins[:2] | ||||
add1 = swh_storage.origin_add([origin, origin2]) | add1 = swh_storage.origin_add([origin, origin2]) | ||||
assert set(swh_storage.journal_writer.journal.objects) == set( | assert set(swh_storage.journal_writer.journal.objects) == set( | ||||
[("origin", origin), ("origin", origin2),] | [("origin", origin), ("origin", origin2),] | ||||
) | ) | ||||
assert add1 == {"origin:add": 2} | assert add1 == {"origin:add": 2} | ||||
▲ Show 20 Lines • Show All 1,154 Lines • ▼ Show 20 Lines | def test_snapshot_add_many(self, swh_storage, sample_data): | ||||
assert swh_storage.snapshot_get(snapshot.id) == { | assert swh_storage.snapshot_get(snapshot.id) == { | ||||
**snapshot.to_dict(), | **snapshot.to_dict(), | ||||
"next_branch": None, | "next_branch": None, | ||||
} | } | ||||
swh_storage.refresh_stat_counters() | swh_storage.refresh_stat_counters() | ||||
assert swh_storage.stat_counters()["snapshot"] == 2 | assert swh_storage.stat_counters()["snapshot"] == 2 | ||||
def test_snapshot_add_many_from_generator(self, swh_storage, sample_data): | |||||
snapshot, _, complete_snapshot = sample_data.snapshots[:3] | |||||
def _snp_gen(): | |||||
yield from [snapshot, complete_snapshot] | |||||
actual_result = swh_storage.snapshot_add(_snp_gen()) | |||||
assert actual_result == {"snapshot:add": 2} | |||||
swh_storage.refresh_stat_counters() | |||||
assert swh_storage.stat_counters()["snapshot"] == 2 | |||||
def test_snapshot_add_many_incremental(self, swh_storage, sample_data): | def test_snapshot_add_many_incremental(self, swh_storage, sample_data): | ||||
snapshot, _, complete_snapshot = sample_data.snapshots[:3] | snapshot, _, complete_snapshot = sample_data.snapshots[:3] | ||||
actual_result = swh_storage.snapshot_add([complete_snapshot]) | actual_result = swh_storage.snapshot_add([complete_snapshot]) | ||||
assert actual_result == {"snapshot:add": 1} | assert actual_result == {"snapshot:add": 1} | ||||
actual_result2 = swh_storage.snapshot_add([snapshot, complete_snapshot]) | actual_result2 = swh_storage.snapshot_add([snapshot, complete_snapshot]) | ||||
assert actual_result2 == {"snapshot:add": 1} | assert actual_result2 == {"snapshot:add": 1} | ||||
▲ Show 20 Lines • Show All 1,815 Lines • Show Last 20 Lines |