Coverage for ivatar/test_digest_cache_invalidation.py: 100%

1""" 

2Unit tests for digest cache invalidation upon User, ConfirmedEmail, ConfirmedOpenId, 

3and Photo deletion/modification (GitLab Issue #109). 

4""" 

5 

6from io import BytesIO 

7from unittest.mock import MagicMock, patch 

8import hashlib 

9 

10from PIL import Image 

11from django.contrib.auth.models import User 

12from django.core.cache import cache, caches 

13from django.test import TestCase, override_settings 

14from django.urls import reverse 

15 

16from ivatar.ivataraccount.models import ConfirmedEmail, ConfirmedOpenId, Photo 

17from ivatar.photo_cache import ( 

18 clear_local_photo_cache, 

19 get_cached_photo, 

20 invalidate_digest_cache, 

21 set_cached_photo, 

22) 

23from ivatar.views import _default_avatars_png_cache, _default_avatars_png_cache_lock 

24 

25 

26@override_settings( 

27 CACHES={ 

28 "default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}, 

29 "filesystem": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}, 

30 }, 

31 AVATAR_CACHE_ALIAS="filesystem", 

32 PHOTO_CACHE_ALIAS="default", 

33 CACHE_RESPONSE=True, 

34 ENABLE_WEBP_CONTENT_NEGOTIATION=True, 

35) 

36class DigestCacheInvalidationTest(TestCase): # pylint: disable=too-many-instance-attributes 

37 """ 

38 Test suite verifying that user, email, openid, and photo deletions 

39 immediately purge all associated digest caches and view response caches. 

40 """ 

41 

42 def setUp(self): 

43 clear_local_photo_cache() 

44 cache.clear() 

45 caches["filesystem"].clear() 

46 with _default_avatars_png_cache_lock: 

47 _default_avatars_png_cache.clear() 

48 

49 self.user = User.objects.create(username="digestuser") 

50 

51 # Create a test 100x100 PNG photo 

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

53 buf = BytesIO() 

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

55 self.png_data = buf.getvalue() 

56 

57 self.photo = Photo.objects.create( # pylint: disable=no-member 

58 user=self.user, 

59 data=self.png_data, 

60 format="png", 

61 ip_address="127.0.0.1", 

62 ) 

63 

64 self.email = "digesttest@example.com" 

65 self.email_md5 = hashlib.md5(self.email.encode("utf-8")).hexdigest() 

66 self.email_sha256 = hashlib.sha256(self.email.encode("utf-8")).hexdigest() 

67 

68 self.confirmed_email = ConfirmedEmail.objects.create( 

69 user=self.user, 

70 email=self.email, 

71 digest=self.email_md5, 

72 digest_sha256=self.email_sha256, 

73 photo=self.photo, 

74 ip_address="127.0.0.1", 

75 ) 

76 

77 self.openid_url = "http://digestuser.example.com/" 

78 self.openid_sha256 = hashlib.sha256(self.openid_url.encode("utf-8")).hexdigest() 

79 self.confirmed_openid = ConfirmedOpenId.objects.create( # pylint: disable=no-member 

80 user=self.user, 

81 openid=self.openid_url, 

82 digest=self.openid_sha256, 

83 photo=self.photo, 

84 ip_address="127.0.0.1", 

85 ) 

86 

87 def tearDown(self): 

88 clear_local_photo_cache() 

89 cache.clear() 

90 caches["filesystem"].clear() 

91 with _default_avatars_png_cache_lock: 

92 _default_avatars_png_cache.clear() 

93 

94 def test_direct_invalidate_digest_cache(self): 

95 """ 

96 Test direct invocation of invalidate_digest_cache removes view responses, 

97 default avatars, and Gravatar keys. 

98 """ 

99 fs_cache = caches["filesystem"] 

100 

101 # Populate view response cache keys 

102 uri_key_std = f"http://testserver/avatar/{self.email_md5}?s=80:standard" 

103 uri_key_webp = f"http://testserver/avatar/{self.email_md5}?s=80:webp" 

104 payload_std = { 

105 "content": b"avatar_png", 

106 "content_type": "image/png", 

107 "status": 200, 

108 "reason": None, 

109 "charset": None, 

110 } 

111 payload_webp = { 

112 "content": b"avatar_webp", 

113 "content_type": "image/webp", 

114 "status": 200, 

115 "reason": None, 

116 "charset": None, 

117 } 

118 fs_cache.set(uri_key_std, payload_std) 

119 fs_cache.set(uri_key_webp, payload_webp) 

120 

121 # Populate default avatar cache in Redis / locmem and in-memory 

122 default_key = f"default_avatar:retro:{self.email_md5}:80" 

123 cache.set(default_key, b"default_retro_bytes") 

124 with _default_avatars_png_cache_lock: 

125 _default_avatars_png_cache[("retro", self.email_md5, 80)] = b"default_retro_bytes" 

126 

127 # Verify they exist 

128 self.assertIsNotNone(fs_cache.get(uri_key_std)) 

129 self.assertIsNotNone(fs_cache.get(uri_key_webp)) 

130 self.assertIsNotNone(cache.get(default_key)) 

131 with _default_avatars_png_cache_lock: 

132 self.assertIn(("retro", self.email_md5, 80), _default_avatars_png_cache) 

133 

134 # Invalidate digest 

135 invalidate_digest_cache(self.email_md5) 

136 

137 # Verify they are purged 

138 self.assertIsNone(fs_cache.get(uri_key_std)) 

139 self.assertIsNone(fs_cache.get(uri_key_webp)) 

140 self.assertIsNone(cache.get(default_key)) 

141 with _default_avatars_png_cache_lock: 

142 self.assertNotIn(("retro", self.email_md5, 80), _default_avatars_png_cache) 

143 

144 def test_invalidate_digest_cache_pattern_deletion(self): 

145 """ 

146 Test that invalidate_digest_cache calls delete_pattern on cache backends that support it. 

147 """ 

148 mock_backend = MagicMock() 

149 mock_backend.delete_pattern = MagicMock() 

150 mock_backend.delete_many = MagicMock() 

151 

152 patch_caches = {"default": mock_backend, "filesystem": mock_backend} 

153 with patch("ivatar.photo_cache.caches", patch_caches): 

154 invalidate_digest_cache([self.email_md5, self.email_sha256]) 

155 

156 mock_backend.delete_pattern.assert_any_call(f"*{self.email_md5}*") 

157 mock_backend.delete_pattern.assert_any_call(f"*{self.email_sha256}*") 

158 

159 def test_user_deletion_invalidates_all_caches(self): 

160 """ 

161 Test that deleting a User invalidates photo caches and digest caches for all emails/openids. 

162 """ 

163 fs_cache = caches["filesystem"] 

164 

165 # 1. Warm up photo cache 

166 set_cached_photo(self.photo.pk, 80, "png", b"testphoto80") 

167 self.assertIsNotNone(get_cached_photo(self.photo.pk, 80, "png")) 

168 

169 # 2. Warm up view response cache for both MD5 and SHA256 and OpenID 

170 uri_email_md5 = f"http://testserver/avatar/{self.email_md5}?s=80:standard" 

171 uri_email_sha = f"http://testserver/avatar/{self.email_sha256}?s=80:standard" 

172 uri_openid = f"http://testserver/avatar/{self.openid_sha256}?s=80:standard" 

173 

174 payload = { 

175 "content": b"img_data", 

176 "content_type": "image/png", 

177 "status": 200, 

178 "reason": None, 

179 "charset": None, 

180 } 

181 fs_cache.set(uri_email_md5, payload) 

182 fs_cache.set(uri_email_sha, payload) 

183 fs_cache.set(uri_openid, payload) 

184 

185 # 3. Delete the user 

186 photo_id = self.photo.pk 

187 self.user.delete() 

188 

189 # 4. Verify photo cache is gone 

190 self.assertIsNone(get_cached_photo(photo_id, 80, "png")) 

191 

192 # 5. Verify response cache entries are purged 

193 self.assertIsNone(fs_cache.get(uri_email_md5)) 

194 self.assertIsNone(fs_cache.get(uri_email_sha)) 

195 self.assertIsNone(fs_cache.get(uri_openid)) 

196 

197 def test_confirmed_email_deletion_invalidates_digest_cache(self): 

198 """ 

199 Test that deleting a ConfirmedEmail invalidates its MD5 and SHA256 response caches. 

200 """ 

201 fs_cache = caches["filesystem"] 

202 uri_email_md5 = f"http://testserver/avatar/{self.email_md5}?s=48:standard" 

203 uri_email_sha = f"http://testserver/avatar/{self.email_sha256}?s=48:standard" 

204 

205 payload = { 

206 "content": b"img_data", 

207 "content_type": "image/png", 

208 "status": 200, 

209 "reason": None, 

210 "charset": None, 

211 } 

212 fs_cache.set(uri_email_md5, payload) 

213 fs_cache.set(uri_email_sha, payload) 

214 

215 self.confirmed_email.delete() 

216 

217 self.assertIsNone(fs_cache.get(uri_email_md5)) 

218 self.assertIsNone(fs_cache.get(uri_email_sha)) 

219 

220 def test_confirmed_openid_deletion_invalidates_digest_cache(self): 

221 """ 

222 Test that deleting a ConfirmedOpenId invalidates its digest response caches. 

223 """ 

224 fs_cache = caches["filesystem"] 

225 uri_openid = f"http://testserver/avatar/{self.openid_sha256}?s=48:standard" 

226 

227 payload = { 

228 "content": b"img_oid", 

229 "content_type": "image/png", 

230 "status": 200, 

231 "reason": None, 

232 "charset": None, 

233 } 

234 fs_cache.set(uri_openid, payload) 

235 

236 self.confirmed_openid.delete() 

237 

238 self.assertIsNone(fs_cache.get(uri_openid)) 

239 

240 def test_avatar_view_end_to_end_user_deletion(self): 

241 """ 

242 End-to-end test of Issue #109: 

243 - Request avatar -> cached in response cache 

244 - Delete user account 

245 - Subsequent request must not return the old cached image 

246 """ 

247 avatar_url = reverse("avatar_view", args=[self.email_md5]) + "?s=80" 

248 

249 # 1. First request -> returns uploaded photo and caches response 

250 response1 = self.client.get(avatar_url, HTTP_ACCEPT="image/png") 

251 self.assertEqual(response1.status_code, 200) 

252 self.assertEqual(response1["Content-Type"], "image/png") 

253 

254 # Verify response is cached 

255 fs_cache = caches["filesystem"] 

256 cached_entry = fs_cache.get(f"http://testserver{avatar_url}:standard") 

257 self.assertIsNotNone(cached_entry) 

258 

259 # 2. Delete the user 

260 self.user.delete() 

261 

262 # 3. Request avatar again -> must not return old cached image; 

263 # since user is deleted, it redirects or serves default avatar 

264 response2 = self.client.get(avatar_url, HTTP_ACCEPT="image/png") 

265 self.assertNotEqual(response2.content, response1.content)