zzstoatzz.io
audio streaming app
plyr.fm
1"""tests for anonymous discovery feed caching."""
2
3from unittest.mock import AsyncMock, patch
4
5from redis.exceptions import ConnectionError as RedisConnectionError
6
7from backend._internal.tasks.hooks import invalidate_tracks_discovery_cache
8from backend.api.tracks.constants import DISCOVERY_CACHE_KEY
9from backend.api.tracks.listing import TracksListResponse
10
11SAMPLE_RESPONSE = TracksListResponse(tracks=[], next_cursor=None, has_more=False)
12
13
14async def test_anonymous_discovery_cache_hit() -> None:
15 """cached response deserializes correctly from Redis."""
16 cached_json = SAMPLE_RESPONSE.model_dump_json()
17 result = TracksListResponse.model_validate_json(cached_json)
18 assert result == SAMPLE_RESPONSE
19
20
21async def test_invalidate_clears_cache() -> None:
22 """invalidate_tracks_discovery_cache deletes the cache key."""
23 mock_redis = AsyncMock()
24 mock_redis.delete = AsyncMock()
25
26 with patch(
27 "backend._internal.tasks.hooks.get_async_redis_client",
28 return_value=mock_redis,
29 ):
30 await invalidate_tracks_discovery_cache()
31
32 mock_redis.delete.assert_awaited_once_with(DISCOVERY_CACHE_KEY)
33
34
35async def test_invalidate_handles_redis_error() -> None:
36 """invalidation silently handles Redis errors."""
37 mock_redis = AsyncMock()
38 mock_redis.delete.side_effect = RedisConnectionError("redis down")
39
40 with patch(
41 "backend._internal.tasks.hooks.get_async_redis_client",
42 return_value=mock_redis,
43 ):
44 # should not raise
45 await invalidate_tracks_discovery_cache()