from decimal import Decimal from django.contrib.auth import get_user_model from django.core.files.uploadedfile import SimpleUploadedFile from django.test import Client, TestCase from rest_framework.authtoken.models import Token from rest_framework.test import APIClient from accounts.models import ShipperProfile, TransporterProfile, UserRole, Vehicle from core.models import Bid, Conversation, Message, Shipment, Trip User = get_user_model() class AdminConversationsMirroringTests(TestCase): def setUp(self): # 1. Admin user self.admin_user = User.objects.create_user( 'admin@truckme.test', email='admin@truckme.test', password='adminpassword', first_name='Admin', last_name='User', is_staff=True, is_superuser=True, ) UserRole.objects.create(user=self.admin_user, role=UserRole.Role.ADMIN) # 2. Shipper user self.shipper = User.objects.create_user( 'shipper@truckme.test', email='shipper@truckme.test', password='secretpassword', first_name='Salim', last_name='Ahmed', ) UserRole.objects.create(user=self.shipper, role=UserRole.Role.SHIPPER, phone='+971501112233') ShipperProfile.objects.create( user=self.shipper, account_type=ShipperProfile.AccountType.INDIVIDUAL, company_name='Emirates Cargo Trading', ) # 3. Transporter user (Fleet Owner) self.transporter = User.objects.create_user( 'transporter@truckme.test', email='transporter@truckme.test', password='secretpassword', first_name='Bilal', last_name='Haider', ) UserRole.objects.create(user=self.transporter, role=UserRole.Role.TRANSPORTER, phone='+971502223344') TransporterProfile.objects.create( user=self.transporter, account_type=TransporterProfile.AccountType.FLEET_OWNER, company_name='Falcon Haulage LLC', documents_verified=True, local=False, country_to_country=True, ) # 4. Assigned Fleet Driver self.driver = User.objects.create_user( 'driver@truckme.test', email='driver@truckme.test', password='secretpassword', first_name='Rashid', last_name='Khan', ) UserRole.objects.create(user=self.driver, role=UserRole.Role.TRANSPORTER, phone='+971503334455') TransporterProfile.objects.create( user=self.driver, account_type=TransporterProfile.AccountType.TRANSPORTER_DRIVER, documents_verified=True, local=False, country_to_country=True, ) # 5. Vehicle self.vehicle = Vehicle.objects.create( owner=self.transporter, assigned_driver=self.driver, vehicle_type='Flatbed', registration_number='DXB-98765', load_capacity=Decimal('10000.00'), is_verified=True, is_active=True, ) # 6. Shipment self.shipment = Shipment.objects.create( shipper=self.shipper, unique_id='TM-90001', pickup_address='Dubai Industrial City, UAE', delivery_address='Riyadh Logistics Zone, KSA', cargo_type='Heavy Equipment', weight='8 tons', suggested_price=Decimal('5000.00'), currency='AED', status=Shipment.Status.ASSIGNED, local=False, country_to_country=True, ) # 7. Bid self.bid = Bid.objects.create( shipment=self.shipment, transporter=self.transporter, amount=Decimal('5500.00'), status=Bid.Status.ACCEPTED, ) # 8. Trip self.trip = Trip.objects.create( shipment=self.shipment, accepted_bid=self.bid, transporter=self.transporter, assigned_driver=self.driver, status=Trip.Status.IN_TRANSIT, ) # 9. Conversation & Messages self.conv = Conversation.objects.create( shipment=self.shipment, shipper=self.shipper, transporter=self.transporter, trip=self.trip, ) # Message 1 (Text from Shipper) self.msg1 = Message.objects.create( conversation=self.conv, sender=self.shipper, message_type=Message.MessageType.TEXT, text='Hello, will the driver reach pickup by 9 AM?', ) # Message 2 (Text from Driver) self.msg2 = Message.objects.create( conversation=self.conv, sender=self.driver, message_type=Message.MessageType.TEXT, text='Yes sir, currently on Sheikh Mohammed Bin Zayed Rd.', ) # Message 3 (Voice note from Driver) sample_audio = SimpleUploadedFile('voice_note.mp3', b'fake-mp3-audio-content', content_type='audio/mpeg') self.msg3 = Message.objects.create( conversation=self.conv, sender=self.driver, message_type=Message.MessageType.VOICE, voice_file=sample_audio, ) # Set up clients self.web_client = Client() self.api_client = APIClient() admin_token, _ = Token.objects.get_or_create(user=self.admin_user) self.admin_token_auth = f'Bearer {admin_token.key}' def test_anonymous_and_non_admin_redirected_from_web_panel(self): # 1. Anonymous access res = self.web_client.get('/conversations') self.assertEqual(res.status_code, 302) self.assertTrue(res.url.startswith('/') or '/login' in res.url) # 2. Non-admin shipper login self.web_client.login(username=self.shipper.email, password='secretpassword') res = self.web_client.get('/conversations') self.assertEqual(res.status_code, 302) self.assertTrue(res.url.startswith('/') or '/login' in res.url) def test_admin_conversations_page_and_datatables_json(self): self.web_client.login(username=self.admin_user.email, password='adminpassword') # 1. Standard HTML render res = self.web_client.get('/conversations') self.assertEqual(res.status_code, 200) self.assertContains(res, 'Driver – Shipper Communications') self.assertContains(res, 'conversations-table') # 2. DataTables AJAX request res_dt = self.web_client.get( '/conversations', {'draw': 1, 'start': 0, 'length': 10}, HTTP_X_REQUESTED_WITH='XMLHttpRequest', ) self.assertEqual(res_dt.status_code, 200) json_data = res_dt.json() self.assertIn('data', json_data) self.assertEqual(json_data['recordsTotal'], 1) row = json_data['data'][0] self.assertEqual(row['id'], self.conv.id) self.assertEqual(row['shipment_label'], 'TM-90001') self.assertEqual(row['shipper_email'], 'shipper@truckme.test') self.assertEqual(row['transporter_email'], 'transporter@truckme.test') self.assertEqual(row['assigned_driver_display'], 'Rashid Khan') self.assertEqual(row['trip_status'], 'IN_TRANSIT') self.assertEqual(row['messages_count'], 3) self.assertEqual(row['last_message_type'], 'VOICE') def test_admin_conversation_modal(self): self.web_client.login(username=self.admin_user.email, password='adminpassword') res = self.web_client.get(f'/conversations/{self.conv.id}/modal') self.assertEqual(res.status_code, 200) self.assertContains(res, f'Communication Thread #{self.conv.id}') self.assertContains(res, 'TM-90001') self.assertContains(res, 'Hello, will the driver reach pickup by 9 AM?') self.assertContains(res, 'Yes sir, currently on Sheikh Mohammed Bin Zayed Rd.') self.assertContains(res, 'Voice Recording') self.assertContains(res, ' returns 0 new res_empty = self.web_client.get( f'/conversations/{self.conv.id}/sync?after_id={self.msg3.id}', HTTP_X_REQUESTED_WITH='XMLHttpRequest', ) self.assertEqual(res_empty.status_code, 200) self.assertEqual(res_empty.json()['total_new'], 0)