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, Vehicle, ) from core.models import Shipment, Zone User = get_user_model() def _make_fleet_driver(owner, *, email, tc_id='drv-1', phone=''): driver = User.objects.create_user( email, email=email, password='secret', first_name='Fleet', last_name='Driver', ) UserRole.objects.create(user=driver, role=UserRole.Role.TRANSPORTER, phone=phone, language='en') TransporterProfile.objects.create( user=driver, account_type=TransporterProfile.AccountType.TRANSPORTER_DRIVER, local=owner.transporter_profile.local, country_to_country=owner.transporter_profile.country_to_country, documents_verified=True, tc_id=tc_id, ) TransporterDriverLink.objects.create(transporter=owner, driver=driver, is_active=True) return driver class TransporterAvailableShipmentsApiTests(TestCase): def setUp(self): self.shipper = User.objects.create_user( 'avail-ship@test.com', email='avail-ship@test.com', password='secret', first_name='Sara', last_name='Shipper', ) UserRole.objects.create(user=self.shipper, role=UserRole.Role.SHIPPER, phone='555-0100', language='en') ShipperProfile.objects.create( user=self.shipper, account_type=ShipperProfile.AccountType.BUSINESS, company_name='Acme Logistics', ) self.fleet_owner = User.objects.create_user( 'avail-fleet@test.com', email='avail-fleet@test.com', password='secret', ) UserRole.objects.create(user=self.fleet_owner, role=UserRole.Role.TRANSPORTER, phone='123', language='en') TransporterProfile.objects.create( user=self.fleet_owner, account_type=TransporterProfile.AccountType.FLEET_OWNER, local=True, country_to_country=False, documents_verified=True, ) Vehicle.objects.create( owner=self.fleet_owner, vehicle_type='Flatbed', registration_number='FLEET-1', load_capacity=Decimal('5000'), is_verified=True, is_active=True, ) self.fleet_driver = _make_fleet_driver( self.fleet_owner, email='fleet-driver1@test.com', phone='555-0200', ) self.shipment = Shipment.objects.create( shipper=self.shipper, pickup_address='A', pickup_lat=Decimal('24.8607000'), pickup_lon=Decimal('67.0011000'), delivery_address='B', pickup_country_code='AE', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', suggested_price=Decimal('1500.00'), local=True, country_to_country=False, status=Shipment.Status.PUBLISHED, ) Zone.objects.get_or_create(country_code='AE', defaults={'radius_km': Decimal('50')}) token, _ = Token.objects.get_or_create(user=self.fleet_owner) self.client = APIClient() self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token.key}') @patch('api.views.get_latest_positions_map', return_value={}) def test_fleet_owner_no_driver_positions_returns_all_loads(self, _mock_pos): r = self.client.get('/api/transporter/available-shipments/?country_code=AE') self.assertEqual(r.status_code, 200) ids = {item['id'] for item in r.json()['data']['shipments']} self.assertIn(self.shipment.id, ids) self.assertEqual(r.json()['data']['driver_location'], None) @patch('api.views.country_code_for_point', return_value='AE') @patch('api.views.get_latest_positions_map') def test_lists_without_query_params(self, mock_positions, _country): mock_positions.return_value = { 'drv-1': {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {'id': 1}}, } response = self.client.get('/api/transporter/available-shipments/') self.assertEqual(response.status_code, 200) ids = {row['id'] for row in response.json()['data']['shipments']} self.assertIn(self.shipment.id, ids) @patch('api.views.country_code_for_point', return_value='PK') @patch('api.views.get_latest_positions_map') def test_fetches_latest_linked_driver_location(self, mock_positions, _country): mock_positions.return_value = { 'drv-1': {'latitude': 31.5204, 'longitude': 74.3587, 'raw': {'id': 1}}, } Zone.objects.get_or_create(country_code='PK', defaults={'radius_km': Decimal('50')}) response = self.client.get('/api/transporter/available-shipments/') self.assertEqual(response.status_code, 200) data = response.json()['data'] self.assertEqual(data['driver_location'], {'lat': 31.5204, 'lon': 74.3587}) self.assertEqual(len(data['driver_locations']), 1) self.assertEqual(data['driver_locations'][0]['id'], self.fleet_driver.id) self.assertEqual( data['driver_locations'][0]['location'], {'lat': 31.5204, 'lon': 74.3587}, ) ids = {row['id'] for row in data['shipments']} self.assertNotIn(self.shipment.id, ids) @patch('api.views.country_code_for_point', return_value='AE') @patch('api.views.get_latest_positions_map') def test_lists_local_shipments_with_shipper_summary(self, mock_positions, _country): mock_positions.return_value = { 'drv-1': {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {'id': 1}}, } distant = Shipment.objects.create( shipper=self.shipper, pickup_address='Far', pickup_lat=Decimal('51.5074000'), pickup_lon=Decimal('-0.1278000'), pickup_country_code='AE', delivery_address='B', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', local=True, country_to_country=False, status=Shipment.Status.PUBLISHED, ) no_coords = Shipment.objects.create( shipper=self.shipper, pickup_address='No GPS', pickup_lat=None, pickup_lon=None, pickup_country_code='AE', delivery_address='B', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', local=True, country_to_country=False, status=Shipment.Status.PUBLISHED, ) response = self.client.get('/api/transporter/available-shipments/') self.assertEqual(response.status_code, 200) data = response.json()['data'] ids = {row['id'] for row in data['shipments']} self.assertIn(self.shipment.id, ids) self.assertNotIn(distant.id, ids) self.assertNotIn(no_coords.id, ids) row = next(item for item in data['shipments'] if item['id'] == self.shipment.id) self.assertEqual(len(row['nearby_drivers']), 1) self.assertEqual(row['nearby_drivers'][0]['id'], self.fleet_driver.id) self.assertIsNotNone(row['nearest_distance_km']) self.assertEqual(row['currency'], 'AED') self.assertEqual( row['shipper'], { 'id': self.shipper.id, 'email': 'avail-ship@test.com', 'first_name': 'Sara', 'last_name': 'Shipper', 'phone': '555-0100', 'account_type': ShipperProfile.AccountType.BUSINESS, 'company_name': 'Acme Logistics', }, ) @patch('api.views.country_code_for_point', return_value='PK') @patch('api.views.get_latest_positions_map') def test_pakistan_drivers_exclude_dubai_pickups(self, mock_positions, _country): mock_positions.return_value = { 'drv-1': {'latitude': 31.5204, 'longitude': 74.3587, 'raw': {'id': 1}}, } Zone.objects.get_or_create(country_code='PK', defaults={'radius_km': Decimal('50')}) lahore = Shipment.objects.create( shipper=self.shipper, pickup_address='Lahore', pickup_lat=Decimal('31.5204000'), pickup_lon=Decimal('74.3587000'), pickup_country_code='PK', delivery_address='B', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', local=True, country_to_country=False, status=Shipment.Status.PUBLISHED, ) dubai = Shipment.objects.create( shipper=self.shipper, pickup_address='Dubai', pickup_lat=Decimal('25.2048000'), pickup_lon=Decimal('55.2708000'), pickup_country_code='AE', delivery_address='B', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', local=True, country_to_country=False, status=Shipment.Status.PUBLISHED, ) response = self.client.get('/api/transporter/available-shipments/') self.assertEqual(response.status_code, 200) ids = {row['id'] for row in response.json()['data']['shipments']} self.assertIn(lahore.id, ids) self.assertNotIn(dubai.id, ids) self.assertNotIn(self.shipment.id, ids) @patch('api.views.country_code_for_point', return_value='AE') @patch('api.views.get_latest_positions_map') def test_excludes_c2c_for_local_fleet_owner(self, mock_positions, _country): mock_positions.return_value = { 'drv-1': {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {'id': 1}}, } c2c = Shipment.objects.create( shipper=self.shipper, pickup_address='C2C', pickup_lat=Decimal('24.8607000'), pickup_lon=Decimal('67.0011000'), pickup_country_code='AE', delivery_address='B', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', local=False, country_to_country=True, status=Shipment.Status.PUBLISHED, ) response = self.client.get('/api/transporter/available-shipments/') self.assertEqual(response.status_code, 200) ids = {row['id'] for row in response.json()['data']['shipments']} self.assertIn(self.shipment.id, ids) self.assertNotIn(c2c.id, ids) @patch('api.views.get_latest_positions_map', return_value={}) def test_no_approved_driver_returns_all_loads(self, _mock_pos): TransporterProfile.objects.filter(user=self.fleet_driver).update(documents_verified=False) response = self.client.get('/api/transporter/available-shipments/?country_code=AE') self.assertEqual(response.status_code, 200) data = response.json()['data'] ids = {row['id'] for row in data['shipments']} self.assertIn(self.shipment.id, ids) self.assertIsNone(data['driver_location']) self.assertEqual(data['driver_locations'], []) self.assertEqual(data['count'], len(data['shipments'])) @patch('api.views.get_latest_positions_map', return_value={}) def test_no_verified_vehicle_returns_all_loads(self, _mock_pos): Vehicle.objects.filter(owner=self.fleet_owner).update(is_verified=False) response = self.client.get('/api/transporter/available-shipments/?country_code=AE') self.assertEqual(response.status_code, 200) data = response.json()['data'] ids = {row['id'] for row in data['shipments']} self.assertIn(self.shipment.id, ids) self.assertIsNone(data['driver_location']) self.assertEqual(data['driver_locations'], []) self.assertEqual(data['count'], len(data['shipments'])) @patch('api.views.country_code_for_point', return_value='AE') @patch('api.views.get_latest_positions_map') def test_c2c_owner_uses_driver_gps_not_query_country(self, mock_positions, _country): """C2C owner with a Dubai driver still sees UAE loads when country_code=PK.""" TransporterProfile.objects.filter(user=self.fleet_owner).update( local=False, country_to_country=True, ) TransporterProfile.objects.filter(user=self.fleet_driver).update( local=False, country_to_country=True, ) mock_positions.return_value = { 'drv-1': {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {'id': 1}}, } pk_local = Shipment.objects.create( shipper=self.shipper, pickup_address='Lahore', pickup_lat=Decimal('31.5204000'), pickup_lon=Decimal('74.3587000'), pickup_country_code='PK', delivery_address='B', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', local=True, country_to_country=False, status=Shipment.Status.PUBLISHED, ) ae_c2c = Shipment.objects.create( shipper=self.shipper, pickup_address='Dubai C2C', pickup_lat=Decimal('24.8607000'), pickup_lon=Decimal('67.0011000'), pickup_country_code='AE', delivery_address='Karachi', delivery_country_code='PK', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', local=False, country_to_country=True, status=Shipment.Status.PUBLISHED, ) response = self.client.get('/api/transporter/available-shipments/?country_code=PK') self.assertEqual(response.status_code, 200) data = response.json()['data'] self.assertIsNotNone(data['shipments']) self.assertIsNotNone(data['driver_location']) ids = {row['id'] for row in data['shipments']} self.assertIn(self.shipment.id, ids) self.assertIn(ae_c2c.id, ids) self.assertNotIn(pk_local.id, ids) @patch('api.views.get_latest_device_position', return_value={'latitude': 31.5204, 'longitude': 74.3587, 'raw': {}}) def test_bid_requires_verified_vehicle(self, _mock_pos): Vehicle.objects.filter(owner=self.fleet_owner).update(is_verified=False) response = self.client.post( f'/api/transporter/shipments/{self.shipment.id}/bid/', {'action': 'COUNTER', 'amount': '1400.00'}, format='json', ) self.assertEqual(response.status_code, 400) self.assertEqual(response.json()['error'], 'Missing verified active vehicle.') @patch('api.views.get_latest_device_position', return_value={'latitude': 31.5204, 'longitude': 74.3587, 'raw': {}}) def test_fleet_owner_bid_requires_own_driver(self, _mock_pos): TransporterDriverLink.objects.filter(transporter=self.fleet_owner).delete() response = self.client.post( f'/api/transporter/shipments/{self.shipment.id}/bid/', {'action': 'COUNTER', 'amount': '1400.00'}, format='json', ) self.assertEqual(response.status_code, 400) self.assertEqual(response.json()['error'], 'Missing fleet driver.') @patch('api.views.get_latest_device_position', return_value={'latitude': 31.5204, 'longitude': 74.3587, 'raw': {}}) def test_fleet_owner_with_driver_can_bid(self, _mock_pos): response = self.client.post( f'/api/transporter/shipments/{self.shipment.id}/bid/', {'action': 'COUNTER', 'amount': '1400.00'}, format='json', ) self.assertEqual(response.status_code, 201, response.content) self.assertEqual(response.json()['data']['currency'], 'AED') class IndividualDriverAvailableShipmentsApiTests(TestCase): def setUp(self): self.shipper = User.objects.create_user( 'avail-drv-ship@test.com', email='avail-drv-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.driver = User.objects.create_user( 'avail-drv@test.com', email='avail-drv@test.com', password='secret', ) UserRole.objects.create(user=self.driver, role=UserRole.Role.TRANSPORTER, phone='123', language='en') TransporterProfile.objects.create( user=self.driver, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, documents_verified=True, tc_id='drv-lhr', ) self.vehicle = Vehicle.objects.create( owner=self.driver, vehicle_type='Flatbed', registration_number='DRV-LHR-1', load_capacity=Decimal('5000'), is_verified=True, is_active=True, ) Zone.objects.get_or_create(country_code='PK', defaults={'radius_km': Decimal('50')}) Zone.objects.get_or_create(country_code='AE', defaults={'radius_km': Decimal('50')}) self.lahore = Shipment.objects.create( shipper=self.shipper, pickup_address='Lahore', pickup_lat=Decimal('31.5204000'), pickup_lon=Decimal('74.3587000'), pickup_country_code='PK', delivery_address='B', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', local=True, country_to_country=False, status=Shipment.Status.PUBLISHED, ) self.lahore_box = Shipment.objects.create( shipper=self.shipper, pickup_address='Lahore box', pickup_lat=Decimal('31.5210000'), pickup_lon=Decimal('74.3590000'), pickup_country_code='PK', delivery_address='B', cargo_type='General', weight='1 ton', vehicle_type_required='Box Trailer 15m', local=True, country_to_country=False, status=Shipment.Status.PUBLISHED, ) self.dubai_ae = Shipment.objects.create( shipper=self.shipper, pickup_address='Dubai', pickup_lat=Decimal('25.2048000'), pickup_lon=Decimal('55.2708000'), pickup_country_code='AE', delivery_address='B', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', local=True, country_to_country=False, status=Shipment.Status.PUBLISHED, ) self.dubai_tagged_pk = Shipment.objects.create( shipper=self.shipper, pickup_address='Dubai mis-tagged', pickup_lat=Decimal('25.2048000'), pickup_lon=Decimal('55.2708000'), pickup_country_code='PK', delivery_address='B', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', local=True, country_to_country=False, status=Shipment.Status.PUBLISHED, ) token, _ = Token.objects.get_or_create(user=self.driver) self.client = APIClient() self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token.key}') @patch('api.views.get_latest_device_position', return_value={'latitude': 31.5204, 'longitude': 74.3587, 'raw': {}}) def test_requires_country_code(self, _mock_pos): r = self.client.get('/api/transporter/available-shipments/') self.assertEqual(r.status_code, 400) self.assertIn('country_code', r.json()['error']) @patch('api.views.country_code_for_point', return_value='PK') @patch('api.views.get_latest_device_position') def test_lahore_driver_excludes_dubai_and_mismatched_vehicle(self, mock_position, _country): mock_position.return_value = {'latitude': 31.5204, 'longitude': 74.3587, 'raw': {}} r = self.client.get('/api/transporter/available-shipments/?country_code=PK') self.assertEqual(r.status_code, 200) ids = {item['id'] for item in r.json()['data']['shipments']} self.assertIn(self.lahore.id, ids) self.assertNotIn(self.lahore_box.id, ids) self.assertNotIn(self.dubai_ae.id, ids) self.assertNotIn(self.dubai_tagged_pk.id, ids) row = next(item for item in r.json()['data']['shipments'] if item['id'] == self.lahore.id) self.assertIsNotNone(row['nearest_distance_km']) self.assertLess(row['nearest_distance_km'], 50) self.assertEqual(r.json()['data']['driver_location'], {'lat': 31.5204, 'lon': 74.3587}) @patch('api.views.country_code_for_point', return_value='PK') @patch('api.views.get_latest_device_position', return_value={'latitude': 31.5204, 'longitude': 74.3587, 'raw': {}}) def test_without_verified_vehicle_returns_all_loads_with_fallback(self, _mock_pos, _country): self.vehicle.is_verified = False self.vehicle.save(update_fields=['is_verified']) r = self.client.get('/api/transporter/available-shipments/?country_code=PK') self.assertEqual(r.status_code, 200) ids = {item['id'] for item in r.json()['data']['shipments']} self.assertIn(self.lahore.id, ids) self.assertEqual(r.json()['data']['driver_location'], None) @patch('api.views.country_code_for_point', return_value='AE') @patch('api.views.get_latest_device_position', return_value={'latitude': 25.2048, 'longitude': 55.2708, 'raw': {}}) def test_gps_outside_query_country_returns_all_country_loads(self, _mock_pos, _country): r = self.client.get('/api/transporter/available-shipments/?country_code=PK') self.assertEqual(r.status_code, 200) ids = {item['id'] for item in r.json()['data']['shipments']} self.assertIn(self.lahore.id, ids) self.assertEqual(r.json()['data']['driver_location'], None) def test_individual_driver_without_tc_id_returns_all_loads(self): self.driver.transporter_profile.tc_id = '' self.driver.transporter_profile.save(update_fields=['tc_id']) r = self.client.get('/api/transporter/available-shipments/?country_code=PK') self.assertEqual(r.status_code, 200) ids = {item['id'] for item in r.json()['data']['shipments']} self.assertIn(self.lahore.id, ids) self.assertEqual(r.json()['data']['driver_location'], None) @patch('api.views.get_latest_device_position', side_effect=Exception('Traccar timeout')) def test_individual_driver_traccar_failure_returns_all_loads(self, _mock_pos): r = self.client.get('/api/transporter/available-shipments/?country_code=PK') self.assertEqual(r.status_code, 200) ids = {item['id'] for item in r.json()['data']['shipments']} self.assertIn(self.lahore.id, ids) self.assertEqual(r.json()['data']['driver_location'], None) @patch('api.views.get_latest_device_position', return_value={'latitude': None, 'longitude': None}) def test_individual_driver_null_coordinates_returns_all_loads(self, _mock_pos): r = self.client.get('/api/transporter/available-shipments/?country_code=PK') self.assertEqual(r.status_code, 200) ids = {item['id'] for item in r.json()['data']['shipments']} self.assertIn(self.lahore.id, ids) self.assertEqual(r.json()['data']['driver_location'], None) @patch('api.views.country_code_for_point', return_value='PK') @patch('api.views.get_latest_device_position') def test_available_shipments_returns_loads_in_created_at_desc_order(self, mock_position, _country): mock_position.return_value = {'latitude': 31.5204, 'longitude': 74.3587, 'raw': {}} second = Shipment.objects.create( shipper=self.shipper, pickup_address='Lahore Second', pickup_lat=Decimal('31.5210000'), pickup_lon=Decimal('74.3590000'), pickup_country_code='PK', delivery_address='B2', cargo_type='General', vehicle_type_required='Flatbed', local=True, status=Shipment.Status.PUBLISHED, ) third = Shipment.objects.create( shipper=self.shipper, pickup_address='Lahore Third', pickup_lat=Decimal('31.5215000'), pickup_lon=Decimal('74.3595000'), pickup_country_code='PK', delivery_address='B3', cargo_type='General', vehicle_type_required='Flatbed', local=True, status=Shipment.Status.PUBLISHED, ) r = self.client.get('/api/transporter/available-shipments/?country_code=PK') self.assertEqual(r.status_code, 200) shipments = r.json()['data']['shipments'] self.assertGreaterEqual(len(shipments), 3) shipment_ids = [s['id'] for s in shipments] self.assertEqual(shipment_ids[:3], [third.id, second.id, self.lahore.id])