Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_init.py
| # Copyright (C) 2019 The Software Heritage developers | # Copyright (C) 2019 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import pytest | import pytest | ||||
| from unittest.mock import patch | from unittest.mock import patch | ||||
| from swh.storage import get_storage | from swh.storage import get_storage | ||||
| from swh.storage.api.client import RemoteStorage | from swh.storage.api.client import RemoteStorage | ||||
| from swh.storage.storage import Storage as DbStorage | from swh.storage.storage import Storage as DbStorage | ||||
| from swh.storage.in_memory import Storage as MemoryStorage | from swh.storage.in_memory import InMemoryStorage | ||||
| from swh.storage.buffer import BufferingProxyStorage | from swh.storage.buffer import BufferingProxyStorage | ||||
| from swh.storage.filter import FilteringProxyStorage | from swh.storage.filter import FilteringProxyStorage | ||||
| from swh.storage.retry import RetryingProxyStorage | from swh.storage.retry import RetryingProxyStorage | ||||
| @patch('swh.storage.storage.psycopg2.pool') | @patch('swh.storage.storage.psycopg2.pool') | ||||
| def test_get_storage(mock_pool): | def test_get_storage(mock_pool): | ||||
| """Instantiating an existing storage should be ok | """Instantiating an existing storage should be ok | ||||
| """ | """ | ||||
| mock_pool.ThreadedConnectionPool.return_value = None | mock_pool.ThreadedConnectionPool.return_value = None | ||||
| for cls, real_class, dummy_args in [ | for cls, real_class, dummy_args in [ | ||||
| ('remote', RemoteStorage, {'url': 'url'}), | ('remote', RemoteStorage, {'url': 'url'}), | ||||
| ('memory', MemoryStorage, {}), | ('memory', InMemoryStorage, {}), | ||||
| ('local', DbStorage, { | ('local', DbStorage, { | ||||
| 'db': 'postgresql://db', 'objstorage': { | 'db': 'postgresql://db', 'objstorage': { | ||||
| 'cls': 'memory', 'args': {}, | 'cls': 'memory', 'args': {}, | ||||
| }, | }, | ||||
| }), | }), | ||||
| ('filter', FilteringProxyStorage, {'storage': { | ('filter', FilteringProxyStorage, {'storage': { | ||||
| 'cls': 'memory'} | 'cls': 'memory'} | ||||
| }), | }), | ||||
| Show All 13 Lines | |||||
| def test_get_storage_legacy_args(mock_pool): | def test_get_storage_legacy_args(mock_pool): | ||||
| """Instantiating an existing storage should be ok even with the legacy | """Instantiating an existing storage should be ok even with the legacy | ||||
| explicit 'args' keys | explicit 'args' keys | ||||
| """ | """ | ||||
| mock_pool.ThreadedConnectionPool.return_value = None | mock_pool.ThreadedConnectionPool.return_value = None | ||||
| for cls, real_class, dummy_args in [ | for cls, real_class, dummy_args in [ | ||||
| ('remote', RemoteStorage, {'url': 'url'}), | ('remote', RemoteStorage, {'url': 'url'}), | ||||
| ('memory', MemoryStorage, {}), | ('memory', InMemoryStorage, {}), | ||||
| ('local', DbStorage, { | ('local', DbStorage, { | ||||
| 'db': 'postgresql://db', 'objstorage': { | 'db': 'postgresql://db', 'objstorage': { | ||||
| 'cls': 'memory', 'args': {}, | 'cls': 'memory', 'args': {}, | ||||
| }, | }, | ||||
| }), | }), | ||||
| ('filter', FilteringProxyStorage, {'storage': { | ('filter', FilteringProxyStorage, {'storage': { | ||||
| 'cls': 'memory', 'args': {}} | 'cls': 'memory', 'args': {}} | ||||
| }), | }), | ||||
| Show All 33 Lines | config = { | ||||
| } | } | ||||
| ] | ] | ||||
| } | } | ||||
| storage = get_storage(**config) | storage = get_storage(**config) | ||||
| assert isinstance(storage, FilteringProxyStorage) | assert isinstance(storage, FilteringProxyStorage) | ||||
| assert isinstance(storage.storage, BufferingProxyStorage) | assert isinstance(storage.storage, BufferingProxyStorage) | ||||
| assert isinstance(storage.storage.storage, MemoryStorage) | assert isinstance(storage.storage.storage, InMemoryStorage) | ||||
| def test_get_storage_pipeline_legacy_args(): | def test_get_storage_pipeline_legacy_args(): | ||||
| config = { | config = { | ||||
| 'cls': 'pipeline', | 'cls': 'pipeline', | ||||
| 'steps': [ | 'steps': [ | ||||
| { | { | ||||
| 'cls': 'filter', | 'cls': 'filter', | ||||
| Show All 12 Lines | config = { | ||||
| ] | ] | ||||
| } | } | ||||
| with pytest.warns(DeprecationWarning): | with pytest.warns(DeprecationWarning): | ||||
| storage = get_storage(**config) | storage = get_storage(**config) | ||||
| assert isinstance(storage, FilteringProxyStorage) | assert isinstance(storage, FilteringProxyStorage) | ||||
| assert isinstance(storage.storage, BufferingProxyStorage) | assert isinstance(storage.storage, BufferingProxyStorage) | ||||
| assert isinstance(storage.storage.storage, MemoryStorage) | assert isinstance(storage.storage.storage, InMemoryStorage) | ||||