Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 1,652 Lines • ▼ Show 20 Lines | def test_origin_visit_add_default_type(self, swh_storage): | ||||
| '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, | ||||
| }, | }, | ||||
| ] | ] | ||||
| assert len(expected_visits) == len(actual_origin_visits) | |||||
| 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.journal.objects) | objects = list(swh_storage.journal_writer.journal.objects) | ||||
| assert ('origin', data.origin2) in objects | assert ('origin', data.origin2) in objects | ||||
| for visit in expected_visits: | for visit in expected_visits: | ||||
| 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(StorageArgumentException) as cm: | with pytest.raises(StorageArgumentException) as cm: | ||||
| swh_storage.origin_visit_add(origin_url, date=[b'foo'], type='git') | swh_storage.origin_visit_add(origin_url, date=[b'foo'], type='git') | ||||
| 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 | ||||
| def test_origin_visit_update(self, swh_storage): | def test_origin_visit_update(self, swh_storage): | ||||
| # given | # given | ||||
| origin_url = swh_storage.origin_add_one(data.origin) | origin_url = swh_storage.origin_add_one(data.origin) | ||||
| origin_url2 = swh_storage.origin_add_one(data.origin2) | origin_url2 = swh_storage.origin_add_one(data.origin2) | ||||
| date_visit = datetime.datetime.now(datetime.timezone.utc) | date_visit = data.date_visit1 | ||||
| date_visit2 = date_visit + datetime.timedelta(minutes=1) | date_visit2 = data.date_visit2 | ||||
| # Round to milliseconds before insertion, so equality doesn't fail | # Round to milliseconds before insertion, so equality doesn't fail | ||||
| # after a round-trip through a DB (eg. Cassandra) | # after a round-trip through a DB (eg. Cassandra) | ||||
| date_visit = date_visit.replace( | date_visit = date_visit.replace( | ||||
| microsecond=round(date_visit.microsecond, -3)) | microsecond=round(date_visit.microsecond, -3)) | ||||
| date_visit2 = date_visit2.replace( | date_visit2 = date_visit2.replace( | ||||
| microsecond=round(date_visit2.microsecond, -3)) | microsecond=round(date_visit2.microsecond, -3)) | ||||
| ▲ Show 20 Lines • Show All 366 Lines • ▼ Show 20 Lines | def test_origin_visit_upsert_existing(self, swh_storage): | ||||
| 'metadata': None, | 'metadata': None, | ||||
| 'snapshot': None, | 'snapshot': None, | ||||
| } | } | ||||
| assert list(swh_storage.journal_writer.journal.objects) == [ | assert list(swh_storage.journal_writer.journal.objects) == [ | ||||
| ('origin', data.origin2), | ('origin', data.origin2), | ||||
| ('origin_visit', data1), | ('origin_visit', data1), | ||||
| ('origin_visit', data2)] | ('origin_visit', data2)] | ||||
| def test_origin_visit_upsert_missing_visit_id(self, swh_storage): | |||||
| # given | |||||
| origin_url = swh_storage.origin_add_one(data.origin2) | |||||
| # then | |||||
| with pytest.raises(StorageArgumentException, match='Missing visit id'): | |||||
| swh_storage.origin_visit_upsert([OriginVisit.from_dict({ | |||||
| 'origin': origin_url, | |||||
| 'date': data.date_visit2, | |||||
| 'visit': None, # <- make the test raise | |||||
| 'type': data.type_visit1, | |||||
| 'status': 'full', | |||||
| 'metadata': None, | |||||
| 'snapshot': None, | |||||
| })]) | |||||
| assert list(swh_storage.journal_writer.journal.objects) == [ | |||||
| ('origin', data.origin2) | |||||
| ] | |||||
| 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 | ||||
| def test_origin_visit_get_latest(self, swh_storage): | def test_origin_visit_get_latest(self, swh_storage): | ||||
| origin_url = swh_storage.origin_add_one(data.origin) | origin_url = swh_storage.origin_add_one(data.origin) | ||||
| ▲ Show 20 Lines • Show All 1,814 Lines • Show Last 20 Lines | |||||