Differential D6249 Diff 22725 swh/storage/tests/storage_tests.py

Changeset 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