Question 277584 in Redis Question posted in
The official documentation can be found here.

writing test for redis.asyncio using fakeredis

I want to test the functionality of action_base_result using fakeredis-py to create tests without a real Redis server.
and don’t want to add a third parameter to action_base_result:

import redis.asyncio as redis
import json
import copy

redis_pool = None

async def create_redis_pool():
    global redis_pool
    redis_pool = redis.from_url("redis://", decode_responses=True)

async def action_base_result(strategy, result):
    async with redis_pool.pipeline(transaction=True) as pipe:
        if result is False:
            pipe.rpush('met_name_actions', json.dumps(['discard', strategy.met_name]))
            pipe.delete(strategy.met_name)
        elif result is True:
            pipe.rpush('met_name_actions', json.dumps(['discard', strategy.prev_met_name]))
            pipe.rpush('met_name_actions', json.dumps(['add', strategy.met_name]))
            pipe.set(strategy.met_name, json.dumps(copy.deepcopy(strategy.__dict__)))
        await pipe.execute()

I tried somethings like this:

from unittest.mock import MagicMock

import pytest
from main import action_base_result
from fakeredis import aioredis

redis_pool = aioredis.FakeRedis()

@pytest.mark.asyncio
async def test_action_base_result():
    strategy = MagicMock()
    strategy.met_name = "met_name"
    strategy.prev_met_name = "prev_met_name"

    await action_base_result(strategy, False)

and I got:

>       with redis_pool.pipeline(transaction=True) as pipe:
E       AttributeError: 'NoneType' object has no attribute 'pipeline'
Tags:

2

Answers


  1. 0

    It seems that you are using the wrong module for fakeredis. You should import fakeredis instead of fakeredis.aioredis, which is a different module that does not support redis-py commands.

    Also, you need to create a redis pool before using it in your action_base_result function. You can use the create_redis_pool function that you have defined, or use the from_url method of fakeredis.FakeStrictRedis class

    I hope this helps you with your testing.

    Login or Signup to reply.
  2. 0

    maintainer of fakeredis here.
    This looks a bit strange, can you share the entire stack trace?
    I suspect the issue comes from the way you are running the test. the line where the FakeRedis is initiated is not executed.

    If you create a fixture, it should work. See examples here

    @pytest_asyncio.fixture()
async def aioredis(request) -> redis.asyncio.Redis:
     return aioredis.FakeRedis() 


@pytest.mark.asyncio
async def test_action_base_result(aioredis: redis.asyncio.Redis):
    strategy = MagicMock()
    strategy.met_name = "met_name"
    strategy.prev_met_name = "prev_met_name"

    await action_base_result(strategy, False)
    Login or Signup to reply.
Please signup or login to give your own answer.

Back To Top
Search