conftest.py

importasyncio
importrandom

importpytest

fromaredis_omimportget_redis_connection


TEST_PREFIX="redis-om:testing"


py_test_mark_asyncio=pytest.mark.asyncio


# "pytest_mark_sync" causes problem in pytest
defpy_test_mark_sync(f):
returnf# no-op decorator


@pytest.fixture(scope="session")
defevent_loop(request):
loop=asyncio.get_event_loop_policy().new_event_loop()
yieldloop
loop.close()


@pytest.fixture(scope="session")
defredis():
yieldget_redis_connection()


def_delete_test_keys(prefix: str, conn):
keys= []
forkeyinconn.scan_iter(f"{prefix}:*"):
keys.append(key)
ifkeys:
conn.delete(*keys)


@pytest.fixture
defkey_prefix(request, redis):
key_prefix=f"{TEST_PREFIX}:{random.random()}"
yieldkey_prefix


@pytest.fixture(scope="session", autouse=True)
defcleanup_keys(request):
# Always use the sync Redis connection with finalizer. Setting up an
# async finalizer should work, but I'm not suer how yet!
fromredis_om.connectionsimportget_redis_connectionasget_sync_redis

# Increment for every pytest-xdist worker
conn=get_sync_redis()
once_key=f"{TEST_PREFIX}:cleanup_keys"
conn.incr(once_key)

yield

# Delete keys only once
ifconn.decr(once_key) ==0:
_delete_test_keys(TEST_PREFIX, conn)