MCPcopy Create free account
hub / github.com/apify/crawlee-python / storage_client

Function storage_client

tests/unit/storages/conftest.py:23–44  ·  view source on GitHub ↗

Parameterized fixture to test with different storage clients.

(
    request: pytest.FixtureRequest,
    redis_client: FakeAsyncRedis,
)

Source from the content-addressed store, hash-verified

21
22@pytest.fixture(params=['memory', 'file_system', 'sql', 'redis'])
23async def storage_client(
24 request: pytest.FixtureRequest,
25 redis_client: FakeAsyncRedis,
26) -> AsyncGenerator[StorageClient, None]:
27 """Parameterized fixture to test with different storage clients."""
28 storage_client: StorageClient
29
30 storage_type = request.param
31
32 if storage_type == 'memory':
33 storage_client = MemoryStorageClient()
34 elif storage_type == 'sql':
35 storage_client = SqlStorageClient()
36 elif storage_type == 'redis':
37 storage_client = RedisStorageClient(redis=redis_client)
38 else:
39 storage_client = FileSystemStorageClient()
40 service_locator.set_storage_client(storage_client)
41 yield storage_client
42
43 if isinstance(storage_client, SqlStorageClient):
44 await storage_client.close()

Callers

nothing calls this directly

Calls 6

MemoryStorageClientClass · 0.90
SqlStorageClientClass · 0.90
RedisStorageClientClass · 0.90
set_storage_clientMethod · 0.80
closeMethod · 0.45

Tested by

no test coverage detected