Repository: redis/redis-py
File: tests/test_asyncio/test_commands.py (Python)

Code before:

if sys.version_info >= (3, 11, 3):
    from asyncio import timeout as async_timeout
else:
    from async_timeout import timeout as async_timeout

REDIS_6_VERSION = "5.9.0"


@pytest_asyncio.fixture()
async def r_teardown(r: redis.Redis):
    """
    A special fixture which removes the provided names from the database after use
    """
    usernames = []

    def factory(username):
        usernames.append(username)
        return r

    yield factory
    try:
        current_user = await r.client_info()
    except exceptions.NoPermissionError:
        current_user = {}
    if "default" != current_user.get("user"):
        await r.auth("", "default")
    for username in usernames:
        await r.acl_deluser(username)


@pytest_asyncio.fixture()
async def slowlog(r: redis.Redis):
    current_config = await r.config_get()
    old_slower_than_value = current_config["slowlog-log-slower-than"]
    old_max_legnth_value = current_config["slowlog-max-len"]

    await r.config_set("slowlog-log-slower-than", 0)
    await r.config_set("slowlog-max-len", 128)

    yield

    await r.config_set("slowlog-log-slower-than", old_slower_than_value)
    await r.config_set("slowlog-max-len", old_max_legnth_value)


async def redis_server_time(client: redis.Redis):
    seconds, milliseconds = await client.time()
    timestamp = float(f"{seconds}.{milliseconds}")
    return datetime.datetime.fromtimestamp(timestamp)


Review comment (suggestion):
Wouldn't it throw an exception if previously the exception was thrown? `current_user.get("user")` should throw `KeyError` if key doesn't exist. You can fix it by falling back to default value like this `current_user.get("user", "default")`.

Code after:

if sys.version_info >= (3, 11, 3):
    from asyncio import timeout as async_timeout
else:
    from async_timeout import timeout as async_timeout

REDIS_6_VERSION = "5.9.0"


@pytest_asyncio.fixture()
async def r_teardown(r: redis.Redis):
    """
    A special fixture which removes the provided names from the database after use
    """
    usernames = []

    def factory(username):
        usernames.append(username)
        return r

    yield factory
    try:
        client_info = await r.client_info()
    except exceptions.NoPermissionError:
        client_info = {}
    if "default" != client_info.get("user", ""):
        await r.auth("", "default")
    for username in usernames:
        await r.acl_deluser(username)


@pytest_asyncio.fixture()
async def slowlog(r: redis.Redis):
    current_config = await r.config_get()
    old_slower_than_value = current_config["slowlog-log-slower-than"]
    old_max_legnth_value = current_config["slowlog-max-len"]

    await r.config_set("slowlog-log-slower-than", 0)
    await r.config_set("slowlog-max-len", 128)

    yield

    await r.config_set("slowlog-log-slower-than", old_slower_than_value)
    await r.config_set("slowlog-max-len", old_max_legnth_value)


async def redis_server_time(client: redis.Redis):
    seconds, milliseconds = await client.time()
    timestamp = float(f"{seconds}.{milliseconds}")
    return datetime.datetime.fromtimestamp(timestamp)