[refactor] SalesOrder allocation fixes (#12379)

* Use API for unit test

* Refactor SalesOrderSerialAllocationSerializer

- Move business logic into the SalesOrder model

* Add unit testing for serial allocation

src/backend/InvenTree/order/models.py

@@ -1633,6 +1633,97 @@ class SalesOrder(TotalPriceMixin, Order):
 
         SalesOrderAllocation.objects.bulk_create(new_allocations, batch_size=250)
 
+    @transaction.atomic
+    def allocate_serial_numbers(
+        self,
+        line_item: 'SalesOrderLineItem',
+        quantity: int,
+        serial_numbers: str,
+        shipment: Optional['SalesOrderShipment'] = None,
+    ) -> list['SalesOrderAllocation']:
+        """Allocate stock items against this SalesOrder, by serial number.
+
+        Arguments:
+            line_item: The SalesOrderLineItem to allocate against
+            quantity: The number of serial numbers expected
+            serial_numbers: A string of serial numbers to allocate (e.g. "1,2,3-5")
+            shipment: Optional shipment to assign the allocations to
+
+        Raises:
+            ValidationError: If the line item does not belong to this order,
+                the serial numbers cannot be parsed, or any of the requested
+                serial numbers do not exist or are unavailable for allocation.
+        """
+        if line_item.order != self:
+            raise ValidationError(_('Line item is not associated with this order'))
+
+        part = line_item.part
+
+        serials = InvenTree.helpers.extract_serial_numbers(
+            serial_numbers, quantity, part.get_latest_serial_number(), part=part
+        )
+
+        serials_not_exist = set()
+        serials_unavailable = set()
+        stock_items_to_allocate = []
+
+        for serial in serials:
+            serial = str(serial).strip()
+
+            items = stock.models.StockItem.objects.filter(
+                part=part, serial=serial, quantity=1
+            )
+
+            if not items.exists():
+                serials_not_exist.add(serial)
+                continue
+
+            stock_item = items[0]
+
+            if get_global_setting('SALESORDER_BLOCK_INCOMPLETE_ITEM_TESTS'):
+                if (
+                    stock_item.hasRequiredTests()
+                    and not stock_item.passedAllRequiredTests()
+                ):
+                    serials_unavailable.add(serial)
+                    continue
+
+            if not stock_item.in_stock:
+                serials_unavailable.add(serial)
+                continue
+
+            if stock_item.unallocated_quantity() < 1:
+                serials_unavailable.add(serial)
+                continue
+
+            # At this point, the serial number is valid, and can be added to the list
+            stock_items_to_allocate.append(stock_item)
+
+        if len(serials_not_exist) > 0:
+            error_msg = _('No match found for the following serial numbers')
+            error_msg += ': '
+            error_msg += ','.join(sorted(serials_not_exist))
+
+            raise ValidationError({'serial_numbers': error_msg})
+
+        if len(serials_unavailable) > 0:
+            error_msg = _('The following serial numbers are unavailable')
+            error_msg += ': '
+            error_msg += ','.join(sorted(serials_unavailable))
+
+            raise ValidationError({'serial_numbers': error_msg})
+
+        allocations = [
+            SalesOrderAllocation(
+                line=line_item, item=stock_item, quantity=1, shipment=shipment
+            )
+            for stock_item in stock_items_to_allocate
+        ]
+
+        SalesOrderAllocation.objects.bulk_create(allocations, batch_size=250)
+
+        return allocations
+
     def is_completed(self) -> bool:
         """Check if this order is "shipped" (all line items delivered).
 

src/backend/InvenTree/order/serializers.py

@@ -1871,104 +1871,23 @@ class SalesOrderSerialAllocationSerializer(serializers.Serializer):
 
         return shipment
 
-    def validate(self, data):
+    def save(self):
-        """Validation for the serializer.
+        """Allocate stock items against the sales order, by serial number."""
-
+        data = self.validated_data
-        - Ensure the serial_numbers and quantity fields match
-        - Check that all serial numbers exist
-        - Check that the serial numbers are not yet allocated
-        """
-        data = super().validate(data)
 
+        sales_order = self.context['order']
         line_item = data['line_item']
         quantity = data['quantity']
         serial_numbers = data['serial_numbers']
-
-        part = line_item.part
-
-        try:
-            data['serials'] = extract_serial_numbers(
-                serial_numbers, quantity, part.get_latest_serial_number(), part=part
-            )
-        except DjangoValidationError as e:
-            raise ValidationError({'serial_numbers': e.messages})
-
-        serials_not_exist = set()
-        serials_unavailable = set()
-        stock_items_to_allocate = []
-
-        for serial in data['serials']:
-            serial = str(serial).strip()
-
-            items = stock.models.StockItem.objects.filter(
-                part=part, serial=serial, quantity=1
-            )
-
-            if not items.exists():
-                serials_not_exist.add(str(serial))
-                continue
-
-            stock_item = items[0]
-
-            if get_global_setting('SALESORDER_BLOCK_INCOMPLETE_ITEM_TESTS'):
-                if (
-                    stock_item.hasRequiredTests()
-                    and not stock_item.passedAllRequiredTests()
-                ):
-                    serials_unavailable.add(str(serial))
-                    continue
-
-            if not stock_item.in_stock:
-                serials_unavailable.add(str(serial))
-                continue
-
-            if stock_item.unallocated_quantity() < 1:
-                serials_unavailable.add(str(serial))
-                continue
-
-            # At this point, the serial number is valid, and can be added to the list
-            stock_items_to_allocate.append(stock_item)
-
-        if len(serials_not_exist) > 0:
-            error_msg = _('No match found for the following serial numbers')
-            error_msg += ': '
-            error_msg += ','.join(sorted(serials_not_exist))
-
-            raise ValidationError({'serial_numbers': error_msg})
-
-        if len(serials_unavailable) > 0:
-            error_msg = _('The following serial numbers are unavailable')
-            error_msg += ': '
-            error_msg += ','.join(sorted(serials_unavailable))
-
-            raise ValidationError({'serial_numbers': error_msg})
-
-        data['stock_items'] = stock_items_to_allocate
-
-        return data
-
-    def save(self):
-        """Allocate stock items against the sales order."""
-        data = self.validated_data
-
-        line_item = data['line_item']
-        stock_items = data['stock_items']
         shipment = data.get('shipment', None)
 
-        allocations = []
+        try:
-
+            return sales_order.allocate_serial_numbers(
-        for stock_item in stock_items:
+                line_item, quantity, serial_numbers, shipment=shipment
-            # Create a new SalesOrderAllocation
-            allocations.append(
-                order.models.SalesOrderAllocation(
-                    line=line_item, item=stock_item, quantity=1, shipment=shipment
-                )
-            )
-
-        with transaction.atomic():
-            order.models.SalesOrderAllocation.objects.bulk_create(
-                allocations, batch_size=250
             )
+        except (ValidationError, DjangoValidationError) as exc:
+            # Catch model errors and re-throw as DRF errors
+            raise ValidationError(detail=serializers.as_serializer_error(exc))
 
 
 class SalesOrderShipmentAllocationSerializer(serializers.Serializer):

src/backend/InvenTree/order/test_api.py

@@ -2698,6 +2698,184 @@ class SalesOrderAllocateTest(OrderTest):
         response = self.post(self.url, data, expected_code=201)
 
 
+class SalesOrderAllocateSerialsTest(OrderTest):
+    """Unit tests for allocating stock items against a SalesOrder, by serial number."""
+
+    @classmethod
+    def setUpTestData(cls):
+        """Init routine for this unit test class."""
+        super().setUpTestData()
+
+    def setUp(self):
+        """Init routines for this unit testing class."""
+        super().setUp()
+
+        self.assignRole('sales_order.add')
+
+        self.url = reverse('api-so-allocate-serials', kwargs={'pk': 1})
+
+        self.order = models.SalesOrder.objects.get(pk=1)
+
+        self.part = Part.objects.create(
+            name='Serial Allocation Part',
+            salable=True,
+            trackable=True,
+            description='A trackable part for serial allocation tests',
+        )
+
+        self.line = models.SalesOrderLineItem.objects.create(
+            order=self.order, part=self.part, quantity=5
+        )
+
+        # Create some serialized stock items for this part
+        self.stock_items = [
+            StockItem.objects.create(part=self.part, quantity=1, serial=str(n))
+            for n in range(1, 6)
+        ]
+
+        self.shipment = models.SalesOrderShipment.objects.create(order=self.order)
+
+    def test_allocate(self):
+        """Test that we can allocate stock items to a SalesOrder line, by serial number."""
+        self.assertEqual(self.order.stock_allocations.count(), 0)
+
+        data = {'line_item': self.line.pk, 'quantity': 3, 'serial_numbers': '1,2,3'}
+
+        self.post(self.url, data, expected_code=201)
+
+        self.assertEqual(self.order.stock_allocations.count(), 3)
+
+        allocated_serials = {
+            allocation.item.serial for allocation in self.order.stock_allocations.all()
+        }
+        self.assertEqual(allocated_serials, {'1', '2', '3'})
+
+        for allocation in self.order.stock_allocations.all():
+            self.assertEqual(allocation.quantity, 1)
+            self.assertIsNone(allocation.shipment)
+
+    def test_allocate_with_shipment(self):
+        """Test that allocations are correctly assigned to a provided shipment."""
+        data = {
+            'line_item': self.line.pk,
+            'quantity': 2,
+            'serial_numbers': '4,5',
+            'shipment': self.shipment.pk,
+        }
+
+        self.post(self.url, data, expected_code=201)
+
+        for allocation in self.order.stock_allocations.all():
+            self.assertEqual(allocation.shipment, self.shipment)
+
+    def test_invalid_line_item(self):
+        """Test that a line item belonging to a different order is rejected."""
+        other_order = models.SalesOrder.objects.exclude(pk=self.order.pk).first()
+        other_line = models.SalesOrderLineItem.objects.create(
+            order=other_order, part=self.part, quantity=5
+        )
+
+        data = {'line_item': other_line.pk, 'quantity': 1, 'serial_numbers': '1'}
+
+        response = self.post(self.url, data, expected_code=400)
+
+        self.assertIn('Line item is not associated with this order', str(response.data))
+
+        self.assertEqual(self.order.stock_allocations.count(), 0)
+
+    def test_shipment_already_shipped(self):
+        """Test that a shipment which has already been shipped is rejected."""
+        self.shipment.shipment_date = date.today()
+        self.shipment.save()
+
+        data = {
+            'line_item': self.line.pk,
+            'quantity': 1,
+            'serial_numbers': '1',
+            'shipment': self.shipment.pk,
+        }
+
+        response = self.post(self.url, data, expected_code=400)
+
+        self.assertIn('Shipment has already been shipped', str(response.data))
+
+        self.assertEqual(self.order.stock_allocations.count(), 0)
+
+    def test_shipment_wrong_order(self):
+        """Test that a shipment belonging to a different order is rejected."""
+        other_order = models.SalesOrder.objects.exclude(pk=self.order.pk).first()
+        other_shipment = models.SalesOrderShipment.objects.create(order=other_order)
+
+        data = {
+            'line_item': self.line.pk,
+            'quantity': 1,
+            'serial_numbers': '1',
+            'shipment': other_shipment.pk,
+        }
+
+        response = self.post(self.url, data, expected_code=400)
+
+        self.assertIn('Shipment is not associated with this order', str(response.data))
+
+        self.assertEqual(self.order.stock_allocations.count(), 0)
+
+    def test_serial_not_exist(self):
+        """Test that non-existent serial numbers are rejected."""
+        data = {'line_item': self.line.pk, 'quantity': 1, 'serial_numbers': '999'}
+
+        response = self.post(self.url, data, expected_code=400)
+
+        self.assertIn(
+            'No match found for the following serial numbers', str(response.data)
+        )
+        self.assertIn('999', str(response.data))
+
+        self.assertEqual(self.order.stock_allocations.count(), 0)
+
+    def test_serial_unavailable(self):
+        """Test that an already-allocated serial number is rejected as unavailable."""
+        # Fully allocate stock item with serial '1' against some other line
+        models.SalesOrderAllocation.objects.create(
+            line=self.line, item=self.stock_items[0], quantity=1
+        )
+
+        data = {'line_item': self.line.pk, 'quantity': 1, 'serial_numbers': '1'}
+
+        response = self.post(self.url, data, expected_code=400)
+
+        self.assertIn(
+            'The following serial numbers are unavailable', str(response.data)
+        )
+        self.assertIn('1', str(response.data))
+
+        # No *additional* allocation should have been created
+        self.assertEqual(self.order.stock_allocations.count(), 1)
+
+    def test_block_on_required_tests(self):
+        """Test the SALESORDER_BLOCK_INCOMPLETE_ITEM_TESTS setting."""
+        from part.models import PartTestTemplate
+
+        self.part.testable = True
+        self.part.save()
+
+        PartTestTemplate.objects.create(
+            part=self.part, test_name='A required test', required=True
+        )
+
+        data = {'line_item': self.line.pk, 'quantity': 1, 'serial_numbers': '1'}
+
+        set_global_setting('SALESORDER_BLOCK_INCOMPLETE_ITEM_TESTS', True)
+
+        response = self.post(self.url, data, expected_code=400)
+        self.assertIn(
+            'The following serial numbers are unavailable', str(response.data)
+        )
+
+        set_global_setting('SALESORDER_BLOCK_INCOMPLETE_ITEM_TESTS', False)
+
+        self.post(self.url, data, expected_code=201)
+
+
 class ReturnOrderTests(InvenTreeAPITestCase):
     """Unit tests for ReturnOrder API endpoints."""
 

src/backend/InvenTree/order/test_sales_order.py

@@ -6,13 +6,18 @@ from django.contrib.auth import get_user_model
 from django.contrib.auth.models import Group
 from django.core.exceptions import ValidationError
 from django.db.models import Sum
+from django.urls import reverse
 
 import order.tasks
 from common.models import InvenTreeSetting, NotificationMessage
 from common.settings import set_global_setting
 from company.models import Address, Company
 from InvenTree import status_codes as status
-from InvenTree.unit_test import InvenTreeTestCase, addUserPermission
+from InvenTree.unit_test import (
+    InvenTreeAPITestCase,
+    InvenTreeTestCase,
+    addUserPermission,
+)
 from order.models import (
     SalesOrder,
     SalesOrderAllocation,
@@ -25,14 +30,17 @@ from stock.models import StockItem, StockItemTracking, StockLocation
 from users.models import Owner
 
 
-class SalesOrderTest(InvenTreeTestCase):
+class SalesOrderTest(InvenTreeAPITestCase):
     """Run tests to ensure that the SalesOrder model is working correctly."""
 
     fixtures = ['company', 'users']
+    roles = ['sales_order.add']
 
     @classmethod
     def setUpTestData(cls):
         """Initial setup for this set of unit tests."""
+        super().setUpTestData()
+
         # Create a Company to ship the goods to
         cls.customer = Company.objects.create(
             name='ABC Co', description='My customer', is_customer=True
@@ -455,8 +463,18 @@ class SalesOrderTest(InvenTreeTestCase):
         self.assertFalse(shipment.is_complete())
         self.assertTrue(shipment.check_can_complete(raise_error=False))
 
-        # Complete the shipment
+        # Complete the shipment via the API
-        shipment.complete_shipment(None)
+        self.assignRole('sales_order.add')
+
+        url = reverse('api-so-shipment-ship', kwargs={'pk': shipment.pk})
+        response = self.post(
+            url,
+            expected_code=200,
+            benchmark=True,
+            max_query_time=100,
+            max_query_count=10000,
+        )
+        self.assertEqual(response.status_code, 200)
 
         shipment.refresh_from_db()
         self.assertIsNotNone(shipment.shipment_date)