from django.test import TestCase, Client from django.urls import reverse from django.utils import timezone from rest_framework import status from rest_framework.test import APIClient from main.user.user import User from bazi.models import Person from liuyao.models import liuyao class TempUserIntegrationTestCase(TestCase): """Base test case with common setup for temp user integration tests""" def setUp(self): """Set up test data""" self.web_client = Client() self.api_client = APIClient() # Test data for web forms self.bazi_web_data = { 'name': 'Web Test Person', 'gender': 'M', 'birth_date': '1990-01-01', 'birth_time': '12:00:00', 'notes': 'Web test notes' } # Test data for API self.bazi_api_data = { 'name': 'API Test Person', 'gender': 'F', 'birth_date': '1985-06-15', 'birth_time': '15:30:00', 'lunar': False, 'notes': 'API test notes' } self.liuyao_data = { 'question': 'Integration test question', 'y1': '1', 'y2': '0', 'y3': '1', 'y4': '000', 'y5': '1', 'y6': '0', 'qdate': timezone.now().isoformat() } # Create regular user 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 WebToAPIIntegrationTests(TempUserIntegrationTestCase): """Test scenarios where user starts with web, then uses API""" def test_web_temp_user_then_api_with_jwt(self): """Test creating temp user via web, then using API with JWT""" # Create temp user via web form response = self.web_client.post(reverse('bazi:calculate_chart'), self.bazi_web_data) # Get temp user temp_user = User.objects.filter(is_temporary_user=True).first() self.assertIsNotNone(temp_user) # Register temp user to get JWT tokens self.web_client.force_login(temp_user) register_data = { 'phone': '9876543210', 'email': 'webtoapi@example.com', 'password': 'newpass123', 'password2': 'newpass123' } register_response = self.web_client.post( reverse('accounts:temp_register'), register_data ) # User should now be permanent temp_user.refresh_from_db() self.assertFalse(temp_user.is_temporary_user) # Login via API to get JWT tokens api_login_response = self.api_client.post( reverse('api:login'), {'phone': '9876543210', 'password': 'newpass123'}, format='json' ) self.assertEqual(api_login_response.status_code, status.HTTP_200_OK) tokens = api_login_response.json() # Use JWT to create more records via API self.api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {tokens['access']}") api_response = self.api_client.post( reverse('api:bazi-list'), self.bazi_api_data, format='json' ) self.assertEqual(api_response.status_code, status.HTTP_201_CREATED) # Check user now has records from both web and API user_records = Person.objects.filter(created_by=temp_user) self.assertEqual(user_records.count(), 2) record_names = [r.name for r in user_records] self.assertIn('Web Test Person', record_names) self.assertIn('API Test Person', record_names) def test_web_temp_user_api_login_migration(self): """Test web temp user data migration when logging in via API""" # Create temp user via web with multiple records self.web_client.post(reverse('bazi:calculate_chart'), self.bazi_web_data) # Create LiuYao record - web uses JavaScript that calls API, so simulate this temp_user = User.objects.filter(is_temporary_user=True).first() self.api_client.force_authenticate(user=temp_user) self.api_client.post('/api/liuyao/liuyao/', { 'question': 'Integration test question', 'y1': 1, 'y2': 0, 'y3': 1, 'y4': 0, 'y5': 1, 'y6': 0, 'qdate': timezone.now().isoformat() }, format='json') # Login to existing account via API login_data = { 'phone': '1234567890', 'password': 'testpass123' } response = self.api_client.post( reverse('api:temp_login'), login_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) # Check data migration transfer = response.json()['transfer_summary'] self.assertEqual(transfer['bazi_records'], 1) self.assertEqual(transfer['liuyao_records'], 1) # Verify data moved to regular user 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) class APIToWebIntegrationTests(TempUserIntegrationTestCase): """Test scenarios where user starts with API, then uses web""" def test_api_temp_user_then_web_register(self): """Test creating temp user via direct API, then registering via web""" # Create temp user via direct API api_response = self.api_client.post( reverse('api:temp_user_create'), self.bazi_api_data, format='json' ) temp_user_data = api_response.json() temp_user = User.objects.get(id=temp_user_data['user_id']) # Log in the temp user for web session # (In real scenario, web app would handle temp user authentication) self.web_client.force_login(temp_user) # Register via web form register_data = { 'phone': '9876543210', 'email': 'apitoweB@example.com', 'first_name': 'API', 'last_name': 'User' } web_response = self.web_client.post( reverse('accounts:temp_register'), register_data ) # Check user was converted temp_user.refresh_from_db() self.assertFalse(temp_user.is_temporary_user) self.assertEqual(temp_user.first_name, 'API') self.assertEqual(temp_user.last_name, 'User') # Check API-created record still exists but name was updated by signal api_record = Person.objects.filter(created_by=temp_user, name='API User').first() self.assertIsNotNone(api_record) class CrossPlatformDataConsistencyTests(TempUserIntegrationTestCase): """Test data consistency across web and API platforms""" def test_mixed_platform_record_creation(self): """Test creating records via both web and API maintains consistency""" # Create temp user via web self.web_client.post(reverse('bazi:calculate_chart'), self.bazi_web_data) temp_user = User.objects.filter(is_temporary_user=True).first() # Simulate API authentication with same temp user self.api_client.force_authenticate(user=temp_user) # Create record via API api_response = self.api_client.post( reverse('api:bazi-list'), self.bazi_api_data, format='json' ) self.assertEqual(api_response.status_code, status.HTTP_201_CREATED) # Check both records belong to same temp user temp_user_records = Person.objects.filter(created_by=temp_user) self.assertEqual(temp_user_records.count(), 2) # Verify record ownership flags self.assertTrue(temp_user_records.filter(owner=True).exists()) self.assertTrue(temp_user_records.filter(owner=False).exists()) def test_platform_switching_owner_consistency(self): """Test that ownership flags remain consistent when switching platforms""" # Create temp user via direct API api_response = self.api_client.post( reverse('api:temp_user_create'), self.bazi_api_data, format='json' ) temp_user_data = api_response.json() temp_user = User.objects.get(id=temp_user_data['user_id']) # Create first BaZi record via API (creates owner record) self.api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") bazi_response = self.api_client.post( reverse('api:bazi-list'), self.bazi_api_data, format='json' ) self.assertEqual(bazi_response.status_code, status.HTTP_201_CREATED) # Switch to web platform self.web_client.force_login(temp_user) web_data = self.bazi_web_data.copy() web_data['name'] = 'Second Web Record' self.web_client.post(reverse('bazi:calculate_chart'), web_data) # Back to API - create third record (already authenticated) third_data = self.bazi_api_data.copy() third_data['name'] = 'Third API Record' self.api_client.post( reverse('api:bazi-list'), third_data, format='json' ) # Check ownership order records = Person.objects.filter(created_by=temp_user).order_by('created_at') self.assertEqual(records.count(), 3) # Only first record should be owner self.assertTrue(records[0].owner) self.assertFalse(records[1].owner) self.assertFalse(records[2].owner) class TempUserSessionSharingTests(TempUserIntegrationTestCase): """Test session sharing between web and API (simulated scenarios)""" def test_session_continuity_web_to_api(self): """Test session continuity from web to API""" # Create temp user via web self.web_client.post(reverse('bazi:calculate_chart'), self.bazi_web_data) temp_user = User.objects.filter(is_temporary_user=True).first() # Register temp user via web to get proper credentials self.web_client.force_login(temp_user) register_data = { 'phone': '9876543210', 'email': 'session@example.com', 'password': 'sessionpass123', 'password2': 'sessionpass123' } self.web_client.post(reverse('accounts:temp_register'), register_data) # Now use API with the converted user api_login = self.api_client.post( reverse('api:login'), {'phone': '9876543210', 'password': 'sessionpass123'}, format='json' ) tokens = api_login.json() self.api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {tokens['access']}") # Check can access web-created data via API api_response = self.api_client.get(reverse('api:bazi-list')) self.assertEqual(api_response.status_code, status.HTTP_200_OK) results = api_response.json()['results'] self.assertEqual(len(results), 1) self.assertEqual(results[0]['name'], 'Web Test Person') class TempUserErrorHandlingIntegrationTests(TempUserIntegrationTestCase): """Test error handling in cross-platform scenarios""" def test_invalid_session_handling(self): """Test handling of invalid/expired temp user sessions""" # Create temp user via direct API api_response = self.api_client.post( reverse('api:temp_user_create'), self.bazi_api_data, format='json' ) temp_user_data = api_response.json() # Try to use web with invalid session ID session = self.web_client.session session['temp_user_id'] = 99999 # Non-existent user ID session.save() # Try to register - should handle gracefully register_data = { 'phone': '9876543210', 'email': 'invalid@example.com' } response = self.web_client.post(reverse('accounts:temp_register'), register_data) # Should redirect or show error, not crash self.assertIn(response.status_code, [302, 400, 403, 404]) def test_concurrent_platform_modifications(self): """Test handling concurrent modifications from different platforms""" # Create temp user via web self.web_client.post(reverse('bazi:calculate_chart'), self.bazi_web_data) temp_user = User.objects.filter(is_temporary_user=True).first() # Simulate concurrent API modification self.api_client.force_authenticate(user=temp_user) # Register via web self.web_client.force_login(temp_user) register_data = { 'phone': '9876543210', 'email': 'concurrent@example.com' } web_response = self.web_client.post( reverse('accounts:temp_register'), register_data ) # Try to use API after web registration (should work with converted user) api_response = self.api_client.post( reverse('api:bazi-list'), self.bazi_api_data, format='json' ) # Should either work with converted user or require re-authentication self.assertIn(api_response.status_code, [201, 401]) class TempUserCompleteWorkflowTests(TempUserIntegrationTestCase): """Test complete workflows spanning both platforms""" def test_complete_mobile_web_workflow(self): """Test complete workflow: mobile app -> web registration -> back to mobile""" # Step 1: Mobile app creates temp user via direct API temp_user_response = self.api_client.post( reverse('api:temp_user_create'), self.bazi_api_data, format='json' ) temp_user_data = temp_user_response.json() temp_user = User.objects.get(id=temp_user_data['user_id']) # Authenticate with JWT tokens and create BaZi record self.api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {temp_user_data['access']}") mobile_bazi = self.api_client.post( reverse('api:bazi-list'), self.bazi_api_data, format='json' ) self.assertEqual(mobile_bazi.status_code, status.HTTP_201_CREATED) mobile_liuyao = self.api_client.post( '/api/liuyao/liuyao/', self.liuyao_data, format='json' ) # Step 2: User goes to web and registers temp_user = User.objects.get(id=temp_user_data['user_id']) self.web_client.force_login(temp_user) register_data = { 'phone': '5551234567', 'email': 'mobile@example.com', 'password': 'mobilepass123', 'password2': 'mobilepass123', 'first_name': 'Mobile', 'last_name': 'User' } web_register = self.web_client.post( reverse('accounts:temp_register'), register_data ) # Step 3: Mobile app logs in with new credentials mobile_login = self.api_client.post( reverse('api:login'), {'phone': '5551234567', 'password': 'mobilepass123'}, format='json' ) self.assertEqual(mobile_login.status_code, status.HTTP_200_OK) tokens = mobile_login.json() # Step 4: Mobile app can access all previous data self.api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {tokens['access']}") bazi_list = self.api_client.get(reverse('api:bazi-list')) liuyao_list = self.api_client.get(reverse('api:liuyao-list')) self.assertEqual(bazi_list.status_code, status.HTTP_200_OK) self.assertEqual(liuyao_list.status_code, status.HTTP_200_OK) self.assertEqual(len(bazi_list.json()['results']), 1) self.assertEqual(len(liuyao_list.json()['results']), 1) # Step 5: User can continue using both platforms web_bazi = self.bazi_web_data.copy() web_bazi['name'] = 'Post-Registration Web Record' # Web usage (need to login first) self.web_client.login(phone='5551234567', password='mobilepass123') self.web_client.post(reverse('bazi:calculate_chart'), web_bazi) # Mobile can see new record updated_list = self.api_client.get(reverse('api:bazi-list')) self.assertEqual(len(updated_list.json()['results']), 2) if __name__ == '__main__': import unittest unittest.main()