Mocking Redis in Python's unittest

Hello folks,

This blog will guide you to mock redis without using any new library. Redis is used as cache in almost every application. It's very likely you will be required to mock redis at some point of writing those testcases. Scanning through solutions available on internet, I felt the need of way to mock redis should be documented in a blog.

So working with the saying 😂,

fake it before you make it
mock it before you rock it

Let's start 💁

Back to Basics ⚡️

In this section let us have a refresher course on patch, mock, side_effect and return_value.

1. Mock

A typical piece of code consists of is objects, function calls and variables. While writing tests we don't want to actual objects/methods/class-methods to execute, so we can replace it with a mock object or mock call.

2. Patch

We now know that we can mock objects and functional calls but how we can establish for which function we have to associate a particular mock with?
Here patch comes into play, patch is a decorator which accepts fully qualified name of method to be mocked as string.

@patch("app.function")
def test_01_another_function(self, mock_function):
    pass

Here we have patched function method which will automatically send a positional argument to the function you're decorating. Usually this position argument is an instance of MagicMock.

3. Return Value

While writing tests we require mock method or mock class method to return a particular value. We can add the expected return value in return_value attribute of MagicMock instance.

Suppose we have a RandomObject class where we have to mock method function.

from unittest.mock import patch, MagicMock
from app.module import RandomObject

@patch("app.module.RandomObject")
def test_01_another_function(self, mock_random):
    mock_random.function.return_value = "test-value"
    random_object = RandomObject()
    self.assertEqual(random_object.function(), "test-value")

4. Side Effect

This is typically used to test if Exceptions are handled correctly in the code. When the patched function is called, the exception mentioned in side_effect is raised.

from unittest.mock import patch, MagicMock
from app.module import RandomObject

@patch("app.module.RandomObject")
def test_01_another_function(self, mock_random):
    mock_random.function.side_effect = Exception("test-message")
    random_object = RandomObject()
    self.assertRaises(random_object.function(), Exception)

Leveraging side_effect ☄️

Another way to use side_effect is we can pass a list of possible values which we want to bind with side effect attribute. Each time the patched function is called the mock will return next element in the list of values. Also we can have any set of data type (not specifically Exceptions).

from unittest.mock import patch, MagicMock

mock = MagicMock()
side_effect_list = ["dummy_val", {"dummy":"value"}]  # list of values on which we want to be returned.
mock.side_effect = side_effect_list

mock("test")
>>> 'dummy_val'
mock("test")
>>> {'dummy': 'value'}

To leverage side_effect even further, we can even bind side_effect attribute with a method.

from unittest.mock import patch, MagicMock

foo_dict = {"foo": "bar", "bar": "foo"}
def foo_function(key):  # function which we need to bind with side effect
    return foo_dict[key]

mock = MagicMock()
mock.side_effect = foo_function
mock("foo")
>>> 'bar'
mock("bar")
>>> 'foo'

Mocking Redis 🔘

Now let's discuss how we can now use above to mock redis. Let us for now consider 5 most common redis methods:

get

set

hset

hget

exists

Since redis is a key-value data store, we can use dictionary for caching these key-value pairs. We can then define above methods in a MockRedis class.

class MockRedis:
    def __init__(self, cache=dict()):
        self.cache = cache

Now let us write function that will mimic the get functionality. The get method will simply take a key and return its value.

def get(self, key):
        if key in self.cache:
            return self.cache[key]
        return None  # return nil

def set(self, key, value, *args, **kwargs):
        if self.cache:
           self.cache[key] = value
           return "OK"
        return None  # return nil in case of some issue

Similarly let us implement hset, hget and exists in the class MockRedis.

class MockRedis:
    def __init__(self, cache=dict()):
        self.cache = cache

    def get(self, key):
        if key in self.cache:
            return self.cache[key]
        return None  # return nil

    def set(self, key, value, *args, **kwargs):
        if self.cache:
           self.cache[key] = value
           return "OK"
        return None  # return nil in case of some issue

    def hget(self, hash, key):
        if hash in self.cache:
            if key in self.cache[hash]:
                return self.cache[hash][key]
        return None  # return nil

    def hset(self, hash, key, value, *args, **kwargs):
        if self.cache:
           self.cache[hash][key] = value
           return 1
        return None  # return nil in case of some issue

    def exists(self, key):
        if key in self.cache:
            return 1
        return 0

    def cache_overwrite(self, cache=dict()):
        self.cache = cache

mock_redis_method.py

So now let us mock redis now, for that we have to patch StrictRedis.

from mock_redis_method import MockRedis
from unittest.mock import patch, MagicMock

@patch("redis.StrictRedis")
def test_01_redis(self, mock_redis):
    # initialising the cache with test values 
    redis_cache = {
        "foo": "bar", 
        "foobar": {"Foo": "Bar"}
    }

    mock_redis_obj = MockRedis(redis_cache)

    # binding a side_effect of a MagicMock instance with redis methods we defined in the MockRedis class.
    mock_redis_method = MagicMock()
    mock_redis_method.hget = Mock(side_effect=mock_redis_obj.get)
    mock_redis_method.hget = Mock(side_effect=mock_redis_obj.hget)
    mock_redis_method.set = Mock(side_effect=mock_redis_obj.set)
    mock_redis_method.hset = Mock(side_effect=mock_redis_obj.hset)
    mock_redis_method.exists = Mock(side_effect=mock_redis_obj.exists)

    # StrictRedis mock return_values is set as above mock_redis_method.
    mock_redis.return_value = mock_redis_method

Voila! it's done 🍸. We have successfully mocked redis.

Bonus Content ✅

We can similarly mock requests library

@patch("requests.get")
@patch("requests.post")
def test_02_external_api_calls(self, *mocks):
    # request and response are mapped in a dict
    request_response_dict = {
        "https://dummy-host?key=foo":  (
            {"key": "foo"}, 200
        ),
        "https://dummy-host?key=bar": (
            {"message": "Not Found"}, 404
        )
    }

    class MockResponse:
        def __init__(self, json_data, status_code):
            self.json_data = json_data
            self.status_code = status_code

        # request.json()
        def json(self):
            return self.json_data

    def mock_request_method(url, *args, **kwargs):
        response = request_response_dict[url]
        return MockResponse(response[0], response[1])

    # get and post method return similar kind of response. 
    mocks[0].side_effect = mock_request_method
    mocks[1].side_effect = mock_request_method