Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
Show First 20 Lines • Show All 1,351 Lines • ▼ Show 20 Lines | def test_origin_add_twice(self, swh_storage): | ||||
add2 = swh_storage.origin_add([data.origin, data.origin2]) | add2 = swh_storage.origin_add([data.origin, data.origin2]) | ||||
assert list(swh_storage.journal_writer.journal.objects) \ | assert list(swh_storage.journal_writer.journal.objects) \ | ||||
== [('origin', data.origin), | == [('origin', data.origin), | ||||
('origin', data.origin2)] | ('origin', data.origin2)] | ||||
assert add1 == add2 | assert add1 == add2 | ||||
def test_origin_add_validation(self, swh_storage): | |||||
with pytest.raises(StorageArgumentException, match='url'): | |||||
swh_storage.origin_add([{'type': 'git'}]) | |||||
def test_origin_get_legacy(self, swh_storage): | def test_origin_get_legacy(self, swh_storage): | ||||
assert swh_storage.origin_get(data.origin) is None | assert swh_storage.origin_get(data.origin) is None | ||||
swh_storage.origin_add_one(data.origin) | swh_storage.origin_add_one(data.origin) | ||||
actual_origin0 = swh_storage.origin_get( | actual_origin0 = swh_storage.origin_get( | ||||
{'url': data.origin['url']}) | {'url': data.origin['url']}) | ||||
assert actual_origin0['url'] == data.origin['url'] | assert actual_origin0['url'] == data.origin['url'] | ||||
▲ Show 20 Lines • Show All 297 Lines • ▼ Show 20 Lines | def test_origin_visit_add_validation(self, swh_storage): | ||||
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 2,167 Lines • Show Last 20 Lines |