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 | |||||