Coverage for ivatar/test_views_stats.py: 99%

1""" 

2Test our StatsView in ivatar.views 

3""" 

4 

5import base64 

6import json 

7import os 

8import django 

9from django.test import TestCase 

10from django.test import Client 

11from django.contrib.auth.models import User 

12from django.core.cache import cache 

13from ivatar.utils import random_string, random_ip_address 

14from ivatar.ivataraccount.models import ( 

15 ConfirmedEmail, 

16 ConfirmedOpenId, 

17 Photo, 

18 UnconfirmedEmail, 

19 UnconfirmedOpenId, 

20 PhotoAccessStat, 

21 ConfirmedEmailAccessStat, 

22 ConfirmedOpenIdAccessStat, 

23) 

24 

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

26django.setup() 

27 

28 

29class StatsTester(TestCase): 

30 """ 

31 Test class for StatsView 

32 """ 

33 

34 client = Client() 

35 user = None 

36 username = random_string() 

37 password = random_string() 

38 

39 def login(self): 

40 """ 

41 Login as user 

42 """ 

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

44 

45 def setUp(self): 

46 """ 

47 Prepare for tests. 

48 - Create user 

49 - Clear the server-side stats cache so each test sees fresh data 

50 """ 

51 self.user = User.objects.create_user( 

52 username=self.username, 

53 password=self.password, 

54 ) 

55 cache.clear() 

56 

57 def test_stats_payload_cached(self): 

58 """ 

59 Verify /stats/ is computed once and then served from cache. 

60 """ 

61 from ivatar.views import STATS_CACHE_KEY 

62 

63 cache.delete(STATS_CACHE_KEY) 

64 self.client.get("/stats/") 

65 cached = cache.get(STATS_CACHE_KEY) 

66 self.assertIsNotNone(cached, "stats payload was not cached") 

67 self.assertEqual(cached["users"], 1, "cached user count incorrect") 

68 

69 # A DB change within the cache TTL must not be reflected 

70 User.objects.create_user(username=random_string(), password=random_string()) 

71 response = self.client.get("/stats/") 

72 self.assertEqual( 

73 json.loads(response.content)["users"], 

74 1, 

75 "stats response did not come from cache", 

76 ) 

77 

78 def test_stats_basic(self): 

79 """ 

80 Test basic stats functionality 

81 """ 

82 response = self.client.get("/stats/", follow=True) 

83 self.assertEqual(response.status_code, 200, "unable to fetch stats!") 

84 j = json.loads(response.content) 

85 self.assertEqual(j["users"], 1, "user count incorrect") 

86 self.assertEqual(j["mails"], 0, "mails count incorrect") 

87 self.assertEqual(j["openids"], 0, "openids count incorrect") 

88 self.assertEqual(j["unconfirmed_mails"], 0, "unconfirmed mails count incorrect") 

89 self.assertEqual( 

90 j["unconfirmed_openids"], 0, "unconfirmed openids count incorrect" 

91 ) 

92 self.assertEqual(j["avatars"], 0, "avatars count incorrect") 

93 self.assertIn("cache_stats", j, "cache_stats missing from stats endpoint") 

94 

95 def test_stats_comprehensive(self): 

96 """ 

97 Test comprehensive stats with actual data 

98 """ 

99 from ivatar.ivataraccount.models import ( 

100 UnconfirmedEmail, 

101 UnconfirmedOpenId, 

102 ) 

103 

104 # Create test data with random values 

105 email1 = ConfirmedEmail.objects.create( 

106 user=self.user, 

107 email=f"{random_string()}@{random_string()}.{random_string(2)}", 

108 ip_address=random_ip_address(), 

109 ) 

110 ConfirmedEmailAccessStat.objects.update_or_create( 

111 email=email1, defaults={"access_count": 100} 

112 ) 

113 

114 email2 = ConfirmedEmail.objects.create( 

115 user=self.user, 

116 email=f"{random_string()}@{random_string()}.{random_string(2)}", 

117 ip_address=random_ip_address(), 

118 ) 

119 ConfirmedEmailAccessStat.objects.update_or_create( 

120 email=email2, defaults={"access_count": 50} 

121 ) 

122 

123 openid1 = ConfirmedOpenId.objects.create( 

124 user=self.user, 

125 openid=f"http://{random_string()}.{random_string()}.org/", 

126 ip_address=random_ip_address(), 

127 ) 

128 ConfirmedOpenIdAccessStat.objects.update_or_create( 

129 openid=openid1, defaults={"access_count": 75} 

130 ) 

131 

132 # Create photos with valid image data (minimal PNG, Base64-encoded for pycco compatibility) 

133 # PNG header + minimal data 

134 png_base64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAIAAACQd1PeAAAACXBIWXMAAAsTAAALEwEAmpwYAAAACklEQVQIHWNgYAAAAAMAAWveibQAAAAASUVORK5CYII=" 

135 png_data = base64.b64decode(png_base64) 

136 

137 photo1 = Photo.objects.create( 

138 user=self.user, data=png_data, format="png", ip_address=random_ip_address() 

139 ) 

140 PhotoAccessStat.objects.update_or_create( 

141 photo=photo1, defaults={"access_count": 200} 

142 ) 

143 

144 photo2 = Photo.objects.create( 

145 user=self.user, 

146 data=png_data, # Same data for testing 

147 format="png", # Same format for testing 

148 ip_address=random_ip_address(), 

149 ) 

150 PhotoAccessStat.objects.update_or_create( 

151 photo=photo2, defaults={"access_count": 150} 

152 ) 

153 

154 # Associate photos with emails/openids 

155 email1.photo = photo1 

156 email1.save() 

157 email2.photo = photo2 

158 email2.save() 

159 openid1.photo = photo1 

160 openid1.save() 

161 

162 # Create unconfirmed entries 

163 UnconfirmedEmail.objects.create( 

164 user=self.user, 

165 email=f"{random_string()}@{random_string()}.{random_string(2)}", 

166 ip_address=random_ip_address(), 

167 ) 

168 

169 UnconfirmedOpenId.objects.create( 

170 user=self.user, 

171 openid=f"http://{random_string()}.{random_string()}.org/", 

172 ip_address=random_ip_address(), 

173 ) 

174 

175 # Test the stats endpoint 

176 response = self.client.get("/stats/") 

177 self.assertEqual(response.status_code, 200, "unable to fetch stats!") 

178 j = json.loads(response.content) 

179 

180 # Test basic counts 

181 self.assertEqual(j["users"], 1, "user count incorrect") 

182 self.assertEqual(j["mails"], 2, "mails count incorrect") 

183 self.assertEqual(j["openids"], 1, "openids count incorrect") 

184 self.assertEqual(j["unconfirmed_mails"], 1, "unconfirmed mails count incorrect") 

185 self.assertEqual( 

186 j["unconfirmed_openids"], 1, "unconfirmed openids count incorrect" 

187 ) 

188 self.assertEqual(j["avatars"], 2, "avatars count incorrect") 

189 

190 # Test top viewed avatars 

191 self.assertIn("top_viewed_avatars", j, "top_viewed_avatars missing") 

192 self.assertEqual( 

193 len(j["top_viewed_avatars"]), 2, "should have 2 top viewed avatars" 

194 ) 

195 # The top viewed avatar should be the one with highest associated email/openid access count 

196 self.assertEqual( 

197 j["top_viewed_avatars"][0]["access_count"], 

198 100, 

199 "top avatar access count incorrect", 

200 ) 

201 # Check that avatar_url is present and starts with the correct base URL 

202 self.assertIn("avatar_url", j["top_viewed_avatars"][0], "avatar_url missing") 

203 self.assertTrue( 

204 j["top_viewed_avatars"][0]["avatar_url"].startswith( 

205 "https://libravatar.org/avatar/" 

206 ), 

207 "avatar_url should start with https://libravatar.org/avatar/", 

208 ) 

209 

210 # Check that avatar_url is present and starts with the correct base URL 

211 self.assertIn("avatar_url", j["top_queried_emails"][0], "avatar_url missing") 

212 self.assertTrue( 

213 j["top_queried_emails"][0]["avatar_url"].startswith( 

214 "https://libravatar.org/avatar/" 

215 ), 

216 "avatar_url should start with https://libravatar.org/avatar/", 

217 ) 

218 

219 # Check that avatar_url is present and starts with the correct base URL 

220 self.assertIn("avatar_url", j["top_queried_openids"][0], "avatar_url missing") 

221 self.assertTrue( 

222 j["top_queried_openids"][0]["avatar_url"].startswith( 

223 "https://libravatar.org/avatar/" 

224 ), 

225 "avatar_url should start with https://libravatar.org/avatar/", 

226 ) 

227 

228 # Test photo format distribution 

229 self.assertIn( 

230 "photo_format_distribution", j, "photo_format_distribution missing" 

231 ) 

232 formats = { 

233 item["format"]: item["count"] for item in j["photo_format_distribution"] 

234 } 

235 self.assertEqual(formats["png"], 2, "png format count incorrect") 

236 

237 # Test user activity stats 

238 self.assertIn("user_activity", j, "user_activity missing") 

239 self.assertEqual( 

240 j["user_activity"]["users_with_multiple_photos"], 

241 1, 

242 "users with multiple photos incorrect", 

243 ) 

244 self.assertEqual( 

245 j["user_activity"]["users_with_both_email_and_openid"], 

246 1, 

247 "users with both email and openid incorrect", 

248 ) 

249 self.assertEqual( 

250 j["user_activity"]["average_photos_per_user"], 

251 2.0, 

252 "average photos per user incorrect", 

253 ) 

254 

255 # Test Bluesky handles (should be empty) 

256 self.assertIn("bluesky_handles", j, "bluesky_handles missing") 

257 self.assertEqual( 

258 j["bluesky_handles"]["total_bluesky_handles"], 

259 0, 

260 "total bluesky handles should be 0", 

261 ) 

262 

263 # Test photo size stats 

264 self.assertIn("photo_size_stats", j, "photo_size_stats missing") 

265 self.assertGreater( 

266 j["photo_size_stats"]["average_size_bytes"], 

267 0, 

268 "average photo size should be > 0", 

269 ) 

270 self.assertEqual( 

271 j["photo_size_stats"]["total_photos_analyzed"], 

272 2, 

273 "total photos analyzed incorrect", 

274 ) 

275 

276 # Test potential duplicate photos 

277 self.assertIn( 

278 "potential_duplicate_photos", j, "potential_duplicate_photos missing" 

279 ) 

280 self.assertEqual( 

281 j["potential_duplicate_photos"]["potential_duplicate_groups"], 

282 1, 

283 "should have 1 duplicate group (same PNG data)", 

284 ) 

285 

286 def test_stats_edge_cases(self): 

287 """ 

288 Test edge cases for stats 

289 """ 

290 # Test with no data 

291 response = self.client.get("/stats/") 

292 self.assertEqual(response.status_code, 200, "unable to fetch stats!") 

293 j = json.loads(response.content) 

294 

295 # All lists should be empty 

296 self.assertEqual( 

297 len(j["top_viewed_avatars"]), 0, "top_viewed_avatars should be empty" 

298 ) 

299 self.assertEqual( 

300 len(j["top_queried_emails"]), 0, "top_queried_emails should be empty" 

301 ) 

302 self.assertEqual( 

303 len(j["top_queried_openids"]), 0, "top_queried_openids should be empty" 

304 ) 

305 self.assertEqual( 

306 len(j["photo_format_distribution"]), 

307 0, 

308 "photo_format_distribution should be empty", 

309 ) 

310 self.assertEqual( 

311 j["bluesky_handles"]["total_bluesky_handles"], 

312 0, 

313 "bluesky_handles should be 0", 

314 ) 

315 self.assertEqual( 

316 j["photo_size_stats"]["total_photos_analyzed"], 

317 0, 

318 "photo_size_stats should be 0", 

319 ) 

320 self.assertEqual( 

321 j["potential_duplicate_photos"]["potential_duplicate_groups"], 

322 0, 

323 "potential_duplicate_photos should be 0", 

324 ) 

325 

326 def test_stats_with_bluesky_handles(self): 

327 """ 

328 Test stats with Bluesky handles 

329 """ 

330 from ivatar.ivataraccount.models import ConfirmedEmail, ConfirmedOpenId 

331 

332 # Create email with Bluesky handle 

333 email = ConfirmedEmail.objects.create( 

334 user=self.user, 

335 email=f"{random_string()}@{random_string()}.{random_string(2)}", 

336 ip_address=random_ip_address(), 

337 ) 

338 email.bluesky_handle = f"{random_string()}.bsky.social" 

339 email.save() 

340 ConfirmedEmailAccessStat.objects.update_or_create( 

341 email=email, defaults={"access_count": 100} 

342 ) 

343 

344 # Create OpenID with Bluesky handle 

345 openid = ConfirmedOpenId.objects.create( 

346 user=self.user, 

347 openid=f"http://{random_string()}.{random_string()}.org/", 

348 ip_address=random_ip_address(), 

349 ) 

350 openid.bluesky_handle = f"{random_string()}.bsky.social" 

351 openid.save() 

352 ConfirmedOpenIdAccessStat.objects.update_or_create( 

353 openid=openid, defaults={"access_count": 50} 

354 ) 

355 

356 response = self.client.get("/stats/") 

357 self.assertEqual(response.status_code, 200, "unable to fetch stats!") 

358 j = json.loads(response.content) 

359 

360 # Test Bluesky handles stats 

361 self.assertEqual( 

362 j["bluesky_handles"]["total_bluesky_handles"], 

363 2, 

364 "total bluesky handles incorrect", 

365 ) 

366 self.assertEqual( 

367 j["bluesky_handles"]["bluesky_emails"], 1, "bluesky emails count incorrect" 

368 ) 

369 self.assertEqual( 

370 j["bluesky_handles"]["bluesky_openids"], 

371 1, 

372 "bluesky openids count incorrect", 

373 ) 

374 

375 def test_stats_photo_duplicates(self): 

376 """ 

377 Test potential duplicate photos detection 

378 """ 

379 from ivatar.ivataraccount.models import Photo 

380 

381 # Create photos with same format and size (potential duplicates) 

382 # PNG header + minimal data 

383 png_data = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x02\x00\x00\x00\x90wS\xde\x00\x00\x00\tpHYs\x00\x00\x0b\x13\x00\x00\x0b\x13\x01\x00\x9a\x9c\x18\x00\x00\x00\nIDATx\x9cc```\x00\x00\x00\x04\x00\x01\xdd\x8d\xb4\x1c\x00\x00\x00\x00IEND\xaeB`\x82" 

384 

385 Photo.objects.create( 

386 user=self.user, data=png_data, format="png", ip_address=random_ip_address() 

387 ) 

388 Photo.objects.create( 

389 user=self.user, 

390 data=png_data, # Same size 

391 format="png", # Same format 

392 ip_address=random_ip_address(), 

393 ) 

394 Photo.objects.create( 

395 user=self.user, 

396 data=png_data, # Same size but different format 

397 format="png", # Same format for testing 

398 ip_address=random_ip_address(), 

399 ) 

400 

401 response = self.client.get("/stats/") 

402 self.assertEqual(response.status_code, 200, "unable to fetch stats!") 

403 j = json.loads(response.content) 

404 

405 # Should detect potential duplicates 

406 self.assertEqual( 

407 j["potential_duplicate_photos"]["potential_duplicate_groups"], 

408 1, 

409 "should have 1 duplicate group", 

410 ) 

411 self.assertEqual( 

412 j["potential_duplicate_photos"]["total_potential_duplicate_photos"], 

413 3, 

414 "should have 3 potential duplicate photos", 

415 ) 

416 self.assertEqual( 

417 len(j["potential_duplicate_photos"]["potential_duplicate_groups_detail"]), 

418 1, 

419 "should have 1 duplicate group detail", 

420 )