Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 28 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,34 @@ Or:

@patch('redis.StrictRedis', mock_strict_redis_client)


### Time expiration

For time expiration management, a custom clock has to be implemented to overwrite
the default internal clock. The default internal clock only uses `datetime.now()`
and won't be able to simulate time jump. A simple clock implementation with the
ability to add or substract time can be found below. Then, instead of using
`mock_redis_client` or `mock_strict_redis_client`, use
`MockRedis(clock=CustomClock())` or `MockRedis(strict=True, clock=CustomClock())`.
The clock is then accessible through `redis.client`.

```python
from datetime import datetime, timedelta
from mockredis import clock


class CustomClock(clock.Clock):
def __init__(self):
self.timeout = 0

def add_timeout(self, timeout):
self.timeout += timeout

def now(self):
return datetime.now() + timedelta(seconds=self.timeout)
```


## Testing

Many unit tests exist to verify correctness of mock functionality. In addition, most
Expand Down
7 changes: 4 additions & 3 deletions mockredis/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,8 +30,9 @@ class MockRedis(object):
A Mock for a redis-py Redis object

Expire functionality must be explicitly
invoked using do_expire(time). Automatic
expiry is NOT supported.
invoked using do_expire(). Automatic
expiry is NOT supported. Please refer to
the README on how to manage time expiration.
"""

def __init__(self,
Expand Down Expand Up @@ -257,7 +258,7 @@ def pttl(self, key):

def do_expire(self):
"""
Expire objects assuming now == time
Expire objects regarding the internal clock 'now()' value
"""
# Deep copy to avoid RuntimeError: dictionary changed size during iteration
_timeouts = deepcopy(self.timeouts)
Expand Down