Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_init.py
| Show All 18 Lines | |||||
| STORAGES = [ | STORAGES = [ | ||||
| pytest.param(cls, real_class, kwargs, id=cls) | pytest.param(cls, real_class, kwargs, id=cls) | ||||
| for (cls, real_class, kwargs) in [ | for (cls, real_class, kwargs) in [ | ||||
| ("remote", client.RemoteStorage, {"url": "url"}), | ("remote", client.RemoteStorage, {"url": "url"}), | ||||
| ("memory", InMemoryStorage, {}), | ("memory", InMemoryStorage, {}), | ||||
| ( | ( | ||||
| "local", | "local", | ||||
| DbStorage, | DbStorage, | ||||
| {"db": "postgresql://db", "objstorage": {"cls": "memory", "args": {}}}, | {"db": "postgresql://db", "objstorage": {"cls": "memory"}}, | ||||
| ), | ), | ||||
| ("filter", FilteringProxyStorage, {"storage": {"cls": "memory"}}), | ("filter", FilteringProxyStorage, {"storage": {"cls": "memory"}}), | ||||
| ("buffer", BufferingProxyStorage, {"storage": {"cls": "memory"}}), | ("buffer", BufferingProxyStorage, {"storage": {"cls": "memory"}}), | ||||
| ("retry", RetryingProxyStorage, {"storage": {"cls": "memory"}}), | ("retry", RetryingProxyStorage, {"storage": {"cls": "memory"}}), | ||||
| ] | ] | ||||
| ] | ] | ||||
| Show All 23 Lines | with pytest.warns(DeprecationWarning): | ||||
| assert isinstance(actual_storage, real_class) | assert isinstance(actual_storage, real_class) | ||||
| def test_get_storage_failure(): | def test_get_storage_failure(): | ||||
| """Instantiating an unknown storage should raise | """Instantiating an unknown storage should raise | ||||
| """ | """ | ||||
| with pytest.raises(ValueError, match="Unknown storage class `unknown`"): | with pytest.raises(ValueError, match="Unknown storage class `unknown`"): | ||||
| get_storage("unknown", args=[]) | get_storage("unknown") | ||||
| def test_get_storage_pipeline(): | def test_get_storage_pipeline(): | ||||
| config = { | config = { | ||||
| "cls": "pipeline", | "cls": "pipeline", | ||||
| "steps": [ | "steps": [ | ||||
| {"cls": "filter",}, | {"cls": "filter",}, | ||||
| {"cls": "buffer", "min_batch_size": {"content": 10,},}, | {"cls": "buffer", "min_batch_size": {"content": 10,},}, | ||||
| ▲ Show 20 Lines • Show All 42 Lines • ▼ Show 20 Lines | |||||
| @patch("swh.storage.postgresql.storage.psycopg2.pool") | @patch("swh.storage.postgresql.storage.psycopg2.pool") | ||||
| def test_get_storage_local_check_config(mock_pool, monkeypatch): | def test_get_storage_local_check_config(mock_pool, monkeypatch): | ||||
| """Instantiating a local storage with check_config should be ok | """Instantiating a local storage with check_config should be ok | ||||
| """ | """ | ||||
| mock_pool.ThreadedConnectionPool.return_value = None | mock_pool.ThreadedConnectionPool.return_value = None | ||||
| check_backend_check_config( | check_backend_check_config( | ||||
| monkeypatch, | monkeypatch, | ||||
| { | {"cls": "local", "db": "postgresql://db", "objstorage": {"cls": "memory"}}, | ||||
| "cls": "local", | |||||
| "db": "postgresql://db", | |||||
| "objstorage": {"cls": "memory", "args": {}}, | |||||
| }, | |||||
| backend_storage_cls=DbStorage, | backend_storage_cls=DbStorage, | ||||
| ) | ) | ||||
| def test_get_storage_pipeline_check_config(monkeypatch): | def test_get_storage_pipeline_check_config(monkeypatch): | ||||
| """Test that the check_config option works as intended for a pipelined storage""" | """Test that the check_config option works as intended for a pipelined storage""" | ||||
| config = { | config = { | ||||
| "cls": "pipeline", | "cls": "pipeline", | ||||
| ▲ Show 20 Lines • Show All 98 Lines • Show Last 20 Lines | |||||