0
# pytest-redis
1

2
Redis fixtures and fixture factories for Pytest. This plugin provides three main fixtures for testing code that relies on Redis databases: `redisdb` (a Redis client fixture that constructs a client and cleans the database after tests), `redis_proc` (a session-scoped fixture that starts/stops Redis instances), and `redis_nooproc` (for connecting to already running Redis servers).
3

4
## Package Information
5

6
- **Package Name**: pytest-redis
7
- **Language**: Python
8
- **Installation**: `pip install pytest-redis`
9
- **Plugin Type**: pytest plugin (automatically registers fixtures)
10

11
## Core Imports
12

13
For creating custom fixtures:
14

15
```python
16
from pytest_redis import factories
17
```
18

19
Individual factory imports:
20

21
```python
22
from pytest_redis.factories import redis_proc, redis_noproc, redisdb
23
```
24

25
Import built-in fixtures (automatically available, no explicit import needed):
26

27
```python
28
# These are automatically registered as pytest fixtures when pytest-redis is installed
29
# Available fixtures: redis_proc, redis_nooproc, redisdb
30
```
31

32
Import internal components (for advanced usage):
33

34
```python
35
from pytest_redis.config import get_config, RedisConfigType
36
from pytest_redis.executor import RedisExecutor, NoopRedis
37
from pytest_redis.exception import RedisUnsupported, RedisMisconfigured, UnixSocketTooLong
38
```
39

40
Note: Factory function naming: `redis_noproc()` (factory function) creates fixtures named ending in `_nooproc` (the actual fixture), while the built-in fixture is named `redis_nooproc`.
41

42
## Type Definitions
43

44
```python { .api }
45
from typing import Callable, Generator, List, Optional, Set, Tuple, TypedDict, Union
46
from pathlib import Path
47
from _pytest.fixtures import FixtureRequest
48
from _pytest.tmpdir import TempPathFactory
49
from packaging.version import Version
50
import redis
51

52
class RedisConfigType(TypedDict):
53
    """Pytest redis config definition type."""
54
    host: str
55
    port: Optional[int]
56
    username: str
57
    password: str
58
    exec: str
59
    timeout: int
60
    loglevel: str
61
    db_count: int
62
    save: str
63
    compression: bool
64
    rdbchecksum: bool
65
    syslog: bool
66
    decode: bool
67
    datadir: str
68
    modules: List[str]
69
```
70

71
## Basic Usage
72

73
### Using Built-in Fixtures
74

75
The plugin automatically provides three fixtures when installed:
76

77
```python
78
def test_redis(redisdb):
79
    """Test using the built-in Redis client fixture."""
80
    # The redisdb fixture provides a Redis client and cleans up after the test
81
    redisdb.set('test_key', 'test_value')
82
    
83
    # Your code that uses Redis
84
    my_component = MyRedisBasedComponent()
85
    my_component.do_something()
86
    
87
    # Verify results
88
    assert redisdb.get('result_key') == b'expected_value'
89
    # Database is automatically cleaned after this test
90
```
91

92
### Creating Custom Fixtures
93

94
```python
95
from pytest_redis import factories
96

97
# Create custom Redis process and client fixtures
98
redis_my_proc = factories.redis_proc(port=None)  # Random port
99
redis_my = factories.redisdb('redis_my_proc')
100

101
def test_custom_redis(redis_my):
102
    """Test using custom Redis fixtures."""
103
    redis_my.set('custom_key', 'custom_value')
104
    assert redis_my.get('custom_key') == b'custom_value'
105
```
106

107
### Connecting to Existing Redis
108

109
```python
110
def test_existing_redis(redis_nooproc):
111
    """Test connecting to an already running Redis server."""
112
    # Connects to Redis server at configured host/port (default: 127.0.0.1:6379)
113
    # Does not start/stop Redis process - expects Redis to already be running
114
    redis_nooproc.set('key', 'value')
115
    assert redis_nooproc.get('key') == b'value'
116
```
117

118
## Capabilities
119

120
### Built-in Fixtures
121

122
Three fixtures are automatically available when pytest-redis is installed:
123

124
```python { .api }
125
# Automatically registered fixtures (no import needed)
126
def redis_proc(request):
127
    """
128
    Session-scoped fixture that starts/stops Redis server process.
129
    
130
    Returns:
131
        RedisExecutor: Redis server process executor
132
    """
133

134
def redis_nooproc(request):
135
    """
136
    Session-scoped fixture for connecting to existing Redis server.
137
    
138
    Returns:
139
        NoopRedis: Connection to existing Redis server
140
    """
141

142
def redisdb(request):
143
    """
144
    Function-scoped Redis client fixture with automatic cleanup.
145
    
146
    Returns:
147
        redis.Redis: Redis client instance
148
    """
149
```
150

151
### Fixture Factories
152

153
Create custom fixtures with specific configurations:
154

155
```python { .api }
156
def redis_proc(
157
    executable: Optional[str] = None,
158
    timeout: Optional[int] = None,
159
    host: Optional[str] = None,
160
    port: Union[
161
        None,
162
        str,
163
        int,
164
        Tuple[int, int],
165
        Set[int],
166
        List[str],
167
        List[int],
168
        List[Tuple[int, int]],
169
        List[Set[int]],
170
        List[Union[Set[int], Tuple[int, int]]],
171
        List[Union[str, int, Tuple[int, int], Set[int]]],
172
    ] = -1,
173
    username: Optional[str] = None,
174
    password: Optional[str] = None,
175
    db_count: Optional[int] = None,
176
    save: Optional[str] = None,
177
    compression: Optional[bool] = None,
178
    checksum: Optional[bool] = None,
179
    syslog: Optional[bool] = None,
180
    loglevel: Optional[str] = None,
181
    datadir: Optional[str] = None,
182
    modules: Optional[List[str]] = None,
183
) -> Callable[[FixtureRequest, TempPathFactory], Generator[RedisExecutor, None, None]]:
184
    """
185
    Factory for creating Redis process fixtures.
186
    
187
    Parameters:
188
    - executable: Path to redis-server executable
189
    - timeout: Client connection timeout in seconds
190
    - host: Hostname for Redis server (default: 127.0.0.1)
191
    - port: Port configuration - exact port, random port (None), port ranges, or sets
192
    - username: Authentication username
193
    - password: Authentication password  
194
    - db_count: Number of Redis databases (default: 8)
195
    - save: Redis persistence configuration (seconds keys format)
196
    - compression: Enable dump file compression
197
    - checksum: Add checksum to RDB files
198
    - syslog: Enable system logger
199
    - loglevel: Log verbosity level (notice, warning, verbose, debug)
200
    - datadir: Directory for Redis data files
201
    - modules: List of Redis extension module paths to load
202
    
203
    Returns:
204
        Callable: Function that creates Redis process fixture
205
    """
206

207
def redis_noproc(
208
    host: Optional[str] = None,
209
    port: Optional[int] = None,
210
    username: Optional[str] = None,
211
    password: Optional[str] = None,
212
    startup_timeout: int = 15,
213
) -> Callable[[FixtureRequest], Generator[NoopRedis, None, None]]:
214
    """
215
    Factory for creating no-process Redis fixtures (connects to existing server).
216
    
217
    Parameters:
218
    - host: Redis server hostname
219
    - port: Redis server port
220
    - username: Authentication username
221
    - password: Authentication password
222
    - startup_timeout: Connection timeout in seconds
223
    
224
    Returns:
225
        Callable: Function that creates no-process Redis fixture
226
    """
227

228
def redisdb(
229
    process_fixture_name: str,
230
    dbnum: int = 0,
231
    decode: Optional[bool] = None,
232
) -> Callable[[FixtureRequest], Generator[redis.Redis, None, None]]:
233
    """
234
    Factory for creating Redis client fixtures.
235
    
236
    Parameters:
237
    - process_fixture_name: Name of process fixture to use for connection
238
    - dbnum: Redis database number to use (default: 0)
239
    - decode: Whether to decode Redis responses to strings
240
    
241
    Returns:
242
        Callable: Function that creates Redis client fixture
243
    """
244
```
245

246
### Configuration
247

248
pytest-redis can be configured via command line options or pytest.ini settings:
249

250
```python { .api }
251
def pytest_addoption(parser: Parser) -> None:
252
    """
253
    Register pytest command line options and ini settings.
254
    
255
    Command line options:
256
    --redis-exec: Redis server executable path
257
    --redis-host: Redis server hostname
258
    --redis-port: Redis server port
259
    --redis-username: Authentication username
260
    --redis-password: Authentication password
261
    --redis-timeout: Client connection timeout
262
    --redis-loglevel: Redis log verbosity level
263
    --redis-db-count: Number of Redis databases
264
    --redis-save: Redis persistence configuration
265
    --redis-compression: Enable dump file compression
266
    --redis-rdbchecksum: Add checksum to RDB files
267
    --redis-syslog: Enable system logger
268
    --redis-client-decode: Decode Redis responses (note: different from ini name)
269
    --redis-datadir: Redis data directory
270
    --redis-modules: Redis extension modules (comma-separated)
271
    
272
    Corresponding ini settings:
273
    redis_exec, redis_host, redis_port, redis_username, redis_password,
274
    redis_timeout, redis_loglevel, redis_db_count, redis_save,
275
    redis_compression, redis_rdbchecksum, redis_syslog, redis_decode,
276
    redis_datadir, redis_modules
277
    """
278
```
279

280
### Executor Classes
281

282
Internal classes that manage Redis processes:
283

284
```python { .api }
285
class RedisExecutor:
286
    """
287
    Redis server process executor.
288
    
289
    Attributes:
290
        host: Redis server hostname
291
        port: Redis server port
292
        username: Authentication username
293
        password: Authentication password
294
        unixsocket: Unix socket path for Redis server
295
        executable: Path to redis-server executable
296
    
297
    Inherits from mirakuru.TCPExecutor for process management.
298
    """
299
    
300
    MIN_SUPPORTED_VERSION: Version = parse("2.6")
301
    
302
    def __init__(
303
        self,
304
        executable: str,
305
        databases: int,
306
        redis_timeout: int,
307
        loglevel: str,
308
        host: str,
309
        port: int,
310
        username: Optional[str] = None,
311
        password: Optional[str] = None,
312
        startup_timeout: int = 60,
313
        save: str = "",
314
        daemonize: str = "no",
315
        rdbcompression: bool = True,
316
        rdbchecksum: bool = False,
317
        syslog_enabled: bool = False,
318
        appendonly: str = "no",
319
        datadir: Optional[Path] = None,
320
        modules: Optional[List[str]] = None,
321
    ) -> None:
322
        """Initialize RedisExecutor."""
323
    
324
    def start(self) -> "RedisExecutor":
325
        """Check supported version before starting."""
326
    
327
    @property
328
    def version(self) -> Version:
329
        """Return redis version."""
330

331
class NoopRedis:
332
    """
333
    No-operation Redis executor for existing servers.
334
    
335
    Attributes:
336
        host: Redis server hostname
337
        port: Redis server port
338
        username: Authentication username
339
        password: Authentication password
340
        unixsocket: Unix socket path (typically None for noop)
341
        timeout: Connection timeout
342
    
343
    Connects to existing Redis instance without managing process lifecycle.
344
    """
345
    
346
    def __init__(
347
        self,
348
        host: str,
349
        port: int,
350
        username: Optional[str] = None,
351
        password: Optional[str] = None,
352
        unixsocket: Optional[str] = None,
353
        startup_timeout: int = 15,
354
    ) -> None:
355
        """Initialize NoopRedis."""
356
    
357
    def start(self) -> "NoopRedis":
358
        """Start is a NOOP - waits for Redis to be available."""
359
    
360
    def redis_available(self) -> bool:
361
        """Return True if connecting to Redis is possible."""
362
```
363

364
### Configuration Helper
365

366
```python { .api }
367
def get_config(request: FixtureRequest) -> RedisConfigType:
368
    """
369
    Extract Redis configuration from pytest options and ini settings.
370
    
371
    Parameters:
372
        request: pytest FixtureRequest object
373
    
374
    Returns:
375
        RedisConfigType: Dictionary containing Redis configuration
376
    """
377
```
378

379
### Exceptions
380

381
Custom exceptions for Redis-specific issues:
382

383
```python { .api }
384
class RedisUnsupported(Exception):
385
    """Raised when Redis version < 2.6 is detected."""
386

387
class RedisMisconfigured(Exception):
388
    """Raised when redis_exec points to non-existing file."""
389

390
class UnixSocketTooLong(Exception):
391
    """Raised when Unix socket path exceeds maximum length."""
392
```
393

394
## Configuration Examples
395

396
### pytest.ini Configuration
397

398
```ini
399
[tool:pytest]
400
redis_host = 127.0.0.1
401
redis_port = 6379
402
redis_timeout = 30
403
redis_loglevel = notice
404
redis_db_count = 16
405
redis_decode = false
406
```
407

408
### Custom Fixture Examples
409

410
```python
411
from pytest_redis import factories
412

413
# Custom Redis with specific port and authentication  
414
redis_proc2 = factories.redis_proc(port=6381)
415
redis_proc3 = factories.redis_proc(port=6385, password="secretpassword")
416

417
# Create client fixtures for the custom processes
418
redisdb2 = factories.redisdb('redis_proc2')
419
redisdb3 = factories.redisdb('redis_proc3')
420

421
# No-process fixtures connecting to existing Redis instances
422
redis_nooproc2 = factories.redis_noproc(port=6381, startup_timeout=1)
423
redis_nooproc3 = factories.redis_noproc(port=6385, password="secretpassword")
424
redisdb2_noop = factories.redisdb('redis_nooproc2')
425
redisdb3_noop = factories.redisdb('redis_nooproc3')
426

427
# Redis with extension modules and custom data directory
428
module_redis_proc = factories.redis_proc(
429
    modules=["/path/to/redis-module.so"],
430
    datadir="/tmp/redis-test-data"
431
)
432
module_redis = factories.redisdb('module_redis_proc')
433
```
434

435
## Advanced Usage Patterns
436

437
### Multiple Redis Instances
438

439
```python
440
from pytest_redis import factories
441

442
# Primary Redis for main data
443
redis_main_proc = factories.redis_proc(port=6379)
444
redis_main = factories.redisdb('redis_main_proc')
445

446
# Secondary Redis for caching
447
redis_cache_proc = factories.redis_proc(port=6380)
448
redis_cache = factories.redisdb('redis_cache_proc', dbnum=1)
449

450
def test_multi_redis(redis_main, redis_cache):
451
    """Test using multiple Redis instances."""
452
    redis_main.set('data_key', 'main_data')
453
    redis_cache.set('cache_key', 'cached_data')
454
    
455
    assert redis_main.get('data_key') == b'main_data'
456
    assert redis_cache.get('cache_key') == b'cached_data'
457
    
458
    # Each Redis instance maintains separate data
459
    assert redis_main.get('cache_key') is None
460
    assert redis_cache.get('data_key') is None
461
```
462

463
### Testing with Database Isolation
464

465
```python
466
def test_database_isolation(redisdb):
467
    """Each test gets a clean Redis database."""
468
    # Database is empty at start of each test
469
    assert redisdb.dbsize() == 0
470
    
471
    # Make changes
472
    redisdb.set('test_key', 'test_value')
473
    redisdb.lpush('test_list', 'item1', 'item2')
474
    
475
    # Changes are visible within the test
476
    assert redisdb.get('test_key') == b'test_value'
477
    assert redisdb.llen('test_list') == 2
478
    
479
    # Database is automatically cleaned after this test
480
```