Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/storage_tests.py
| Show First 20 Lines • Show All 1,406 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): | ||||
| origins = list(sample_data.origins[:2]) | origins = list(sample_data.origins) | ||||
| origin_urls = [o.url for o in origins] | origin_urls = [o.url for o in origins] | ||||
| assert swh_storage.origin_get(origin_urls) == [None, None] | assert swh_storage.origin_get(origin_urls) == [None] * len(origins) | ||||
| stats = swh_storage.origin_add(origins) | stats = swh_storage.origin_add(origins) | ||||
| assert stats == {"origin:add": 2} | assert stats == {"origin:add": len(origin_urls)} | ||||
| actual_origins = swh_storage.origin_get(origin_urls) | actual_origins = swh_storage.origin_get(origin_urls) | ||||
| assert actual_origins == origins | assert actual_origins == origins | ||||
| 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", origin) for origin in origins] | ||||
| ) | ) | ||||
| swh_storage.refresh_stat_counters() | swh_storage.refresh_stat_counters() | ||||
| assert swh_storage.stat_counters()["origin"] == 2 | assert swh_storage.stat_counters()["origin"] == len(origins) | ||||
| 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),] | ||||
| ) | ) | ||||
| ▲ Show 20 Lines • Show All 2,986 Lines • ▼ Show 20 Lines | class TestStorageGeneratedData: | ||||
| ) | ) | ||||
| @given(strategies.lists(objects(split_content=True), max_size=2)) | @given(strategies.lists(objects(split_content=True), max_size=2)) | ||||
| def test_add_arbitrary(self, swh_storage, objects): | def test_add_arbitrary(self, swh_storage, objects): | ||||
| for (obj_type, obj) in objects: | for (obj_type, obj) in objects: | ||||
| if obj.object_type == "origin_visit": | if obj.object_type == "origin_visit": | ||||
| swh_storage.origin_add([Origin(url=obj.origin)]) | swh_storage.origin_add([Origin(url=obj.origin)]) | ||||
| visit = OriginVisit(origin=obj.origin, date=obj.date, type=obj.type,) | visit = OriginVisit(origin=obj.origin, date=obj.date, type=obj.type,) | ||||
| swh_storage.origin_visit_add([visit]) | swh_storage.origin_visit_add([visit]) | ||||
| elif obj.object_type == "raw_extrinsic_metadata": | |||||
| swh_storage.metadata_authority_add([obj.authority]) | |||||
| swh_storage.metadata_fetcher_add([obj.fetcher]) | |||||
| swh_storage.raw_extrinsic_metadata_add([obj]) | |||||
| else: | else: | ||||
| method = getattr(swh_storage, obj_type + "_add") | method = getattr(swh_storage, obj_type + "_add") | ||||
| try: | try: | ||||
| method([obj]) | method([obj]) | ||||
| except HashCollision: | except HashCollision: | ||||
| pass | pass | ||||