"""Ensure unique_id is returned on all shipper, transporter, and driver trip APIs.""" from decimal import Decimal from django.contrib.auth import get_user_model from rest_framework.authtoken.models import Token from rest_framework.test import APIClient, APITestCase from accounts.models import ShipperProfile, TransporterDriverLink, TransporterProfile, UserRole from core.models import Bid, Shipment, Trip User = get_user_model() class TripUniqueIdApiCoverageTests(APITestCase): @classmethod def setUpTestData(cls): cls.shipper = User.objects.create_user( 'uid-ship@test.com', email='uid-ship@test.com', password='secret', ) UserRole.objects.create(user=cls.shipper, role=UserRole.Role.SHIPPER, phone='0300', language='en') ShipperProfile.objects.create(user=cls.shipper, account_type=ShipperProfile.AccountType.INDIVIDUAL) cls.fleet_owner = User.objects.create_user( 'uid-fleet@test.com', email='uid-fleet@test.com', password='secret', ) UserRole.objects.create(user=cls.fleet_owner, role=UserRole.Role.TRANSPORTER, phone='0301', language='en') TransporterProfile.objects.create( user=cls.fleet_owner, account_type=TransporterProfile.AccountType.FLEET_OWNER, local=True, country_to_country=False, documents_verified=True, tc_id='101', ) cls.fleet_driver = User.objects.create_user( 'uid-driver@test.com', email='uid-driver@test.com', password='secret', ) UserRole.objects.create(user=cls.fleet_driver, role=UserRole.Role.TRANSPORTER, phone='0302', language='en') TransporterProfile.objects.create( user=cls.fleet_driver, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, documents_verified=True, tc_id='102', ) TransporterDriverLink.objects.create(transporter=cls.fleet_owner, driver=cls.fleet_driver, is_active=True) cls.individual_driver = User.objects.create_user( 'uid-ind@test.com', email='uid-ind@test.com', password='secret', ) UserRole.objects.create(user=cls.individual_driver, role=UserRole.Role.TRANSPORTER, phone='0303', language='en') TransporterProfile.objects.create( user=cls.individual_driver, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, documents_verified=True, tc_id='103', ) def make_trip(transporter, suffix, *, status=Trip.Status.ASSIGNED, driver=None): shipment = Shipment.objects.create( shipper=cls.shipper, pickup_address=f'A-{suffix}', delivery_address=f'B-{suffix}', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', local=True, country_to_country=False, status=Shipment.Status.ASSIGNED, ) bid = Bid.objects.create( shipment=shipment, transporter=transporter, amount=Decimal('100.00'), status=Bid.Status.ACCEPTED, ) trip = Trip.objects.create( shipment=shipment, accepted_bid=bid, transporter=transporter, status=status, assigned_driver=driver, ) return trip, shipment.unique_id cls.fleet_trip, cls.fleet_unique_id = make_trip( cls.fleet_owner, 'fleet', driver=cls.fleet_driver, status=Trip.Status.EN_ROUTE, ) cls.individual_trip, cls.individual_unique_id = make_trip( cls.individual_driver, 'ind', status=Trip.Status.ASSIGNED, ) cls.completed_trip, cls.completed_unique_id = make_trip( cls.individual_driver, 'done', status=Trip.Status.COMPLETED, ) def _client(self, user): token, _ = Token.objects.get_or_create(user=user) client = APIClient() client.credentials(HTTP_AUTHORIZATION=f'Bearer {token.key}') return client def _assert_trip_payload_has_unique_id(self, payload, expected): self.assertEqual(payload['unique_id'], expected) def test_shipper_trip_list_and_detail(self): client = self._client(self.shipper) listing = client.get('/api/shipper/trips/') self.assertEqual(listing.status_code, 200) rows = listing.json()['data'] self.assertTrue(rows) for row in rows: self.assertIn('unique_id', row) self.assertTrue(row['unique_id'].startswith('SH-')) detail = client.get(f'/api/shipper/trips/{self.fleet_trip.id}/') self.assertEqual(detail.status_code, 200) self._assert_trip_payload_has_unique_id(detail.json()['data'], self.fleet_unique_id) self.assertEqual(detail.json()['data']['shipment']['unique_id'], self.fleet_unique_id) def test_fleet_owner_trip_lists(self): client = self._client(self.fleet_owner) for path in ( '/api/transporter/trips/my/', '/api/transporter/trips/assigned/', '/api/transporter/trips/active/', ): r = client.get(path) self.assertEqual(r.status_code, 200, path) rows = r.json()['data'] self.assertTrue(any(row['id'] == self.fleet_trip.id for row in rows), path) match = next(row for row in rows if row['id'] == self.fleet_trip.id) self._assert_trip_payload_has_unique_id(match, self.fleet_unique_id) def test_fleet_owner_trip_detail_and_navigation(self): client = self._client(self.fleet_owner) detail = client.get(f'/api/transporter/trips/{self.fleet_trip.id}/') self.assertEqual(detail.status_code, 200) self._assert_trip_payload_has_unique_id(detail.json()['data'], self.fleet_unique_id) nav = client.get(f'/api/transporter/trips/{self.fleet_trip.id}/navigation/') self.assertEqual(nav.status_code, 200) self.assertEqual(nav.json()['data']['unique_id'], self.fleet_unique_id) def test_individual_driver_trip_lists(self): client = self._client(self.individual_driver) endpoints = ( ('/api/transporter/my-trips/assigned/', self.individual_trip.id, self.individual_unique_id), ('/api/transporter/my-trips/completed/', self.completed_trip.id, self.completed_unique_id), ) for path, trip_id, expected_uid in endpoints: r = client.get(path) self.assertEqual(r.status_code, 200, path) trips = r.json()['data']['trips'] match = next(t for t in trips if t['id'] == trip_id) self._assert_trip_payload_has_unique_id(match, expected_uid) def test_fleet_driver_trip_lists(self): client = self._client(self.fleet_driver) r = client.get('/api/transporter/driver/trips/active/') self.assertEqual(r.status_code, 200) rows = r.json()['data'] match = next(row for row in rows if row['id'] == self.fleet_trip.id) self._assert_trip_payload_has_unique_id(match, self.fleet_unique_id) def test_driver_status_update_returns_unique_id(self): shipment = Shipment.objects.create( shipper=self.shipper, pickup_address='A-st', delivery_address='B-st', cargo_type='General', weight='1 ton', vehicle_type_required='Flatbed', local=True, country_to_country=False, status=Shipment.Status.ASSIGNED, ) bid = Bid.objects.create( shipment=shipment, transporter=self.individual_driver, amount=Decimal('50.00'), status=Bid.Status.ACCEPTED, ) trip = Trip.objects.create( shipment=shipment, accepted_bid=bid, transporter=self.individual_driver, status=Trip.Status.ASSIGNED, ) expected = shipment.unique_id client = self._client(self.individual_driver) r = client.patch( f'/api/transporter/trips/{trip.id}/status/', {'status': Trip.Status.EN_ROUTE}, format='json', ) self.assertEqual(r.status_code, 200) self._assert_trip_payload_has_unique_id(r.json()['data'], expected)