From 6a3a9020451320cf86a4d91f81f59328860d3108 Mon Sep 17 00:00:00 2001 From: ixgadmin Date: Thu, 17 Sep 2026 14:48:04 +0000 Subject: [PATCH] Enforce the intake allocation limit in the lineage API The ops lineage screen refuses to allocate more of an intake than was received, counting every batch, but POST /api/intake-batch-lineage/ accepted 9999 kg against a 100 kg intake. Move the rule into a shared helper so the screen and the serializer refuse the same allocations. --- apps/compliance/serializers.py | 16 +++++++++ apps/compliance/services.py | 38 ++++++++++++++++++++++ apps/compliance/tests.py | 59 ++++++++++++++++++++++++++++++++++ 3 files changed, 113 insertions(+) diff --git a/apps/compliance/serializers.py b/apps/compliance/serializers.py index 4ae3f1d..f2728af 100644 --- a/apps/compliance/serializers.py +++ b/apps/compliance/serializers.py @@ -1,5 +1,6 @@ from rest_framework import serializers +from . import services from .models import ( ProductSpecification, SpecParameter, Facility, OriginZone, CollectorGroup, HarvestIntake, IntakeBatchLineage, @@ -68,6 +69,21 @@ class IntakeBatchLineageSerializer(serializers.ModelSerializer): fields = "__all__" read_only_fields = ["id", "created_at"] + def validate(self, attrs): + intake = attrs.get("harvest_intake") or getattr( + self.instance, "harvest_intake", None + ) + weight = attrs.get("weight_allocated_kg") + if weight is None and self.instance is not None: + weight = self.instance.weight_allocated_kg + if intake is not None and weight is not None: + message = services.intake_allocation_error( + intake, weight, exclude_lineage=self.instance + ) + if message: + raise serializers.ValidationError({"weight_allocated_kg": message}) + return attrs + class EUDREvidenceSerializer(serializers.ModelSerializer): verified_by_name = serializers.CharField(source="verified_by.full_name", read_only=True) diff --git a/apps/compliance/services.py b/apps/compliance/services.py index ee219c5..0fe6094 100644 --- a/apps/compliance/services.py +++ b/apps/compliance/services.py @@ -1,6 +1,7 @@ from decimal import Decimal from django.db import transaction +from django.db.models import Sum from django.utils import timezone from apps.batches.models import ProcessingBatch @@ -177,6 +178,43 @@ def ensure_due_diligence_case(batch): return case +def intake_allocation_error(intake, weight, *, exclude_batch=None, exclude_lineage=None): + """Return a message when ``weight`` cannot be allocated from ``intake``. + + An intake is a finite quantity of shea, so the rule counts **every** batch + that has already drawn from it, not just the one being saved. ``exclude_batch`` + and ``exclude_lineage`` keep the caller's own allocation out of that total. + Shared by the ops lineage formset and the lineage API serializer so the screen + and the API refuse the same allocations. + """ + from apps.compliance.models import IntakeBatchLineage + + received = intake.weight_kg + if received is None: + # Records that arrived through the API can be missing their weight. + # Nothing to check against, so nothing is claimed. + return None + if weight > received: + return ( + f"{intake.intake_reference} received {received} kg; " + f"{weight} kg cannot be allocated from it." + ) + elsewhere_qs = IntakeBatchLineage.objects.filter(harvest_intake=intake) + if exclude_batch is not None: + elsewhere_qs = elsewhere_qs.exclude(batch=exclude_batch) + if exclude_lineage is not None: + elsewhere_qs = elsewhere_qs.exclude(pk=exclude_lineage.pk) + elsewhere = elsewhere_qs.aggregate(total=Sum("weight_allocated_kg"))["total"] or Decimal("0") + if elsewhere + weight > received: + others = ", ".join(sorted({row.batch.batch_reference for row in elsewhere_qs})) + return ( + f"{intake.intake_reference} received {received} kg and " + f"{elsewhere} kg of it is already allocated to {others}. " + f"Only {received - elsewhere} kg is left." + ) + return None + + def _lineage(batch): return list(batch.intake_lineage.select_related("harvest_intake", "harvest_intake__origin_zone", "harvest_intake__collector_group").all()) diff --git a/apps/compliance/tests.py b/apps/compliance/tests.py index 668d748..58c375b 100644 --- a/apps/compliance/tests.py +++ b/apps/compliance/tests.py @@ -240,6 +240,65 @@ class EUDRWorkflowTests(TestCase): self.assertEqual(case.statement_references.count(), 1) +class LineageAllocationAPITests(TestCase): + """The API refuses the same over-allocation the ops lineage screen refuses.""" + + def setUp(self): + self.ops = make_user("lineage.ops@ixg.local", "ops") + self.batch = make_batch(self.ops) + facility = Facility.objects.create( + name="Lineage Facility", country="Nigeria", region="Katsina" + ) + zone = OriginZone.objects.create( + facility=facility, zone_code="LIN-01", country="Nigeria", + region="Katsina", verification_status=OriginZone.VerificationStatus.VERIFIED, + ) + group = CollectorGroup.objects.create(name="Lineage Group", origin_zone=zone) + self.intake = HarvestIntake.objects.create( + intake_reference="INT-LIN-001", collector_group=group, + origin_zone=zone, intake_date="2026-01-10", + weight_kg=Decimal("100.00"), + ) + self.client = APIClient() + self.client.force_authenticate(user=self.ops) + + def post(self, weight): + return self.client.post( + "/api/intake-batch-lineage/", + { + "batch": str(self.batch.id), + "harvest_intake": str(self.intake.id), + "weight_allocated_kg": weight, + }, + format="json", + ) + + def test_an_allocation_larger_than_the_intake_is_refused(self): + response = self.post("9999.00") + self.assertEqual(response.status_code, 400) + self.assertFalse(IntakeBatchLineage.objects.exists()) + + def test_an_allocation_that_overdraws_with_another_batch_is_refused(self): + other = make_batch(self.ops) + IntakeBatchLineage.objects.create( + batch=other, harvest_intake=self.intake, + weight_allocated_kg=Decimal("80.00"), + ) + response = self.post("30.00") + self.assertEqual(response.status_code, 400) + self.assertEqual(IntakeBatchLineage.objects.count(), 1) + + def test_what_is_left_of_the_intake_can_be_allocated(self): + other = make_batch(self.ops) + IntakeBatchLineage.objects.create( + batch=other, harvest_intake=self.intake, + weight_allocated_kg=Decimal("80.00"), + ) + response = self.post("20.00") + self.assertEqual(response.status_code, 201) + self.assertEqual(IntakeBatchLineage.objects.count(), 2) + + class BuyerIsolationTests(TestCase): def setUp(self): self.buyer = make_user("iso.buyer@ixg.local", "buyer")