""" Management command to test web activity tracking functionality. This command verifies that the UserActivityMiddleware properly tracks activity for both API (JWT) and web (session) authenticated users. """ from django.core.management.base import BaseCommand from django.contrib.auth import get_user_model from django.test import Client from django.core.cache import cache from rest_framework_simplejwt.tokens import RefreshToken from rest_framework.test import APIClient import json User = get_user_model() class Command(BaseCommand): help = 'Test web activity tracking functionality' def add_arguments(self, parser): parser.add_argument( '--inspect', action='store_true', help='Show detailed inspection of tracking behavior', ) def handle(self, *args, **options): self.stdout.write(self.style.SUCCESS('Testing web activity tracking...')) # Create test user test_user, created = User.objects.get_or_create( phone="1234567890", defaults={ 'email': "webtest@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 any existing activity User.objects.filter(pk=test_user.pk).update(last_active_date=None) cache.clear() test_user.refresh_from_db() self.stdout.write(f"Initial last_active_date: {test_user.last_active_date}") # Test 1: Web session authentication self.stdout.write("\n" + "="*50) self.stdout.write(self.style.WARNING("Test 1: Web Session Authentication")) self.stdout.write("="*50) web_client = Client() # Test unauthenticated request (should not track) self.stdout.write("Testing unauthenticated web request...") response = web_client.get('/') test_user.refresh_from_db() self.stdout.write(f"After unauthenticated request: {test_user.last_active_date}") # Test authenticated web session self.stdout.write("Testing web login...") login_success = web_client.login(phone='1234567890', password='testpass123') self.stdout.write(f"Login successful: {login_success}") if login_success: # Test authenticated web request self.stdout.write("Testing authenticated web request to /bazi/...") try: response = web_client.get('/bazi/') self.stdout.write(f"Response status: {response.status_code}") test_user.refresh_from_db() self.stdout.write(f"After web session request: {test_user.last_active_date}") if test_user.last_active_date: self.stdout.write(self.style.SUCCESS("✅ Web session tracking WORKS!")) else: self.stdout.write(self.style.ERROR("❌ Web session tracking FAILED")) except Exception as e: self.stdout.write(f"Web request failed: {e}") # Test 2: API JWT authentication self.stdout.write("\n" + "="*50) self.stdout.write(self.style.WARNING("Test 2: API JWT Authentication")) self.stdout.write("="*50) # Clear activity and cache for clean test User.objects.filter(pk=test_user.pk).update(last_active_date=None) cache.clear() test_user.refresh_from_db() api_client = APIClient() # Generate JWT token refresh = RefreshToken.for_user(test_user) access_token = str(refresh.access_token) self.stdout.write(f"Generated JWT token: {access_token[:50]}...") # Test authenticated API request api_client.credentials(HTTP_AUTHORIZATION=f'Bearer {access_token}') try: response = api_client.get('/api/bazi/') self.stdout.write(f"API response status: {response.status_code}") test_user.refresh_from_db() self.stdout.write(f"After API JWT request: {test_user.last_active_date}") if test_user.last_active_date: self.stdout.write(self.style.SUCCESS("✅ API JWT tracking WORKS!")) else: self.stdout.write(self.style.ERROR("❌ API JWT tracking FAILED")) except Exception as e: self.stdout.write(f"API request failed: {e}") # Test 3: Mixed requests self.stdout.write("\n" + "="*50) self.stdout.write(self.style.WARNING("Test 3: Mixed Web + API Requests")) self.stdout.write("="*50) # Clear activity for clean test User.objects.filter(pk=test_user.pk).update(last_active_date=None) cache.clear() test_user.refresh_from_db() # Web request first web_client.login(phone='1234567890', password='testpass123') web_response = web_client.get('/bazi/') test_user.refresh_from_db() web_time = test_user.last_active_date self.stdout.write(f"After web request: {web_time}") # Clear cache and make API request cache.clear() api_response = api_client.get('/api/bazi/') test_user.refresh_from_db() api_time = test_user.last_active_date self.stdout.write(f"After API request: {api_time}") if web_time and api_time: self.stdout.write(self.style.SUCCESS("✅ Mixed tracking WORKS!")) else: self.stdout.write(self.style.ERROR("❌ Mixed tracking has issues")) # Test 4: Request filtering if options['inspect']: self.stdout.write("\n" + "="*50) self.stdout.write(self.style.WARNING("Test 4: Request Filtering Inspection")) self.stdout.write("="*50) from main.middleware import UserActivityMiddleware from django.test import RequestFactory factory = RequestFactory() middleware = UserActivityMiddleware(get_response=lambda r: None) test_requests = [ '/', '/bazi/', '/api/bazi/', '/static/css/style.css', '/media/image.png', '/favicon.ico', '/health/', ] for path in test_requests: request = factory.get(path) should_track = middleware.should_track_request(request) status = "✅ TRACK" if should_track else "❌ SKIP" self.stdout.write(f"{status}: {path}") self.stdout.write("\n" + "="*50) self.stdout.write(self.style.SUCCESS("Activity tracking test completed!")) self.stdout.write("="*50) # Cleanup test user if created if created: test_user.delete() self.stdout.write("Cleaned up test user")