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 core.models import Shipment, Zone User = get_user_model() MATCHING_VEHICLE_TYPE = 'Flat Bed 12m' OTHER_VEHICLE_TYPE = 'Box Trailer 15m' class ShipperDriverLocationsTests(TestCase): def setUp(self): self.shipper = User.objects.create_user('shiploc@test.com', email='shiploc@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.shipment = Shipment.objects.create( shipper=self.shipper, pickup_address='A', pickup_lat=Decimal('24.8607000'), pickup_lon=Decimal('67.0011000'), delivery_address='B', cargo_type='General', weight='1 ton', vehicle_type_required=MATCHING_VEHICLE_TYPE, local=True, country_to_country=False, status=Shipment.Status.PUBLISHED, ) self.driver_near = User.objects.create_user('drv1@test.com', email='drv1@test.com', password='secret') UserRole.objects.create(user=self.driver_near, role=UserRole.Role.TRANSPORTER, phone='111', language='en') TransporterProfile.objects.create( user=self.driver_near, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, documents_verified=True, tc_id='111', ) Vehicle.objects.create( owner=self.driver_near, vehicle_type=MATCHING_VEHICLE_TYPE, registration_number='DRV-111', load_capacity=Decimal('5000'), is_verified=True, is_active=True, ) self.driver_far = User.objects.create_user('drv2@test.com', email='drv2@test.com', password='secret') UserRole.objects.create(user=self.driver_far, role=UserRole.Role.TRANSPORTER, phone='222', language='en') TransporterProfile.objects.create( user=self.driver_far, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, documents_verified=True, tc_id='222', ) Vehicle.objects.create( owner=self.driver_far, vehicle_type=OTHER_VEHICLE_TYPE, registration_number='DRV-222', load_capacity=Decimal('5000'), is_verified=True, is_active=True, ) self.driver_wrong_type_near = User.objects.create_user( 'drv3@test.com', email='drv3@test.com', password='secret', ) UserRole.objects.create( user=self.driver_wrong_type_near, role=UserRole.Role.TRANSPORTER, phone='444', language='en', ) TransporterProfile.objects.create( user=self.driver_wrong_type_near, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, documents_verified=True, tc_id='444', ) Vehicle.objects.create( owner=self.driver_wrong_type_near, vehicle_type='Flatbed', registration_number='DRV-444', load_capacity=Decimal('5000'), is_verified=True, is_active=True, ) self.fleet_owner = User.objects.create_user('owner@test.com', email='owner@test.com', password='secret') UserRole.objects.create(user=self.fleet_owner, role=UserRole.Role.TRANSPORTER, phone='333', 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='333', ) Vehicle.objects.create( owner=self.fleet_owner, vehicle_type=MATCHING_VEHICLE_TYPE, registration_number='OWN-333', load_capacity=Decimal('5000'), is_verified=True, is_active=True, ) token, _ = Token.objects.get_or_create(user=self.shipper) self.client = APIClient() self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token.key}') self.ref_lat = 24.8607000 self.ref_lon = 67.0011000 self.payload = { 'shipment_id': self.shipment.id, 'country_code': 'AE', 'lat': str(self.ref_lat), 'lng': str(self.ref_lon), } @patch('api.views.get_latest_positions_map') def test_uses_zone_radius_when_country_matches(self, mock_positions): Zone.objects.create(country_code='AE', radius_km=Decimal('50'), allow_return_trips=False) mock_positions.return_value = { '111': {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {'id': 10}}, '222': {'latitude': 25.5000, 'longitude': 68.0000, 'raw': {'id': 11}}, '333': {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {'id': 12}}, '444': {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {'id': 13}}, } r = self.client.post('/api/shipper/drivers/locations/', self.payload, format='json') self.assertEqual(r.status_code, 200) data = r.json()['data'] self.assertEqual(data['shipment_id'], self.shipment.id) self.assertEqual(data['vehicle_type_required'], MATCHING_VEHICLE_TYPE) self.assertEqual(data['radius_km'], 50.0) self.assertEqual(data['radius_source'], 'zone') self.assertEqual(data['count'], 1) self.assertEqual(data['drivers'][0]['transporter_id'], self.driver_near.id) @patch('api.views.get_latest_positions_map') def test_uses_default_radius_when_zone_not_matched(self, mock_positions): mock_positions.return_value = { '111': {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {'id': 10}}, '222': {'latitude': 25.5000, 'longitude': 68.0000, 'raw': {'id': 11}}, } r = self.client.post( '/api/shipper/drivers/locations/', {**self.payload, 'country_code': 'ZA'}, format='json', ) self.assertEqual(r.status_code, 200) data = r.json()['data'] self.assertEqual(data['radius_km'], 25.0) self.assertEqual(data['radius_source'], 'default') self.assertEqual(data['count'], 1) self.assertEqual(data['drivers'][0]['transporter_id'], self.driver_near.id) @patch('api.views.get_latest_positions_map') def test_excludes_fleet_owner_even_within_radius(self, mock_positions): mock_positions.return_value = { '111': {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {'id': 10}}, '333': {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {'id': 12}}, } r = self.client.post('/api/shipper/drivers/locations/', self.payload, format='json') self.assertEqual(r.status_code, 200) ids = {d['transporter_id'] for d in r.json()['data']['drivers']} self.assertIn(self.driver_near.id, ids) self.assertNotIn(self.fleet_owner.id, ids) @patch('api.views.get_latest_positions_map') def test_excludes_driver_with_non_matching_vehicle_type(self, mock_positions): mock_positions.return_value = { '111': {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {'id': 10}}, '444': {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {'id': 13}}, } r = self.client.post('/api/shipper/drivers/locations/', self.payload, format='json') self.assertEqual(r.status_code, 200) ids = {d['transporter_id'] for d in r.json()['data']['drivers']} self.assertEqual(ids, {self.driver_near.id}) def test_rejects_non_shipper(self): transporter = User.objects.create_user('tr@test.com', email='tr@test.com', password='secret') UserRole.objects.create(user=transporter, role=UserRole.Role.TRANSPORTER, phone='', language='en') token, _ = Token.objects.get_or_create(user=transporter) client = APIClient() client.credentials(HTTP_AUTHORIZATION=f'Bearer {token.key}') r = client.post('/api/shipper/drivers/locations/', self.payload, format='json') self.assertEqual(r.status_code, 403) def test_rejects_invalid_country_code(self): r = self.client.post( '/api/shipper/drivers/locations/', {**self.payload, 'country_code': 'INVALID'}, format='json', ) self.assertEqual(r.status_code, 400) def test_rejects_missing_shipment_id(self): r = self.client.post( '/api/shipper/drivers/locations/', {'country_code': 'AE', 'lat': str(self.ref_lat), 'lng': str(self.ref_lon)}, format='json', ) self.assertEqual(r.status_code, 400) def test_rejects_other_shippers_shipment(self): other = User.objects.create_user('other@test.com', email='other@test.com', password='secret') UserRole.objects.create(user=other, role=UserRole.Role.SHIPPER, phone='', language='en') ShipperProfile.objects.create(user=other, account_type=ShipperProfile.AccountType.INDIVIDUAL) foreign_shipment = Shipment.objects.create( shipper=other, pickup_address='X', delivery_address='Y', cargo_type='General', weight='1 ton', vehicle_type_required=MATCHING_VEHICLE_TYPE, local=True, country_to_country=False, status=Shipment.Status.PUBLISHED, ) r = self.client.post( '/api/shipper/drivers/locations/', {**self.payload, 'shipment_id': foreign_shipment.id}, format='json', ) self.assertEqual(r.status_code, 404) def test_rejects_missing_coordinates(self): r = self.client.post( '/api/shipper/drivers/locations/', {'country_code': 'AE', 'shipment_id': self.shipment.id}, format='json', ) self.assertEqual(r.status_code, 400) @patch('api.views.get_latest_positions_map') def test_traccar_failure_returns_400(self, mock_positions): mock_positions.side_effect = ValueError('Traccar credentials are not configured.') r = self.client.post('/api/shipper/drivers/locations/', self.payload, format='json') self.assertEqual(r.status_code, 400) self.assertIn('Traccar', r.json()['error']) @patch('api.views.get_latest_device_position') @patch('api.views.get_latest_positions_map') def test_falls_back_to_device_position_when_bulk_map_misses(self, mock_positions, mock_latest): Zone.objects.create(country_code='AE', radius_km=Decimal('50'), allow_return_trips=False) mock_positions.return_value = {} mock_latest.side_effect = [ {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {'id': 10}}, {'latitude': 25.5000, 'longitude': 68.0000, 'raw': {'id': 11}}, ] r = self.client.post('/api/shipper/drivers/locations/', self.payload, format='json') self.assertEqual(r.status_code, 200) self.assertEqual(r.json()['data']['count'], 1)