Coverage for ivatar/test_photo_cache.py: 100%

1""" 

2Tests for user photo caching and invalidation. 

3""" 

4 

5from io import BytesIO 

6from unittest.mock import patch 

7import hashlib 

8 

9from PIL import Image 

10from django.contrib.auth.models import User 

11from django.core.cache import cache 

12from django.test import TestCase, override_settings 

13 

14from ivatar.ivataraccount.models import ConfirmedEmail, Photo 

15from ivatar.photo_cache import ( 

16 clear_local_photo_cache, 

17 get_cached_photo, 

18 invalidate_photo_cache, 

19 set_cached_photo, 

20 pregenerate_photo_thumbnails, 

21 render_photo_thumbnail, 

22 DEFAULT_PREGENERATE_SIZES, 

23 _user_photos_cache, 

24 _user_photos_cache_lock, 

25) 

26 

27 

28@override_settings( 

29 CACHES={"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}}, 

30 CACHE_RESPONSE=False, 

31) 

32class UserPhotoCacheTest(TestCase): 

33 """ 

34 Test suite for user photo Redis & in-memory caching with model invalidation. 

35 """ 

36 

37 def setUp(self): 

38 clear_local_photo_cache() 

39 cache.clear() 

40 

41 self.user = User.objects.create(username="cachetestuser") 

42 

43 # Create a test 100x100 PNG 

44 img = Image.new("RGBA", (100, 100), (255, 0, 0, 255)) 

45 buf = BytesIO() 

46 img.save(buf, format="PNG") 

47 self.png_data = buf.getvalue() 

48 

49 self.photo = Photo.objects.create( 

50 user=self.user, 

51 data=self.png_data, 

52 format="png", 

53 ip_address="127.0.0.1", 

54 ) 

55 

56 self.email = "cachetest@example.com" 

57 self.digest = hashlib.md5(self.email.encode("utf-8")).hexdigest() 

58 self.digest_sha256 = hashlib.sha256(self.email.encode("utf-8")).hexdigest() 

59 

60 self.confirmed_email = ConfirmedEmail.objects.create( 

61 user=self.user, 

62 email=self.email, 

63 digest=self.digest, 

64 digest_sha256=self.digest_sha256, 

65 photo=self.photo, 

66 ip_address="127.0.0.1", 

67 ) 

68 

69 def tearDown(self): 

70 clear_local_photo_cache() 

71 cache.clear() 

72 

73 def test_cache_set_and_get_lifecycle(self): 

74 """ 

75 Test caching resized photo in L1 and falling back to L2 cache. 

76 """ 

77 test_photo_id = 99999 

78 sample_bytes = b"resized_photo_payload" 

79 

80 # Initially empty 

81 self.assertIsNone(get_cached_photo(test_photo_id, 80, "png")) 

82 

83 # Set cache 

84 set_cached_photo(test_photo_id, 80, "png", sample_bytes) 

85 

86 # L1 hit 

87 cached_stream = get_cached_photo(test_photo_id, 80, "png") 

88 self.assertIsNotNone(cached_stream) 

89 self.assertEqual(cached_stream.getvalue(), sample_bytes) 

90 

91 # Clear L1 to force L2 hit 

92 clear_local_photo_cache() 

93 with _user_photos_cache_lock: 

94 self.assertEqual(len(_user_photos_cache), 0) 

95 

96 # L2 hit 

97 cached_stream_l2 = get_cached_photo(test_photo_id, 80, "png") 

98 self.assertIsNotNone(cached_stream_l2) 

99 self.assertEqual(cached_stream_l2.getvalue(), sample_bytes) 

100 

101 # Invalidate 

102 invalidate_photo_cache(test_photo_id) 

103 self.assertIsNone(get_cached_photo(test_photo_id, 80, "png")) 

104 

105 def test_avatar_image_view_uses_cache_and_invalidates_on_save(self): 

106 """ 

107 Test that AvatarImageView populates cache, serves from cache, 

108 and invalidates on photo update. 

109 """ 

110 # First request: cache miss, renders and populates cache 

111 response = self.client.get(f"/avatar/{self.digest}?s=48") 

112 self.assertEqual(response.status_code, 200) 

113 self.assertEqual(response["Content-Type"], "image/png") 

114 initial_content = response.content 

115 

116 # Verify photo is now in cache 

117 cached = get_cached_photo(self.photo.pk, 48, "png") 

118 self.assertIsNotNone(cached) 

119 self.assertEqual(cached.getvalue(), initial_content) 

120 

121 # Second request: served from cache 

122 response2 = self.client.get(f"/avatar/{self.digest}?s=48") 

123 self.assertEqual(response2.status_code, 200) 

124 self.assertEqual(response2.content, initial_content) 

125 

126 # Set an un-pregenerated size to test explicit invalidation 

127 set_cached_photo(self.photo.pk, 64, "png", b"old_unpregenerated_size") 

128 

129 # Update the photo with a blue image 

130 img_blue = Image.new("RGBA", (100, 100), (0, 0, 255, 255)) 

131 buf_blue = BytesIO() 

132 img_blue.save(buf_blue, format="PNG") 

133 self.photo.data = buf_blue.getvalue() 

134 self.photo.save() 

135 

136 # Non-pregenerated size should be completely invalidated 

137 self.assertIsNone(get_cached_photo(self.photo.pk, 64, "png")) 

138 

139 # Common size (s=48) should have the old content invalidated and new content pregenerated 

140 cached_pregenerated = get_cached_photo(self.photo.pk, 48, "png") 

141 self.assertIsNotNone(cached_pregenerated) 

142 self.assertNotEqual(cached_pregenerated.getvalue(), initial_content) 

143 

144 # Third request: served from pregenerated cache with new blue image 

145 response3 = self.client.get(f"/avatar/{self.digest}?s=48") 

146 self.assertEqual(response3.status_code, 200) 

147 self.assertEqual(response3.content, cached_pregenerated.getvalue()) 

148 

149 def test_invalidation_on_photo_delete(self): 

150 """ 

151 Test that deleting a photo purges the cache. 

152 """ 

153 # Populate cache 

154 set_cached_photo(self.photo.pk, 64, "png", b"test64") 

155 set_cached_photo(self.photo.pk, 128, "png", b"test128") 

156 

157 self.assertIsNotNone(get_cached_photo(self.photo.pk, 64, "png")) 

158 self.assertIsNotNone(get_cached_photo(self.photo.pk, 128, "png")) 

159 

160 # Delete photo 

161 photo_id = self.photo.pk 

162 self.photo.delete() 

163 

164 # Should be invalidated 

165 self.assertIsNone(get_cached_photo(photo_id, 64, "png")) 

166 self.assertIsNone(get_cached_photo(photo_id, 128, "png")) 

167 

168 def test_cache_graceful_degradation(self): 

169 """ 

170 Test graceful degradation when cache raises an unexpected exception. 

171 """ 

172 with patch.object(cache, "get", side_effect=Exception("Redis down")): 

173 clear_local_photo_cache() 

174 result = get_cached_photo(self.photo.pk, 80, "png") 

175 self.assertIsNone(result) 

176 

177 with patch.object(cache, "set", side_effect=Exception("Redis down")): 

178 # Should not raise exception 

179 set_cached_photo(self.photo.pk, 80, "png", b"sample") 

180 

181 def test_pregenerate_photo_thumbnails_populates_cache(self): 

182 """ 

183 Test pregenerate_photo_thumbnails directly generates and caches all target sizes. 

184 """ 

185 clear_local_photo_cache() 

186 cache.clear() 

187 

188 # Generate thumbnails for standard sizes 

189 target_sizes = (28, 48, 80, 128, 256) 

190 generated = pregenerate_photo_thumbnails(self.photo, sizes=target_sizes) 

191 self.assertEqual(generated, list(target_sizes)) 

192 

193 # Check each size in cache 

194 for size in target_sizes: 

195 stream = get_cached_photo(self.photo.pk, size, "png") 

196 self.assertIsNotNone(stream, f"Size {size} should be cached") 

197 img = Image.open(stream) 

198 self.assertEqual(img.size, (size, size)) 

199 

200 def test_render_photo_thumbnail_format_handling(self): 

201 """ 

202 Test render_photo_thumbnail produces valid thumbnails for PNG, JPEG, and GIF. 

203 """ 

204 # 1. PNG 

205 png_bytes = render_photo_thumbnail(self.png_data, "png", 48) 

206 img_png = Image.open(BytesIO(png_bytes)) 

207 self.assertEqual(img_png.size, (48, 48)) 

208 self.assertEqual(img_png.format, "PNG") 

209 

210 # 2. JPEG from RGBA (tests automatic mode conversion to RGB) 

211 jpeg_bytes = render_photo_thumbnail(self.png_data, "jpg", 80) 

212 img_jpeg = Image.open(BytesIO(jpeg_bytes)) 

213 self.assertEqual(img_jpeg.size, (80, 80)) 

214 self.assertEqual(img_jpeg.format, "JPEG") 

215 

216 def test_pregeneration_on_photo_save_and_update(self): 

217 """ 

218 Test that saving a photo automatically pregenerates thumbnails in cache. 

219 """ 

220 clear_local_photo_cache() 

221 cache.clear() 

222 

223 # Create a new photo 

224 img = Image.new("RGB", (200, 200), (0, 255, 0)) 

225 buf = BytesIO() 

226 img.save(buf, format="JPEG") 

227 

228 new_photo = Photo.objects.create( 

229 user=self.user, 

230 data=buf.getvalue(), 

231 format="jpg", 

232 ip_address="127.0.0.1", 

233 ) 

234 

235 # Target sizes should be immediately cached 

236 for size in DEFAULT_PREGENERATE_SIZES: 

237 cached = get_cached_photo(new_photo.pk, size, "jpeg") 

238 self.assertIsNotNone(cached, f"Size {size} should be pre-cached on save") 

239 thumb = Image.open(cached) 

240 self.assertEqual(thumb.size, (size, size)) 

241 

242 def test_pregenerate_photo_thumbnails_async(self): 

243 """ 

244 Test that asynchronous thumbnail pregeneration returns target sizes and populates cache. 

245 """ 

246 import time 

247 

248 clear_local_photo_cache() 

249 cache.clear() 

250 

251 target_sizes = (28, 48) 

252 generated = pregenerate_photo_thumbnails( 

253 self.photo, sizes=target_sizes, async_generation=True 

254 ) 

255 self.assertEqual(generated, list(target_sizes)) 

256 

257 # Allow daemon thread a short window to finish 

258 for _ in range(50): 

259 if get_cached_photo(self.photo.pk, 28, "png") and get_cached_photo( 

260 self.photo.pk, 48, "png" 

261 ): 

262 break 

263 time.sleep(0.02) 

264 

265 self.assertIsNotNone(get_cached_photo(self.photo.pk, 28, "png")) 

266 self.assertIsNotNone(get_cached_photo(self.photo.pk, 48, "png"))