Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 1,094 Lines • ▼ Show 20 Lines | def test_origin_visit_add(self, swh_storage, use_url): | ||||
| assert origin_id is not None | assert origin_id is not None | ||||
| origin_id_or_url = data.origin2['url'] if use_url else origin_id | origin_id_or_url = data.origin2['url'] if use_url else origin_id | ||||
| # when | # when | ||||
| date_visit = datetime.datetime.now(datetime.timezone.utc) | date_visit = datetime.datetime.now(datetime.timezone.utc) | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| origin_id_or_url, | origin_id_or_url, | ||||
| type='git', | type=data.type_visit1, | ||||
| date=date_visit) | date=date_visit) | ||||
| actual_origin_visits = list(swh_storage.origin_visit_get( | actual_origin_visits = list(swh_storage.origin_visit_get( | ||||
| origin_id_or_url)) | origin_id_or_url)) | ||||
| assert { | assert { | ||||
| 'origin': origin_id, | 'origin': origin_id, | ||||
| 'date': date_visit, | 'date': date_visit, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': 'git', | '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 | expected_origin = data.origin2 | ||||
| origin_visit = { | origin_visit = { | ||||
| 'origin': expected_origin, | 'origin': expected_origin, | ||||
| 'date': date_visit, | 'date': date_visit, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': 'git', | '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', expected_origin) in objects | ||||
| assert ('origin_visit', origin_visit) in objects | assert ('origin_visit', origin_visit) in objects | ||||
| Show All 11 Lines | def test_origin_visit_add_default_type(self, swh_storage, use_url): | ||||
| origin_id_or_url = data.origin2['url'] if use_url else origin_id | origin_id_or_url = data.origin2['url'] if use_url else origin_id | ||||
| assert origin_id is not None | assert origin_id is not None | ||||
| # when | # when | ||||
| date_visit = datetime.datetime.now(datetime.timezone.utc) | date_visit = datetime.datetime.now(datetime.timezone.utc) | ||||
| date_visit2 = date_visit + datetime.timedelta(minutes=1) | date_visit2 = date_visit + datetime.timedelta(minutes=1) | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| origin_id_or_url, | origin_id_or_url, | ||||
| date=date_visit) | date=date_visit, | ||||
| type=data.type_visit1, | |||||
| ) | |||||
| origin_visit2 = swh_storage.origin_visit_add( | origin_visit2 = swh_storage.origin_visit_add( | ||||
| origin_id_or_url, | origin_id_or_url, | ||||
| date=date_visit2) | date=date_visit2, | ||||
| type=data.type_visit2, | |||||
| ) | |||||
| # then | # then | ||||
| assert origin_visit1['origin'] == origin_id | assert origin_visit1['origin'] == origin_id | ||||
| assert origin_visit1['visit'] is not None | assert origin_visit1['visit'] is not None | ||||
| actual_origin_visits = list(swh_storage.origin_visit_get( | actual_origin_visits = list(swh_storage.origin_visit_get( | ||||
| origin_id_or_url)) | origin_id_or_url)) | ||||
| expected_visits = [ | expected_visits = [ | ||||
| { | { | ||||
| 'origin': origin_id, | 'origin': origin_id, | ||||
| 'date': date_visit, | 'date': date_visit, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': 'hg', | 'type': data.type_visit1, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }, | }, | ||||
| { | { | ||||
| 'origin': origin_id, | 'origin': origin_id, | ||||
| 'date': date_visit2, | 'date': date_visit2, | ||||
| 'visit': origin_visit2['visit'], | 'visit': origin_visit2['visit'], | ||||
| 'type': 'hg', | '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 | ||||
| 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 | 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_id_or_url = swh_storage.origin_add_one(data.origin2) | origin_id_or_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_id_or_url, date=[b'foo']) | swh_storage.origin_visit_add(origin_id_or_url, date=[b'foo'], | ||||
| type=data.type_visit1) | |||||
| if type(cm.value) == psycopg2.ProgrammingError: | if type(cm.value) == psycopg2.ProgrammingError: | ||||
| assert cm.value.pgcode \ | assert cm.value.pgcode \ | ||||
| == psycopg2.errorcodes.UNDEFINED_FUNCTION | == psycopg2.errorcodes.UNDEFINED_FUNCTION | ||||
| @pytest.mark.parametrize('use_url', [True, False]) | @pytest.mark.parametrize('use_url', [True, False]) | ||||
| def test_origin_visit_update(self, swh_storage, use_url): | def test_origin_visit_update(self, swh_storage, use_url): | ||||
| if not self._test_origin_ids and not use_url: | if not self._test_origin_ids and not use_url: | ||||
| return | return | ||||
| # given | # given | ||||
| swh_storage.origin_add_one(data.origin) | swh_storage.origin_add_one(data.origin) | ||||
| origin_url = data.origin['url'] | origin_url = data.origin['url'] | ||||
| date_visit = datetime.datetime.now(datetime.timezone.utc) | date_visit = datetime.datetime.now(datetime.timezone.utc) | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| origin_url, | origin_url, | ||||
| date=date_visit) | date=date_visit, | ||||
| type=data.type_visit1, | |||||
| ) | |||||
| date_visit2 = date_visit + datetime.timedelta(minutes=1) | date_visit2 = date_visit + datetime.timedelta(minutes=1) | ||||
| origin_visit2 = swh_storage.origin_visit_add( | origin_visit2 = swh_storage.origin_visit_add( | ||||
| origin_url, | origin_url, | ||||
| date=date_visit2) | date=date_visit2, | ||||
| type=data.type_visit2 | |||||
| ) | |||||
| swh_storage.origin_add_one(data.origin2) | swh_storage.origin_add_one(data.origin2) | ||||
| origin_url2 = data.origin2['url'] | origin_url2 = data.origin2['url'] | ||||
| origin_visit3 = swh_storage.origin_visit_add( | origin_visit3 = swh_storage.origin_visit_add( | ||||
| origin_url2, | origin_url2, | ||||
| date=date_visit2) | date=date_visit2, | ||||
| type=data.type_visit3 | |||||
| ) | |||||
| # when | # when | ||||
| visit1_metadata = { | visit1_metadata = { | ||||
| 'contents': 42, | 'contents': 42, | ||||
| 'directories': 22, | 'directories': 22, | ||||
| } | } | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin_url, | origin_url, | ||||
| origin_visit1['visit'], status='full', | origin_visit1['visit'], status='full', | ||||
| metadata=visit1_metadata) | metadata=visit1_metadata) | ||||
| 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_visit2['origin'], | ||||
| 'date': date_visit, | 'date': date_visit, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.origin['type'], | 'type': data.type_visit1, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': visit1_metadata, | 'metadata': visit1_metadata, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }, { | }, { | ||||
| 'origin': origin_visit2['origin'], | 'origin': origin_visit2['origin'], | ||||
| 'date': date_visit2, | 'date': date_visit2, | ||||
| 'visit': origin_visit2['visit'], | 'visit': origin_visit2['visit'], | ||||
| 'type': data.origin['type'], | '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_visit2['origin'], | ||||
| 'date': date_visit, | 'date': date_visit, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.origin['type'], | '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_visit2['origin'], | ||||
| 'date': date_visit2, | 'date': date_visit2, | ||||
| 'visit': origin_visit2['visit'], | 'visit': origin_visit2['visit'], | ||||
| 'type': data.origin['type'], | '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_visit3['origin'], | ||||
| 'date': date_visit2, | 'date': date_visit2, | ||||
| 'visit': origin_visit3['visit'], | 'visit': origin_visit3['visit'], | ||||
| 'type': data.origin2['type'], | 'type': data.type_visit3, | ||||
| 'status': 'partial', | 'status': 'partial', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }] | }] | ||||
| expected_origin = data.origin.copy() | expected_origin = data.origin.copy() | ||||
| expected_origin2 = data.origin2.copy() | expected_origin2 = data.origin2.copy() | ||||
| data1 = { | data1 = { | ||||
| 'origin': expected_origin, | 'origin': expected_origin, | ||||
| 'date': date_visit, | 'date': date_visit, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.origin['type'], | 'type': data.type_visit1, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data2 = { | data2 = { | ||||
| 'origin': expected_origin, | 'origin': expected_origin, | ||||
| 'date': date_visit2, | 'date': date_visit2, | ||||
| 'visit': origin_visit2['visit'], | 'visit': origin_visit2['visit'], | ||||
| 'type': data.origin['type'], | 'type': data.type_visit2, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data3 = { | data3 = { | ||||
| 'origin': expected_origin2, | 'origin': expected_origin2, | ||||
| 'date': date_visit2, | 'date': date_visit2, | ||||
| 'visit': origin_visit3['visit'], | 'visit': origin_visit3['visit'], | ||||
| 'type': data.origin2['type'], | 'type': data.type_visit3, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data4 = { | data4 = { | ||||
| 'origin': expected_origin, | 'origin': expected_origin, | ||||
| 'date': date_visit, | 'date': date_visit, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.origin['type'], | 'type': data.type_visit1, | ||||
| 'metadata': visit1_metadata, | 'metadata': visit1_metadata, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data5 = { | data5 = { | ||||
| 'origin': expected_origin2, | 'origin': expected_origin2, | ||||
| 'date': date_visit2, | 'date': date_visit2, | ||||
| 'visit': origin_visit3['visit'], | 'visit': origin_visit3['visit'], | ||||
| 'type': data.origin2['type'], | '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', expected_origin) in objects | ||||
| assert ('origin', expected_origin2) in objects | assert ('origin', expected_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_id = swh_storage.origin_add_one(data.origin) | origin_id = swh_storage.origin_add_one(data.origin) | ||||
| visit = swh_storage.origin_visit_add( | visit = swh_storage.origin_visit_add( | ||||
| origin_id, | origin_id, | ||||
| date=data.date_visit2) | date=data.date_visit2, | ||||
| type=data.type_visit2, | |||||
| ) | |||||
| with pytest.raises((ValueError, psycopg2.DataError), | with pytest.raises((ValueError, psycopg2.DataError), | ||||
| match='status') as cm: | match='status') as cm: | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin_id, visit['visit'], status='foobar') | origin_id, visit['visit'], status='foobar') | ||||
| if type(cm.value) == psycopg2.DataError: | if type(cm.value) == psycopg2.DataError: | ||||
| assert cm.value.pgcode == \ | assert cm.value.pgcode == \ | ||||
| psycopg2.errorcodes.INVALID_TEXT_REPRESENTATION | psycopg2.errorcodes.INVALID_TEXT_REPRESENTATION | ||||
| def test_origin_visit_find_by_date(self, swh_storage): | def test_origin_visit_find_by_date(self, swh_storage): | ||||
| # given | # given | ||||
| swh_storage.origin_add_one(data.origin) | swh_storage.origin_add_one(data.origin) | ||||
| swh_storage.origin_visit_add( | swh_storage.origin_visit_add( | ||||
| data.origin['url'], | data.origin['url'], | ||||
| date=data.date_visit2) | date=data.date_visit2, | ||||
| type=data.type_visit1, | |||||
| ) | |||||
| origin_visit2 = swh_storage.origin_visit_add( | origin_visit2 = swh_storage.origin_visit_add( | ||||
| data.origin['url'], | data.origin['url'], | ||||
| date=data.date_visit3) | date=data.date_visit3, | ||||
| type=data.type_visit2, | |||||
| ) | |||||
| origin_visit3 = swh_storage.origin_visit_add( | origin_visit3 = swh_storage.origin_visit_add( | ||||
| data.origin['url'], | data.origin['url'], | ||||
| date=data.date_visit2) | date=data.date_visit2, | ||||
| type=data.type_visit3, | |||||
| ) | |||||
| # Simple case | # Simple case | ||||
| visit = swh_storage.origin_visit_find_by_date( | visit = swh_storage.origin_visit_find_by_date( | ||||
| data.origin['url'], data.date_visit3) | data.origin['url'], data.date_visit3) | ||||
| assert visit['visit'] == origin_visit2['visit'] | assert visit['visit'] == origin_visit2['visit'] | ||||
| # There are two visits at the same date, the latest must be returned | # There are two visits at the same date, the latest must be returned | ||||
| visit = swh_storage.origin_visit_find_by_date( | visit = swh_storage.origin_visit_find_by_date( | ||||
| data.origin['url'], data.date_visit2) | data.origin['url'], data.date_visit2) | ||||
| assert visit['visit'] == origin_visit3['visit'] | assert visit['visit'] == origin_visit3['visit'] | ||||
| def test_origin_visit_find_by_date__unknown_origin(self, swh_storage): | def test_origin_visit_find_by_date__unknown_origin(self, swh_storage): | ||||
| swh_storage.origin_visit_find_by_date('foo', data.date_visit2) | swh_storage.origin_visit_find_by_date('foo', data.date_visit2) | ||||
| @pytest.mark.parametrize('use_url', [True, False]) | @pytest.mark.parametrize('use_url', [True, False]) | ||||
| def test_origin_visit_update_missing_snapshot(self, swh_storage, use_url): | def test_origin_visit_update_missing_snapshot(self, swh_storage, use_url): | ||||
| if not self._test_origin_ids and not use_url: | if not self._test_origin_ids and not use_url: | ||||
| return | return | ||||
| # given | # given | ||||
| origin_id = swh_storage.origin_add_one(data.origin) | origin_id = swh_storage.origin_add_one(data.origin) | ||||
| origin_id_or_url = data.origin['url'] if use_url else origin_id | origin_id_or_url = data.origin['url'] if use_url else origin_id | ||||
| origin_visit = swh_storage.origin_visit_add( | origin_visit = swh_storage.origin_visit_add( | ||||
| origin_id_or_url, | origin_id_or_url, | ||||
| date=data.date_visit1) | date=data.date_visit1, | ||||
| type=data.type_visit1, | |||||
| ) | |||||
| # when | # when | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin_id_or_url, | origin_id_or_url, | ||||
| origin_visit['visit'], | origin_visit['visit'], | ||||
| snapshot=data.snapshot['id']) | snapshot=data.snapshot['id']) | ||||
| # then | # then | ||||
| Show All 13 Lines | def test_origin_visit_get_by(self, swh_storage, use_url): | ||||
| origin_id = swh_storage.origin_add_one(data.origin) | origin_id = swh_storage.origin_add_one(data.origin) | ||||
| origin_id2 = swh_storage.origin_add_one(data.origin2) | origin_id2 = swh_storage.origin_add_one(data.origin2) | ||||
| origin_id_or_url = data.origin['url'] if use_url else origin_id | origin_id_or_url = data.origin['url'] if use_url else origin_id | ||||
| origin2_id_or_url = data.origin2['url'] if use_url else origin_id2 | origin2_id_or_url = data.origin2['url'] if use_url else origin_id2 | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| origin_id_or_url, | origin_id_or_url, | ||||
| date=data.date_visit2) | date=data.date_visit2, | ||||
| type=data.type_visit2, | |||||
| ) | |||||
| swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin_id_or_url, | origin_id_or_url, | ||||
| origin_visit1['visit'], | origin_visit1['visit'], | ||||
| snapshot=data.snapshot['id']) | snapshot=data.snapshot['id']) | ||||
| # Add some other {origin, visit} entries | # Add some other {origin, visit} entries | ||||
| swh_storage.origin_visit_add( | swh_storage.origin_visit_add( | ||||
| origin_id_or_url, | origin_id_or_url, | ||||
| date=data.date_visit3) | date=data.date_visit3, | ||||
| type=data.type_visit3, | |||||
| ) | |||||
| swh_storage.origin_visit_add( | swh_storage.origin_visit_add( | ||||
| origin2_id_or_url, | origin2_id_or_url, | ||||
| date=data.date_visit3) | date=data.date_visit3, | ||||
| type=data.type_visit3, | |||||
| ) | |||||
| # when | # when | ||||
| visit1_metadata = { | visit1_metadata = { | ||||
| 'contents': 42, | 'contents': 42, | ||||
| 'directories': 22, | 'directories': 22, | ||||
| } | } | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin_id_or_url, | origin_id_or_url, | ||||
| origin_visit1['visit'], status='full', | origin_visit1['visit'], status='full', | ||||
| metadata=visit1_metadata) | metadata=visit1_metadata) | ||||
| expected_origin_visit = origin_visit1.copy() | expected_origin_visit = origin_visit1.copy() | ||||
| expected_origin_visit.update({ | expected_origin_visit.update({ | ||||
| 'origin': origin_id, | 'origin': origin_id, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'date': data.date_visit2, | 'date': data.date_visit2, | ||||
| 'type': data.origin['type'], | 'type': data.type_visit2, | ||||
| 'metadata': visit1_metadata, | 'metadata': visit1_metadata, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'snapshot': data.snapshot['id'], | 'snapshot': data.snapshot['id'], | ||||
| }) | }) | ||||
| # when | # when | ||||
| actual_origin_visit1 = swh_storage.origin_visit_get_by( | actual_origin_visit1 = swh_storage.origin_visit_get_by( | ||||
| origin_id_or_url, | origin_id_or_url, | ||||
| ▲ Show 20 Lines • Show All 93 Lines • ▼ Show 20 Lines | def test_origin_visit_upsert_existing(self, swh_storage, use_url): | ||||
| # given | # given | ||||
| origin_id = swh_storage.origin_add_one(data.origin2) | origin_id = swh_storage.origin_add_one(data.origin2) | ||||
| origin_url = data.origin2['url'] | origin_url = data.origin2['url'] | ||||
| assert origin_id is not None | assert origin_id is not None | ||||
| # 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, | |||||
| ) | |||||
| swh_storage.origin_visit_upsert([{ | swh_storage.origin_visit_upsert([{ | ||||
| 'origin': data.origin2, | 'origin': data.origin2, | ||||
| 'date': data.date_visit2, | 'date': data.date_visit2, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.origin2['type'], | 'type': data.type_visit1, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }]) | }]) | ||||
| # then | # then | ||||
| assert origin_visit1['origin'] == origin_id | assert origin_visit1['origin'] == origin_id | ||||
| assert origin_visit1['visit'] is not None | assert origin_visit1['visit'] is not None | ||||
| actual_origin_visits = list(swh_storage.origin_visit_get( | actual_origin_visits = list(swh_storage.origin_visit_get( | ||||
| origin_url)) | origin_url)) | ||||
| assert actual_origin_visits == [ | assert actual_origin_visits == [ | ||||
| { | { | ||||
| 'origin': origin_id, | 'origin': origin_id, | ||||
| 'date': data.date_visit2, | 'date': data.date_visit2, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.origin2['type'], | 'type': data.type_visit1, | ||||
| 'status': 'full', | 'status': 'full', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| }] | }] | ||||
| expected_origin = data.origin2 | expected_origin = data.origin2 | ||||
| data1 = { | data1 = { | ||||
| 'origin': expected_origin, | 'origin': expected_origin, | ||||
| 'date': data.date_visit2, | 'date': data.date_visit2, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.origin2['type'], | 'type': data.type_visit1, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data2 = { | data2 = { | ||||
| 'origin': expected_origin, | 'origin': expected_origin, | ||||
| 'date': data.date_visit2, | 'date': data.date_visit2, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.origin2['type'], | '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', expected_origin), | ||||
| ('origin_visit', data1), | ('origin_visit', data1), | ||||
| ('origin_visit', data2)] | ('origin_visit', data2)] | ||||
| Show All 11 Lines | |||||
| @pytest.mark.parametrize('use_url', [True, False]) | @pytest.mark.parametrize('use_url', [True, False]) | ||||
| def test_origin_visit_get_latest(self, swh_storage, use_url): | def test_origin_visit_get_latest(self, swh_storage, use_url): | ||||
| if not self._test_origin_ids and not use_url: | if not self._test_origin_ids and not use_url: | ||||
| return | return | ||||
| swh_storage.origin_add_one(data.origin) | swh_storage.origin_add_one(data.origin) | ||||
| origin_url = data.origin['url'] | origin_url = data.origin['url'] | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| origin_url, | origin=origin_url, | ||||
| data.date_visit1) | date=data.date_visit1, | ||||
| type=data.type_visit1, | |||||
| ) | |||||
| visit1_id = origin_visit1['visit'] | visit1_id = origin_visit1['visit'] | ||||
| origin_visit2 = swh_storage.origin_visit_add( | origin_visit2 = swh_storage.origin_visit_add( | ||||
| origin_url, | origin=origin_url, | ||||
| data.date_visit2) | date=data.date_visit2, | ||||
| type=data.type_visit2, | |||||
| ) | |||||
| visit2_id = origin_visit2['visit'] | visit2_id = origin_visit2['visit'] | ||||
| # Add a visit with the same date as the previous one | # Add a visit with the same date as the previous one | ||||
| origin_visit3 = swh_storage.origin_visit_add( | origin_visit3 = swh_storage.origin_visit_add( | ||||
| origin_url, | origin=origin_url, | ||||
| data.date_visit2) | date=data.date_visit2, | ||||
| type=data.type_visit2, | |||||
| ) | |||||
| visit3_id = origin_visit3['visit'] | visit3_id = origin_visit3['visit'] | ||||
| origin_visit1 = swh_storage.origin_visit_get_by(origin_url, visit1_id) | origin_visit1 = swh_storage.origin_visit_get_by(origin_url, visit1_id) | ||||
| origin_visit2 = swh_storage.origin_visit_get_by(origin_url, visit2_id) | origin_visit2 = swh_storage.origin_visit_get_by(origin_url, visit2_id) | ||||
| origin_visit3 = swh_storage.origin_visit_get_by(origin_url, visit3_id) | origin_visit3 = swh_storage.origin_visit_get_by(origin_url, visit3_id) | ||||
| # Two visits, both with no snapshot | # Two visits, both with no snapshot | ||||
| assert origin_visit3 == swh_storage.origin_visit_get_latest(origin_url) | assert origin_visit3 == swh_storage.origin_visit_get_latest(origin_url) | ||||
| ▲ Show 20 Lines • Show All 91 Lines • ▼ Show 20 Lines | def test_person_fullname_unicity(self, swh_storage): | ||||
| # then | # then | ||||
| # check committers are the same | # check committers are the same | ||||
| assert revisions[0]['committer'] == revisions[1]['committer'] | assert revisions[0]['committer'] == revisions[1]['committer'] | ||||
| def test_snapshot_add_get_empty(self, swh_storage): | def test_snapshot_add_get_empty(self, swh_storage): | ||||
| origin_id = swh_storage.origin_add_one(data.origin) | origin_id = swh_storage.origin_add_one(data.origin) | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| origin_id, data.date_visit1) | origin=origin_id, | ||||
| date=data.date_visit1, | |||||
| type=data.type_visit1, | |||||
| ) | |||||
| visit_id = origin_visit1['visit'] | visit_id = origin_visit1['visit'] | ||||
| actual_result = swh_storage.snapshot_add([data.empty_snapshot]) | actual_result = swh_storage.snapshot_add([data.empty_snapshot]) | ||||
| assert actual_result == {'snapshot:add': 1} | assert actual_result == {'snapshot:add': 1} | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin_id, visit_id, snapshot=data.empty_snapshot['id']) | origin_id, 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_id, visit_id) | by_ov = swh_storage.snapshot_get_by_origin_visit(origin_id, 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() | expected_origin = data.origin.copy() | ||||
| data1 = { | data1 = { | ||||
| 'origin': expected_origin, | 'origin': expected_origin, | ||||
| 'date': data.date_visit1, | 'date': data.date_visit1, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.origin['type'], | 'type': data.type_visit1, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data2 = { | data2 = { | ||||
| 'origin': expected_origin, | 'origin': expected_origin, | ||||
| 'date': data.date_visit1, | 'date': data.date_visit1, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.origin['type'], | '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', expected_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_id = swh_storage.origin_add_one(data.origin) | origin_id = swh_storage.origin_add_one(data.origin) | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| origin_id, data.date_visit1) | origin=origin_id, | ||||
| date=data.date_visit1, | |||||
| type=data.type_visit1, | |||||
| ) | |||||
| visit_id = origin_visit1['visit'] | visit_id = origin_visit1['visit'] | ||||
| actual_result = swh_storage.snapshot_add([data.complete_snapshot]) | actual_result = swh_storage.snapshot_add([data.complete_snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin_id, visit_id, snapshot=data.complete_snapshot['id']) | origin_id, visit_id, snapshot=data.complete_snapshot['id']) | ||||
| assert actual_result == {'snapshot:add': 1} | assert actual_result == {'snapshot:add': 1} | ||||
| by_id = swh_storage.snapshot_get(data.complete_snapshot['id']) | by_id = swh_storage.snapshot_get(data.complete_snapshot['id']) | ||||
| ▲ Show 20 Lines • Show All 123 Lines • ▼ Show 20 Lines | def test_snapshot_add_get_paginated(self, swh_storage): | ||||
| 'next_branch': branch_names[dir_idx + 3], | 'next_branch': branch_names[dir_idx + 3], | ||||
| } | } | ||||
| assert snapshot == expected_snapshot | assert snapshot == expected_snapshot | ||||
| def test_snapshot_add_get_filtered(self, swh_storage): | def test_snapshot_add_get_filtered(self, swh_storage): | ||||
| origin_id = swh_storage.origin_add_one(data.origin) | origin_id = swh_storage.origin_add_one(data.origin) | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| origin_id, data.date_visit1) | origin=origin_id, | ||||
| date=data.date_visit1, | |||||
| type=data.type_visit1, | |||||
| ) | |||||
| visit_id = origin_visit1['visit'] | visit_id = origin_visit1['visit'] | ||||
| swh_storage.snapshot_add([data.complete_snapshot]) | swh_storage.snapshot_add([data.complete_snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin_id, visit_id, snapshot=data.complete_snapshot['id']) | origin_id, visit_id, snapshot=data.complete_snapshot['id']) | ||||
| snp_id = data.complete_snapshot['id'] | snp_id = data.complete_snapshot['id'] | ||||
| branches = data.complete_snapshot['branches'] | branches = data.complete_snapshot['branches'] | ||||
| ▲ Show 20 Lines • Show All 98 Lines • ▼ Show 20 Lines | def test_snapshot_add_get_filtered_and_paginated(self, swh_storage): | ||||
| 'next_branch': b'release', | 'next_branch': b'release', | ||||
| } | } | ||||
| assert snapshot == expected_snapshot | assert snapshot == expected_snapshot | ||||
| def test_snapshot_add_get(self, swh_storage): | def test_snapshot_add_get(self, swh_storage): | ||||
| origin_id = swh_storage.origin_add_one(data.origin) | origin_id = swh_storage.origin_add_one(data.origin) | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| origin_id, data.date_visit1) | origin=origin_id, | ||||
| date=data.date_visit1, | |||||
| type=data.type_visit1, | |||||
| ) | |||||
| visit_id = origin_visit1['visit'] | visit_id = origin_visit1['visit'] | ||||
| swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin_id, visit_id, snapshot=data.snapshot['id']) | origin_id, visit_id, snapshot=data.snapshot['id']) | ||||
| by_id = swh_storage.snapshot_get(data.snapshot['id']) | by_id = swh_storage.snapshot_get(data.snapshot['id']) | ||||
| assert by_id == {**data.snapshot, 'next_branch': None} | assert by_id == {**data.snapshot, 'next_branch': None} | ||||
| Show All 18 Lines | def test_snapshot_add_nonexistent_visit(self, swh_storage): | ||||
| origin_id, visit_id, snapshot=data.snapshot['id']) | origin_id, visit_id, snapshot=data.snapshot['id']) | ||||
| assert list(swh_storage.journal_writer.objects) == [ | assert list(swh_storage.journal_writer.objects) == [ | ||||
| ('snapshot', data.snapshot)] | ('snapshot', data.snapshot)] | ||||
| def test_snapshot_add_twice__by_origin_visit(self, swh_storage): | def test_snapshot_add_twice__by_origin_visit(self, swh_storage): | ||||
| origin_id = swh_storage.origin_add_one(data.origin) | origin_id = swh_storage.origin_add_one(data.origin) | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| origin_id, data.date_visit1) | origin=origin_id, | ||||
| date=data.date_visit1, | |||||
| type=data.type_visit1, | |||||
| ) | |||||
| visit1_id = origin_visit1['visit'] | visit1_id = origin_visit1['visit'] | ||||
| swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin_id, visit1_id, snapshot=data.snapshot['id']) | origin_id, visit1_id, snapshot=data.snapshot['id']) | ||||
| by_ov1 = swh_storage.snapshot_get_by_origin_visit( | by_ov1 = swh_storage.snapshot_get_by_origin_visit( | ||||
| origin_id, visit1_id) | origin_id, visit1_id) | ||||
| assert by_ov1 == {**data.snapshot, 'next_branch': None} | assert by_ov1 == {**data.snapshot, 'next_branch': None} | ||||
| origin_visit2 = swh_storage.origin_visit_add( | origin_visit2 = swh_storage.origin_visit_add( | ||||
| origin_id, data.date_visit2) | origin=origin_id, | ||||
| date=data.date_visit2, | |||||
| type=data.type_visit2, | |||||
| ) | |||||
| visit2_id = origin_visit2['visit'] | visit2_id = origin_visit2['visit'] | ||||
| swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| origin_id, visit2_id, snapshot=data.snapshot['id']) | origin_id, 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_id, visit2_id) | origin_id, visit2_id) | ||||
| assert by_ov2 == {**data.snapshot, 'next_branch': None} | assert by_ov2 == {**data.snapshot, 'next_branch': None} | ||||
| expected_origin = data.origin.copy() | expected_origin = data.origin.copy() | ||||
| data1 = { | data1 = { | ||||
| 'origin': expected_origin, | 'origin': expected_origin, | ||||
| 'date': data.date_visit1, | 'date': data.date_visit1, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.origin['type'], | 'type': data.type_visit1, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data2 = { | data2 = { | ||||
| 'origin': expected_origin, | 'origin': expected_origin, | ||||
| 'date': data.date_visit1, | 'date': data.date_visit1, | ||||
| 'visit': origin_visit1['visit'], | 'visit': origin_visit1['visit'], | ||||
| 'type': data.origin['type'], | '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': expected_origin, | ||||
| 'date': data.date_visit2, | 'date': data.date_visit2, | ||||
| 'visit': origin_visit2['visit'], | 'visit': origin_visit2['visit'], | ||||
| 'type': data.origin['type'], | 'type': data.type_visit2, | ||||
| 'status': 'ongoing', | 'status': 'ongoing', | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| data4 = { | data4 = { | ||||
| 'origin': expected_origin, | 'origin': expected_origin, | ||||
| 'date': data.date_visit2, | 'date': data.date_visit2, | ||||
| 'visit': origin_visit2['visit'], | 'visit': origin_visit2['visit'], | ||||
| 'type': data.origin['type'], | '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', expected_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)] | ||||
| @pytest.mark.parametrize('use_url', [True, False]) | @pytest.mark.parametrize('use_url', [True, False]) | ||||
| def test_snapshot_get_latest(self, swh_storage, use_url): | def test_snapshot_get_latest(self, swh_storage, use_url): | ||||
| if not self._test_origin_ids and not use_url: | if not self._test_origin_ids and not use_url: | ||||
| return | return | ||||
| origin_id = swh_storage.origin_add_one(data.origin) | origin_id = swh_storage.origin_add_one(data.origin) | ||||
| origin_url = data.origin['url'] | origin_url = data.origin['url'] | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| origin_id, data.date_visit1) | origin=origin_id, | ||||
| date=data.date_visit1, | |||||
| type=data.type_visit1, | |||||
| ) | |||||
| visit1_id = origin_visit1['visit'] | visit1_id = origin_visit1['visit'] | ||||
| origin_visit2 = swh_storage.origin_visit_add( | origin_visit2 = swh_storage.origin_visit_add( | ||||
| origin_id, data.date_visit2) | origin=origin_id, | ||||
| date=data.date_visit2, | |||||
| type=data.type_visit2, | |||||
| ) | |||||
| visit2_id = origin_visit2['visit'] | visit2_id = origin_visit2['visit'] | ||||
| # Add a visit with the same date as the previous one | # Add a visit with the same date as the previous one | ||||
| origin_visit3 = swh_storage.origin_visit_add( | origin_visit3 = swh_storage.origin_visit_add( | ||||
| origin_id, data.date_visit2) | origin=origin_id, | ||||
| date=data.date_visit2, | |||||
| type=data.type_visit3, | |||||
| ) | |||||
| visit3_id = origin_visit3['visit'] | visit3_id = origin_visit3['visit'] | ||||
| # Two visits, both with no snapshot: latest snapshot is None | # Two visits, both with no snapshot: latest snapshot is None | ||||
| assert swh_storage.snapshot_get_latest(origin_url) is None | assert swh_storage.snapshot_get_latest(origin_url) is None | ||||
| # Add snapshot to visit1, latest snapshot = visit 1 snapshot | # Add snapshot to visit1, latest snapshot = visit 1 snapshot | ||||
| swh_storage.snapshot_add([data.complete_snapshot]) | swh_storage.snapshot_add([data.complete_snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| Show All 40 Lines | def test_snapshot_get_latest__missing_snapshot(self, swh_storage, use_url): | ||||
| if not self._test_origin_ids and not use_url: | if not self._test_origin_ids and not use_url: | ||||
| return | return | ||||
| # Origin does not exist | # Origin does not exist | ||||
| origin_url = data.origin['url'] | origin_url = data.origin['url'] | ||||
| assert swh_storage.snapshot_get_latest(origin_url) is None | assert swh_storage.snapshot_get_latest(origin_url) is None | ||||
| 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( | ||||
| origin_url, | origin=origin_url, | ||||
| data.date_visit1) | date=data.date_visit1, | ||||
| type=data.type_visit1, | |||||
| ) | |||||
| visit1_id = origin_visit1['visit'] | visit1_id = origin_visit1['visit'] | ||||
| origin_visit2 = swh_storage.origin_visit_add( | origin_visit2 = swh_storage.origin_visit_add( | ||||
| origin_url, | origin=origin_url, | ||||
| data.date_visit2) | date=data.date_visit2, | ||||
| type=data.type_visit2, | |||||
| ) | |||||
| visit2_id = origin_visit2['visit'] | visit2_id = origin_visit2['visit'] | ||||
| # Two visits, both with no snapshot: latest snapshot is None | # Two visits, both with no snapshot: latest snapshot is None | ||||
| assert swh_storage.snapshot_get_latest(origin_url) is None | assert swh_storage.snapshot_get_latest(origin_url) is None | ||||
| # Add unknown snapshot to visit1, check that the inconsistency is | # Add unknown snapshot to visit1, check that the inconsistency is | ||||
| # detected | # detected | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| ▲ Show 20 Lines • Show All 61 Lines • ▼ Show 20 Lines | def test_stat_counters(self, swh_storage): | ||||
| if key != 'content': | if key != 'content': | ||||
| assert counters[key] == 0 | assert counters[key] == 0 | ||||
| assert counters['content'] == 1 | assert counters['content'] == 1 | ||||
| # Add other objects. Check their counter increased as well. | # Add other objects. Check their counter increased as well. | ||||
| swh_storage.origin_add_one(data.origin2) | swh_storage.origin_add_one(data.origin2) | ||||
| origin_visit1 = swh_storage.origin_visit_add( | origin_visit1 = swh_storage.origin_visit_add( | ||||
| data.origin2['url'], date=data.date_visit2) | origin=data.origin2['url'], | ||||
| date=data.date_visit2, | |||||
| type=data.type_visit2, | |||||
| ) | |||||
| swh_storage.snapshot_add([data.snapshot]) | swh_storage.snapshot_add([data.snapshot]) | ||||
| swh_storage.origin_visit_update( | swh_storage.origin_visit_update( | ||||
| data.origin2['url'], origin_visit1['visit'], | data.origin2['url'], origin_visit1['visit'], | ||||
| snapshot=data.snapshot['id']) | snapshot=data.snapshot['id']) | ||||
| swh_storage.directory_add([data.dir]) | swh_storage.directory_add([data.dir]) | ||||
| swh_storage.revision_add([data.revision]) | swh_storage.revision_add([data.revision]) | ||||
| swh_storage.release_add([data.release]) | swh_storage.release_add([data.release]) | ||||
| ▲ Show 20 Lines • Show All 1,110 Lines • Show Last 20 Lines | |||||