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 | |||||