#!/usr/bin/env python """ Comprehensive unit tests for temporary user API flows. Tests all the scenarios for temporary user creation, data management, and conversion via API. """ import os import django from django.test import TestCase, Client from rest_framework.test import APIClient from rest_framework import status from django.urls import reverse from django.contrib.auth import get_user_model from django.contrib.auth.models import AnonymousUser from django.utils import timezone from datetime import datetime, date, time import json import uuid # Setup Django os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'iching.settings') os.environ.setdefault('DJANGO_ENV', 'development') from main.models import User from accounts.models import UserProfile from bazi.models import Person from liuyao.models import liuyao from accounts.utils import create_temporary_user class TempUserAPIFlowTestCase(TestCase): """Base test case with common setup for temporary user API flows""" def setUp(self): """Set up test data""" self.client = APIClient() # Test data for BaZi API self.bazi_api_data = { 'name': 'API Test Person', 'gender': 'M', 'birth_date': '1990-01-01', 'birth_time': '12:00:00', 'lunar': False, 'notes': 'API test notes' } # Test data for Number API self.number_api_data = { 'name': 'API Number Test', 'gender': 'F', 'birth_date': '1985-06-15', 'lunar': False } # Test data for LiuYao API self.liuyao_api_data = { 'question': 'API test question', 'y1': '0', 'y2': '0', 'y3': '0', 'y4': '000', 'y5': '0', 'y6': '1', 'usecur': True # Use current time instead of qdate } # Create a regular user for login tests self.regular_user = User.objects.create_user( phone='1234567890', email='regular@example.com', password='testpass123', first_name='Regular', last_name='User' ) # UserProfile automatically created by signals class TempUserCreationAPITests(TempUserAPIFlowTestCase): """Test temporary user creation via API endpoints""" def test_bazi_api_unauthenticated_calculation_only(self): """Test that BaZi API only does calculation for unauthenticated users""" initial_user_count = User.objects.count() initial_person_count = Person.objects.count() response = self.client.post( reverse('api:bazi-list'), self.bazi_api_data, format='json' ) # Should return calculation-only result (HTTP 200) self.assertEqual(response.status_code, status.HTTP_200_OK) response_data = response.json() # Should contain calculation fields but no temp_user data self.assertIn('message', response_data) self.assertNotIn('temp_user', response_data) self.assertNotIn('id', response_data) # No record saved # Verify no new users or records created self.assertEqual(User.objects.count(), initial_user_count) self.assertEqual(Person.objects.count(), initial_person_count) def test_number_api_unauthenticated_calculation_only(self): """Test that Number API (via BaZi) only does calculation for unauthenticated users""" initial_user_count = User.objects.count() initial_person_count = Person.objects.count() response = self.client.post( reverse('api:bazi-list'), self.number_api_data, format='json' ) # Should return calculation-only result self.assertEqual(response.status_code, status.HTTP_200_OK) response_data = response.json() # Should not create temp user or save records self.assertNotIn('temp_user', response_data) self.assertEqual(User.objects.count(), initial_user_count) self.assertEqual(Person.objects.count(), initial_person_count) def test_liuyao_api_unauthenticated_calculation_only(self): """Test that LiuYao API only does calculation for unauthenticated users""" initial_user_count = User.objects.count() initial_liuyao_count = liuyao.objects.count() response = self.client.post( reverse('api:liuyao-calc'), self.liuyao_api_data, format='json' ) # Should return calculation-only result self.assertEqual(response.status_code, status.HTTP_200_OK) response_data = response.json() # Should contain calculation fields but no temp_user data self.assertIn('gua', response_data) self.assertIn('question', response_data) self.assertNotIn('temp_user', response_data) self.assertNotIn('id', response_data) # No record saved # Verify no new users or records created self.assertEqual(User.objects.count(), initial_user_count) self.assertEqual(liuyao.objects.count(), initial_liuyao_count) def test_direct_temp_user_creation(self): """Test creating temp user via direct API endpoint""" response = self.client.post( reverse('api:temp_user_create'), {'name': 'Direct API Test User'}, format='json' ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) response_data = response.json() # Check response contains JWT tokens and user info self.assertIn('user_id', response_data) self.assertIn('access', response_data) self.assertIn('refresh', response_data) self.assertIn('is_temporary', response_data) self.assertTrue(response_data['is_temporary']) # Verify temp user in database temp_user = User.objects.get(id=response_data['user_id']) self.assertTrue(temp_user.is_temporary_user) self.assertTrue(temp_user.phone.startswith('temp_')) class TempUserJWTAuthAPITests(TempUserAPIFlowTestCase): """Test JWT authentication for temporary users""" def _create_temp_user_with_jwt(self): """Helper to create temp user via direct API and get JWT tokens""" response = self.client.post( reverse('api:temp_user_create'), {'name': 'JWT Test User'}, format='json' ) return response.json() def test_temp_user_jwt_authentication(self): """Test using JWT tokens for subsequent API calls""" # Create temp user via direct API temp_user_data = self._create_temp_user_with_jwt() # Use access token for authenticated request self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") # Create a BaZi record using authenticated API response = self.client.post( reverse('api:bazi-list'), self.bazi_api_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) # Try to list user's BaZi records response = self.client.get(reverse('api:bazi-list')) self.assertEqual(response.status_code, status.HTTP_200_OK) response_data = response.json() self.assertEqual(len(response_data['results']), 1) # Should see the created record def test_temp_user_jwt_token_refresh(self): """Test refreshing JWT tokens for temporary user""" # Create temp user via direct API temp_user_data = self._create_temp_user_with_jwt() # Use refresh token to get new access token refresh_response = self.client.post( reverse('api:token_refresh'), {'refresh': temp_user_data['refresh']}, format='json' ) self.assertEqual(refresh_response.status_code, status.HTTP_200_OK) refresh_data = refresh_response.json() self.assertIn('access', refresh_data) # Use new access token to create record self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {refresh_data['access']}") response = self.client.post( reverse('api:bazi-list'), self.bazi_api_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) class TempUserOwnershipAPITests(TempUserAPIFlowTestCase): """Test owner flag behavior for temporary users via API""" def test_first_bazi_record_owner_true_api(self): """Test first BaZi record via authenticated API has owner=True""" # Create temp user via direct API temp_user_response = self.client.post( reverse('api:temp_user_create'), {'name': 'Owner Test User'}, format='json' ) temp_user_data = temp_user_response.json() # Authenticate and create first record self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") response = self.client.post( reverse('api:bazi-list'), self.bazi_api_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) temp_user = User.objects.get(id=temp_user_data['user_id']) first_record = Person.objects.filter(created_by=temp_user).first() self.assertTrue(first_record.owner) def test_subsequent_bazi_records_owner_false_api(self): """Test subsequent BaZi records via authenticated API have owner=False""" # Create temp user via direct API temp_user_response = self.client.post( reverse('api:temp_user_create'), {'name': 'Multiple Records Test'}, format='json' ) temp_user_data = temp_user_response.json() # Authenticate with temp user self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") # Create first record response = self.client.post( reverse('api:bazi-list'), self.bazi_api_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) # Create second record second_data = self.bazi_api_data.copy() second_data['name'] = 'Second API Person' response = self.client.post( reverse('api:bazi-list'), second_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) temp_user = User.objects.get(id=temp_user_data['user_id']) records = Person.objects.filter(created_by=temp_user).order_by('created_at') self.assertEqual(records.count(), 2) self.assertTrue(records[0].owner) # First record self.assertFalse(records[1].owner) # Second record def test_multiple_api_records_not_overwritten(self): """Test multiple API records with same data are not overwritten""" # Create temp user via direct API temp_user_response = self.client.post( reverse('api:temp_user_create'), {'name': 'Duplicate Test User'}, format='json' ) temp_user_data = temp_user_response.json() # Authenticate and create first record self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") response = self.client.post( reverse('api:bazi-list'), self.bazi_api_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) # Create second record with same data response = self.client.post( reverse('api:bazi-list'), self.bazi_api_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) temp_user = User.objects.get(id=temp_user_data['user_id']) records = Person.objects.filter(created_by=temp_user) # Should have 2 separate records self.assertEqual(records.count(), 2) self.assertEqual(records.filter(owner=True).count(), 1) # Only one owner self.assertEqual(records.filter(owner=False).count(), 1) # One non-owner class TempUserRegistrationAPITests(TempUserAPIFlowTestCase): """Test temporary user registration/conversion via API""" def _create_temp_user_with_data_api(self): """Helper to create temp user with data via direct API""" response = self.client.post( reverse('api:temp_user_create'), { 'name': self.bazi_api_data['name'], 'gender': self.bazi_api_data['gender'], 'birth_date': self.bazi_api_data['birth_date'], 'birth_time': self.bazi_api_data['birth_time'] }, format='json' ) return response.json() def test_temp_register_api_minimal_data(self): """Test temp user registration via API with minimal data""" temp_user_data = self._create_temp_user_with_data_api() # Authenticate as temp user self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") # Register with minimal data register_data = { 'phone': '9876543210', 'email': 'converted@example.com' } response = self.client.post( reverse('api:temp_register'), register_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) response_data = response.json() # Check user was converted temp_user = User.objects.get(id=temp_user_data['user_id']) self.assertFalse(temp_user.is_temporary_user) self.assertEqual(temp_user.phone, '9876543210') self.assertEqual(temp_user.email, 'converted@example.com') # Check response contains user data self.assertIn('user', response_data) self.assertIn('message', response_data) def test_temp_register_api_with_password(self): """Test temp user registration via API with password""" temp_user_data = self._create_temp_user_with_data_api() # Authenticate as temp user self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") # Register with password register_data = { 'phone': '9876543210', 'email': 'converted@example.com', 'password': 'newpass123', 'password2': 'newpass123' } response = self.client.post( reverse('api:temp_register'), register_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) response_data = response.json() # Check new JWT tokens returned self.assertIn('access', response_data) self.assertIn('refresh', response_data) # Check password was set temp_user = User.objects.get(id=temp_user_data['user_id']) self.assertTrue(temp_user.check_password('newpass123')) def test_temp_register_api_with_optional_fields(self): """Test temp user registration via API with all optional fields""" temp_user_data = self._create_temp_user_with_data_api() # Authenticate as temp user self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") # Register with all optional fields register_data = { 'phone': '9876543210', 'email': 'converted@example.com', 'first_name': 'John', 'last_name': 'Doe', 'gender': 'M', 'profile': { 'birth_date': '1990-01-01', 'birth_time': '12:00:00', 'twin_type': 0, 'father_dob': '1965-01-01', 'mother_dob': '1967-01-01' } } response = self.client.post( reverse('api:temp_register'), register_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) # Check all fields were saved temp_user = User.objects.get(id=temp_user_data['user_id']) self.assertEqual(temp_user.first_name, 'John') self.assertEqual(temp_user.last_name, 'Doe') self.assertEqual(temp_user.gender, 'M') # Get or create profile if it doesn't exist (signal might not have fired) profile, created = UserProfile.objects.get_or_create(user=temp_user) self.assertEqual(profile.birth_date, date(1990, 1, 1)) self.assertEqual(profile.birth_time, time(12, 0, 0)) self.assertEqual(profile.twin_type, 0) self.assertEqual(profile.father_dob, date(1965, 1, 1)) self.assertEqual(profile.mother_dob, date(1967, 1, 1)) def test_temp_register_comprehensive_profile_update(self): """Test comprehensive profile data update in temp user registration""" temp_user_data = self._create_temp_user_with_data_api() # Authenticate as temp user self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") # Register with comprehensive profile data register_data = { 'phone': '9876543210', 'email': 'profile-test@example.com', 'first_name': 'Profile', 'last_name': 'Test', 'gender': 'F', 'profile': { 'birth_date': '1995-06-15', 'birth_time': '14:30:00', 'twin_type': 1, # Elder twin 'father_dob': '1970-03-20', 'mother_dob': '1972-08-10', 'can_regenerate_ai': False } } response = self.client.post( reverse('api:temp_register'), register_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) response_data = response.json() # Verify all user data is in response self.assertIn('user', response_data) user_data = response_data['user'] self.assertEqual(user_data['first_name'], 'Profile') self.assertEqual(user_data['last_name'], 'Test') self.assertEqual(user_data['gender'], 'F') # Verify profile data is in response self.assertIn('profile', user_data) profile_data = user_data['profile'] self.assertEqual(profile_data['birth_date'], '1995-06-15') self.assertEqual(profile_data['birth_time'], '14:30:00') self.assertEqual(profile_data['twin_type'], 1) self.assertEqual(profile_data['father_dob'], '1970-03-20') self.assertEqual(profile_data['mother_dob'], '1972-08-10') self.assertFalse(profile_data['can_regenerate_ai']) # Verify data is actually saved in database temp_user = User.objects.get(id=temp_user_data['user_id']) profile = UserProfile.objects.get(user=temp_user) self.assertEqual(profile.birth_date, date(1995, 6, 15)) self.assertEqual(profile.birth_time, time(14, 30, 0)) self.assertEqual(profile.twin_type, 1) self.assertEqual(profile.father_dob, date(1970, 3, 20)) self.assertEqual(profile.mother_dob, date(1972, 8, 10)) self.assertFalse(profile.can_regenerate_ai) def test_temp_register_with_password_returns_tokens(self): """Test that providing password in temp register returns new JWT tokens""" temp_user_data = self._create_temp_user_with_data_api() # Authenticate as temp user self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") # Register with password register_data = { 'phone': '9876543210', 'email': 'password-test@example.com', 'password': 'newpassword123', 'password2': 'newpassword123', 'first_name': 'Password', 'last_name': 'Test' } response = self.client.post( reverse('api:temp_register'), register_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) response_data = response.json() # Verify new JWT tokens are returned when password is provided self.assertIn('access', response_data) self.assertIn('refresh', response_data) self.assertIsNotNone(response_data['access']) self.assertIsNotNone(response_data['refresh']) # Verify message indicates tokens were updated self.assertIn('new authentication tokens', response_data['message']) # Verify password was set temp_user = User.objects.get(id=temp_user_data['user_id']) self.assertTrue(temp_user.check_password('newpassword123')) # Verify tokens are different from original temp user tokens self.assertNotEqual(response_data['access'], temp_user_data['access']) self.assertNotEqual(response_data['refresh'], temp_user_data['refresh']) def test_temp_register_without_password_no_tokens(self): """Test that not providing password in temp register does not return new JWT tokens""" temp_user_data = self._create_temp_user_with_data_api() # Authenticate as temp user self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") # Register without password register_data = { 'phone': '9876543210', 'email': 'no-password-test@example.com', 'first_name': 'NoPassword', 'last_name': 'Test' } response = self.client.post( reverse('api:temp_register'), register_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) response_data = response.json() # Verify no JWT tokens are returned when password is not provided self.assertNotIn('access', response_data) self.assertNotIn('refresh', response_data) # Verify message does not mention tokens self.assertNotIn('authentication tokens', response_data['message']) # Verify user data is still returned self.assertIn('user', response_data) self.assertEqual(response_data['user']['first_name'], 'NoPassword') def test_temp_register_partial_profile_update(self): """Test that partial profile data updates work correctly""" temp_user_data = self._create_temp_user_with_data_api() # Authenticate as temp user self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") # Register with only some profile fields register_data = { 'phone': '9876543210', 'email': 'partial-profile@example.com', 'profile': { 'twin_type': 2, # Younger twin 'mother_dob': '1975-12-25' # Deliberately omit father_dob, birth_date, birth_time } } response = self.client.post( reverse('api:temp_register'), register_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) response_data = response.json() # Verify only provided profile fields are updated temp_user = User.objects.get(id=temp_user_data['user_id']) profile = UserProfile.objects.get(user=temp_user) self.assertEqual(profile.twin_type, 2) self.assertEqual(profile.mother_dob, date(1975, 12, 25)) # father_dob should remain None since it wasn't provided self.assertIsNone(profile.father_dob) # Verify response contains profile data user_data = response_data['user'] self.assertIn('profile', user_data) profile_data = user_data['profile'] self.assertEqual(profile_data['twin_type'], 2) self.assertEqual(profile_data['mother_dob'], '1975-12-25') self.assertIsNone(profile_data['father_dob']) class TempUserLoginMigrationAPITests(TempUserAPIFlowTestCase): """Test data migration when temp user logs in to existing account via API""" def test_temp_login_api_data_migration(self): """Test temp user data migration via API login""" # Create temp user via direct API temp_user_response = self.client.post( reverse('api:temp_user_create'), self.bazi_api_data, format='json' ) temp_user_data = temp_user_response.json() # Authenticate as temp user and create BaZi and LiuYao data self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") # Create BaZi record bazi_response = self.client.post( reverse('api:bazi-list'), self.bazi_api_data, format='json' ) self.assertEqual(bazi_response.status_code, status.HTTP_201_CREATED) # Create LiuYao record liuyao_response = self.client.post( reverse('api:liuyao-calc'), self.liuyao_api_data, format='json' ) self.assertEqual(liuyao_response.status_code, status.HTTP_201_CREATED) # Login to regular user account login_data = { 'phone': '1234567890', 'password': 'testpass123' } response = self.client.post( reverse('api:temp_login'), login_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) response_data = response.json() # Check transfer summary self.assertIn('transfer_summary', response_data) transfer = response_data['transfer_summary'] self.assertEqual(transfer['bazi_records'], 1) self.assertEqual(transfer['liuyao_records'], 1) # Check data was migrated bazi_records = Person.objects.filter(created_by=self.regular_user) liuyao_records = liuyao.objects.filter(user=self.regular_user) self.assertEqual(bazi_records.count(), 1) self.assertEqual(liuyao_records.count(), 1) # Check temp user was deleted after data transfer with self.assertRaises(User.DoesNotExist): User.objects.get(id=temp_user_data['user_id']) def test_temp_login_api_owner_conflict_resolution(self): """Test owner flag handling in API login migration""" # Create owner record for regular user Person.objects.create( name='Regular User Record', gender='M', birth_date=date(1985, 1, 1), created_by=self.regular_user, owner=True ) # Create temp user via direct API temp_user_response = self.client.post( reverse('api:temp_user_create'), self.bazi_api_data, format='json' ) temp_user_data = temp_user_response.json() # Authenticate and create BaZi record self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") bazi_response = self.client.post( reverse('api:bazi-list'), self.bazi_api_data, format='json' ) self.assertEqual(bazi_response.status_code, status.HTTP_201_CREATED) # Login and migrate via API login_data = { 'phone': '1234567890', 'password': 'testpass123' } response = self.client.post( reverse('api:temp_login'), login_data, format='json' ) # Check owner flag handling transfer = response.json()['transfer_summary'] self.assertTrue(transfer['owner_flag_cleared']) # Should be True since flag was cleared # Verify only one owner record owner_records = Person.objects.filter(created_by=self.regular_user, owner=True) self.assertEqual(owner_records.count(), 1) # Check migrated record has owner=False migrated_record = Person.objects.filter( created_by=self.regular_user, name='API Test Person' ).first() self.assertFalse(migrated_record.owner) class TempUserAPIEdgeCasesTests(TempUserAPIFlowTestCase): """Test edge cases and error conditions for API""" def test_invalid_temp_register_api_data(self): """Test temp register API with invalid data""" temp_user_data = self._create_temp_user_with_data_api() self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") # Try to register with invalid data invalid_data = { 'phone': '123', # Phone validation removed, so this should be acceptable now 'email': 'invalid-email' # Invalid format } response = self.client.post( reverse('api:temp_register'), invalid_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) # Phone validation removed, so only email should have errors self.assertNotIn('phone', response.json()) self.assertIn('email', response.json()) def test_duplicate_phone_email_api_registration(self): """Test API registration with existing phone/email""" temp_user_data = self._create_temp_user_with_data_api() self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") # Try to register with existing user's data duplicate_data = { 'phone': '1234567890', # Already used by regular_user 'email': 'regular@example.com' # Already used by regular_user } response = self.client.post( reverse('api:temp_register'), duplicate_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_unauthorized_api_access(self): """Test accessing temp user API endpoints without authentication""" # Try to access temp register without auth response = self.client.post( reverse('api:temp_register'), {'phone': '123', 'email': 'test@example.com'}, format='json' ) self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) def test_regular_user_temp_api_access(self): """Test regular user trying to access temp user APIs""" # Login as regular user self.client.force_authenticate(user=self.regular_user) # Try to access temp register response = self.client.post( reverse('api:temp_register'), {'phone': '123', 'email': 'test@example.com'}, format='json' ) # Should be forbidden since not a temp user self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) def _create_temp_user_with_data_api(self): """Helper method for creating temp user data""" response = self.client.post( reverse('api:temp_user_create'), { 'name': self.bazi_api_data['name'], 'gender': self.bazi_api_data['gender'], 'birth_date': self.bazi_api_data['birth_date'], 'birth_time': self.bazi_api_data['birth_time'] }, format='json' ) return response.json() class TempUserAPINormalFlowTests(TempUserAPIFlowTestCase): """Test normal user API flows are not affected""" def test_regular_user_api_bazi_creation(self): """Test regular logged-in user creating BaZi via API works normally""" # Authenticate as regular user self.client.force_authenticate(user=self.regular_user) # Create BaZi record via API response = self.client.post( reverse('api:bazi-list'), self.bazi_api_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_201_CREATED) response_data = response.json() # Should not contain temp_user data self.assertNotIn('temp_user', response_data) # Check record created for regular user bazi_record = Person.objects.filter(created_by=self.regular_user).first() self.assertIsNotNone(bazi_record) self.assertEqual(bazi_record.name, 'API Test Person') # No temp user should be created temp_users = User.objects.filter(is_temporary_user=True) self.assertEqual(temp_users.count(), 0) def test_regular_user_api_token_refresh(self): """Test regular user JWT token refresh works normally""" # Get initial tokens for regular user login_response = self.client.post( reverse('api:login'), {'phone': '1234567890', 'password': 'testpass123'}, format='json' ) tokens = login_response.json() # Use refresh token refresh_response = self.client.post( reverse('api:token_refresh'), {'refresh': tokens['refresh']}, format='json' ) self.assertEqual(refresh_response.status_code, status.HTTP_200_OK) self.assertIn('access', refresh_response.json()) class TempUserAPISecurityTests(TempUserAPIFlowTestCase): """Test security aspects of temporary user API""" def test_temp_user_jwt_isolation(self): """Test temp users can only access their own data""" # Create first temp user via direct API temp1_response = self.client.post( reverse('api:temp_user_create'), {'name': 'First Temp User'}, format='json' ) temp1_data = temp1_response.json() # Authenticate as first temp user and create BaZi record self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp1_data['access']}") bazi1_response = self.client.post( reverse('api:bazi-list'), self.bazi_api_data, format='json' ) self.assertEqual(bazi1_response.status_code, status.HTTP_201_CREATED) # Clear authentication and create second temp user self.client.credentials() temp2_response = self.client.post( reverse('api:temp_user_create'), {'name': 'Second Temp User'}, format='json' ) temp2_data = temp2_response.json() # Authenticate as second temp user and create BaZi record self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp2_data['access']}") temp2_bazi_data = self.bazi_api_data.copy() temp2_bazi_data['name'] = 'Second Temp User Record' bazi2_response = self.client.post( reverse('api:bazi-list'), temp2_bazi_data, format='json' ) self.assertEqual(bazi2_response.status_code, status.HTTP_201_CREATED) # Authenticate as temp user 1 again self.client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp1_data['access']}") # Try to access data - should only see own record response = self.client.get(reverse('api:bazi-list')) self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.json()['results'] self.assertEqual(len(results), 1) self.assertEqual(results[0]['name'], 'API Test Person') # Only sees own data def test_temp_user_token_expiry_handling(self): """Test behavior with expired temp user tokens""" # Create temp user temp_user_data = self._create_temp_user_with_data_api() # Use an obviously invalid token self.client.credentials(HTTP_AUTHORIZATION="Bearer invalid_token_123") # Try to access protected endpoint response = self.client.get(reverse('api:bazi-list')) self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) def _create_temp_user_with_data_api(self): """Helper method for creating temp user data""" response = self.client.post( reverse('api:temp_user_create'), { 'name': self.bazi_api_data['name'], 'gender': self.bazi_api_data['gender'], 'birth_date': self.bazi_api_data['birth_date'], 'birth_time': self.bazi_api_data['birth_time'] }, format='json' ) return response.json() if __name__ == '__main__': import unittest unittest.main()