pssecret-server/tests/integration/test_utils.py
Ivan Golikov 6804360352 Added support for all Redis versions (>=1.0.0) (#5)
Previously support was provided for Redis>=6.2.0

Reviewed-on: #5
Co-authored-by: Ivan Golikov <root@ivnglkv.me>
Co-committed-by: Ivan Golikov <root@ivnglkv.me>
2025-01-08 21:22:25 +00:00

55 lines
1.6 KiB
Python

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)