Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
Show All 20 Lines | |||||
import pytest | import pytest | ||||
from hypothesis import given, strategies, settings, HealthCheck | from hypothesis import given, strategies, settings, HealthCheck | ||||
from typing import ClassVar, Optional | from typing import ClassVar, Optional | ||||
from swh.model import from_disk, identifiers | from swh.model import from_disk, identifiers | ||||
from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
from swh.model.model import Content, Release, Revision | from swh.model.model import Content, OriginVisit, Release, Revision | ||||
from swh.model.hypothesis_strategies import objects | from swh.model.hypothesis_strategies import objects | ||||
from swh.storage import HashCollision, get_storage | from swh.storage import HashCollision, get_storage | ||||
from swh.storage.converters import origin_url_to_sha1 as sha1 | from swh.storage.converters import origin_url_to_sha1 as sha1 | ||||
from swh.storage.exc import StorageArgumentException | from swh.storage.exc import StorageArgumentException | ||||
from swh.storage.interface import StorageInterface | from swh.storage.interface import StorageInterface | ||||
from .storage_data import data | from .storage_data import data | ||||
▲ Show 20 Lines • Show All 1,943 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([ | ||||
{ | OriginVisit.from_dict({ | ||||
'origin': origin_url, | '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, | ||||
}, | }), | ||||
{ | OriginVisit.from_dict({ | ||||
'origin': origin_url, | '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, | ||||
}, | }), | ||||
]) | ]) | ||||
# then | # then | ||||
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_url, | 'origin': origin_url, | ||||
▲ Show 20 Lines • Show All 44 Lines • ▼ Show 20 Lines | def test_origin_visit_upsert_existing(self, swh_storage): | ||||
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([OriginVisit.from_dict({ | ||||
'origin': origin_url, | '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, | ||||
}]) | })]) | ||||
# then | # then | ||||
assert origin_visit1['origin'] == origin_url | assert origin_visit1['origin'] == origin_url | ||||
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 == [ | ||||
▲ Show 20 Lines • Show All 1,891 Lines • Show Last 20 Lines |