pssecret-server/tests/integration/test_utils.py

from unittest.mock import AsyncMock, patch

import pytest
from redis.asyncio import Redis

from pssecret_server.utils import get_new_key, getdel, save_secret

from ..factories import SecretFactory


async def test_get_new_key_returns_key(redis_server: Redis) -> None:
    assert isinstance(await get_new_key(redis_server), str)


async def test_get_new_key_returns_free_key(redis_server: Redis) -> None:
    new_key = await get_new_key(redis_server)
    res = not await redis_server.exists(new_key)
    assert res


@patch("pssecret_server.utils.uuid4", side_effect=("used", "free"))
async def test_get_new_key_skips_used_keys(_, redis_server: Redis) -> None:
    await redis_server.set("used", "")

    assert await get_new_key(redis_server) == "free"


async def test_save_secret_data(redis_server: Redis) -> None:
    secret = SecretFactory.build()

    key = await save_secret(secret, redis_server)

    redis_data: bytes | None = await redis_server.get(key)

    assert redis_data is not None
    assert redis_data.decode() == secret.data


@pytest.mark.parametrize("getdel_available", [True, False])
@patch("pssecret_server.utils._is_getdel_available", new_callable=AsyncMock)
async def test_getdel(
    mock_is_getdel_available: AsyncMock,
    getdel_available: bool,
    redis_server: Redis,
) -> None:
    mock_is_getdel_available.return_value = getdel_available

    test_value = "test_data"
    test_key = "test_key"
    await redis_server.set(test_key, test_value)

    result = await getdel(redis_server, test_key)

    assert result.decode() == test_value  # pyright: ignore[reportAttributeAccessIssue]
    assert not await redis_server.exists(test_key)