Coverage for ivatar/test_views.py: 100%

1""" 

2Test our views in ivatar.ivataraccount.views and ivatar.views 

3""" 

4 

5# pylint: disable=too-many-lines 

6import os 

7import django 

8from django.urls import reverse 

9from django.test import TestCase, Client, override_settings 

10from django.contrib.auth.models import User 

11from ivatar.utils import random_string, Bluesky 

12 

13from django.conf import settings 

14 

15BLUESKY_APP_PASSWORD = getattr(settings, "BLUESKY_APP_PASSWORD", None) 

16BLUESKY_IDENTIFIER = getattr(settings, "BLUESKY_IDENTIFIER", None) 

17os.environ["DJANGO_SETTINGS_MODULE"] = "ivatar.settings" 

18django.setup() 

19 

20 

21class Tester(TestCase): # pylint: disable=too-many-public-methods 

22 """ 

23 Main test class 

24 """ 

25 

26 client = Client() 

27 user = None 

28 username = random_string() 

29 password = random_string() 

30 email = "{}@{}.{}".format(username, random_string(), random_string(2)) 

31 # Dunno why random tld doesn't work, but I'm too lazy now to investigate 

32 openid = "http://{}.{}.{}/".format(username, random_string(), "org") 

33 

34 def login(self): 

35 """ 

36 Login as user 

37 """ 

38 self.client.login(username=self.username, password=self.password) 

39 

40 def setUp(self): 

41 """ 

42 Prepare for tests. 

43 - Create user 

44 """ 

45 self.user = User.objects.create_user( 

46 username=self.username, 

47 password=self.password, 

48 ) 

49 

50 def test_incorrect_digest(self): 

51 """ 

52 Test incorrect digest 

53 """ 

54 response = self.client.get("/avatar/" + "x" * 65, follow=True) 

55 self.assertEqual( 

56 response.redirect_chain[2][0], 

57 "/static/img/nobody/80.png", 

58 "Doesn't redirect to static?", 

59 ) 

60 # self.assertRedirects( 

61 # response=response, 

62 # expected_url="/static/img/nobody/80.png", 

63 # msg_prefix="Why does an invalid hash not redirect to deadbeef?", 

64 # ) 

65 

66 def test_logout(self): 

67 """ 

68 Test if logout works correctly 

69 """ 

70 self.login() 

71 response = self.client.get(reverse("logout"), follow=True) 

72 self.assertEqual( 

73 response.status_code, 405, "logout with get should lead to http error 405" 

74 ) 

75 response = self.client.post(reverse("logout"), follow=True) 

76 self.assertEqual(response.status_code, 200, "logout with post should logout") 

77 

78 def test_Bluesky_client(self): 

79 """ 

80 Bluesky client needs credentials, so it's limited with testing here now 

81 """ 

82 

83 if BLUESKY_APP_PASSWORD and BLUESKY_IDENTIFIER: 

84 b = Bluesky() 

85 profile = b.get_profile("libravatar.org") 

86 self.assertEqual(profile["handle"], "libravatar.org") 

87 # As long as I don't change my avatar, this should stay the same 

88 self.assertEqual( 

89 profile["avatar"], 

90 "https://cdn.bsky.app/img/avatar/plain/did:plc:pnzkvz5q5xg72bzkxrucalje/bafkreic456md3b46sm5sh3rsrgis26njy2d4p6u5yqxpz5bpqa6yk6ceni", 

91 ) 

92 self.assertEqual(True, True) 

93 

94 @override_settings( 

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

96 ) 

97 def test_default_avatar_caching(self): 

98 """ 

99 Test in-memory and shared Redis caching for default avatars 

100 """ 

101 from ivatar.views import ( 

102 _get_cached_default_avatar, 

103 _set_cached_default_avatar, 

104 _default_avatars_png_cache, 

105 _default_avatars_png_cache_lock, 

106 ) 

107 from django.core.cache import cache 

108 

109 test_digest = "abcdef1234567890abcdef1234567890" 

110 test_png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDRtestdata" 

111 

112 # Clear any existing cache for this key 

113 with _default_avatars_png_cache_lock: 

114 _default_avatars_png_cache.pop(("retro", test_digest, 28), None) 

115 cache.delete(f"default_avatar:retro:{test_digest}:28") 

116 

117 # Initially not cached 

118 self.assertIsNone(_get_cached_default_avatar("retro", test_digest, 28)) 

119 

120 # Store in cache 

121 _set_cached_default_avatar("retro", test_digest, 28, test_png) 

122 

123 # Retrieve from in-process memory cache 

124 result = _get_cached_default_avatar("retro", test_digest, 28) 

125 self.assertIsNotNone(result) 

126 self.assertEqual(result.getvalue(), test_png) 

127 

128 # Evict in-memory process cache to test shared Django/Redis cache fallback 

129 with _default_avatars_png_cache_lock: 

130 _default_avatars_png_cache.pop(("retro", test_digest, 28), None) 

131 

132 result_fallback = _get_cached_default_avatar("retro", test_digest, 28) 

133 self.assertIsNotNone(result_fallback) 

134 self.assertEqual(result_fallback.getvalue(), test_png) 

135 

136 def test_retro_avatar_request(self): 

137 """ 

138 Test requesting an uncached retro avatar 

139 """ 

140 digest = "63a75a80e6b1f4adfdb04c1ca02e596c" 

141 response = self.client.get(f"/avatar/{digest}?d=retro&s=28&forcedefault=y") 

142 self.assertEqual(response.status_code, 200) 

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

144 self.assertGreater(len(response.content), 0) 

145 

146 def test_default_avatar_caching_graceful_degradation(self): 

147 """ 

148 Test that default avatar caching gracefully degrades if Redis or Django cache fails 

149 """ 

150 from unittest.mock import patch 

151 from ivatar.views import ( 

152 _get_cached_default_avatar, 

153 _set_cached_default_avatar, 

154 _default_avatars_png_cache, 

155 _default_avatars_png_cache_lock, 

156 ) 

157 

158 test_digest = "deadbeef1234567890deadbeef123456" 

159 test_png = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDRgraceful" 

160 

161 with patch("django.core.cache.cache.get", side_effect=ConnectionError("Redis connection refused")), \ 

162 patch("django.core.cache.cache.set", side_effect=ConnectionError("Redis connection refused")): 

163 with _default_avatars_png_cache_lock: 

164 _default_avatars_png_cache.pop(("retro", test_digest, 28), None) 

165 

166 # Get with Redis failing should return None instead of raising an exception 

167 self.assertIsNone(_get_cached_default_avatar("retro", test_digest, 28)) 

168 

169 # Set with Redis failing should still store in local in-memory cache without raising 

170 _set_cached_default_avatar("retro", test_digest, 28, test_png) 

171 

172 # Retrieval from local memory cache works despite Redis failure 

173 cached = _get_cached_default_avatar("retro", test_digest, 28) 

174 self.assertIsNotNone(cached) 

175 self.assertEqual(cached.getvalue(), test_png) 

176 

177 def test_gravatar_fallback_with_encoded_spaces_in_default_url(self): 

178 """ 

179 Test that default URLs containing spaces (e.g. ui-avatars.com name) 

180 do not trigger InvalidURL or skip Gravatar fallback (Issue #111). 

181 """ 

182 from unittest.mock import patch 

183 from io import BytesIO 

184 from PIL import Image 

185 

186 digest = "7487baad7b087cb03cc4ccf7d044c150" 

187 img = Image.new("RGBA", (10, 10), (255, 0, 0, 255)) 

188 buf = BytesIO() 

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

190 valid_png = buf.getvalue() 

191 

192 class MockResponse: 

193 def __init__(self, content): 

194 self.content = content 

195 def read(self): 

196 return self.content 

197 

198 # Test request with URL-encoded space in default 

199 url = f"/avatar/{digest}?d=https%3A%2F%2Fui-avatars.com%2Fapi%2FSteve%2520Lacey" 

200 with patch("ivatar.views.urlopen", side_effect=lambda u, **kw: MockResponse(valid_png)) as mock_urlopen: 

201 response = self.client.get(url, follow=True) 

202 self.assertEqual(response.status_code, 200) 

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

204 # Verify urlopen was called with properly quoted URL without raw spaces 

205 gravatar_call_url = mock_urlopen.call_args_list[-1][0][0] 

206 self.assertNotIn(" ", gravatar_call_url) 

207 self.assertIn("Steve%20Lacey", gravatar_call_url) 

208 

209 def test_trusted_default_urls_ui_avatars_regional_subdomains(self): 

210 """ 

211 Test that regional subdomains of ui-avatars.com are accepted as trusted (Issue #112). 

212 """ 

213 from ivatar.utils import is_trusted_url 

214 from django.conf import settings 

215 

216 self.assertTrue(is_trusted_url("https://ui-avatars.com/api/Test", settings.TRUSTED_DEFAULT_URLS)) 

217 self.assertTrue(is_trusted_url("https://eu.ui-avatars.com/api/Test", settings.TRUSTED_DEFAULT_URLS)) 

218 self.assertTrue(is_trusted_url("https://na.ui-avatars.com/api/Test", settings.TRUSTED_DEFAULT_URLS)) 

219 self.assertTrue(is_trusted_url("https://ap.ui-avatars.com/api/Test", settings.TRUSTED_DEFAULT_URLS)) 

220 self.assertFalse(is_trusted_url("https://fake-ui-avatars.com/api/Test", settings.TRUSTED_DEFAULT_URLS)) 

221 

222 def test_bluesky_proxy_caches_and_supports_etag(self): 

223 """ 

224 Verify BlueskyProxyView caches the proxied payload, advertises an 

225 ETag/Cache-Control, skips re-downloads on later hits, and answers 

226 conditional requests with 304. 

227 """ 

228 from unittest.mock import patch 

229 from io import BytesIO 

230 from PIL import Image 

231 from django.core.cache import cache 

232 from ivatar.ivataraccount.models import ConfirmedEmail 

233 

234 email = ConfirmedEmail.objects.create( 

235 user=self.user, 

236 email="bsky-proxy@example.com", 

237 ip_address="127.0.0.1", 

238 ) 

239 email.bluesky_handle = "test.bsky.social" 

240 email.save() 

241 email.refresh_from_db() 

242 

243 img = Image.new("RGB", (200, 200), (10, 20, 30)) 

244 buf = BytesIO() 

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

246 png = buf.getvalue() 

247 

248 class MockResponse: 

249 def read(self): 

250 return png 

251 

252 class MockBluesky: 

253 def get_avatar(self, handle): 

254 return "https://cdn.bsky.test/avatar.png" 

255 

256 cache.clear() 

257 url = f"/blueskyproxy/{email.digest}?s=80" 

258 with patch("ivatar.views.Bluesky", MockBluesky), patch( 

259 "ivatar.views.urlopen", return_value=MockResponse() 

260 ) as mock_urlopen: 

261 response = self.client.get(url) 

262 self.assertEqual(response.status_code, 200) 

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

264 self.assertIn("Etag", response) 

265 self.assertIn("Cache-Control", response) 

266 self.assertEqual(mock_urlopen.call_count, 1) 

267 etag = response["Etag"] 

268 

269 # Second plain request is served from the payload cache 

270 response2 = self.client.get(url) 

271 self.assertEqual(response2.status_code, 200) 

272 self.assertEqual(mock_urlopen.call_count, 1) 

273 

274 # Conditional request short-circuits to 304 without downloading 

275 response3 = self.client.get(url, HTTP_IF_NONE_MATCH=etag) 

276 self.assertEqual(response3.status_code, 304) 

277 self.assertEqual(response3["Etag"], etag) 

278 self.assertEqual(mock_urlopen.call_count, 1) 

279 

280 

281