from decimal import Decimal from unittest.mock import patch 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 accounts.models import TransporterDriverLink from core.models import Bid, Shipment, Trip User = get_user_model() class TransporterTripDetailApiTests(TestCase): def setUp(self): self.shipper = User.objects.create_user( 'ship-trip-detail@test.com', email='ship-trip-detail@test.com', password='secret', first_name='Ali', last_name='Raza', ) UserRole.objects.create( user=self.shipper, role=UserRole.Role.SHIPPER, phone='03001234567', language='en', ) ShipperProfile.objects.create( user=self.shipper, account_type=ShipperProfile.AccountType.BUSINESS, company_name='Acme Logistics', ) self.transporter = User.objects.create_user( 'driver-trip-detail@test.com', email='driver-trip-detail@test.com', password='secret', ) UserRole.objects.create( user=self.transporter, role=UserRole.Role.TRANSPORTER, phone='03111234567', language='en', ) TransporterProfile.objects.create( user=self.transporter, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, documents_verified=True, tc_id='222', ) 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.IN_TRANSIT, ) bid = Bid.objects.create( shipment=shipment, transporter=self.transporter, amount=Decimal('100.00'), status=Bid.Status.ACCEPTED, ) self.trip = Trip.objects.create( shipment=shipment, accepted_bid=bid, transporter=self.transporter, status=Trip.Status.IN_TRANSIT, ) token, _ = Token.objects.get_or_create(user=self.transporter) self.client = APIClient() self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token.key}') def test_returns_shipper_info_in_trip_detail(self): response = self.client.get(f'/api/transporter/trips/{self.trip.id}/') self.assertEqual(response.status_code, 200) shipper = response.json()['data']['shipper'] self.assertEqual(shipper['id'], self.shipper.id) self.assertEqual(shipper['email'], self.shipper.email) self.assertEqual(shipper['first_name'], 'Ali') self.assertEqual(shipper['last_name'], 'Raza') self.assertEqual(shipper['phone'], '03001234567') self.assertEqual(shipper['account_type'], ShipperProfile.AccountType.BUSINESS) self.assertEqual(shipper['company_name'], 'Acme Logistics') def test_fleet_owner_can_assign_driver_to_trip_and_driver_can_access(self): fleet_owner = User.objects.create_user( 'fleet@test.com', email='fleet@test.com', password='secret', ) UserRole.objects.create(user=fleet_owner, role=UserRole.Role.TRANSPORTER, phone='', language='en') TransporterProfile.objects.create( user=fleet_owner, account_type=TransporterProfile.AccountType.FLEET_OWNER, local=True, country_to_country=False, documents_verified=True, tc_id='333', ) driver = User.objects.create_user( 'fleet-driver@test.com', email='fleet-driver@test.com', password='secret', ) UserRole.objects.create(user=driver, role=UserRole.Role.TRANSPORTER, phone='', language='en') TransporterProfile.objects.create( user=driver, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, documents_verified=True, tc_id='444', ) TransporterDriverLink.objects.create(transporter=fleet_owner, driver=driver, is_active=True) # Trip belongs to fleet owner (winning bidder/acceptor) self.trip.transporter = fleet_owner self.trip.assigned_driver = None self.trip.save(update_fields=['transporter', 'assigned_driver', 'updated_at']) fleet_token, _ = Token.objects.get_or_create(user=fleet_owner) fleet_client = APIClient() fleet_client.credentials(HTTP_AUTHORIZATION=f'Bearer {fleet_token.key}') r = fleet_client.post( f'/api/transporter/trips/{self.trip.id}/assign-driver/', {'driver_id': driver.id}, format='json', ) self.assertEqual(r.status_code, 200) self.assertEqual(r.json()['data']['assigned_driver']['id'], driver.id) # Driver can access trip detail (assigned_driver-based authorization) driver_token, _ = Token.objects.get_or_create(user=driver) driver_client = APIClient() driver_client.credentials(HTTP_AUTHORIZATION=f'Bearer {driver_token.key}') r2 = driver_client.get(f'/api/transporter/trips/{self.trip.id}/') self.assertEqual(r2.status_code, 200) self.assertEqual(r2.json()['data']['assigned_driver']['id'], driver.id) def test_returns_assigned_vehicle_for_fleet_driver(self): fleet_owner = User.objects.create_user( 'fleet-veh@test.com', email='fleet-veh@test.com', password='secret', ) UserRole.objects.create(user=fleet_owner, role=UserRole.Role.TRANSPORTER, phone='', language='en') TransporterProfile.objects.create( user=fleet_owner, account_type=TransporterProfile.AccountType.FLEET_OWNER, local=True, country_to_country=False, documents_verified=True, ) driver = User.objects.create_user( 'fleet-driver-veh@test.com', email='fleet-driver-veh@test.com', password='secret', ) UserRole.objects.create(user=driver, role=UserRole.Role.TRANSPORTER, phone='', language='en') TransporterProfile.objects.create( user=driver, account_type=TransporterProfile.AccountType.TRANSPORTER_DRIVER, local=True, country_to_country=False, documents_verified=True, ) TransporterDriverLink.objects.create(transporter=fleet_owner, driver=driver, is_active=True) vehicle = Vehicle.objects.create( owner=fleet_owner, assigned_driver=driver, vehicle_type='Flatbed', registration_number='FLEET-VEH-1', load_capacity=Decimal('10000'), is_verified=True, is_active=True, ) self.trip.transporter = fleet_owner self.trip.assigned_driver = driver self.trip.save(update_fields=['transporter', 'assigned_driver', 'updated_at']) fleet_token, _ = Token.objects.get_or_create(user=fleet_owner) fleet_client = APIClient() fleet_client.credentials(HTTP_AUTHORIZATION=f'Bearer {fleet_token.key}') response = fleet_client.get(f'/api/transporter/trips/{self.trip.id}/') self.assertEqual(response.status_code, 200) assigned_vehicle = response.json()['data']['assigned_vehicle'] self.assertIsNotNone(assigned_vehicle) self.assertEqual(assigned_vehicle['id'], vehicle.id) self.assertEqual(assigned_vehicle['registration_number'], 'FLEET-VEH-1') self.assertEqual(assigned_vehicle['assigned_driver_id'], driver.id) def test_returns_owned_vehicle_for_individual_driver(self): vehicle = Vehicle.objects.create( owner=self.transporter, vehicle_type='Flatbed', registration_number='IND-VEH-1', load_capacity=Decimal('5000'), is_verified=True, is_active=True, ) response = self.client.get(f'/api/transporter/trips/{self.trip.id}/') self.assertEqual(response.status_code, 200) assigned_vehicle = response.json()['data']['assigned_vehicle'] self.assertIsNotNone(assigned_vehicle) self.assertEqual(assigned_vehicle['id'], vehicle.id) self.assertEqual(assigned_vehicle['registration_number'], 'IND-VEH-1') @patch('api.views.notify_users') def test_assign_driver_notifies_driver(self, mock_notify): fleet_owner = User.objects.create_user( 'fleet-notif@test.com', email='fleet-notif@test.com', password='secret', ) UserRole.objects.create(user=fleet_owner, role=UserRole.Role.TRANSPORTER, phone='', language='en') TransporterProfile.objects.create( user=fleet_owner, account_type=TransporterProfile.AccountType.FLEET_OWNER, local=True, country_to_country=False, documents_verified=True, ) driver = User.objects.create_user( 'fleet-driver-notif@test.com', email='fleet-driver-notif@test.com', password='secret', ) UserRole.objects.create(user=driver, role=UserRole.Role.TRANSPORTER, phone='', language='en') TransporterProfile.objects.create( user=driver, account_type=TransporterProfile.AccountType.TRANSPORTER_DRIVER, local=True, country_to_country=False, documents_verified=True, ) TransporterDriverLink.objects.create(transporter=fleet_owner, driver=driver, is_active=True) self.trip.transporter = fleet_owner self.trip.assigned_driver = None self.trip.save(update_fields=['transporter', 'assigned_driver', 'updated_at']) fleet_token, _ = Token.objects.get_or_create(user=fleet_owner) fleet_client = APIClient() fleet_client.credentials(HTTP_AUTHORIZATION=f'Bearer {fleet_token.key}') response = fleet_client.post( f'/api/transporter/trips/{self.trip.id}/assign-driver/', {'driver_id': driver.id}, format='json', ) self.assertEqual(response.status_code, 200) mock_notify.assert_called_once() self.assertEqual(mock_notify.call_args[0][0], [driver.id]) self.assertEqual(mock_notify.call_args[0][1], 'TRIP_ASSIGNED') payload = mock_notify.call_args[0][2] self.assertEqual(payload['trip_id'], str(self.trip.id)) self.assertEqual(payload['shipment_id'], str(self.trip.shipment_id)) self.assertEqual(payload['assigned_by'], str(fleet_owner.id))