Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
Show All 11 Lines | |||||
from unittest.mock import Mock, patch | from unittest.mock import Mock, patch | ||||
import pytest | import pytest | ||||
from hypothesis import given, strategies | from hypothesis import given, strategies | ||||
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.hypothesis_strategies import origins | from swh.model.hypothesis_strategies import origins, objects | ||||
from swh.storage.tests.storage_testing import StorageTestFixture | from swh.storage.tests.storage_testing import StorageTestFixture | ||||
from swh.storage import HashCollision | from swh.storage import HashCollision | ||||
from .generate_data_test import gen_contents | from .generate_data_test import gen_contents | ||||
@pytest.mark.db | @pytest.mark.db | ||||
class StorageTestDbFixture(StorageTestFixture): | class StorageTestDbFixture(StorageTestFixture): | ||||
▲ Show 20 Lines • Show All 3,044 Lines • ▼ Show 20 Lines | def test_origin_count(self): | ||||
self.storage.origin_count('.*user.*', regexp=True), 5) | self.storage.origin_count('.*user.*', regexp=True), 5) | ||||
self.assertEqual( | self.assertEqual( | ||||
self.storage.origin_count('.*user.*', regexp=False), 0) | self.storage.origin_count('.*user.*', regexp=False), 0) | ||||
self.assertEqual( | self.assertEqual( | ||||
self.storage.origin_count('.*user1.*', regexp=True), 2) | self.storage.origin_count('.*user1.*', regexp=True), 2) | ||||
self.assertEqual( | self.assertEqual( | ||||
self.storage.origin_count('.*user1.*', regexp=False), 0) | self.storage.origin_count('.*user1.*', regexp=False), 0) | ||||
@given(strategies.lists(objects(), max_size=2)) | |||||
def test_add_arbitrary(self, objects): | |||||
self.reset_storage_tables() | |||||
for (obj_type, obj) in objects: | |||||
obj = obj.to_dict() | |||||
if obj_type == 'origin_visit': | |||||
origin_id = self.storage.origin_add_one(obj.pop('origin')) | |||||
if 'visit' in obj: | |||||
del obj['visit'] | |||||
self.storage.origin_visit_add(origin_id, **obj) | |||||
else: | |||||
method = getattr(self.storage, obj_type + '_add') | |||||
try: | |||||
method([obj]) | |||||
except HashCollision: | |||||
pass | |||||
@pytest.mark.db | @pytest.mark.db | ||||
class TestLocalStorage(CommonTestStorage, StorageTestDbFixture, | class TestLocalStorage(CommonTestStorage, StorageTestDbFixture, | ||||
unittest.TestCase): | unittest.TestCase): | ||||
"""Test the local storage""" | """Test the local storage""" | ||||
# Can only be tested with local storage as you can't mock | # Can only be tested with local storage as you can't mock | ||||
# datetimes for the remote server | # datetimes for the remote server | ||||
▲ Show 20 Lines • Show All 107 Lines • Show Last 20 Lines |