Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_api_client.py
| Show All 12 Lines | |||||
| from swh.core.api.tests.server_testing import ServerTestFixture | from swh.core.api.tests.server_testing import ServerTestFixture | ||||
| from swh.journal.writer import get_journal_writer | from swh.journal.writer import get_journal_writer | ||||
| from swh.storage.api.client import RemoteStorage | from swh.storage.api.client import RemoteStorage | ||||
| import swh.storage.api.server as server | import swh.storage.api.server as server | ||||
| from swh.storage.api.server import app | from swh.storage.api.server import app | ||||
| from swh.storage.in_memory import Storage as InMemoryStorage | from swh.storage.in_memory import Storage as InMemoryStorage | ||||
| import swh.storage.storage | import swh.storage.storage | ||||
| from swh.storage.db import Db | from swh.storage.db import Db | ||||
| from swh.storage.tests.test_storage import \ | from swh.storage.tests.test_storage import ( | ||||
| CommonTestStorage, CommonPropTestStorage, StorageTestDbFixture | TestStorage as _TestStorage, | ||||
| TestStorageCommonProp as _TestStorageCommonProp, | |||||
| StorageTestDbFixture) | |||||
| @pytest.mark.xfail | |||||
| class RemotePgStorageFixture(StorageTestDbFixture, ServerTestFixture, | class RemotePgStorageFixture(StorageTestDbFixture, ServerTestFixture, | ||||
| unittest.TestCase): | unittest.TestCase): | ||||
| def setUp(self): | def setUp(self): | ||||
| journal_writer = get_journal_writer(cls='memory') | journal_writer = get_journal_writer(cls='memory') | ||||
| def mock_get_journal_writer(cls, args=None): | def mock_get_journal_writer(cls, args=None): | ||||
| assert cls == 'memory' | assert cls == 'memory' | ||||
| return journal_writer | return journal_writer | ||||
| ▲ Show 20 Lines • Show All 42 Lines • ▼ Show 20 Lines | def reset_storage(self): | ||||
| self.reset_db_tables(self.TEST_DB_NAME, excluded=excluded) | self.reset_db_tables(self.TEST_DB_NAME, excluded=excluded) | ||||
| self.journal_writer.objects[:] = [] | self.journal_writer.objects[:] = [] | ||||
| @contextmanager | @contextmanager | ||||
| def get_db(self): | def get_db(self): | ||||
| yield Db(self.conn) | yield Db(self.conn) | ||||
| @pytest.mark.xfail | |||||
| class RemoteMemStorageFixture(ServerTestFixture, unittest.TestCase): | class RemoteMemStorageFixture(ServerTestFixture, unittest.TestCase): | ||||
| def setUp(self): | def setUp(self): | ||||
| self.config = { | self.config = { | ||||
| 'storage': { | 'storage': { | ||||
| 'cls': 'memory', | 'cls': 'memory', | ||||
| 'args': { | 'args': { | ||||
| 'journal_writer': { | 'journal_writer': { | ||||
| 'cls': 'memory', | 'cls': 'memory', | ||||
| Show All 16 Lines | def tearDown(self): | ||||
| super().tearDown() | super().tearDown() | ||||
| self._get_storage_patcher.stop() | self._get_storage_patcher.stop() | ||||
| def reset_storage(self): | def reset_storage(self): | ||||
| self.storage.reset() | self.storage.reset() | ||||
| self.journal_writer.objects[:] = [] | self.journal_writer.objects[:] = [] | ||||
| @pytest.mark.xfail | |||||
| @pytest.mark.network | @pytest.mark.network | ||||
| class TestRemoteMemStorage(CommonTestStorage, RemoteMemStorageFixture): | class TestRemoteMemStorage(_TestStorage, RemoteMemStorageFixture): | ||||
| @pytest.mark.skip('refresh_stat_counters not available in the remote api.') | @pytest.mark.skip('refresh_stat_counters not available in the remote api.') | ||||
| def test_stat_counters(self): | def test_stat_counters(self): | ||||
| pass | pass | ||||
| @pytest.mark.skip('postgresql-specific test') | @pytest.mark.skip('postgresql-specific test') | ||||
| def test_content_add_db(self): | def test_content_add_db(self): | ||||
| pass | pass | ||||
| @pytest.mark.skip('postgresql-specific test') | @pytest.mark.skip('postgresql-specific test') | ||||
| def test_skipped_content_add_db(self): | def test_skipped_content_add_db(self): | ||||
| pass | pass | ||||
| @pytest.mark.skip('postgresql-specific test') | @pytest.mark.skip('postgresql-specific test') | ||||
| def test_content_add_metadata_db(self): | def test_content_add_metadata_db(self): | ||||
| pass | pass | ||||
| @pytest.mark.skip( | @pytest.mark.skip( | ||||
| 'not implemented, see https://forge.softwareheritage.org/T1633') | 'not implemented, see https://forge.softwareheritage.org/T1633') | ||||
| def test_skipped_content_add(self): | def test_skipped_content_add(self): | ||||
| pass | pass | ||||
| @pytest.mark.xfail | |||||
| @pytest.mark.db | @pytest.mark.db | ||||
| @pytest.mark.network | @pytest.mark.network | ||||
| class TestRemotePgStorage(CommonTestStorage, RemotePgStorageFixture): | class TestRemotePgStorage(_TestStorage, RemotePgStorageFixture): | ||||
| @pytest.mark.skip('refresh_stat_counters not available in the remote api.') | @pytest.mark.skip('refresh_stat_counters not available in the remote api.') | ||||
| def test_stat_counters(self): | def test_stat_counters(self): | ||||
| pass | pass | ||||
| @pytest.mark.xfail | |||||
| @pytest.mark.db | @pytest.mark.db | ||||
| @pytest.mark.property_based | @pytest.mark.property_based | ||||
| class PropTestRemotePgStorage(CommonPropTestStorage, RemotePgStorageFixture): | class PropTestRemotePgStorage(_TestStorageCommonProp, RemotePgStorageFixture): | ||||
| @pytest.mark.skip('too slow') | @pytest.mark.skip('too slow') | ||||
| def test_add_arbitrary(self): | def test_add_arbitrary(self): | ||||
| pass | pass | ||||