Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/storage_tests.py
Show All 11 Lines | |||||
import random | import random | ||||
from typing import Any, ClassVar, Dict, Iterator, Optional | from typing import Any, ClassVar, Dict, Iterator, Optional | ||||
from unittest.mock import MagicMock | from unittest.mock import MagicMock | ||||
import attr | import attr | ||||
from hypothesis import HealthCheck, given, settings, strategies | from hypothesis import HealthCheck, given, settings, strategies | ||||
import pytest | import pytest | ||||
from swh.core.api import RemoteException | |||||
from swh.core.api.classes import stream_results | from swh.core.api.classes import stream_results | ||||
from swh.model import from_disk | from swh.model import from_disk | ||||
from swh.model.hashutil import DEFAULT_ALGORITHMS, hash_to_bytes | from swh.model.hashutil import DEFAULT_ALGORITHMS, hash_to_bytes | ||||
from swh.model.hypothesis_strategies import objects | from swh.model.hypothesis_strategies import objects | ||||
from swh.model.identifiers import CoreSWHID, ObjectType | from swh.model.identifiers import CoreSWHID, ObjectType | ||||
from swh.model.model import ( | from swh.model.model import ( | ||||
Content, | Content, | ||||
Directory, | Directory, | ||||
▲ Show 20 Lines • Show All 1,317 Lines • ▼ Show 20 Lines | def test_extid_version_behavior(self, swh_storage, sample_data): | ||||
if revision.type.value == "git" | if revision.type.value == "git" | ||||
] | ] | ||||
# Insert extids with several different versions | # Insert extids with several different versions | ||||
extids = [ | extids = [ | ||||
ExtID( | ExtID( | ||||
extid=extid, | extid=extid, | ||||
extid_type="git", | extid_type="git", | ||||
extid_version=0, | |||||
target=CoreSWHID(object_id=extid, object_type=ObjectType.REVISION,), | target=CoreSWHID(object_id=extid, object_type=ObjectType.REVISION,), | ||||
) | ) | ||||
for extid in ids | for extid in ids | ||||
] + [ | ] + [ | ||||
ExtID( | ExtID( | ||||
extid=extid, | extid=extid, | ||||
extid_type="git", | extid_type="git", | ||||
extid_version=1, | extid_version=1, | ||||
target=CoreSWHID(object_id=extid, object_type=ObjectType.REVISION,), | target=CoreSWHID(object_id=extid, object_type=ObjectType.REVISION,), | ||||
) | ) | ||||
for extid in ids | for extid in ids | ||||
] | ] | ||||
swh_storage.extid_add(extids) | swh_storage.extid_add(extids) | ||||
# Check that both versions get returned | # Check that both versions get returned | ||||
for git_id in ids: | for git_id in ids: | ||||
objs = swh_storage.extid_get_from_extid("git", [git_id]) | objs = swh_storage.extid_get_from_extid("git", [git_id]) | ||||
assert len(objs) == 2 | assert len(objs) == 2 | ||||
assert set(obj.extid_version for obj in objs) == {0, 1} | assert set(obj.extid_version for obj in objs) == {0, 1} | ||||
for swhid in ids: | for swhid in ids: | ||||
objs = swh_storage.extid_get_from_target(ObjectType.REVISION, [swhid]) | objs = swh_storage.extid_get_from_target(ObjectType.REVISION, [swhid]) | ||||
assert len(objs) == 2 | assert len(objs) == 2 | ||||
assert set(obj.extid_version for obj in objs) == {0, 1} | assert set(obj.extid_version for obj in objs) == {0, 1} | ||||
for version in [0, 1]: | |||||
for git_id in ids: | |||||
objs = swh_storage.extid_get_from_extid( | |||||
"git", [git_id], version=version | |||||
) | |||||
assert len(objs) == 1 | |||||
assert objs[0].extid_version == version | |||||
for swhid in ids: | |||||
objs = swh_storage.extid_get_from_target( | |||||
ObjectType.REVISION, | |||||
[swhid], | |||||
extid_version=version, | |||||
extid_type="git", | |||||
) | |||||
assert len(objs) == 1 | |||||
assert objs[0].extid_version == version | |||||
assert objs[0].extid_type == "git" | |||||
def test_extid_version_behavior_failure(self, swh_storage, sample_data): | |||||
"""Calls with wrong input should raise""" | |||||
ids = [ | |||||
revision.id | |||||
for revision in sample_data.revisions | |||||
if revision.type.value == "git" | |||||
] | |||||
# Other edge cases | |||||
with pytest.raises( | |||||
(ValueError, RemoteException), match="both extid_type and extid_version" | |||||
): | |||||
swh_storage.extid_get_from_target( | |||||
ObjectType.REVISION, [ids[0]], extid_version=0 | |||||
) | |||||
with pytest.raises( | |||||
(ValueError, RemoteException), match="both extid_type and extid_version" | |||||
): | |||||
swh_storage.extid_get_from_target( | |||||
ObjectType.REVISION, [ids[0]], extid_type="git" | |||||
) | |||||
def test_release_add(self, swh_storage, sample_data): | def test_release_add(self, swh_storage, sample_data): | ||||
release, release2 = sample_data.releases[:2] | release, release2 = sample_data.releases[:2] | ||||
init_missing = swh_storage.release_missing([release.id, release2.id]) | init_missing = swh_storage.release_missing([release.id, release2.id]) | ||||
assert list(init_missing) == [release.id, release2.id] | assert list(init_missing) == [release.id, release2.id] | ||||
actual_result = swh_storage.release_add([release, release2]) | actual_result = swh_storage.release_add([release, release2]) | ||||
▲ Show 20 Lines • Show All 3,270 Lines • Show Last 20 Lines |