From 6be654954ecfdaf3f079d99a97f3def6e109cbbb Mon Sep 17 00:00:00 2001 From: Oliver Date: Tue, 14 Jul 2026 10:34:54 +1000 Subject: [PATCH] [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 | 91 +++++++++ src/backend/InvenTree/order/serializers.py | 101 +--------- src/backend/InvenTree/order/test_api.py | 178 ++++++++++++++++++ .../InvenTree/order/test_sales_order.py | 26 ++- 4 files changed, 301 insertions(+), 95 deletions(-) diff --git a/src/backend/InvenTree/order/models.py b/src/backend/InvenTree/order/models.py index 4c609b6b33..53caaa16d1 100644 --- a/src/backend/InvenTree/order/models.py +++ b/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). diff --git a/src/backend/InvenTree/order/serializers.py b/src/backend/InvenTree/order/serializers.py index c69045a205..067d78a169 100644 --- a/src/backend/InvenTree/order/serializers.py +++ b/src/backend/InvenTree/order/serializers.py @@ -1871,104 +1871,23 @@ class SalesOrderSerialAllocationSerializer(serializers.Serializer): return shipment - def validate(self, data): - """Validation for the serializer. - - - 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) + def save(self): + """Allocate stock items against the sales order, by serial number.""" + data = self.validated_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 = [] - - for stock_item in stock_items: - # 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 + try: + return sales_order.allocate_serial_numbers( + line_item, quantity, serial_numbers, shipment=shipment ) + 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): diff --git a/src/backend/InvenTree/order/test_api.py b/src/backend/InvenTree/order/test_api.py index de4805bef1..2911978c70 100644 --- a/src/backend/InvenTree/order/test_api.py +++ b/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.""" diff --git a/src/backend/InvenTree/order/test_sales_order.py b/src/backend/InvenTree/order/test_sales_order.py index c293eb57be..2d07a406c4 100644 --- a/src/backend/InvenTree/order/test_sales_order.py +++ b/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 - shipment.complete_shipment(None) + # Complete the shipment via the API + 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)