Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
Show First 20 Lines • Show All 1,163 Lines • ▼ Show 20 Lines | def test_release_get_random(self, swh_storage, sample_data): | ||||
assert swh_storage.release_get_random() in { | assert swh_storage.release_get_random() in { | ||||
release.id, | release.id, | ||||
release2.id, | release2.id, | ||||
release3.id, | release3.id, | ||||
} | } | ||||
def test_origin_add(self, swh_storage, sample_data): | def test_origin_add(self, swh_storage, sample_data): | ||||
origin, origin2 = sample_data.origins[:2] | origins = list(sample_data.origins[:2]) | ||||
origin_dict, origin2_dict = [o.to_dict() for o in [origin, origin2]] | origin_urls = [o.url for o in origins] | ||||
assert swh_storage.origin_get([origin_dict])[0] is None | assert swh_storage.origin_get(origin_urls) == [None, None] | ||||
stats = swh_storage.origin_add([origin, origin2]) | stats = swh_storage.origin_add(origins) | ||||
assert stats == {"origin:add": 2} | assert stats == {"origin:add": 2} | ||||
actual_origin = swh_storage.origin_get([origin_dict])[0] | actual_origins = swh_storage.origin_get(origin_urls) | ||||
assert actual_origin["url"] == origin.url | assert actual_origins == origins | ||||
actual_origin2 = swh_storage.origin_get([origin2_dict])[0] | |||||
assert actual_origin2["url"] == origin2.url | |||||
assert set(swh_storage.journal_writer.journal.objects) == set( | assert set(swh_storage.journal_writer.journal.objects) == set( | ||||
[("origin", origin), ("origin", origin2),] | [("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): | def test_origin_add_from_generator(self, swh_storage, sample_data): | ||||
origin, origin2 = sample_data.origins[:2] | origin, origin2 = sample_data.origins[:2] | ||||
origin_dict, origin2_dict = [o.to_dict() for o in [origin, origin2]] | |||||
def _ori_gen(): | def _ori_gen(): | ||||
yield origin | yield origin | ||||
yield origin2 | yield origin2 | ||||
stats = swh_storage.origin_add(_ori_gen()) | stats = swh_storage.origin_add(_ori_gen()) | ||||
assert stats == {"origin:add": 2} | assert stats == {"origin:add": 2} | ||||
actual_origin = swh_storage.origin_get([origin_dict])[0] | actual_origins = swh_storage.origin_get([origin.url, origin2.url]) | ||||
assert actual_origin["url"] == origin.url | assert actual_origins == [origin, origin2] | ||||
actual_origin2 = swh_storage.origin_get([origin2_dict])[0] | |||||
assert actual_origin2["url"] == origin2.url | |||||
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),] | ||||
) | ) | ||||
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_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] | ||||
origin_dict, origin2_dict = [o.to_dict() for o in [origin, origin2]] | |||||
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} | ||||
add2 = swh_storage.origin_add([origin, origin2]) | add2 = 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 add2 == {"origin:add": 0} | assert add2 == {"origin:add": 0} | ||||
def test_origin_get_legacy(self, swh_storage, sample_data): | |||||
origin, origin2 = sample_data.origins[:2] | |||||
origin_dict, origin2_dict = [o.to_dict() for o in [origin, origin2]] | |||||
assert swh_storage.origin_get(origin_dict) is None | |||||
swh_storage.origin_add([origin]) | |||||
actual_origin0 = swh_storage.origin_get(origin_dict) | |||||
assert actual_origin0["url"] == origin.url | |||||
def test_origin_get(self, swh_storage, sample_data): | def test_origin_get(self, swh_storage, sample_data): | ||||
origin, origin2 = sample_data.origins[:2] | origin, origin2 = sample_data.origins[:2] | ||||
origin_dict, origin2_dict = [o.to_dict() for o in [origin, origin2]] | |||||
assert swh_storage.origin_get(origin_dict) is None | assert swh_storage.origin_get([origin.url]) == [None] | ||||
assert swh_storage.origin_get([origin_dict]) == [None] | |||||
swh_storage.origin_add([origin]) | swh_storage.origin_add([origin]) | ||||
actual_origins = swh_storage.origin_get([origin_dict]) | actual_origins = swh_storage.origin_get([origin.url]) | ||||
assert len(actual_origins) == 1 | assert actual_origins == [origin] | ||||
actual_origin0 = swh_storage.origin_get(origin_dict) | |||||
assert actual_origin0 == actual_origins[0] | |||||
assert actual_origin0["url"] == origin.url | |||||
actual_origins = swh_storage.origin_get([origin_dict, {"url": "not://exists"}]) | actual_origins = swh_storage.origin_get([origin.url, "not://exists"]) | ||||
assert actual_origins == [origin_dict, None] | assert actual_origins == [origin, None] | ||||
def _generate_random_visits(self, nb_visits=100, start=0, end=7): | def _generate_random_visits(self, nb_visits=100, start=0, end=7): | ||||
"""Generate random visits within the last 2 months (to avoid | """Generate random visits within the last 2 months (to avoid | ||||
computations) | computations) | ||||
""" | """ | ||||
visits = [] | visits = [] | ||||
today = now() | today = now() | ||||
▲ Show 20 Lines • Show All 149 Lines • ▼ Show 20 Lines | ): | ||||
] | ] | ||||
) | ) | ||||
random_origin_visit = swh_storage.origin_visit_status_get_random(visit_type) | random_origin_visit = swh_storage.origin_visit_status_get_random(visit_type) | ||||
assert random_origin_visit is None | assert random_origin_visit is None | ||||
def test_origin_get_by_sha1(self, swh_storage, sample_data): | def test_origin_get_by_sha1(self, swh_storage, sample_data): | ||||
origin = sample_data.origin | origin = sample_data.origin | ||||
assert swh_storage.origin_get(origin.to_dict()) is None | assert swh_storage.origin_get([origin.url])[0] is None | ||||
swh_storage.origin_add([origin]) | swh_storage.origin_add([origin]) | ||||
origins = list(swh_storage.origin_get_by_sha1([sha1(origin.url)])) | origins = list(swh_storage.origin_get_by_sha1([sha1(origin.url)])) | ||||
assert len(origins) == 1 | assert len(origins) == 1 | ||||
assert origins[0]["url"] == origin.url | assert origins[0]["url"] == origin.url | ||||
def test_origin_get_by_sha1_not_found(self, swh_storage, sample_data): | def test_origin_get_by_sha1_not_found(self, swh_storage, sample_data): | ||||
origin = sample_data.origin | origin = sample_data.origin | ||||
assert swh_storage.origin_get(origin.to_dict()) is None | assert swh_storage.origin_get([origin.url])[0] is None | ||||
origins = list(swh_storage.origin_get_by_sha1([sha1(origin.url)])) | origins = list(swh_storage.origin_get_by_sha1([sha1(origin.url)])) | ||||
assert len(origins) == 1 | assert len(origins) == 1 | ||||
assert origins[0] is None | assert origins[0] is None | ||||
def test_origin_search_single_result(self, swh_storage, sample_data): | def test_origin_search_single_result(self, swh_storage, sample_data): | ||||
origin, origin2 = sample_data.origins[:2] | origin, origin2 = sample_data.origins[:2] | ||||
found_origins = list(swh_storage.origin_search(origin.url)) | found_origins = list(swh_storage.origin_search(origin.url)) | ||||
▲ Show 20 Lines • Show All 2,737 Lines • Show Last 20 Lines |