""" Management command to test the systematic activity tracking system. This command verifies that the signal-based activity tracking works for both web (session) and API (JWT) authentication methods. """ from django.core.management.base import BaseCommand from django.contrib.auth import get_user_model, login, logout from django.test import Client, RequestFactory from django.core.cache import cache from rest_framework_simplejwt.tokens import RefreshToken from rest_framework.test import APIClient from main.signals import fire_jwt_authenticated_signal import json User = get_user_model() class Command(BaseCommand): help = 'Test the systematic activity tracking functionality' def add_arguments(self, parser): parser.add_argument( '--comprehensive', action='store_true', help='Run comprehensive tests including signal verification', ) def handle(self, *args, **options): self.stdout.write(self.style.SUCCESS('Testing systematic activity tracking...')) # Create or get test user test_user, created = User.objects.get_or_create( phone="1234567890", defaults={ 'email': "systemtest@example.com", 'is_active': True } ) if created: test_user.set_password("testpass123") test_user.save() self.stdout.write(f"Created test user: {test_user.phone}") else: self.stdout.write(f"Using existing test user: {test_user.phone}") # Clear initial state self._clear_user_activity(test_user) # Test 1: Signal-based tracking self.stdout.write("\n" + "="*60) self.stdout.write(self.style.WARNING("Test 1: Signal-Based Activity Tracking")) self.stdout.write("="*60) self._test_django_signals(test_user) # Test 2: JWT Authentication Backend self.stdout.write("\n" + "="*60) self.stdout.write(self.style.WARNING("Test 2: JWT Authentication Backend")) self.stdout.write("="*60) self._test_jwt_authentication_backend(test_user) # Test 3: Web session authentication self.stdout.write("\n" + "="*60) self.stdout.write(self.style.WARNING("Test 3: Web Session Authentication")) self.stdout.write("="*60) self._test_web_session_authentication(test_user) if options['comprehensive']: # Test 4: Comprehensive API testing self.stdout.write("\n" + "="*60) self.stdout.write(self.style.WARNING("Test 4: Comprehensive API Testing")) self.stdout.write("="*60) self._test_comprehensive_api(test_user) # Final summary self.stdout.write("\n" + "="*60) self.stdout.write(self.style.SUCCESS("Systematic Activity Tracking Test Summary")) self.stdout.write("="*60) test_user.refresh_from_db() self.stdout.write(f"Final user activity state: {test_user.last_active_date}") if test_user.last_active_date: self.stdout.write(self.style.SUCCESS("✅ SYSTEMATIC TRACKING IS WORKING!")) self.stdout.write("All authentication methods automatically track activity.") else: self.stdout.write(self.style.ERROR("❌ No activity tracked - check signal configuration")) # Cleanup if created: test_user.delete() self.stdout.write("Cleaned up test user") def _clear_user_activity(self, user): """Clear user activity and cache for clean testing.""" User.objects.filter(pk=user.pk).update(last_active_date=None) cache.clear() user.refresh_from_db() self.stdout.write(f"Cleared activity for user {user.pk}") def _test_django_signals(self, user): """Test Django's built-in authentication signals.""" from django.contrib.auth.signals import user_logged_in, user_logged_out from django.test import RequestFactory factory = RequestFactory() request = factory.get('/') request.user = user # Clear state self._clear_user_activity(user) # Test user_logged_in signal self.stdout.write("Testing user_logged_in signal...") user_logged_in.send(sender=User, request=request, user=user) user.refresh_from_db() if user.last_active_date: self.stdout.write(self.style.SUCCESS("✅ user_logged_in signal tracking works")) else: self.stdout.write(self.style.ERROR("❌ user_logged_in signal tracking failed")) # Test custom JWT signal self.stdout.write("Testing custom jwt_authenticated signal...") self._clear_user_activity(user) fire_jwt_authenticated_signal(user, request) user.refresh_from_db() if user.last_active_date: self.stdout.write(self.style.SUCCESS("✅ jwt_authenticated signal tracking works")) else: self.stdout.write(self.style.ERROR("❌ jwt_authenticated signal tracking failed")) def _test_jwt_authentication_backend(self, user): """Test the custom JWT authentication backend.""" try: from main.authentication import ActivityTrackingJWTAuthentication from django.test import RequestFactory factory = RequestFactory() auth_backend = ActivityTrackingJWTAuthentication() # Generate a token refresh = RefreshToken.for_user(user) access_token = str(refresh.access_token) # Create request with JWT token request = factory.get('/api/test/') request.META['HTTP_AUTHORIZATION'] = f'Bearer {access_token}' # Clear state self._clear_user_activity(user) # Test authentication self.stdout.write("Testing JWT authentication backend...") result = auth_backend.authenticate(request) if result: authenticated_user, token = result self.stdout.write(f"JWT authentication successful for user: {authenticated_user.pk}") # Check if activity was tracked user.refresh_from_db() if user.last_active_date: self.stdout.write(self.style.SUCCESS("✅ JWT backend activity tracking works")) else: self.stdout.write(self.style.ERROR("❌ JWT backend activity tracking failed")) else: self.stdout.write(self.style.ERROR("❌ JWT authentication failed")) except Exception as e: self.stdout.write(self.style.ERROR(f"❌ JWT backend test error: {e}")) def _test_web_session_authentication(self, user): """Test web session-based authentication.""" self._clear_user_activity(user) # Create a web client client = Client() self.stdout.write("Testing web session login...") try: # Attempt login (this should fire the user_logged_in signal) login_success = client.login(phone='1234567890', password='testpass123') if login_success: self.stdout.write("Web login successful") # Check if signal was fired and activity tracked user.refresh_from_db() if user.last_active_date: self.stdout.write(self.style.SUCCESS("✅ Web session activity tracking works")) else: self.stdout.write(self.style.WARNING("⚠️ Web login signal may not have fired")) # Test a protected view access response = client.get('/') if response.status_code in [200, 302]: # 302 for redirects self.stdout.write("Web view access successful") else: self.stdout.write(self.style.ERROR("❌ Web login failed")) except Exception as e: self.stdout.write(self.style.ERROR(f"❌ Web session test error: {e}")) def _test_comprehensive_api(self, user): """Test comprehensive API functionality.""" api_client = APIClient() # Test API login endpoint self.stdout.write("Testing API login endpoint...") try: login_data = { 'phone': '1234567890', 'password': 'testpass123' } self._clear_user_activity(user) response = api_client.post('/api/user/login/', login_data, format='json') self.stdout.write(f"API login response: {response.status_code}") if response.status_code == 200: try: response_data = json.loads(response.content.decode()) if 'access' in response_data: token = response_data['access'] self.stdout.write("API login successful, token received") # Test authenticated API request api_client.credentials(HTTP_AUTHORIZATION=f'Bearer {token}') api_response = api_client.get('/api/user/profile/') self.stdout.write(f"Authenticated API request: {api_response.status_code}") # Check activity tracking user.refresh_from_db() if user.last_active_date: self.stdout.write(self.style.SUCCESS("✅ API authentication activity tracking works")) else: self.stdout.write(self.style.ERROR("❌ API authentication activity tracking failed")) except json.JSONDecodeError: self.stdout.write("API login response not valid JSON") except Exception as e: self.stdout.write(self.style.ERROR(f"❌ API test error: {e}")) def _check_signal_registration(self): """Check if signals are properly registered.""" from django.dispatch import Signal from main.signals import jwt_authenticated self.stdout.write("Checking signal registration...") # Check if our custom signal exists if isinstance(jwt_authenticated, Signal): self.stdout.write(self.style.SUCCESS("✅ Custom JWT signal is registered")) else: self.stdout.write(self.style.ERROR("❌ Custom JWT signal not found")) # Check if signal receivers are connected from django.contrib.auth.signals import user_logged_in receivers = user_logged_in._live_receivers(sender=User) if receivers: self.stdout.write(f"✅ Found {len(receivers)} receivers for user_logged_in signal") else: self.stdout.write(self.style.ERROR("❌ No receivers found for user_logged_in signal"))