""" Tests for Redis user activity caching functionality. """ from datetime import timedelta from unittest.mock import MagicMock, patch from django.contrib.auth import get_user_model from django.test import TestCase from django.utils import timezone from users.functions.last_activity import ( SYNC_THRESHOLD, USER_ACTIVITY_COUNTER_KEY, _bulk_update_user_activities, _get_user_activity_key, cleanup_redis_activity_data, clear_batch_user_ids, get_activity_counter, get_batch_user_ids, get_user_last_activity, increment_activity_counter, reset_activity_counter, set_user_last_activity, should_sync_activities, ) User = get_user_model() class TestRedisActivity(TestCase): """Test Redis activity functions.""" def setUp(self): self.user = User.objects.create_user(email='test@example.com', username='testuser', password='testpass123') self.test_time = timezone.now() def tearDown(self): # Clean up Redis data after each test with patch('users.functions.last_activity.redis_connected', return_value=True): cleanup_redis_activity_data({self.user.id}) reset_activity_counter() @patch('users.functions.last_activity.redis_connected', return_value=True) @patch('users.functions.last_activity.get_connection') def test_set_user_last_activity_success(self, mock_get_connection, mock_redis_connected): """Test successful setting of user activity.""" mock_redis_client = MagicMock() mock_get_connection.return_value = mock_redis_client mock_redis_client.incr.return_value = 1 result = set_user_last_activity(self.user.id, self.test_time) self.assertTrue(result) mock_redis_client.setex.assert_called_once() mock_redis_client.sadd.assert_called_once() self.assertEqual(mock_redis_client.expire.call_count, 2) # One for batch key, one for counter mock_redis_client.incr.assert_called_once() @patch('users.functions.last_activity.redis_connected', return_value=False) def test_set_user_last_activity_redis_disconnected(self, mock_redis_connected): """Test setting activity when Redis is disconnected.""" result = set_user_last_activity(self.user.id, self.test_time) self.assertFalse(result) @patch('users.functions.last_activity.redis_connected', return_value=True) @patch('users.functions.last_activity.get_connection') def test_get_user_last_activity_from_redis(self, mock_get_connection, mock_redis_connected): """Test getting user activity from Redis.""" with patch('users.functions.last_activity._redis', MagicMock()): mock_redis_client = MagicMock() mock_get_connection.return_value = mock_redis_client mock_redis_client.get.return_value = self.test_time.isoformat().encode('utf-8') result = get_user_last_activity(self.user.id) self.assertEqual(result, self.test_time) mock_redis_client.get.assert_called_once() @patch('users.functions.last_activity.redis_connected', return_value=True) @patch('users.functions.last_activity.get_connection') def test_get_user_last_activity_no_redis_data(self, mock_get_connection, mock_redis_connected): """Test getting activity when Redis has no data (returns None).""" with patch('users.functions.last_activity._redis', MagicMock()): mock_redis_client = MagicMock() mock_get_connection.return_value = mock_redis_client mock_redis_client.get.return_value = None result = get_user_last_activity(self.user.id) self.assertIsNone(result) mock_redis_client.get.assert_called_once() @patch('users.functions.last_activity.redis_connected', return_value=False) def test_get_user_last_activity_redis_disconnected(self, mock_redis_connected): """Test getting activity when Redis is disconnected (returns None).""" with patch('users.functions.last_activity._redis', None): result = get_user_last_activity(self.user.id) self.assertIsNone(result) @patch('users.functions.last_activity.redis_connected', return_value=True) @patch('users.functions.last_activity.get_connection') def test_increment_activity_counter(self, mock_get_connection, mock_redis_connected): """Test incrementing activity counter.""" mock_redis_client = MagicMock() mock_get_connection.return_value = mock_redis_client mock_redis_client.incr.return_value = 5 result = increment_activity_counter() self.assertEqual(result, 5) mock_redis_client.incr.assert_called_with(USER_ACTIVITY_COUNTER_KEY) mock_redis_client.expire.assert_called_once() @patch('users.functions.last_activity.redis_connected', return_value=True) @patch('users.functions.last_activity.get_connection') def test_get_activity_counter(self, mock_get_connection, mock_redis_connected): """Test getting activity counter.""" mock_redis_client = MagicMock() mock_get_connection.return_value = mock_redis_client mock_redis_client.get.return_value = b'10' result = get_activity_counter() self.assertEqual(result, 10) mock_redis_client.get.assert_called_once() @patch('users.functions.last_activity.redis_connected', return_value=True) @patch('users.functions.last_activity.get_connection') def test_reset_activity_counter(self, mock_get_connection, mock_redis_connected): """Test resetting activity counter.""" mock_redis_client = MagicMock() mock_get_connection.return_value = mock_redis_client result = reset_activity_counter() self.assertTrue(result) mock_redis_client.set.assert_called_once_with(USER_ACTIVITY_COUNTER_KEY, 0) mock_redis_client.expire.assert_called_once() @patch('users.functions.last_activity.redis_connected', return_value=True) @patch('users.functions.last_activity.get_connection') def test_get_batch_user_ids(self, mock_get_connection, mock_redis_connected): """Test getting batch user IDs.""" mock_redis_client = MagicMock() mock_get_connection.return_value = mock_redis_client mock_redis_client.smembers.return_value = {b'1', b'2', b'3'} result = get_batch_user_ids() self.assertEqual(result, {1, 2, 3}) mock_redis_client.smembers.assert_called_once() @patch('users.functions.last_activity.redis_connected', return_value=True) @patch('users.functions.last_activity.get_connection') def test_clear_batch_user_ids(self, mock_get_connection, mock_redis_connected): """Test clearing batch user IDs.""" mock_redis_client = MagicMock() mock_get_connection.return_value = mock_redis_client result = clear_batch_user_ids({1, 2, 3}) self.assertTrue(result) mock_redis_client.srem.assert_called_once() @patch('users.functions.last_activity.get_activity_counter') def test_should_sync_activities(self, mock_get_counter): """Test sync threshold check.""" # Below threshold mock_get_counter.return_value = SYNC_THRESHOLD - 1 self.assertFalse(should_sync_activities()) # At threshold mock_get_counter.return_value = SYNC_THRESHOLD self.assertTrue(should_sync_activities()) # Above threshold mock_get_counter.return_value = SYNC_THRESHOLD + 1 self.assertTrue(should_sync_activities()) def test_get_user_activity_key(self): """Test Redis key generation.""" expected_key = f'user_activity:{self.user.id}' result = _get_user_activity_key(self.user.id) self.assertEqual(result, expected_key) class TestUserLastActivityMixin(TestCase): """Test UserLastActivityMixin methods.""" def setUp(self): self.user = User.objects.create_user(email='test@example.com', username='testuser', password='testpass123') @patch('users.models.set_user_last_activity') @patch('users.models.schedule_activity_sync') def test_update_last_activity_redis_success(self, mock_schedule, mock_set_activity): """Test updating last activity with Redis success.""" mock_set_activity.return_value = True self.user.update_last_activity() mock_set_activity.assert_called_once() mock_schedule.assert_called_once() @patch('users.models.set_user_last_activity') def test_update_last_activity_redis_failure(self, mock_set_activity): """Test updating last activity with Redis failure (fallback to DB).""" mock_set_activity.return_value = False original_activity = self.user.last_activity self.user.update_last_activity() mock_set_activity.assert_called_once() # Check that database was updated self.user.refresh_from_db() self.assertNotEqual(self.user.last_activity, original_activity) @patch('users.models.get_user_last_activity') def test_get_last_activity_cached(self, mock_get_activity): """Test getting cached last activity.""" test_time = timezone.now() mock_get_activity.return_value = test_time result = self.user.get_last_activity() self.assertEqual(result, test_time) mock_get_activity.assert_called_once_with(self.user.id) @patch('users.models.get_user_last_activity') def test_last_activity_cached_property(self, mock_get_activity): """Test last_activity_cached property.""" test_time = timezone.now() mock_get_activity.return_value = test_time result = self.user.last_activity_cached self.assertEqual(result, test_time) mock_get_activity.assert_called_once_with(self.user.id) class TestSyncTasks(TestCase): """Test synchronization tasks.""" def setUp(self): self.user = User.objects.create_user(email='test@example.com', username='testuser', password='testpass123') def test_bulk_update_user_activities_success(self): """Test bulk updating user activities.""" test_time = timezone.now() activities = [{'user_id': self.user.id, 'last_activity': test_time}] result = _bulk_update_user_activities(activities) self.assertTrue(result['success']) self.assertEqual(result['processed'], 1) self.assertEqual(result['errors'], 0) # Check database was updated self.user.refresh_from_db() self.assertEqual(self.user.last_activity, test_time) def test_bulk_update_user_activities_nonexistent_user(self): """Test bulk update with non-existent user.""" test_time = timezone.now() activities = [{'user_id': 99999, 'last_activity': test_time}] result = _bulk_update_user_activities(activities) self.assertTrue(result['success']) self.assertEqual(result['processed'], 0) self.assertEqual(result['errors'], 1) def test_bulk_update_user_activities_outdated_activity(self): """Test bulk update with outdated activity.""" # Set user's current activity to a future time future_time = timezone.now() + timedelta(hours=1) self.user.last_activity = future_time self.user.save() # Try to update with older time past_time = timezone.now() - timedelta(hours=1) activities = [{'user_id': self.user.id, 'last_activity': past_time}] result = _bulk_update_user_activities(activities) self.assertTrue(result['success']) self.assertEqual(result['processed'], 1) self.assertEqual(result['errors'], 0) self.assertEqual(result['updated'], 0) # No actual update performed # Check database was not updated self.user.refresh_from_db() self.assertEqual(self.user.last_activity, future_time) def test_bulk_update_user_activities_empty_list(self): """Test bulk update with empty activities list.""" result = _bulk_update_user_activities([]) self.assertTrue(result['success']) self.assertEqual(result['processed'], 0) self.assertEqual(result['errors'], 0)