A0 Implementation Engineer
e62ece1c06
- FastAPI app with CORS, lifespan handlers - Pydantic Settings config (DB, Redis, CORS, SMTP, JWT, Rentman) - SQLAlchemy async engine + session (DeclarativeBase) - 6 DB models: EquipmentCache, RentalRequest, RentalRequestItem, Contact, AdminUser, SyncLog - Pydantic schemas: EquipmentItem, EquipmentDetail, PaginatedEquipment, ContactCreate, ContactResponse - Redis cache helper: set/get/delete_pattern, rate limiting, equipment key builders - Equipment router: list (search/category/sort/pagination), detail, categories – all cached - Contact router: POST with Pydantic validation + rate limiting (5/min) - Health router: GET /api/health with DB + Redis status - 28 pytest tests (all pass, 90% coverage) - Dockerfile, requirements.txt, pytest.ini, test_report.md
85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
"""Tests for Redis cache helpers."""
|
|
|
|
import pytest
|
|
import fakeredis.aioredis
|
|
|
|
from app.cache import (
|
|
_cache_client,
|
|
check_rate_limit,
|
|
delete_pattern,
|
|
equipment_categories_key,
|
|
equipment_detail_key,
|
|
equipment_list_key,
|
|
get_cache,
|
|
get_cached,
|
|
set_cache,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_and_get_cache(fake_redis):
|
|
"""set_cache stores JSON and get_cached retrieves it."""
|
|
await set_cache("test:key", {"foo": "bar"}, ttl=60)
|
|
result = await get_cached("test:key")
|
|
assert result == {"foo": "bar"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_cached_miss(fake_redis):
|
|
"""get_cached returns None for non-existent key."""
|
|
result = await get_cached("nonexistent:key")
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_pattern(fake_redis):
|
|
"""delete_pattern removes matching keys."""
|
|
await set_cache("equipment:list:1:all:default:all", ["a"], ttl=60)
|
|
await set_cache("equipment:list:2:all:default:all", ["b"], ttl=60)
|
|
await delete_pattern("equipment:list:*")
|
|
assert await get_cached("equipment:list:1:all:default:all") is None
|
|
assert await get_cached("equipment:list:2:all:default:all") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rate_limit_allows_under_limit(fake_redis):
|
|
"""check_rate_limit allows requests under the limit."""
|
|
for _ in range(5):
|
|
allowed = await check_rate_limit("test_ip", max_requests=5, window=60)
|
|
assert allowed is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rate_limit_blocks_over_limit(fake_redis):
|
|
"""check_rate_limit blocks requests over the limit."""
|
|
for _ in range(5):
|
|
await check_rate_limit("blocked_ip", max_requests=5, window=60)
|
|
allowed = await check_rate_limit("blocked_ip", max_requests=5, window=60)
|
|
assert allowed is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_equipment_list_key(fake_redis):
|
|
"""equipment_list_key builds correct key format."""
|
|
key = equipment_list_key(1, "Lautsprecher", "name_asc", "K2")
|
|
assert "equipment" in key
|
|
assert "list" in key
|
|
assert "1" in key
|
|
assert "Lautsprecher" in key
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_equipment_detail_key(fake_redis):
|
|
"""equipment_detail_key builds correct key format."""
|
|
key = equipment_detail_key(42)
|
|
assert "equipment" in key
|
|
assert "detail" in key
|
|
assert "42" in key
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_equipment_categories_key(fake_redis):
|
|
"""equipment_categories_key builds correct key format."""
|
|
key = equipment_categories_key()
|
|
assert key == "equipment:categories"
|