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 Bid, Shipment, Zone User = get_user_model() def _register_payload(**overrides): payload = { 'email': 'newdriver@test.com', 'password': 'secret123', 'account_type': TransporterProfile.AccountType.DRIVER, 'country_to_country': False, 'local': True, } payload.update(overrides) return payload class TransporterLoadTypeSignupTests(TestCase): @patch('api.serializers.create_traccar_device', return_value={'id': 99}) @patch('api.serializers.generate_unique_tc_uid', return_value='TCUID1234') def test_register_requires_xor_load_type(self, _mock_uid, _mock_traccar): client = APIClient() response = client.post( '/api/transporter/register/', { 'email': 'xor@test.com', 'password': 'secret123', 'account_type': TransporterProfile.AccountType.DRIVER, 'country_to_country': True, 'local': True, }, format='json', ) self.assertEqual(response.status_code, 400) @patch('api.serializers.create_traccar_device', return_value={'id': 99}) @patch('api.serializers.generate_unique_tc_uid', return_value='TCUID1234') def test_register_persists_load_type(self, _mock_uid, _mock_traccar): client = APIClient() response = client.post( '/api/transporter/register/', _register_payload( email='c2c@test.com', country_to_country=True, local=False, ), format='json', ) self.assertEqual(response.status_code, 201) profile = TransporterProfile.objects.get(user__email='c2c@test.com') self.assertTrue(profile.country_to_country) self.assertFalse(profile.local) def test_fleet_owner_register_requires_company_name(self): client = APIClient() response = client.post( '/api/transporter/register/', { 'email': 'fleet-nocompany@test.com', 'password': 'secret123', 'account_type': TransporterProfile.AccountType.FLEET_OWNER, 'country_to_country': False, 'local': True, }, format='json', ) self.assertEqual(response.status_code, 400) self.assertIn('company_name', response.json()['error']) def test_fleet_owner_register_persists_company_name(self): client = APIClient() response = client.post( '/api/transporter/register/', { 'email': 'fleet-co@test.com', 'password': 'secret123', 'account_type': TransporterProfile.AccountType.FLEET_OWNER, 'company_name': ' Acme Fleet LLC ', 'country_to_country': False, 'local': True, }, format='json', ) self.assertEqual(response.status_code, 201) profile = TransporterProfile.objects.get(user__email='fleet-co@test.com') self.assertEqual(profile.company_name, 'Acme Fleet LLC') @patch('api.serializers.create_traccar_device', return_value={'id': 9001}) @patch('api.serializers.generate_unique_tc_uid', return_value='TC-DRV-CO') def test_individual_driver_register_optional_company_name(self, _uid, _traccar): client = APIClient() response = client.post( '/api/transporter/register/', { 'email': 'drv-co@test.com', 'password': 'secret123', 'account_type': TransporterProfile.AccountType.DRIVER, 'company_name': ' Solo Haul Co ', 'country_to_country': False, 'local': True, }, format='json', ) self.assertEqual(response.status_code, 201, response.content) profile = TransporterProfile.objects.get(user__email='drv-co@test.com') self.assertEqual(profile.company_name, 'Solo Haul Co') self.assertEqual(profile.account_type, TransporterProfile.AccountType.DRIVER) def test_profile_patch_requires_both_load_type_fields(self): user = User.objects.create_user('patch@test.com', email='patch@test.com', password='secret') UserRole.objects.create(user=user, role=UserRole.Role.TRANSPORTER, phone='', language='en') TransporterProfile.objects.create( user=user, account_type=TransporterProfile.AccountType.FLEET_OWNER, local=True, country_to_country=False, ) token, _ = Token.objects.get_or_create(user=user) client = APIClient() client.credentials(HTTP_AUTHORIZATION=f'Bearer {token.key}') response = client.patch( '/api/transporter/profile/', {'country_to_country': True}, format='json', ) self.assertEqual(response.status_code, 400) class ShipmentLoadTypeApiTests(TestCase): def setUp(self): self.shipper = User.objects.create_user('loadtype@test.com', email='loadtype@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) token, _ = Token.objects.get_or_create(user=self.shipper) self.client = APIClient() self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {token.key}') def _payload(self, **overrides): payload = { 'pickup_address': 'Pickup', 'delivery_address': 'Delivery', 'cargo_type': 'General', 'weight': '1 ton', 'vehicle_type_required': 'Flatbed', 'local': True, 'country_to_country': False, } payload.update(overrides) return payload def test_create_rejects_both_true(self): response = self.client.post( '/api/shipper/shipments/', self._payload(local=True, country_to_country=True), format='json', ) self.assertEqual(response.status_code, 400) def test_create_accepts_local_shipment(self): response = self.client.post( '/api/shipper/shipments/', self._payload(), format='json', ) self.assertEqual(response.status_code, 201) data = response.json()['data'] self.assertTrue(data['local']) self.assertFalse(data['country_to_country']) def test_patch_requires_both_load_type_fields(self): shipment = Shipment.objects.create( shipper=self.shipper, pickup_address='Pickup', delivery_address='Delivery', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', local=True, country_to_country=False, status=Shipment.Status.DRAFT, ) response = self.client.patch( f'/api/shipper/shipments/{shipment.id}/', {'country_to_country': True, 'local': False}, format='json', ) self.assertEqual(response.status_code, 200) shipment.refresh_from_db() self.assertTrue(shipment.country_to_country) self.assertFalse(shipment.local) class TransporterLoadTypeVisibilityTests(TestCase): def setUp(self): self.shipper = User.objects.create_user('vis@test.com', email='vis@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.local_shipment = Shipment.objects.create( shipper=self.shipper, pickup_address='A', 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', suggested_price=Decimal('1000.00'), local=True, country_to_country=False, status=Shipment.Status.PUBLISHED, ) self.c2c_shipment = Shipment.objects.create( shipper=self.shipper, pickup_address='C', pickup_lat=Decimal('24.8700000'), pickup_lon=Decimal('67.0100000'), pickup_country_code='AE', delivery_address='D', delivery_country_code='SA', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', suggested_price=Decimal('2000.00'), local=False, country_to_country=True, status=Shipment.Status.PUBLISHED, ) Zone.objects.create(country_code='AE', radius_km=Decimal('50')) def _make_fleet_owner_with_driver(self, *, local=True, country_to_country=False, tc_id='fleet-drv-1'): owner = User.objects.create_user( f'fleet-{local}-{country_to_country}@test.com', email=f'fleet-{local}-{country_to_country}@test.com', password='secret', ) UserRole.objects.create(user=owner, role=UserRole.Role.TRANSPORTER, phone='', language='en') TransporterProfile.objects.create( user=owner, account_type=TransporterProfile.AccountType.FLEET_OWNER, documents_verified=True, local=local, country_to_country=country_to_country, ) Vehicle.objects.create( owner=owner, vehicle_type='Flatbed', registration_number=f'FLEET-{tc_id}', load_capacity=Decimal('5000'), is_verified=True, is_active=True, ) driver = User.objects.create_user( f'drv-{tc_id}@test.com', email=f'drv-{tc_id}@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, documents_verified=True, local=local, country_to_country=country_to_country, tc_id=tc_id, ) TransporterDriverLink.objects.create(transporter=owner, driver=driver, is_active=True) Vehicle.objects.create( owner=owner, assigned_driver=driver, vehicle_type='Flatbed', registration_number=f'ASGN-{tc_id}', load_capacity=Decimal('5000'), is_verified=True, is_active=True, ) token, _ = Token.objects.get_or_create(user=owner) client = APIClient() client.credentials(HTTP_AUTHORIZATION=f'Bearer {token.key}') return client, owner, tc_id def _make_local_transporter(self): user = User.objects.create_user('localtr@test.com', email='localtr@test.com', password='secret') UserRole.objects.create(user=user, role=UserRole.Role.TRANSPORTER, phone='', language='en') TransporterProfile.objects.create( user=user, account_type=TransporterProfile.AccountType.DRIVER, documents_verified=True, local=True, country_to_country=False, tc_id='loc-111', ) Vehicle.objects.create( owner=user, vehicle_type='Flatbed', registration_number='LOC-1', load_capacity=Decimal('5000'), is_verified=True, is_active=True, ) token, _ = Token.objects.get_or_create(user=user) client = APIClient() client.credentials(HTTP_AUTHORIZATION=f'Bearer {token.key}') return client, user def _make_c2c_transporter(self): user = User.objects.create_user('c2ctr@test.com', email='c2ctr@test.com', password='secret') UserRole.objects.create(user=user, role=UserRole.Role.TRANSPORTER, phone='', language='en') TransporterProfile.objects.create( user=user, account_type=TransporterProfile.AccountType.DRIVER, documents_verified=True, local=False, country_to_country=True, tc_id='222', ) Vehicle.objects.create( owner=user, vehicle_type='Flatbed', registration_number='C2C-1', load_capacity=Decimal('5000'), is_verified=True, is_active=True, ) token, _ = Token.objects.get_or_create(user=user) client = APIClient() client.credentials(HTTP_AUTHORIZATION=f'Bearer {token.key}') return client, user @patch('api.views.country_code_for_point', return_value='AE') @patch('api.views.get_latest_positions_map') def test_local_transporter_available_shipments_excludes_c2c(self, mock_positions, _country): mock_positions.return_value = { 'fleet-drv-local': {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {'id': 1}}, } client, _owner, _tc_id = self._make_fleet_owner_with_driver( local=True, country_to_country=False, tc_id='fleet-drv-local', ) response = 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.local_shipment.id, ids) self.assertNotIn(self.c2c_shipment.id, ids) @patch('api.views.country_code_for_point', return_value='AE') @patch('api.views.get_latest_positions_map') def test_c2c_transporter_available_shipments_includes_both(self, mock_positions, _country): mock_positions.return_value = { 'fleet-drv-c2c': {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {'id': 1}}, } client, _owner, _tc_id = self._make_fleet_owner_with_driver( local=False, country_to_country=True, tc_id='fleet-drv-c2c', ) other_country_c2c = Shipment.objects.create( shipper=self.shipper, pickup_address='Doha', pickup_lat=Decimal('25.2854000'), pickup_lon=Decimal('51.5310000'), pickup_country_code='QA', delivery_address='Dubai', delivery_country_code='AE', cargo_type='General', weight='1 ton', vehicle_type_required='Box Trailer 15m', suggested_price=Decimal('3000.00'), local=False, country_to_country=True, status=Shipment.Status.PUBLISHED, ) response = 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.local_shipment.id, ids) self.assertIn(self.c2c_shipment.id, ids) self.assertNotIn(other_country_c2c.id, ids) @patch('api.views.get_latest_device_position') def test_local_transporter_cannot_bid_on_c2c_shipment(self, mock_position): mock_position.return_value = {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {}} client, user = self._make_local_transporter() response = client.post( f'/api/transporter/shipments/{self.c2c_shipment.id}/bid/', {'action': 'counter', 'amount': '1500.00'}, format='json', ) self.assertEqual(response.status_code, 403) self.assertFalse(Bid.objects.filter(shipment=self.c2c_shipment, transporter=user).exists()) @patch('api.views.get_latest_device_position') def test_load_discovery_respects_local_profile(self, mock_position): mock_position.return_value = {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {}} client, _user = self._make_local_transporter() TransporterProfile.objects.filter(user__email='localtr@test.com').update(tc_id='333') response = client.get('/api/transporter/load-discovery/?country_code=AE') self.assertEqual(response.status_code, 200) ids = {row['id'] for row in response.json()['data']['loads']} self.assertIn(self.local_shipment.id, ids) self.assertNotIn(self.c2c_shipment.id, ids) @patch('api.views.get_latest_device_position') def test_load_discovery_c2c_profile_includes_local_and_c2c(self, mock_position): mock_position.return_value = {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {}} client, _user = self._make_c2c_transporter() response = client.get('/api/transporter/load-discovery/?country_code=AE') self.assertEqual(response.status_code, 200) ids = {row['id'] for row in response.json()['data']['loads']} self.assertIn(self.local_shipment.id, ids) self.assertIn(self.c2c_shipment.id, ids) @patch('api.views.get_latest_device_position') def test_load_discovery_load_type_local_excludes_c2c(self, mock_position): mock_position.return_value = {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {}} client, _user = self._make_c2c_transporter() response = client.get('/api/transporter/load-discovery/?country_code=AE&load_type=local') self.assertEqual(response.status_code, 200) ids = {row['id'] for row in response.json()['data']['loads']} self.assertIn(self.local_shipment.id, ids) self.assertNotIn(self.c2c_shipment.id, ids) @patch('api.views.get_latest_device_position') def test_load_discovery_load_type_c2c_excludes_local(self, mock_position): mock_position.return_value = {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {}} client, _user = self._make_c2c_transporter() response = client.get( '/api/transporter/load-discovery/?country_code=AE&load_type=country_to_country' ) self.assertEqual(response.status_code, 200) ids = {row['id'] for row in response.json()['data']['loads']} self.assertNotIn(self.local_shipment.id, ids) self.assertIn(self.c2c_shipment.id, ids) @patch('api.views.country_code_for_point', return_value='AE') @patch('api.views.get_latest_device_position') def test_available_shipments_load_type_local_excludes_c2c(self, mock_position, _country): mock_position.return_value = {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {}} client, _user = self._make_c2c_transporter() response = client.get('/api/transporter/available-shipments/?country_code=AE&load_type=local') self.assertEqual(response.status_code, 200) ids = {row['id'] for row in response.json()['data']['shipments']} self.assertIn(self.local_shipment.id, ids) self.assertNotIn(self.c2c_shipment.id, ids) @patch('api.views.country_code_for_point', return_value='AE') @patch('api.views.get_latest_device_position') def test_available_shipments_load_type_c2c_excludes_local(self, mock_position, _country): mock_position.return_value = {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {}} client, _user = self._make_c2c_transporter() response = client.get( '/api/transporter/available-shipments/?country_code=AE&load_type=country_to_country' ) self.assertEqual(response.status_code, 200) ids = {row['id'] for row in response.json()['data']['shipments']} self.assertNotIn(self.local_shipment.id, ids) self.assertIn(self.c2c_shipment.id, ids) @patch('api.views.get_latest_device_position') def test_load_discovery_invalid_load_type_returns_400(self, mock_position): mock_position.return_value = {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {}} client, _user = self._make_c2c_transporter() response = client.get('/api/transporter/load-discovery/?country_code=AE&load_type=invalid') self.assertEqual(response.status_code, 400) self.assertIn('load_type', response.json()['error']) def test_available_shipments_invalid_load_type_returns_400(self): client, _user = self._make_c2c_transporter() response = client.get( '/api/transporter/available-shipments/?country_code=AE&load_type=bogus' ) self.assertEqual(response.status_code, 400) self.assertIn('load_type', response.json()['error']) @patch('api.views.get_latest_device_position') def test_local_profile_load_type_c2c_returns_empty(self, mock_position): mock_position.return_value = {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {}} client, _user = self._make_local_transporter() TransporterProfile.objects.filter(user__email='localtr@test.com').update(tc_id='333') response = client.get( '/api/transporter/load-discovery/?country_code=AE&load_type=country_to_country' ) self.assertEqual(response.status_code, 200) ids = {row['id'] for row in response.json()['data']['loads']} self.assertEqual(ids, set()) @patch('api.views.get_latest_device_position') def test_load_discovery_c2c_filter_excludes_same_country_flagged_c2c(self, mock_position): """Same-country routes must not appear under load_type=country_to_country.""" mock_position.return_value = {'latitude': 24.8610, 'longitude': 67.0020, 'raw': {}} same_country = Shipment.objects.create( shipper=self.shipper, pickup_address='Dubai A', pickup_lat=Decimal('24.8650000'), pickup_lon=Decimal('67.0050000'), pickup_country_code='AE', delivery_address='Dubai B', delivery_country_code='AE', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', suggested_price=Decimal('500.00'), local=False, country_to_country=True, status=Shipment.Status.PUBLISHED, ) client, _user = self._make_c2c_transporter() response = client.get( '/api/transporter/load-discovery/?country_code=AE&load_type=country_to_country' ) self.assertEqual(response.status_code, 200) ids = {row['id'] for row in response.json()['data']['loads']} self.assertNotIn(same_country.id, ids) self.assertNotIn(self.local_shipment.id, ids) self.assertIn(self.c2c_shipment.id, ids) for row in response.json()['data']['loads']: self.assertFalse(row.get('local')) self.assertTrue(row.get('country_to_country'))