diff --git a/apps/batches/serializers.py b/apps/batches/serializers.py index 06d20e3..6468264 100644 --- a/apps/batches/serializers.py +++ b/apps/batches/serializers.py @@ -49,8 +49,37 @@ class ProcessingBatchDetailSerializer(serializers.ModelSerializer): return DocumentSerializer(docs, many=True).data +# Mirrors apps/ops/forms.py: a QC status is set by recording a QC result, not by +# hand, and a failed batch is sticky — only a new result moves it. The portal +# enforces this in BatchForm; without the same rule here the write serializer is +# a way around the screen. +QC_SET_STATUSES = { + ProcessingBatch.Status.QC_PASS, + ProcessingBatch.Status.QC_FAIL, +} + + class ProcessingBatchWriteSerializer(serializers.ModelSerializer): class Meta: model = ProcessingBatch fields = ["id", "batch_reference", "harvest_date", "weight_kg", "grade", "origin_facility", "notes", "status"] read_only_fields = ["id"] + + def validate_status(self, status): + instance = self.instance + unchanged = instance is not None and status == instance.status + if status in QC_SET_STATUSES and not unchanged: + raise serializers.ValidationError( + f"Batch status {ProcessingBatch.Status(status).label} is set by the " + "QC result, not by hand." + ) + if ( + instance is not None + and instance.status == ProcessingBatch.Status.QC_FAIL + and status != ProcessingBatch.Status.QC_FAIL + ): + raise serializers.ValidationError( + "A batch that has failed QC cannot be moved by hand; a new QC " + "result is what moves it." + ) + return status diff --git a/apps/batches/tests.py b/apps/batches/tests.py new file mode 100644 index 0000000..f34f02b --- /dev/null +++ b/apps/batches/tests.py @@ -0,0 +1,92 @@ +"""The processing-batch API. + +The portal screens have their own tests; these pin the API boundary that used to +be missing. ``ProcessingBatchViewSet`` was open to any authenticated user and the +write serializer accepted ``status``, so a buyer could PATCH a failed batch to +``delivered``. The write rules mirror ``apps/ops/forms.py``: a QC status is set by +a QC result, and a failed batch is sticky. +""" +from django.contrib.auth import get_user_model +from django.test import TestCase +from rest_framework.test import APIClient + +from .models import ProcessingBatch + +User = get_user_model() + + +class ProcessingBatchAPITests(TestCase): + def setUp(self): + self.ops = User.objects.create_user( + email="batch-api-ops@ixg.local", password="pass1234", role="ops", + full_name="Batch API Ops", is_staff=True, + ) + self.buyer = User.objects.create_user( + email="batch-api-buyer@ixg.local", password="pass1234", role="buyer", + full_name="Batch API Buyer", is_staff=True, + ) + self.failed = ProcessingBatch.objects.create( + batch_reference="IXG-BATCH-API-001", + status=ProcessingBatch.Status.QC_FAIL, + ) + self.client = APIClient() + + def test_a_buyer_cannot_patch_a_batch(self): + self.client.force_authenticate(user=self.buyer) + response = self.client.patch( + f"/api/batches/{self.failed.id}/", {"status": "delivered"}, format="json" + ) + self.assertEqual(response.status_code, 403) + self.failed.refresh_from_db() + self.assertEqual(self.failed.status, ProcessingBatch.Status.QC_FAIL) + + def test_a_buyer_cannot_read_the_batch_api(self): + """The buyer portal reads batches through the ORM, not this API, so the + API needs no buyer read and is closed to one.""" + self.client.force_authenticate(user=self.buyer) + self.assertEqual(self.client.get("/api/batches/").status_code, 403) + + def test_ops_can_read_the_batch_api(self): + self.client.force_authenticate(user=self.ops) + self.assertEqual(self.client.get("/api/batches/").status_code, 200) + + def test_a_qc_status_cannot_be_hand_set(self): + self.client.force_authenticate(user=self.ops) + journey = ProcessingBatch.objects.create( + batch_reference="IXG-BATCH-API-002", + status=ProcessingBatch.Status.HARVESTED, + ) + response = self.client.patch( + f"/api/batches/{journey.id}/", {"status": "qc_pass"}, format="json" + ) + self.assertEqual(response.status_code, 400) + journey.refresh_from_db() + self.assertEqual(journey.status, ProcessingBatch.Status.HARVESTED) + + def test_a_batch_cannot_be_created_with_a_qc_status(self): + self.client.force_authenticate(user=self.ops) + response = self.client.post( + "/api/batches/", + {"batch_reference": "IXG-BATCH-API-003", "status": "qc_pass"}, + format="json", + ) + self.assertEqual(response.status_code, 400) + self.assertFalse( + ProcessingBatch.objects.filter(batch_reference="IXG-BATCH-API-003").exists() + ) + + def test_a_failed_batch_cannot_be_moved_out_of_qc_fail(self): + self.client.force_authenticate(user=self.ops) + response = self.client.patch( + f"/api/batches/{self.failed.id}/", {"status": "delivered"}, format="json" + ) + self.assertEqual(response.status_code, 400) + self.failed.refresh_from_db() + self.assertEqual(self.failed.status, ProcessingBatch.Status.QC_FAIL) + + def test_a_failed_batch_can_be_saved_with_its_own_status(self): + self.client.force_authenticate(user=self.ops) + response = self.client.patch( + f"/api/batches/{self.failed.id}/", {"status": "qc_fail"}, format="json" + ) + self.assertEqual(response.status_code, 200) diff --git a/apps/batches/views.py b/apps/batches/views.py index 9d37411..0e29a67 100644 --- a/apps/batches/views.py +++ b/apps/batches/views.py @@ -1,12 +1,18 @@ -from rest_framework import viewsets, permissions, filters +from rest_framework import viewsets, filters from django_filters.rest_framework import DjangoFilterBackend + +from apps.compliance.permissions import IsOpsOrAdmin + from .models import ProcessingBatch from .serializers import ProcessingBatchListSerializer, ProcessingBatchDetailSerializer, ProcessingBatchWriteSerializer class ProcessingBatchViewSet(viewsets.ModelViewSet): queryset = ProcessingBatch.objects.prefetch_related("qc_results", "shipments", "documents").all() - permission_classes = [permissions.IsAuthenticated] + # The buyer portal reads batches through the ORM (apps/buyer/views.py), not + # this API, so nothing outside ops/admin needs it — including for reads. That + # closes the hole where a buyer could PATCH a batch's status. + permission_classes = [IsOpsOrAdmin] filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_fields = ["status", "grade"] search_fields = ["batch_reference"] @@ -22,10 +28,3 @@ class ProcessingBatchViewSet(viewsets.ModelViewSet): def perform_create(self, serializer): serializer.save(created_by=self.request.user) - - def get_queryset(self): - user = self.request.user - qs = super().get_queryset() - if user.role == "buyer": - return qs # Buyers can read all batches per spec - return qs