Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 1,065 Lines • ▼ Show 20 Lines | def test_origin_visit_add(self, swh_storage): | ||||
| 'date': date_visit, | 'date': date_visit, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.type_visit1, | 'type': data.type_visit1, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } in actual_origin_visits | } in actual_origin_visits | ||||
| expected_origin = data.origin2 | |||||
| origin_visit = { | origin_visit = { | ||||
| 'origin': expected_origin, | 'origin': origin_url, | ||||
| 'date': date_visit, | 'date': date_visit, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.type_visit1, | 'type': data.type_visit1, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| objects = list(swh_storage.journal_writer.objects) | objects = list(swh_storage.journal_writer.objects) | ||||
| assert ('origin', expected_origin) in objects | assert ('origin', data.origin2) in objects | ||||
| assert ('origin_visit', origin_visit) in objects | assert ('origin_visit', origin_visit) in objects | ||||
| def test_origin_visit_get__unknown_origin(self, swh_storage): | def test_origin_visit_get__unknown_origin(self, swh_storage): | ||||
| assert [] == list(swh_storage.origin_visit_get('foo')) | assert [] == list(swh_storage.origin_visit_get('foo')) | ||||
| def test_origin_visit_add_default_type(self, swh_storage): | def test_origin_visit_add_default_type(self, swh_storage): | ||||
| # given | # given | ||||
| swh_storage.origin_add_one(data.origin2) | swh_storage.origin_add_one(data.origin2) | ||||
| ▲ Show 20 Lines • Show All 41 Lines • ▼ Show 20 Lines | def test_origin_visit_add_default_type(self, swh_storage): | ||||
| ] | ] | ||||
| for visit in expected_visits: | for visit in expected_visits: | ||||
| assert visit in actual_origin_visits | assert visit in actual_origin_visits | ||||
| objects = list(swh_storage.journal_writer.objects) | objects = list(swh_storage.journal_writer.objects) | ||||
| assert ('origin', data.origin2) in objects | assert ('origin', data.origin2) in objects | ||||
| for visit in expected_visits: | for visit in expected_visits: | ||||
| visit['origin'] = data.origin2 | |||||
| assert ('origin_visit', visit) in objects | assert ('origin_visit', visit) in objects | ||||
| def test_origin_visit_add_validation(self, swh_storage): | def test_origin_visit_add_validation(self, swh_storage): | ||||
| origin_url = swh_storage.origin_add_one(data.origin2) | origin_url = swh_storage.origin_add_one(data.origin2) | ||||
| with pytest.raises((TypeError, psycopg2.ProgrammingError)) as cm: | with pytest.raises((TypeError, psycopg2.ProgrammingError)) as cm: | ||||
| swh_storage.origin_visit_add(origin_url, date=[b'foo']) | swh_storage.origin_visit_add(origin_url, date=[b'foo']) | ||||
| Show All 40 Lines | def test_origin_visit_update(self, swh_storage): | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin_url2, | origin_url2, | ||||
| origin_visit3['visit'], status='partial') | origin_visit3['visit'], status='partial') | ||||
| # then | # then | ||||
| actual_origin_visits = list(swh_storage.origin_visit_get( | actual_origin_visits = list(swh_storage.origin_visit_get( | ||||
| origin_url)) | origin_url)) | ||||
| expected_visits = [{ | expected_visits = [{ | ||||
| 'origin': origin_visit2['origin'], | 'origin': origin_url, | ||||
| 'date': date_visit, | 'date': date_visit, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.type_visit1, | 'type': data.type_visit1, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': visit1_metadata, | 'metadata': visit1_metadata, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }, { | }, { | ||||
| 'origin': origin_visit2['origin'], | 'origin': origin_url, | ||||
| 'date': date_visit2, | 'date': date_visit2, | ||||
| 'visit': origin_visit2['visit'], | 'visit': origin_visit2['visit'], | ||||
| 'type': data.type_visit2, | 'type': data.type_visit2, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }] | }] | ||||
| for visit in expected_visits: | for visit in expected_visits: | ||||
| assert visit in actual_origin_visits | assert visit in actual_origin_visits | ||||
| actual_origin_visits_bis = list(swh_storage.origin_visit_get( | actual_origin_visits_bis = list(swh_storage.origin_visit_get( | ||||
| origin_url, | origin_url, | ||||
| limit=1)) | limit=1)) | ||||
| assert actual_origin_visits_bis == [ | assert actual_origin_visits_bis == [ | ||||
| { | { | ||||
| 'origin': origin_visit2['origin'], | 'origin': origin_url, | ||||
| 'date': date_visit, | 'date': date_visit, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.type_visit1, | 'type': data.type_visit1, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': visit1_metadata, | 'metadata': visit1_metadata, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }] | }] | ||||
| actual_origin_visits_ter = list(swh_storage.origin_visit_get( | actual_origin_visits_ter = list(swh_storage.origin_visit_get( | ||||
| origin_url, | origin_url, | ||||
| last_visit=origin_visit1['visit'])) | last_visit=origin_visit1['visit'])) | ||||
| assert actual_origin_visits_ter == [ | assert actual_origin_visits_ter == [ | ||||
| { | { | ||||
| 'origin': origin_visit2['origin'], | 'origin': origin_url, | ||||
| 'date': date_visit2, | 'date': date_visit2, | ||||
| 'visit': origin_visit2['visit'], | 'visit': origin_visit2['visit'], | ||||
| 'type': data.type_visit2, | 'type': data.type_visit2, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }] | }] | ||||
| actual_origin_visits2 = list(swh_storage.origin_visit_get( | actual_origin_visits2 = list(swh_storage.origin_visit_get( | ||||
| origin_url2)) | origin_url2)) | ||||
| assert actual_origin_visits2 == [ | assert actual_origin_visits2 == [ | ||||
| { | { | ||||
| 'origin': origin_visit3['origin'], | 'origin': origin_url2, | ||||
| 'date': date_visit2, | 'date': date_visit2, | ||||
| 'visit': origin_visit3['visit'], | 'visit': origin_visit3['visit'], | ||||
| 'type': data.type_visit3, | 'type': data.type_visit3, | ||||
| 'status': 'partial', | 'status': 'partial', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }] | }] | ||||
| expected_origin = data.origin.copy() | |||||
| expected_origin2 = data.origin2.copy() | |||||
| data1 = { | data1 = { | ||||
| 'origin': expected_origin, | 'origin': origin_url, | ||||
| 'date': date_visit, | 'date': date_visit, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.type_visit1, | 'type': data.type_visit1, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data2 = { | data2 = { | ||||
| 'origin': expected_origin, | 'origin': origin_url, | ||||
| 'date': date_visit2, | 'date': date_visit2, | ||||
| 'visit': origin_visit2['visit'], | 'visit': origin_visit2['visit'], | ||||
| 'type': data.type_visit2, | 'type': data.type_visit2, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data3 = { | data3 = { | ||||
| 'origin': expected_origin2, | 'origin': origin_url2, | ||||
| 'date': date_visit2, | 'date': date_visit2, | ||||
| 'visit': origin_visit3['visit'], | 'visit': origin_visit3['visit'], | ||||
| 'type': data.type_visit3, | 'type': data.type_visit3, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data4 = { | data4 = { | ||||
| 'origin': expected_origin, | 'origin': origin_url, | ||||
| 'date': date_visit, | 'date': date_visit, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.type_visit1, | 'type': data.type_visit1, | ||||
| 'metadata': visit1_metadata, | 'metadata': visit1_metadata, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data5 = { | data5 = { | ||||
| 'origin': expected_origin2, | 'origin': origin_url2, | ||||
| 'date': date_visit2, | 'date': date_visit2, | ||||
| 'visit': origin_visit3['visit'], | 'visit': origin_visit3['visit'], | ||||
| 'type': data.type_visit3, | 'type': data.type_visit3, | ||||
| 'status': 'partial', | 'status': 'partial', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| objects = list(swh_storage.journal_writer.objects) | objects = list(swh_storage.journal_writer.objects) | ||||
| assert ('origin', expected_origin) in objects | assert ('origin', data.origin) in objects | ||||
| assert ('origin', expected_origin2) in objects | assert ('origin', data.origin2) in objects | ||||
| assert ('origin_visit', data1) in objects | assert ('origin_visit', data1) in objects | ||||
| assert ('origin_visit', data2) in objects | assert ('origin_visit', data2) in objects | ||||
| assert ('origin_visit', data3) in objects | assert ('origin_visit', data3) in objects | ||||
| assert ('origin_visit', data4) in objects | assert ('origin_visit', data4) in objects | ||||
| assert ('origin_visit', data5) in objects | assert ('origin_visit', data5) in objects | ||||
| def test_origin_visit_update_validation(self, swh_storage): | def test_origin_visit_update_validation(self, swh_storage): | ||||
| origin_url = data.origin['url'] | origin_url = data.origin['url'] | ||||
| ▲ Show 20 Lines • Show All 142 Lines • ▼ Show 20 Lines | |||||
| def test_origin_visit_upsert_new(self, swh_storage): | def test_origin_visit_upsert_new(self, swh_storage): | ||||
| # given | # given | ||||
| swh_storage.origin_add_one(data.origin2) | swh_storage.origin_add_one(data.origin2) | ||||
| origin_url = data.origin2['url'] | origin_url = data.origin2['url'] | ||||
| # when | # when | ||||
| swh_storage.origin_visit_upsert([ | swh_storage.origin_visit_upsert([ | ||||
| { | { | ||||
| 'origin': data.origin2, | 'origin': origin_url, | ||||
| 'date': data.date_visit2, | 'date': data.date_visit2, | ||||
| 'visit': 123, | 'visit': 123, | ||||
| 'type': data.type_visit2, | 'type': data.type_visit2, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }, | }, | ||||
| { | { | ||||
| 'origin': data.origin2, | 'origin': origin_url, | ||||
| 'date': '2018-01-01 23:00:00+00', | 'date': '2018-01-01 23:00:00+00', | ||||
| 'visit': 1234, | 'visit': 1234, | ||||
| 'type': data.type_visit2, | 'type': data.type_visit2, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }, | }, | ||||
| ]) | ]) | ||||
| Show All 17 Lines | def test_origin_visit_upsert_new(self, swh_storage): | ||||
| 'visit': 1234, | 'visit': 1234, | ||||
| 'type': data.type_visit2, | 'type': data.type_visit2, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }, | }, | ||||
| ] | ] | ||||
| expected_origin = data.origin2 | |||||
| data1 = { | data1 = { | ||||
| 'origin': expected_origin, | 'origin': origin_url, | ||||
| 'date': data.date_visit2, | 'date': data.date_visit2, | ||||
| 'visit': 123, | 'visit': 123, | ||||
| 'type': data.type_visit2, | 'type': data.type_visit2, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data2 = { | data2 = { | ||||
| 'origin': expected_origin, | 'origin': origin_url, | ||||
| 'date': data.date_visit3, | 'date': data.date_visit3, | ||||
| 'visit': 1234, | 'visit': 1234, | ||||
| 'type': data.type_visit2, | 'type': data.type_visit2, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| assert list(swh_storage.journal_writer.objects) == [ | assert list(swh_storage.journal_writer.objects) == [ | ||||
| ('origin', expected_origin), | ('origin', data.origin2), | ||||
| ('origin_visit', data1), | ('origin_visit', data1), | ||||
| ('origin_visit', data2)] | ('origin_visit', data2)] | ||||
| def test_origin_visit_upsert_existing(self, swh_storage): | def test_origin_visit_upsert_existing(self, swh_storage): | ||||
| # given | # given | ||||
| swh_storage.origin_add_one(data.origin2) | swh_storage.origin_add_one(data.origin2) | ||||
| origin_url = data.origin2['url'] | origin_url = data.origin2['url'] | ||||
| # when | # when | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| origin_url, | origin_url, | ||||
| date=data.date_visit2, | date=data.date_visit2, | ||||
| type=data.type_visit1, | type=data.type_visit1, | ||||
| ) | ) | ||||
| swh_storage.origin_visit_upsert([{ | swh_storage.origin_visit_upsert([{ | ||||
| 'origin': data.origin2, | 'origin': origin_url, | ||||
| 'date': data.date_visit2, | 'date': data.date_visit2, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.type_visit1, | 'type': data.type_visit1, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }]) | }]) | ||||
| Show All 9 Lines | def test_origin_visit_upsert_existing(self, swh_storage): | ||||
| 'date': data.date_visit2, | 'date': data.date_visit2, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.type_visit1, | 'type': data.type_visit1, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }] | }] | ||||
| expected_origin = data.origin2 | |||||
| data1 = { | data1 = { | ||||
| 'origin': expected_origin, | 'origin': origin_url, | ||||
| 'date': data.date_visit2, | 'date': data.date_visit2, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.type_visit1, | 'type': data.type_visit1, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data2 = { | data2 = { | ||||
| 'origin': expected_origin, | 'origin': origin_url, | ||||
| 'date': data.date_visit2, | 'date': data.date_visit2, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.type_visit1, | 'type': data.type_visit1, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| assert list(swh_storage.journal_writer.objects) == [ | assert list(swh_storage.journal_writer.objects) == [ | ||||
| ('origin', expected_origin), | ('origin', data.origin2), | ||||
| ('origin_visit', data1), | ('origin_visit', data1), | ||||
| ('origin_visit', data2)] | ('origin_visit', data2)] | ||||
| def test_origin_visit_get_by_no_result(self, swh_storage): | def test_origin_visit_get_by_no_result(self, swh_storage): | ||||
| swh_storage.origin_add([data.origin]) | swh_storage.origin_add([data.origin]) | ||||
| actual_origin_visit = swh_storage.origin_visit_get_by( | actual_origin_visit = swh_storage.origin_visit_get_by( | ||||
| data.origin['url'], 999) | data.origin['url'], 999) | ||||
| assert actual_origin_visit is None | assert actual_origin_visit is None | ||||
| ▲ Show 20 Lines • Show All 141 Lines • ▼ Show 20 Lines | def test_snapshot_add_get_empty(self, swh_storage): | ||||
| origin_url, visit_id, snapshot=data.empty_snapshot['id']) | origin_url, visit_id, snapshot=data.empty_snapshot['id']) | ||||
| by_id = swh_storage.snapshot_get(data.empty_snapshot['id']) | by_id = swh_storage.snapshot_get(data.empty_snapshot['id']) | ||||
| assert by_id == {**data.empty_snapshot, 'next_branch': None} | assert by_id == {**data.empty_snapshot, 'next_branch': None} | ||||
| by_ov = swh_storage.snapshot_get_by_origin_visit(origin_url, visit_id) | by_ov = swh_storage.snapshot_get_by_origin_visit(origin_url, visit_id) | ||||
| assert by_ov == {**data.empty_snapshot, 'next_branch': None} | assert by_ov == {**data.empty_snapshot, 'next_branch': None} | ||||
| expected_origin = data.origin.copy() | |||||
| data1 = { | data1 = { | ||||
| 'origin': expected_origin, | 'origin': origin_url, | ||||
| 'date': data.date_visit1, | 'date': data.date_visit1, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.type_visit1, | 'type': data.type_visit1, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data2 = { | data2 = { | ||||
| 'origin': expected_origin, | 'origin': origin_url, | ||||
| 'date': data.date_visit1, | 'date': data.date_visit1, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.type_visit1, | 'type': data.type_visit1, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': data.empty_snapshot['id'], | 'snapshot': data.empty_snapshot['id'], | ||||
| } | } | ||||
| assert list(swh_storage.journal_writer.objects) == \ | assert list(swh_storage.journal_writer.objects) == \ | ||||
| [('origin', expected_origin), | [('origin', data.origin), | ||||
| ('origin_visit', data1), | ('origin_visit', data1), | ||||
| ('snapshot', data.empty_snapshot), | ('snapshot', data.empty_snapshot), | ||||
| ('origin_visit', data2)] | ('origin_visit', data2)] | ||||
| def test_snapshot_add_get_complete(self, swh_storage): | def test_snapshot_add_get_complete(self, swh_storage): | ||||
| origin_url = data.origin['url'] | origin_url = data.origin['url'] | ||||
| swh_storage.origin_add_one(data.origin) | swh_storage.origin_add_one(data.origin) | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| ▲ Show 20 Lines • Show All 323 Lines • ▼ Show 20 Lines | def test_snapshot_add_twice__by_origin_visit(self, swh_storage): | ||||
| swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin_url, visit2_id, snapshot=data.snapshot['id']) | origin_url, visit2_id, snapshot=data.snapshot['id']) | ||||
| by_ov2 = swh_storage.snapshot_get_by_origin_visit( | by_ov2 = swh_storage.snapshot_get_by_origin_visit( | ||||
| origin_url, visit2_id) | origin_url, visit2_id) | ||||
| assert by_ov2 == {**data.snapshot, 'next_branch': None} | assert by_ov2 == {**data.snapshot, 'next_branch': None} | ||||
| expected_origin = data.origin.copy() | |||||
| data1 = { | data1 = { | ||||
| 'origin': expected_origin, | 'origin': origin_url, | ||||
| 'date': data.date_visit1, | 'date': data.date_visit1, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.type_visit1, | 'type': data.type_visit1, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data2 = { | data2 = { | ||||
| 'origin': expected_origin, | 'origin': origin_url, | ||||
| 'date': data.date_visit1, | 'date': data.date_visit1, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.type_visit1, | 'type': data.type_visit1, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': data.snapshot['id'], | 'snapshot': data.snapshot['id'], | ||||
| } | } | ||||
| data3 = { | data3 = { | ||||
| 'origin': expected_origin, | 'origin': origin_url, | ||||
| 'date': data.date_visit2, | 'date': data.date_visit2, | ||||
| 'visit': origin_visit2['visit'], | 'visit': origin_visit2['visit'], | ||||
| 'type': data.type_visit2, | 'type': data.type_visit2, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data4 = { | data4 = { | ||||
| 'origin': expected_origin, | 'origin': origin_url, | ||||
| 'date': data.date_visit2, | 'date': data.date_visit2, | ||||
| 'visit': origin_visit2['visit'], | 'visit': origin_visit2['visit'], | ||||
| 'type': data.type_visit2, | 'type': data.type_visit2, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': data.snapshot['id'], | 'snapshot': data.snapshot['id'], | ||||
| } | } | ||||
| assert list(swh_storage.journal_writer.objects) \ | assert list(swh_storage.journal_writer.objects) \ | ||||
| == [('origin', expected_origin), | == [('origin', data.origin), | ||||
| ('origin_visit', data1), | ('origin_visit', data1), | ||||
| ('snapshot', data.snapshot), | ('snapshot', data.snapshot), | ||||
| ('origin_visit', data2), | ('origin_visit', data2), | ||||
| ('origin_visit', data3), | ('origin_visit', data3), | ||||
| ('origin_visit', data4)] | ('origin_visit', data4)] | ||||
| def test_snapshot_get_latest(self, swh_storage): | def test_snapshot_get_latest(self, swh_storage): | ||||
| origin_url = data.origin['url'] | origin_url = data.origin['url'] | ||||
| ▲ Show 20 Lines • Show All 1,050 Lines • ▼ Show 20 Lines | class TestStorageGeneratedData: | ||||
| @settings(suppress_health_check=[HealthCheck.too_slow]) | @settings(suppress_health_check=[HealthCheck.too_slow]) | ||||
| @given(strategies.lists(objects(), max_size=2)) | @given(strategies.lists(objects(), 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: | ||||
| obj = obj.to_dict() | obj = obj.to_dict() | ||||
| if obj_type == 'origin_visit': | if obj_type == 'origin_visit': | ||||
| origin = obj.pop('origin') | origin = obj.pop('origin') | ||||
| swh_storage.origin_add_one(origin) | swh_storage.origin_add_one({'url': origin}) | ||||
| if 'visit' in obj: | if 'visit' in obj: | ||||
| del obj['visit'] | del obj['visit'] | ||||
| swh_storage.origin_visit_add( | swh_storage.origin_visit_add( | ||||
| origin['url'], obj['date'], obj['type']) | origin, obj['date'], obj['type']) | ||||
| 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 | ||||
| ▲ Show 20 Lines • Show All 196 Lines • Show Last 20 Lines | |||||