Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_api_client.py
Show First 20 Lines • Show All 60 Lines • ▼ Show 20 Lines | def setUp(self): | ||||
super().setUp() | super().setUp() | ||||
self.storage = RemoteStorage(self.url()) | self.storage = RemoteStorage(self.url()) | ||||
def tearDown(self): | def tearDown(self): | ||||
super().tearDown() | super().tearDown() | ||||
shutil.rmtree(self.storage_base) | shutil.rmtree(self.storage_base) | ||||
storage.get_journal_writer = get_journal_writer | storage.get_journal_writer = get_journal_writer | ||||
def reset_storage(self): | |||||
excluded = {'dbversion', 'tool'} | |||||
self.reset_db_tables(self.TEST_DB_NAME, excluded=excluded) | |||||
self.journal_writer.objects[:] = [] | |||||
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': 'inmemory', | 'cls': 'inmemory', | ||||
} | } | ||||
} | } | ||||
} | } | ||||
} | } | ||||
self.__storage = InMemoryStorage(journal_writer={'cls': 'inmemory'}) | self.__storage = InMemoryStorage( | ||||
journal_writer={'cls': 'inmemory'}) | |||||
self._get_storage_patcher = unittest.mock.patch( | self._get_storage_patcher = unittest.mock.patch( | ||||
'swh.storage.api.server.get_storage', return_value=self.__storage) | 'swh.storage.api.server.get_storage', return_value=self.__storage) | ||||
self._get_storage_patcher.start() | self._get_storage_patcher.start() | ||||
self.app = app | self.app = app | ||||
super().setUp() | super().setUp() | ||||
self.storage = RemoteStorage(self.url()) | self.storage = RemoteStorage(self.url()) | ||||
self.journal_writer = self.__storage.journal_writer | self.journal_writer = self.__storage.journal_writer | ||||
def tearDown(self): | def tearDown(self): | ||||
super().tearDown() | super().tearDown() | ||||
self._get_storage_patcher.stop() | self._get_storage_patcher.stop() | ||||
def reset_storage(self): | |||||
self.storage.reset() | |||||
self.journal_writer.objects[:] = [] | |||||
@pytest.mark.network | @pytest.mark.network | ||||
class TestRemoteMemStorage(CommonTestStorage, RemoteMemStorageFixture): | class TestRemoteMemStorage(CommonTestStorage, 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') | ||||
Show All 31 Lines |