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, TransporterDriverLink, TransporterProfile, UserRole from core.models import Bid, Shipment, Trip User = get_user_model() class TripDriverLocationApiTests(TestCase): def setUp(self): self.shipper = User.objects.create_user( 'trip-loc-ship@test.com', email='trip-loc-ship@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.fleet_owner = User.objects.create_user( 'trip-loc-owner@test.com', email='trip-loc-owner@test.com', password='secret', ) UserRole.objects.create( user=self.fleet_owner, role=UserRole.Role.TRANSPORTER, phone='', language='en', ) TransporterProfile.objects.create( user=self.fleet_owner, account_type=TransporterProfile.AccountType.FLEET_OWNER, local=True, country_to_country=False, documents_verified=True, tc_id='owner-device', ) self.fleet_driver = User.objects.create_user( 'trip-loc-driver@test.com', email='trip-loc-driver@test.com', password='secret', ) UserRole.objects.create( user=self.fleet_driver, role=UserRole.Role.TRANSPORTER, phone='', language='en', ) TransporterProfile.objects.create( user=self.fleet_driver, account_type=TransporterProfile.AccountType.TRANSPORTER_DRIVER, local=True, country_to_country=False, documents_verified=True, tc_id='driver-device-99', tc_u_id='DRV99', ) TransporterDriverLink.objects.create( transporter=self.fleet_owner, driver=self.fleet_driver, is_active=True, ) self.individual_driver = User.objects.create_user( 'trip-loc-ind@test.com', email='trip-loc-ind@test.com', password='secret', ) UserRole.objects.create( user=self.individual_driver, role=UserRole.Role.TRANSPORTER, phone='', language='en', ) TransporterProfile.objects.create( user=self.individual_driver, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, documents_verified=True, tc_id='individual-device', ) self.other_shipper = User.objects.create_user( 'trip-loc-other@test.com', email='trip-loc-other@test.com', password='secret', ) UserRole.objects.create(user=self.other_shipper, role=UserRole.Role.SHIPPER, phone='', language='en') def _shipment(self): return 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.ASSIGNED, ) def _fleet_trip(self): shipment = self._shipment() bid = Bid.objects.create( shipment=shipment, transporter=self.fleet_owner, amount=Decimal('100.00'), status=Bid.Status.ACCEPTED, ) return Trip.objects.create( shipment=shipment, accepted_bid=bid, transporter=self.fleet_owner, assigned_driver=self.fleet_driver, status=Trip.Status.IN_TRANSIT, ) def _individual_trip(self): shipment = self._shipment() bid = Bid.objects.create( shipment=shipment, transporter=self.individual_driver, amount=Decimal('150.00'), status=Bid.Status.ACCEPTED, ) return Trip.objects.create( shipment=shipment, accepted_bid=bid, transporter=self.individual_driver, status=Trip.Status.IN_TRANSIT, ) def _client_for(self, user): token, _ = Token.objects.get_or_create(user=user) client = APIClient() client.credentials(HTTP_AUTHORIZATION=f'Bearer {token.key}') return client @patch('api.views.get_latest_device_position') def test_shipper_gets_assigned_driver_traccar_location(self, mock_position): mock_position.return_value = { 'latitude': 25.1234, 'longitude': 55.5678, 'raw': {'deviceId': 99, 'fixTime': '2026-07-13T08:00:00.000+00:00'}, } trip = self._fleet_trip() client = self._client_for(self.shipper) response = client.get(f'/api/shipper/trips/{trip.id}/driver-location/') self.assertEqual(response.status_code, 200) data = response.json()['data'] self.assertEqual(data['trip_id'], trip.id) self.assertEqual(data['driver']['id'], self.fleet_driver.id) self.assertEqual(data['tc_id'], 'driver-device-99') self.assertEqual(data['tc_u_id'], 'DRV99') self.assertEqual(data['position']['lat'], 25.1234) self.assertEqual(data['position']['lon'], 55.5678) mock_position.assert_called_once_with('driver-device-99') @patch('api.views.get_latest_device_position') def test_fleet_owner_can_use_transporter_trip_driver_location(self, mock_position): mock_position.return_value = { 'latitude': 24.0, 'longitude': 67.0, 'raw': {'deviceId': 99}, } trip = self._fleet_trip() client = self._client_for(self.fleet_owner) response = client.get(f'/api/transporter/trips/{trip.id}/driver-location/') self.assertEqual(response.status_code, 200) self.assertEqual(response.json()['data']['driver']['id'], self.fleet_driver.id) @patch('api.views.get_latest_device_position') def test_individual_driver_trip_uses_transporter_device(self, mock_position): mock_position.return_value = { 'latitude': 23.5, 'longitude': 66.5, 'raw': {'deviceId': 1}, } trip = self._individual_trip() client = self._client_for(self.shipper) response = client.get(f'/api/shipper/trips/{trip.id}/driver-location/') self.assertEqual(response.status_code, 200) self.assertEqual(response.json()['data']['driver']['id'], self.individual_driver.id) mock_position.assert_called_once_with('individual-device') def test_other_shipper_gets_404(self): trip = self._fleet_trip() client = self._client_for(self.other_shipper) response = client.get(f'/api/shipper/trips/{trip.id}/driver-location/') self.assertEqual(response.status_code, 404) def test_missing_tc_id_returns_400(self): self.fleet_driver.transporter_profile.tc_id = '' self.fleet_driver.transporter_profile.save(update_fields=['tc_id']) trip = self._fleet_trip() client = self._client_for(self.shipper) response = client.get(f'/api/shipper/trips/{trip.id}/driver-location/') self.assertEqual(response.status_code, 400) self.assertEqual(response.json()['error'], 'Missing tc_id on driver profile.')