I have a class

class ReferenceData(celery.Task):
    def __init__(self):
        redis_pool = redis.ConnectionPool()
        self.redis = redis.StrictRedis(connection_pool=redis_pool)

    def get_agents(self,id):
        if condition:
            return self.agents.get(id)
        return self.agents.get(external_id

and there is celery task which calls up the following method

def save_raisr_event(self, event):
    # Get attributes:
    transcriptions = []
    if trans is not None:
       transcriptions.append(trans)
        self.redis.append(f"{crn}-transcription", ' '   transcription)
        return 'transcription saved'

In the unit testing how to mock the redis connection and mock data from redis in this case the value pf transcription saved in redis

CodePudding user response：

Let's say this is the source code:

src.py

import celery
import redis


class ReferenceData(celery.Task):
    def __init__(self):
        redis_pool = redis.ConnectionPool()
        self.redis = redis.StrictRedis(connection_pool=redis_pool)

    def save_raisr_event(self, event):
        self.redis.append("crn", 'transcription')
        return self.redis.keys()

If you just want to patch the response of the methods within redis with static values:

test_src.py

from src import ReferenceData


def test_redis(mocker):  # Requires <pip install pytest-mock>
    mock_redis_pkg = mocker.patch("src.redis")
    mock_redis = mock_redis_pkg.StrictRedis.return_value

    mocked_keys = ["some", "transactions", "here"]
    mock_redis.keys.return_value = mocked_keys
    assert ReferenceData().save_raisr_event(None) == mocked_keys

If you want to store the actual transactions appended, then setup a dict that will be updated:

test_src.py

from src import ReferenceData


def test_redis(mocker):  # Requires <pip install pytest-mock>
    mock_redis_pkg = mocker.patch("src.redis")
    mock_redis = mock_redis_pkg.StrictRedis.return_value

    storage = {}
    mock_redis.append.side_effect = lambda key, value: storage.update({key: value})
    mock_redis.keys.side_effect = lambda: storage.keys()
    assert ReferenceData().save_raisr_event(None) == storage.keys()
    assert storage == {'crn': 'transcription'}