from decimal import Decimal 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, TransporterDriverLink, UserRole from core.models import Bid, Shipment, Trip, TripShipperReview User = get_user_model() class TransporterShipperReviewApiTests(TestCase): def setUp(self): self.shipper = User.objects.create_user('ship-review@test.com', email='ship-review@test.com', password='secret') UserRole.objects.create(user=self.shipper, role=UserRole.Role.SHIPPER) ShipperProfile.objects.create(user=self.shipper, account_type=ShipperProfile.AccountType.INDIVIDUAL) self.transporter = User.objects.create_user('tr-review@test.com', email='tr-review@test.com', password='secret') UserRole.objects.create(user=self.transporter, role=UserRole.Role.TRANSPORTER) TransporterProfile.objects.create( user=self.transporter, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, documents_verified=True, ) self.assigned_driver = User.objects.create_user('assigned-review@test.com', email='assigned-review@test.com', password='secret') UserRole.objects.create(user=self.assigned_driver, role=UserRole.Role.TRANSPORTER) TransporterProfile.objects.create( user=self.assigned_driver, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, documents_verified=True, ) self.other_transporter = User.objects.create_user('other-review@test.com', email='other-review@test.com', password='secret') UserRole.objects.create(user=self.other_transporter, role=UserRole.Role.TRANSPORTER) TransporterProfile.objects.create( user=self.other_transporter, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, documents_verified=True, ) self.trip = self._trip(status=Trip.Status.COMPLETED) TransporterDriverLink.objects.create( transporter=self.transporter, driver=self.assigned_driver, is_active=True, ) 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 def _trip(self, status): shipment = 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.COMPLETED, ) bid = Bid.objects.create( shipment=shipment, transporter=self.transporter, amount=Decimal('100.00'), status=Bid.Status.ACCEPTED, ) return Trip.objects.create( shipment=shipment, accepted_bid=bid, transporter=self.transporter, assigned_driver=self.assigned_driver, status=status, ) def test_transporter_can_rate_trip_shipper_once_trip_completed(self): response = self._client_for(self.transporter).post( f'/api/transporter/trips/{self.trip.id}/shipper-review/', {'shipper_rating': 5, 'comment': 'Easy pickup and quick payment.'}, format='json', ) self.assertEqual(response.status_code, 201) data = response.json()['data'] self.assertEqual(data['trip'], self.trip.id) self.assertEqual(data['shipper'], self.shipper.id) self.assertEqual(data['transporter'], self.transporter.id) self.assertEqual(data['reviewer'], self.transporter.id) self.assertEqual(data['shipper_rating'], 5) self.assertTrue(TripShipperReview.objects.filter(trip=self.trip).exists()) def test_trip_detail_includes_shipper_rating_boolean(self): client = self._client_for(self.transporter) before = client.get(f'/api/transporter/trips/{self.trip.id}/') self.assertEqual(before.status_code, 200) self.assertIs(before.json()['data']['shipper_rating'], False) client.post( f'/api/transporter/trips/{self.trip.id}/shipper-review/', {'shipper_rating': 5}, format='json', ) after = client.get(f'/api/transporter/trips/{self.trip.id}/') self.assertEqual(after.status_code, 200) self.assertIs(after.json()['data']['shipper_rating'], True) def test_assigned_driver_can_rate_via_driver_endpoint(self): trip = self._trip(status=Trip.Status.COMPLETED) response = self._client_for(self.assigned_driver).post( f'/api/transporter/driver/trips/{trip.id}/shipper-review/', {'shipper_rating': 4, 'comment': 'Good shipper.'}, format='json', ) self.assertEqual(response.status_code, 201) self.assertEqual(response.json()['data']['reviewer'], self.assigned_driver.id) def test_assigned_driver_can_rate_trip_shipper(self): response = self._client_for(self.assigned_driver).post( f'/api/transporter/trips/{self.trip.id}/shipper-review/', {'shipper_rating': 4}, format='json', ) self.assertEqual(response.status_code, 201) self.assertEqual(response.json()['data']['reviewer'], self.assigned_driver.id) def test_individual_driver_can_rate_via_my_trips_endpoint(self): solo_driver = User.objects.create_user( 'solo-ind-review@test.com', email='solo-ind-review@test.com', password='secret', ) UserRole.objects.create(user=solo_driver, role=UserRole.Role.TRANSPORTER) TransporterProfile.objects.create( user=solo_driver, account_type=TransporterProfile.AccountType.DRIVER, local=True, country_to_country=False, documents_verified=True, ) shipment = 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.COMPLETED, ) bid = Bid.objects.create( shipment=shipment, transporter=solo_driver, amount=Decimal('100.00'), status=Bid.Status.ACCEPTED, ) solo_trip = Trip.objects.create( shipment=shipment, accepted_bid=bid, transporter=solo_driver, assigned_driver=solo_driver, status=Trip.Status.COMPLETED, ) response = self._client_for(solo_driver).post( f'/api/transporter/my-trips/{solo_trip.id}/shipper-review/', {'shipper_rating': 5}, format='json', ) self.assertEqual(response.status_code, 201) self.assertEqual(response.json()['data']['reviewer'], solo_driver.id) def test_fleet_driver_endpoint_rejects_non_assigned_trip(self): solo_trip = self._trip(status=Trip.Status.COMPLETED) solo_trip.assigned_driver = None solo_trip.save(update_fields=['assigned_driver']) response = self._client_for(self.assigned_driver).post( f'/api/transporter/driver/trips/{solo_trip.id}/shipper-review/', {'shipper_rating': 5}, format='json', ) self.assertEqual(response.status_code, 404) def test_rejects_duplicate_shipper_review_for_trip(self): TripShipperReview.objects.create( trip=self.trip, shipper=self.shipper, transporter=self.transporter, reviewer=self.transporter, shipper_rating=5, ) response = self._client_for(self.transporter).post( f'/api/transporter/trips/{self.trip.id}/shipper-review/', {'shipper_rating': 4}, format='json', ) self.assertEqual(response.status_code, 400) self.assertEqual(response.json()['error'], 'Duplicate.') def test_rejects_review_before_terminal_trip_status(self): trip = self._trip(status=Trip.Status.IN_TRANSIT) response = self._client_for(self.transporter).post( f'/api/transporter/trips/{trip.id}/shipper-review/', {'shipper_rating': 5}, format='json', ) self.assertEqual(response.status_code, 400) self.assertEqual(response.json()['error'], 'Invalid trip status.') def test_hides_trip_from_unrelated_transporter(self): response = self._client_for(self.other_transporter).post( f'/api/transporter/trips/{self.trip.id}/shipper-review/', {'shipper_rating': 5}, format='json', ) self.assertEqual(response.status_code, 404)