from decimal import Decimal from django.core.files.uploadedfile import SimpleUploadedFile from django.contrib.auth import get_user_model from django.test import 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, Notification, Shipment, Trip User = get_user_model() class UnifiedChatFlowTests(TestCase): def setUp(self): self.shipper = User.objects.create_user('shipper2@test.com', email='shipper2@test.com', password='secret') UserRole.objects.create(user=self.shipper, role=UserRole.Role.SHIPPER, phone='', language='en') ShipperProfile.objects.create(user=self.shipper, account_type=ShipperProfile.AccountType.INDIVIDUAL) self.transporter = User.objects.create_user( 'transporter2@test.com', email='transporter2@test.com', password='secret', ) UserRole.objects.create(user=self.transporter, role=UserRole.Role.TRANSPORTER, phone='', language='en') TransporterProfile.objects.create( user=self.transporter, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, documents_verified=True, ) Vehicle.objects.create( owner=self.transporter, vehicle_type='Flatbed', registration_number='ABC-123', load_capacity=Decimal('2000'), is_verified=True, is_active=True, ) self.other_user = User.objects.create_user('other@test.com', email='other@test.com', password='secret') UserRole.objects.create(user=self.other_user, role=UserRole.Role.SHIPPER, phone='', language='en') ShipperProfile.objects.create(user=self.other_user, account_type=ShipperProfile.AccountType.INDIVIDUAL) self.shipment = Shipment.objects.create( shipper=self.shipper, pickup_address='A', delivery_address='B', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', local=True, country_to_country=False, status=Shipment.Status.PUBLISHED, ) self.shipper_client = APIClient() shipper_token, _ = Token.objects.get_or_create(user=self.shipper) self.shipper_client.credentials(HTTP_AUTHORIZATION=f'Bearer {shipper_token.key}') self.transporter_client = APIClient() transporter_token, _ = Token.objects.get_or_create(user=self.transporter) self.transporter_client.credentials(HTTP_AUTHORIZATION=f'Bearer {transporter_token.key}') self.other_client = APIClient() other_token, _ = Token.objects.get_or_create(user=self.other_user) self.other_client.credentials(HTTP_AUTHORIZATION=f'Bearer {other_token.key}') def test_conversation_created_and_reused_for_same_pair(self): r1 = self.transporter_client.post( f'/api/transporter/shipments/{self.shipment.id}/bid/', {'action': 'COUNTER', 'amount': '100', 'message': 'first offer'}, format='json', ) self.assertEqual(r1.status_code, 201) self.assertTrue( Notification.objects.filter( user=self.shipper, type='new_bid', ).exists() ) conv1 = Conversation.objects.get(shipment=self.shipment, transporter=self.transporter) r2 = self.transporter_client.post( f'/api/transporter/shipments/{self.shipment.id}/bid/', {'action': 'COUNTER', 'amount': '120', 'message': 'second offer'}, format='json', ) self.assertEqual(r2.status_code, 200) conv2 = Conversation.objects.get(shipment=self.shipment, transporter=self.transporter) self.assertEqual(conv1.id, conv2.id) def test_accept_from_chat_creates_trip_on_same_conversation(self): self.transporter_client.post( f'/api/transporter/shipments/{self.shipment.id}/bid/', {'action': 'COUNTER', 'amount': '100', 'message': 'offer'}, format='json', ) conv = Conversation.objects.get(shipment=self.shipment, transporter=self.transporter) send_res = self.shipper_client.post( f'/api/chats/{conv.id}/messages/', {'text': 'accepting this', 'message_type': 'TEXT'}, format='json', ) self.assertEqual(send_res.status_code, 201) send_data = send_res.json()['data'] self.assertEqual(send_data['sender_email'], self.shipper.email) self.assertIn('sender_name', send_data) self.assertIn('sender_account_type', send_data) notif = Notification.objects.filter( user=self.transporter, type='new_message', data__conversation_id=str(conv.id), ).latest('id') self.assertEqual(notif.data.get('type'), 'New Message') self.assertIn('conversation_id', notif.data.get('tag', '')) self.assertEqual(notif.title, 'New Message') accept_res = self.shipper_client.post(f'/api/chats/{conv.id}/accept-bid/', {}, format='json') self.assertEqual(accept_res.status_code, 200) trip_id = accept_res.json()['data']['trip_id'] conv.refresh_from_db() self.assertEqual(conv.trip_id, trip_id) self.assertTrue(Trip.objects.filter(pk=trip_id, shipment=self.shipment).exists()) self.assertTrue( Notification.objects.filter( user=self.transporter, type='bid_accepted', data__bid_id=str(Bid.objects.filter(shipment=self.shipment, transporter=self.transporter).latest('id').id), ).exists() ) get_res = self.shipper_client.get(f'/api/chats/{conv.id}/messages/') self.assertEqual(get_res.status_code, 200) self.assertEqual(get_res.json()['data'][0]['conversation_id'], conv.id) def test_only_participants_can_access_chat_messages(self): self.transporter_client.post( f'/api/transporter/shipments/{self.shipment.id}/bid/', {'action': 'COUNTER', 'amount': '100', 'message': 'offer'}, format='json', ) conv = Conversation.objects.get(shipment=self.shipment, transporter=self.transporter) denied = self.other_client.get(f'/api/chats/{conv.id}/messages/') self.assertEqual(denied.status_code, 403) def test_voice_message_multipart_returns_voice_url(self): self.transporter_client.post( f'/api/transporter/shipments/{self.shipment.id}/bid/', {'action': 'COUNTER', 'amount': '100', 'message': 'offer'}, format='json', ) conv = Conversation.objects.get(shipment=self.shipment, transporter=self.transporter) audio = SimpleUploadedFile('note.m4a', b'fakeaudio', content_type='audio/mp4') res = self.shipper_client.post( f'/api/chats/{conv.id}/messages/', {'message_type': 'VOICE', 'voice_file': audio}, format='multipart', ) self.assertEqual(res.status_code, 201) data = res.json()['data'] self.assertEqual(data['message_type'], 'VOICE') self.assertIsNotNone(data.get('voice_url')) n = Notification.objects.filter( user=self.transporter, type='new_message', data__chat_message_type='VOICE', ) self.assertTrue(n.exists()) def test_text_message_requires_non_empty_text(self): self.transporter_client.post( f'/api/transporter/shipments/{self.shipment.id}/bid/', {'action': 'COUNTER', 'amount': '100', 'message': 'offer'}, format='json', ) conv = Conversation.objects.get(shipment=self.shipment, transporter=self.transporter) res = self.shipper_client.post( f'/api/chats/{conv.id}/messages/', {'text': ' ', 'message_type': 'TEXT'}, format='json', ) self.assertEqual(res.status_code, 400) self.assertIn('text', res.json()['error']) def test_voice_message_requires_file(self): self.transporter_client.post( f'/api/transporter/shipments/{self.shipment.id}/bid/', {'action': 'COUNTER', 'amount': '100', 'message': 'offer'}, format='json', ) conv = Conversation.objects.get(shipment=self.shipment, transporter=self.transporter) res = self.shipper_client.post( f'/api/chats/{conv.id}/messages/', {'message_type': 'VOICE', 'text': 'oops'}, format='json', ) self.assertEqual(res.status_code, 400) self.assertIn('voice_file', res.json()['error']) def test_shipper_trip_apis_include_status_timeline(self): self.transporter_client.post( f'/api/transporter/shipments/{self.shipment.id}/bid/', {'action': 'COUNTER', 'amount': '100', 'message': 'offer'}, format='json', ) bid = Bid.objects.filter(shipment=self.shipment, transporter=self.transporter).latest('id') accept = self.shipper_client.post(f'/api/shipper/bids/{bid.id}/accept/', {}, format='json') self.assertEqual(accept.status_code, 200) trip_id = accept.json()['data']['trip_id'] detail = self.shipper_client.get(f'/api/shipper/trips/{trip_id}/') self.assertEqual(detail.status_code, 200) timeline = detail.json()['data']['status_timeline'] self.assertEqual(timeline[-1]['status'], 'ASSIGNED') self.assertIsNotNone(timeline[-1]['recorded_at']) update = self.transporter_client.patch( f'/api/transporter/trips/{trip_id}/status/', {'status': 'EN_ROUTE'}, format='json', ) self.assertEqual(update.status_code, 200) detail2 = self.shipper_client.get(f'/api/shipper/trips/{trip_id}/') self.assertEqual(detail2.status_code, 200) timeline2 = detail2.json()['data']['status_timeline'] self.assertEqual([x['status'] for x in timeline2][-2:], ['ASSIGNED', 'EN_ROUTE']) listing = self.shipper_client.get('/api/shipper/trips/') self.assertEqual(listing.status_code, 200) row = next(item for item in listing.json()['data'] if item['id'] == trip_id) self.assertEqual([x['status'] for x in row['status_timeline']][-2:], ['ASSIGNED', 'EN_ROUTE']) class AssignedFleetDriverChatTests(TestCase): def setUp(self): self.shipper = User.objects.create_user( 'shipper-fd-chat@test.com', email='shipper-fd-chat@test.com', password='secret', ) UserRole.objects.create(user=self.shipper, role=UserRole.Role.SHIPPER, phone='', language='en') ShipperProfile.objects.create(user=self.shipper, account_type=ShipperProfile.AccountType.INDIVIDUAL) self.owner = User.objects.create_user( 'owner-fd-chat@test.com', email='owner-fd-chat@test.com', password='secret', ) UserRole.objects.create(user=self.owner, role=UserRole.Role.TRANSPORTER, phone='', language='en') TransporterProfile.objects.create( user=self.owner, account_type=TransporterProfile.AccountType.FLEET_OWNER, company_name='Fleet Co', local=True, country_to_country=False, documents_verified=True, ) Vehicle.objects.create( owner=self.owner, vehicle_type='Flatbed', registration_number='FD-CHAT-1', load_capacity=Decimal('2000'), is_verified=True, is_active=True, ) self.driver = User.objects.create_user( 'driver-fd-chat@test.com', email='driver-fd-chat@test.com', password='secret', first_name='Fleet', last_name='Driver', ) UserRole.objects.create(user=self.driver, role=UserRole.Role.TRANSPORTER, phone='', language='en') TransporterProfile.objects.create( user=self.driver, account_type=TransporterProfile.AccountType.TRANSPORTER_DRIVER, local=True, country_to_country=False, documents_verified=True, ) from accounts.models import TransporterDriverLink TransporterDriverLink.objects.create(transporter=self.owner, driver=self.driver, is_active=True) self.shipment = Shipment.objects.create( shipper=self.shipper, pickup_address='A', delivery_address='B', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', local=True, country_to_country=False, status=Shipment.Status.PUBLISHED, ) self.shipper_client = APIClient() t, _ = Token.objects.get_or_create(user=self.shipper) self.shipper_client.credentials(HTTP_AUTHORIZATION=f'Bearer {t.key}') self.owner_client = APIClient() t, _ = Token.objects.get_or_create(user=self.owner) self.owner_client.credentials(HTTP_AUTHORIZATION=f'Bearer {t.key}') self.driver_client = APIClient() t, _ = Token.objects.get_or_create(user=self.driver) self.driver_client.credentials(HTTP_AUTHORIZATION=f'Bearer {t.key}') def _accept_trip_and_assign(self): self.owner_client.post( f'/api/transporter/shipments/{self.shipment.id}/bid/', {'action': 'COUNTER', 'amount': '100', 'message': 'offer'}, format='json', ) bid = Bid.objects.filter(shipment=self.shipment, transporter=self.owner).latest('id') accept = self.shipper_client.post(f'/api/shipper/bids/{bid.id}/accept/', {}, format='json') self.assertEqual(accept.status_code, 200, accept.content) trip_id = accept.json()['data']['trip_id'] assign = self.owner_client.post( f'/api/transporter/trips/{trip_id}/assign-driver/', {'driver_id': self.driver.id}, format='json', ) self.assertEqual(assign.status_code, 200, assign.content) conv = Conversation.objects.get(shipment=self.shipment, transporter=self.owner) return trip_id, conv def test_assigned_driver_can_list_and_message_trip_chat(self): trip_id, conv = self._accept_trip_and_assign() denied_before_list_shape = self.driver_client.get('/api/chats/') self.assertEqual(denied_before_list_shape.status_code, 200) ids = {row['id'] for row in denied_before_list_shape.json()['data']} self.assertIn(conv.id, ids) row = next(r for r in denied_before_list_shape.json()['data'] if r['id'] == conv.id) self.assertEqual(row['assigned_driver_id'], self.driver.id) self.assertEqual(row['counterparty']['id'], self.shipper.id) get_msgs = self.driver_client.get(f'/api/chats/{conv.id}/messages/') self.assertEqual(get_msgs.status_code, 200) send = self.driver_client.post( f'/api/chats/{conv.id}/messages/', {'text': 'On my way', 'message_type': 'TEXT'}, format='json', ) self.assertEqual(send.status_code, 201, send.content) self.assertTrue( Notification.objects.filter( user=self.shipper, type='new_message', data__conversation_id=str(conv.id), ).exists() ) self.assertTrue( Notification.objects.filter( user=self.owner, type='new_message', data__conversation_id=str(conv.id), ).exists() ) shipper_msg = self.shipper_client.post( f'/api/chats/{conv.id}/messages/', {'text': 'Thanks', 'message_type': 'TEXT'}, format='json', ) self.assertEqual(shipper_msg.status_code, 201) self.assertTrue( Notification.objects.filter( user=self.driver, type='new_message', data__conversation_id=str(conv.id), ).exists() ) open_res = self.driver_client.post( '/api/chats/open/', {'shipment_id': self.shipment.id, 'transporter_id': self.owner.id}, format='json', ) self.assertEqual(open_res.status_code, 200) self.assertEqual(open_res.json()['data']['conversation_id'], conv.id) self.assertEqual(open_res.json()['data']['assigned_driver_id'], self.driver.id) self.assertEqual(open_res.json()['data']['trip_id'], trip_id) def test_unassigned_driver_loses_chat_access(self): _, conv = self._accept_trip_and_assign() trip = Trip.objects.get(conversation=conv) unassign = self.owner_client.post( f'/api/transporter/trips/{trip.id}/assign-driver/', {'driver_id': None}, format='json', ) self.assertEqual(unassign.status_code, 200) denied = self.driver_client.get(f'/api/chats/{conv.id}/messages/') self.assertEqual(denied.status_code, 403) listed = self.driver_client.get('/api/chats/') self.assertNotIn(conv.id, {row['id'] for row in listed.json()['data']})