Redis

Users can test Redis dependent code using the create_redis_fixture.

pytest_mock_resources.create_redis_fixture(scope='function')

Produce a Redis fixture.

Any number of fixture functions can be created. Under the hood they will all share the same database server.

Note

If running tests in parallel, the implementation fans out to different redis “database”s, up to a 16 (which is the default container fixed limit). This means you can only run up to 16 simultaneous tests.

Additionally, any calls to flushall or any other cross-database calls will still represent cross-test state.

Finally, the above notes are purely describing the current implementation, and should not be assumed. In the future, the current database selection mechanism may change, or databases may not be used altogether.

Parameters

scope (str) – The scope of the fixture can be specified by the user, defaults to “function”.

Raises

KeyError – If any additional arguments are provided to the function than what is necessary.

Consider the following example:

# src/some_module.py

def insert_into_friends(redis_client):
    redis_client.sadd("friends:leto", "ghanima")
    redis_client.sadd("friends:leto", "duncan")
    redis_client.sadd("friends:paul", "duncan")
    redis_client.sadd("friends:paul", "gurney")

A user can test this as follows:

# tests/some_test.py

from pytest_mock_resources import create_redis_fixture
from some_module import insert_into_friends

redis = create_redis_fixture()

def test_insert_into_friends(redis):
    insert_into_friends(redis)

    friends_leto = redis.smembers("friends:leto")
    friends_paul = redis.smembers("friends:paul")

    assert friends_leto == {b"duncan", b"ghanima"}
    assert friends_paul == {b"gurney", b"duncan"}

Manual Engine Creation

Engines can be created manually via the fixture’s yielded attributes/REDIS_* fixtures:

# tests/some_test.py

from redis import Redis

from pytest_mock_resources import create_redis_fixture

redis = create_redis_fixture()


def test_create_custom_connection(redis):
    client = Redis(**redis.pmr_credentials.as_redis_kwargs())
    client.set("foo", "bar")
    client.append("foo", "baz")
    value = client.get("foo").decode("utf-8")
    assert value == "barbaz"