Coverage for ivatar/test_access_stats.py: 98%

1import base64 

2from django.test import TransactionTestCase, Client, override_settings 

3from django.contrib.auth.models import User 

4from ivatar.ivataraccount.models import ( 

5 ConfirmedEmail, 

6 Photo, 

7 ConfirmedEmailAccessStat, 

8) 

9from ivatar.access_stats import stats_manager 

10from django.core.cache import cache 

11import time 

12from unittest.mock import patch 

13 

14 

15@override_settings( 

16 CACHES={ 

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

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

19 }, 

20 CACHE_RESPONSE=False, 

21 STATS_BATCH_SIZE=100, 

22 STATS_FLUSH_TIMEOUT=5, # Short timeout for testing 

23 ASYNC_ACCESS_COUNT=True, 

24) 

25class BatchStatsTest(TransactionTestCase): 

26 def setUp(self): 

27 self.username = "testuser_" + str(time.time()) 

28 self.password = "password" 

29 self.user = User.objects.create_user( 

30 username=self.username, password=self.password 

31 ) 

32 self.client = Client() 

33 

34 # Create a confirmed email and photo 

35 self.email_address = f"test_{time.time()}@example.com" 

36 self.email = ConfirmedEmail.objects.create( 

37 user=self.user, email=self.email_address, ip_address="127.0.0.1" 

38 ) 

39 

40 # Create a dummy photo (minimal PNG, Base64-encoded for pycco compatibility) 

41 png_base64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAIAAACQd1PeAAAACXBIWXMAAAsTAAALEwEAmpwYAAAACklEQVQIHWNgYAAAAAMAAWveibQAAAAASUVORK5CYII=" 

42 png_data = base64.b64decode(png_base64) 

43 self.photo = Photo.objects.create( 

44 user=self.user, data=png_data, format="png", ip_address="127.0.0.1" 

45 ) 

46 self.email.photo = self.photo 

47 self.email.save() 

48 

49 # Clear keys for this test 

50 model_name = self.email.__class__.__name__ 

51 pk = self.email.pk 

52 self.email_key = f"stats:access_count:{model_name}:{pk}" 

53 self.start_time_key = f"stats:start_time:{model_name}:{pk}" 

54 

55 cache.delete(self.email_key) 

56 cache.delete(self.start_time_key) 

57 

58 # Ensure stats_manager has clean state 

59 stats_manager._local_dirty_keys.clear() 

60 stats_manager._local_start_times.clear() 

61 

62 # Mock threading.Thread to run synchronously 

63 self.thread_patcher = patch("ivatar.access_stats.threading.Thread") 

64 self.mock_thread_cls = self.thread_patcher.start() 

65 

66 # Set up side effect to run target immediately 

67 def side_effect(target=None, args=(), **kwargs): 

68 class SyncThread: 

69 def start(self): 

70 if target: 

71 target(*args) 

72 

73 return SyncThread() 

74 

75 self.mock_thread_cls.side_effect = side_effect 

76 

77 def tearDown(self): 

78 self.thread_patcher.stop() 

79 

80 def _get_email_access_count(self): 

81 """Read access_count from the stat table for self.email.""" 

82 try: 

83 return ConfirmedEmailAccessStat.objects.get(email=self.email).access_count 

84 except ConfirmedEmailAccessStat.DoesNotExist: 

85 return 0 

86 

87 @override_settings(ASYNC_ACCESS_COUNT=False) 

88 def test_sync_update_when_disabled(self): 

89 """ 

90 Verify that updates are synchronous when feature flag is disabled. 

91 """ 

92 stats_manager.update_access_count(self.email) 

93 

94 # Check stat table updated immediately 

95 self.assertEqual(self._get_email_access_count(), 1) 

96 

97 # Check cache should be empty (or at least not used for this logic) 

98 self.assertIsNone(cache.get(self.email_key)) 

99 

100 def test_normal_increment_cached(self): 

101 """ 

102 Verify that a single request increments cache but does not update DB immediately. 

103 """ 

104 stats_manager.update_access_count(self.email) 

105 

106 # Check cache 

107 self.assertEqual(cache.get(self.email_key), 1) 

108 

109 # Check stat table (should be 0) 

110 self.assertEqual(self._get_email_access_count(), 0) 

111 

112 def test_batch_flush(self): 

113 """ 

114 Verify that hitting the batch limit triggers a DB update. 

115 """ 

116 # Increment 99 times (limit is 100) 

117 for _ in range(99): 

118 stats_manager.update_access_count(self.email) 

119 

120 self.assertEqual(cache.get(self.email_key), 99) 

121 self.assertEqual(self._get_email_access_count(), 0) 

122 

123 # 100th increment -> Triggers flush (synchronously due to mock) 

124 stats_manager.update_access_count(self.email) 

125 

126 # No wait needed 

127 

128 # Check stat table 

129 self.assertEqual(self._get_email_access_count(), 100) 

130 

131 # Check cache reset 

132 self.assertEqual(cache.get(self.email_key), 0) 

133 

134 def test_time_flush(self): 

135 """ 

136 Verify that the time-based flush works. 

137 Timeout set to 5 seconds in override_settings. 

138 """ 

139 # 1. First increment sets start time (worker-local) 

140 stats_manager.update_access_count(self.email) 

141 

142 # Verify start time is tracked locally 

143 self.assertIsNotNone(stats_manager._local_start_times.get(self.email_key)) 

144 

145 # Verify no DB update yet 

146 self.assertEqual(self._get_email_access_count(), 0) 

147 

148 # 2. Wait for timeout > 5s 

149 time.sleep(6) 

150 

151 # 3. Next increment should trigger flush due to timeout (synchronously) 

152 stats_manager.update_access_count(self.email) 

153 

154 # No wait needed 

155 

156 # Check stat table: should have 2 updates (1 from before wait + 1 from trigger) 

157 self.assertEqual(self._get_email_access_count(), 2) 

158 

159 # Check cache reset 

160 self.assertEqual(cache.get(self.email_key), 0) 

161 # Start time should be cleared 

162 self.assertIsNone(stats_manager._local_start_times.get(self.email_key)) 

163 

164 def test_time_flush_rearms_after_flush(self): 

165 """ 

166 Verify the time-based trigger fires repeatedly, not just once per key. 

167 The old cache-based start time was never rewritten after the first 

168 flush (incr on the now-zero key returns 1 without raising), so 

169 subsequent timeout windows silently never flushed. 

170 """ 

171 # Window 1: arm + trigger 

172 stats_manager.update_access_count(self.email) 

173 time.sleep(6) 

174 stats_manager.update_access_count(self.email) 

175 self.assertEqual(self._get_email_access_count(), 2) 

176 

177 # Window 2: must flush again 

178 stats_manager.update_access_count(self.email) 

179 time.sleep(6) 

180 stats_manager.update_access_count(self.email) 

181 self.assertEqual(self._get_email_access_count(), 4) 

182 

183 def test_atexit_handler(self): 

184 """ 

185 Verify that the cleanup handler flushes remaining counts. 

186 """ 

187 # Add some counts 

188 stats_manager.update_access_count(self.email) 

189 stats_manager.update_access_count(self.email) 

190 stats_manager.update_access_count(self.email) 

191 

192 self.assertEqual(cache.get(self.email_key), 3) 

193 self.assertEqual(self._get_email_access_count(), 0) 

194 

195 # Manually call the handler 

196 stats_manager.flush_all_dirty_keys() 

197 

198 # Check stat table 

199 self.assertEqual(self._get_email_access_count(), 3) 

200 

201 # Check cache decremented 

202 self.assertEqual(cache.get(self.email_key), 0)