diff --git a/src/backend/InvenTree/InvenTree/helpers_db.py b/src/backend/InvenTree/InvenTree/helpers_db.py index 1dfaeea581..1d9b1ec328 100644 --- a/src/backend/InvenTree/InvenTree/helpers_db.py +++ b/src/backend/InvenTree/InvenTree/helpers_db.py @@ -11,7 +11,7 @@ from django.db.models import QuerySet def bulk_create_and_fetch( model, items, id_field: str = 'pk', filters: Optional[dict] = None ) -> QuerySet: - """Bulk create items in the database, and return a queryset of the created items. + """Bulk create items in the database, and return a list of the created items. Arguments: model: The Django model class to create instances of. @@ -20,7 +20,7 @@ def bulk_create_and_fetch( filters: Optional dictionary of filters to apply when fetching the created items. Returns: - A Django QuerySet containing the created items. + A QuerySet containing the created items. This helper method is required because the Django bulk_create() method does not guarantee that the ID values of the created items will be populated in the returned objects. diff --git a/src/backend/InvenTree/InvenTree/test_helpers_db.py b/src/backend/InvenTree/InvenTree/test_helpers_db.py index 6a899f0bc2..de4da41b50 100644 --- a/src/backend/InvenTree/InvenTree/test_helpers_db.py +++ b/src/backend/InvenTree/InvenTree/test_helpers_db.py @@ -22,7 +22,7 @@ class BulkCreateAndFetchTest(TestCase): result = bulk_create_and_fetch(Contact, items) - self.assertEqual(result.count(), 10) + self.assertEqual(len(result), 10) pks = [c.pk for c in result] @@ -55,8 +55,8 @@ class BulkCreateAndFetchTest(TestCase): Contact, items_b, filters={'company': company_b} ) - self.assertEqual(result_a.count(), 3) - self.assertEqual(result_b.count(), 4) + self.assertEqual(len(result_a), 3) + self.assertEqual(len(result_b), 4) self.assertTrue(all(c.company_id == company_a.pk for c in result_a)) self.assertTrue(all(c.company_id == company_b.pk for c in result_b)) @@ -71,13 +71,13 @@ class BulkCreateAndFetchTest(TestCase): items = [Contact(company=company, name=f'New {i}') for i in range(2)] result = bulk_create_and_fetch(Contact, items) - self.assertEqual(result.count(), 2) + self.assertEqual(len(result), 2) self.assertNotIn('Existing', [c.name for c in result]) def test_empty_list(self): """Calling with an empty list of items should not raise, and return no items.""" result = bulk_create_and_fetch(Contact, []) - self.assertEqual(result.count(), 0) + self.assertEqual(len(result), 0) def test_does_not_mutate_caller_filters(self): """The caller-provided 'filters' dict should not be mutated as a side effect.""" @@ -100,7 +100,7 @@ class BulkCreateAndFetchTest(TestCase): result = bulk_create_and_fetch(PurchaseOrder, items) - self.assertEqual(result.count(), 10) + self.assertEqual(len(result), 10) pks = [o.pk for o in result] self.assertTrue(all(pk is not None for pk in pks)) @@ -121,7 +121,7 @@ class BulkCreateAndFetchTest(TestCase): result = bulk_create_and_fetch(StockItem, items) - self.assertEqual(result.count(), 10) + self.assertEqual(len(result), 10) pks = [si.pk for si in result] self.assertTrue(all(pk is not None for pk in pks)) @@ -144,7 +144,7 @@ class BulkCreateAndFetchTest(TestCase): with CaptureQueriesContext(connection) as ctx: result = bulk_create_and_fetch(Contact, items) - self.assertEqual(result.count(), n) + self.assertEqual(len(result), n) MAX_QUERIES = 25 if n > 100 else 10 diff --git a/src/backend/InvenTree/order/models.py b/src/backend/InvenTree/order/models.py index 42419c0c16..a9ec1990c4 100644 --- a/src/backend/InvenTree/order/models.py +++ b/src/backend/InvenTree/order/models.py @@ -1,5 +1,6 @@ """Order model definitions.""" +import copy from decimal import Decimal from typing import Any, Optional, TypedDict @@ -8,6 +9,7 @@ from django.core.exceptions import ValidationError from django.core.validators import MaxValueValidator, MinValueValidator from django.db import models, transaction from django.db.models import F, Q, QuerySet, Sum +from django.db.models.base import ModelState from django.db.models.functions import Coalesce from django.db.models.signals import post_delete, post_save from django.dispatch.dispatcher import receiver @@ -64,7 +66,8 @@ from order.status_codes import ( TransferOrderStatusGroups, ) from part import models as PartModels -from plugin.events import trigger_event +from plugin.events import bulk_trigger_event, trigger_event +from stock.events import StockEvents from stock.status_codes import StockHistoryCode, StockStatus logger = structlog.get_logger('inventree') @@ -1289,7 +1292,9 @@ class PurchaseOrder(TotalPriceMixin, Order): item.run_plugin_validation() # Bulk create the stock items and fetch the newly created instances - new_items = bulk_create_and_fetch(stock.models.StockItem, bulk_create_items) + new_items = list( + bulk_create_and_fetch(stock.models.StockItem, bulk_create_items) + ) stock_items.extend(new_items) @@ -2851,6 +2856,199 @@ class SalesOrderShipment( return True + @transaction.atomic + def complete_allocations( + self, allocations: QuerySet, user: Optional[User] = None + ) -> None: + """Complete a set of SalesOrderAllocation objects, marking their stock as shipped to the customer. + + Arguments: + allocations: QuerySet of SalesOrderAllocation objects to complete + user: The user completing the allocations + + Notes: + This unrolls what would otherwise be a per-allocation call, so that the underlying + StockItem, StockItemTracking, SalesOrderLineItem and SalesOrderAllocation writes can + be batched into a handful of bulk queries instead of several per allocation. + """ + import part.tasks + + order = self.order + customer = order.customer + + # Preselect related fields to avoid per-row database queries below + allocations = allocations.select_related('line', 'item', 'item__part') + + split_items = [] # (source_item, new_item, quantity) - stock to split off + shipped_items = [] # (target_item, quantity) - stock to mark as shipped + + # Canonical (mutable) copy of each distinct StockItem being drawn from - multiple + # allocations may draw from the same StockItem, so track running state + seen_stock_items: dict = {} + + # Track the lines which have already been processed, to avoid double counting + seen_lines: dict = {} + + # Allocations whose 'item' now points at a newly-split-off StockItem + allocations_to_update = [] + + for allocation in allocations: + stock_item = seen_stock_items.get(allocation.item_id) or allocation.item + quantity = allocation.quantity + + seen_stock_items[stock_item.pk] = stock_item + + if quantity < stock_item.quantity: + # Split off exactly the shipped quantity into a new StockItem, + # leaving the remainder in place as available stock + new_item = copy.copy(stock_item) + new_item._state = ModelState() + new_item.pk = None + new_item.quantity = quantity + new_item.parent = stock_item + + stock_item.quantity -= quantity + + target_item = new_item + split_items.append((stock_item, new_item, quantity)) + + allocation.item = new_item + allocations_to_update.append(allocation) + else: + target_item = stock_item + + # Resolve the final resting state of the target item now + target_item.sales_order = order + target_item.customer = customer + target_item.location = None + + shipped_items.append((target_item, quantity)) + + # Increase the "shipped" quantity for the associated line + line = seen_lines.get(allocation.line_id) or allocation.line + line.shipped += quantity + seen_lines[line.pk] = line + + # Nothing to do? + if not seen_stock_items: + return + + # Bulk-create the newly split-off stock items - this resolves their primary keys, + # which the tracking entries below need + new_stock_item_data = [new_item for _, new_item, _ in split_items] + + # Evaluate the queryset immediately (into a list) - otherwise, indexing into it + # below would re-query the database once per split item + new_stock_items = list( + bulk_create_and_fetch(stock.models.StockItem, new_stock_item_data) + ) + + # Backfill the newly created StockItem objects into the split_items list, + # then repoint every other reference to a placeholder (pk=None) copy at the + # newly persisted instance instead - shipped_items and allocations_to_update + # were populated with the pre-creation copies, which never gain a primary key + split_item_map = {} + + for i, (source, placeholder, quantity) in enumerate(split_items): + persisted = new_stock_items[i] + split_item_map[id(placeholder)] = persisted + split_items[i] = (source, persisted, quantity) + + shipped_items = [ + (split_item_map.get(id(item), item), quantity) + for item, quantity in shipped_items + ] + + for allocation in allocations_to_update: + allocation.item = split_item_map[id(allocation.item)] + + tracking_entries = [] + split_events = [] + customer_events = [] + + # Split stock items for "split_items" + for source_item, new_item, quantity in split_items: + tracking_entries.append( + stock.models.StockItemTracking( + item_id=new_item.pk, + part_id=new_item.part_id, + tracking_type=StockHistoryCode.SPLIT_FROM_PARENT.value, + user=user, + deltas={'stockitem': source_item.pk, 'quantity': float(quantity)}, + ) + ) + tracking_entries.append( + stock.models.StockItemTracking( + item_id=source_item.pk, + part_id=source_item.part_id, + tracking_type=StockHistoryCode.SPLIT_CHILD_ITEM.value, + user=user, + deltas={ + 'removed': float(quantity), + 'quantity': float(source_item.quantity), + }, + ) + ) + + split_events.append(((), {'id': new_item.pk, 'parent': source_item.pk})) + + # Ship stock items for "shipped_items" + for target_item, quantity in shipped_items: + deltas = {'quantity': float(quantity), 'salesorder': order.pk} + + if customer is not None: + deltas['customer'] = customer.pk + deltas['customer_name'] = customer.name + + tracking_entries.append( + stock.models.StockItemTracking( + item_id=target_item.pk, + part_id=target_item.part_id, + tracking_type=StockHistoryCode.SHIPPED_AGAINST_SALES_ORDER.value, + user=user, + deltas=deltas, + ) + ) + + customer_events.append(( + (), + {'id': target_item.pk, 'customer': customer.pk if customer else None}, + )) + + # Flush all StockItem field changes (quantity reductions, and shipment details) + stock.models.StockItem.objects.bulk_update( + seen_stock_items.values(), ['quantity'] + ) + stock.models.StockItem.objects.bulk_update( + [item for item, _ in shipped_items], ['sales_order', 'customer', 'location'] + ) + + stock.models.StockItemTracking.objects.bulk_create(tracking_entries) + + # Update sales order lines for "seen_lines" + SalesOrderLineItem.objects.bulk_update(seen_lines.values(), ['shipped']) + + # Repoint allocations onto their (possibly newly split) StockItem + if allocations_to_update: + SalesOrderAllocation.objects.bulk_update(allocations_to_update, ['item']) + + # Queue the ITEM_SPLIT / ITEM_ASSIGNED_TO_CUSTOMER plugin events in bulk, + # rather than one offload_task() call (and one OrmQ insert) per item + bulk_trigger_event(StockEvents.ITEM_SPLIT, split_events) + bulk_trigger_event(StockEvents.ITEM_ASSIGNED_TO_CUSTOMER, customer_events) + + # bulk_update()/bulk_create() above do not fire StockItem's post_save signal, + # which normally triggers a low-stock check for the affected part - so queue + # that check explicitly, once per distinct part touched by this call + touched_part_ids = {item.part_id for item in seen_stock_items.values()} + + InvenTree.tasks.bulk_offload_task( + part.tasks.notify_low_stock_if_required, + [((part_id,), {}) for part_id in touched_part_ids], + group='notification', + force_async=True, + ) + @transaction.atomic def complete_shipment(self, user, **kwargs): """Complete this particular shipment. @@ -3053,29 +3251,18 @@ class SalesOrderAllocation(models.Model): """Return the PurchaseOrder associated with this allocation.""" return self.item.purchase_order - def complete_allocation(self, user): + def complete_allocation(self, user=None): """Complete this allocation (called when the parent SalesOrder is marked as "shipped"). - Executes: - - Determine if the referenced StockItem needs to be "split" (if allocated quantity != stock quantity) - - Mark the StockItem as belonging to the Customer (this will remove it from stock) + Retained for backwards compatibility with external callers (e.g. plugins) which + complete allocations one at a time - delegates to the bulk + SalesOrderShipment.complete_allocations() implementation. """ - order = self.line.order - - item = self.item.allocateToCustomer( - order.customer, quantity=self.quantity, order=order, user=user + self.shipment.complete_allocations( + SalesOrderAllocation.objects.filter(pk=self.pk), user=user ) - # Update the 'shipped' quantity - # Increment at the database level to prevent lost updates - self.line.shipped = F('shipped') + self.quantity - self.line.save(update_fields=['shipped']) - self.line.refresh_from_db(fields=['shipped']) - - # Update our own reference to the StockItem - # (It may have changed if the stock was split) - self.item = item - self.save() + self.refresh_from_db() class ReturnOrder(TotalPriceMixin, Order): diff --git a/src/backend/InvenTree/order/tasks.py b/src/backend/InvenTree/order/tasks.py index 34e2fa033b..49636e63d7 100644 --- a/src/backend/InvenTree/order/tasks.py +++ b/src/backend/InvenTree/order/tasks.py @@ -277,8 +277,7 @@ def complete_sales_order_shipment( ) return - for allocation in shipment.allocations.all(): - allocation.complete_allocation(user=user) + shipment.complete_allocations(shipment.allocations.all(), user=user) # Once all allocations have been completed, we can mark the shipment as complete shipment.shipment_date = shipment_date or datetime.now().date() diff --git a/src/backend/InvenTree/order/test_sales_order.py b/src/backend/InvenTree/order/test_sales_order.py index 99a13dba3e..74358eb9aa 100644 --- a/src/backend/InvenTree/order/test_sales_order.py +++ b/src/backend/InvenTree/order/test_sales_order.py @@ -9,6 +9,8 @@ from django.core.exceptions import ValidationError from django.db.models import Sum from django.urls import reverse +from django_q.models import OrmQ + import order.tasks from common.models import InvenTreeSetting, NotificationMessage from common.settings import set_global_setting @@ -27,7 +29,9 @@ from order.models import ( SalesOrderShipment, ) from part.models import Part +from stock.events import StockEvents from stock.models import StockItem, StockItemTracking, StockLocation +from stock.status_codes import StockHistoryCode from users.models import Owner @@ -188,6 +192,24 @@ class SalesOrderTest(InvenTreeAPITestCase): quantity=25 if full else 20, ) + def test_scratch_complete_allocation_single(self): + """Scratch check: SalesOrderAllocation.complete_allocation() backwards-compat shim.""" + self.allocate_stock(True) + + allocations = list(self.shipment.allocations.all()) + self.assertEqual(len(allocations), 2) + + for allocation in allocations: + allocation.complete_allocation(self.user) + + self.line.refresh_from_db() + self.assertEqual(self.line.shipped, 50) + + for allocation in allocations: + allocation.item.refresh_from_db() + self.assertEqual(allocation.item.sales_order, self.order) + self.assertEqual(allocation.item.customer, self.order.customer) + def test_over_allocate(self): """Test that over allocation logic works.""" SA = StockItem.objects.create(part=self.part, quantity=9) @@ -245,8 +267,9 @@ class SalesOrderTest(InvenTreeAPITestCase): self.assertEqual(alloc_a.line.shipped, 0) self.assertEqual(alloc_b.line.shipped, 0) - alloc_a.complete_allocation(None) - alloc_b.complete_allocation(None) + allocations = SalesOrderAllocation.objects.filter(line=self.line).order_by('pk') + + self.shipment.complete_allocations(allocations) # Both shipped quantities must be counted self.line.refresh_from_db() @@ -486,6 +509,13 @@ class SalesOrderTest(InvenTreeAPITestCase): This test is designed to test that the database does not error out, even when a large number of items are assigned to a shipment. + It also checks that: + - Stock items are split (or fully consumed) correctly + - The correct stock tracking entries are created for each item + - The associated ITEM_SPLIT / ITEM_ASSIGNED_TO_CUSTOMER events are queued + - A low-stock check is queued for the (single) affected part + - Line 'shipped' quantities and allocation->item references are updated correctly + Ref: https://github.com/inventree/InvenTree/pull/11500 """ customer = Company.objects.create(name='Customer 2', is_customer=True) @@ -508,6 +538,8 @@ class SalesOrderTest(InvenTreeAPITestCase): line = SalesOrderLineItem.objects.create(part=part, order=so, quantity=N_ITEMS) # Create stock items, and assign to shipment + # Every 5th item has exactly the allocated quantity (no split required); + # the rest have excess quantity (requiring a split) allocations = [] stock_items = [] @@ -516,6 +548,17 @@ class SalesOrderTest(InvenTreeAPITestCase): StockItem.objects.bulk_create(stock_items) + # Re-fetch the created items, as bulk_create() does not guarantee that the + # pk values of the passed-in objects are populated (e.g. on MySQL) + stock_items = list(StockItem.objects.filter(part=part)) + + # Original quantity of each stock item, keyed by pk (used for verification below) + original_quantities = {item.pk: item.quantity for item in stock_items} + split_pks = {pk for pk, qty in original_quantities.items() if qty > 1} + exact_pks = {pk for pk, qty in original_quantities.items() if qty == 1} + self.assertEqual(len(split_pks), N_ITEMS * 4 // 5) + self.assertEqual(len(exact_pks), N_ITEMS // 5) + # Check expected available quantity self.assertEqual(part.total_stock, 2250) @@ -538,14 +581,26 @@ class SalesOrderTest(InvenTreeAPITestCase): # Complete the shipment via the API self.assignRole('sales_order.add') + # Enable plugin events, and queue them (rather than firing synchronously), + # so we can inspect exactly what was queued once the shipment is completed + set_global_setting('ENABLE_PLUGINS_EVENTS', True, change_user=None) + + # Start with a fresh slate for the OrmQ queue + OrmQ.objects.all().delete() + 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, - ) + + with self.settings( + PLUGIN_TESTING_EVENTS=True, PLUGIN_TESTING_EVENTS_ASYNC=True + ): + response = self.post( + url, + expected_code=200, + benchmark=True, + max_query_time=5.0, + max_query_count=100, + ) + self.assertEqual(response.status_code, 200) shipment.refresh_from_db() @@ -556,6 +611,143 @@ class SalesOrderTest(InvenTreeAPITestCase): part.refresh_from_db() self.assertEqual(part.total_stock, 2250 - N_ITEMS) + # The line item should now show the full quantity as 'shipped' + line.refresh_from_db() + self.assertEqual(line.shipped, N_ITEMS) + + # Every allocation should now point to a StockItem which has been assigned to the customer + self.assertEqual( + shipment.allocations.filter( + item__sales_order=so, item__customer=customer + ).count(), + N_ITEMS, + ) + + queued_tasks = list(OrmQ.objects.all()) + + # register_event tasks queued for StockEvents.ITEM_SPLIT, one per split item + split_event_tasks = [ + task + for task in queued_tasks + if task.func() == 'plugin.base.event.events.register_event' + and task.args() == (StockEvents.ITEM_SPLIT,) + ] + self.assertEqual(len(split_event_tasks), len(split_pks)) + + # register_event tasks queued for StockEvents.ITEM_ASSIGNED_TO_CUSTOMER, one per item + customer_event_tasks = [ + task + for task in queued_tasks + if task.func() == 'plugin.base.event.events.register_event' + and task.args() == (StockEvents.ITEM_ASSIGNED_TO_CUSTOMER,) + ] + self.assertEqual(len(customer_event_tasks), N_ITEMS) + + # 'notify_low_stock_if_required' should be queued exactly once - there is only + # a single distinct part involved, despite the large number of stock items + low_stock_tasks = [ + task + for task in queued_tasks + if task.func() == 'part.tasks.notify_low_stock_if_required' + ] + self.assertEqual(len(low_stock_tasks), 1) + self.assertEqual(low_stock_tasks[0].args(), (part.pk,)) + + # Exactly one StockItemTracking entry should exist per split, per side + self.assertEqual( + StockItemTracking.objects.filter( + tracking_type=StockHistoryCode.SPLIT_FROM_PARENT.value + ).count(), + len(split_pks), + ) + self.assertEqual( + StockItemTracking.objects.filter( + tracking_type=StockHistoryCode.SPLIT_CHILD_ITEM.value + ).count(), + len(split_pks), + ) + + # Exactly one 'shipped' tracking entry should exist per allocated item + self.assertEqual( + StockItemTracking.objects.filter( + tracking_type=StockHistoryCode.SHIPPED_AGAINST_SALES_ORDER.value + ).count(), + N_ITEMS, + ) + + # Check the original (exact-quantity) stock items were consumed directly, with no split + for pk in exact_pks: + item = StockItem.objects.get(pk=pk) + self.assertEqual(item.quantity, 1) + self.assertEqual(item.sales_order, so) + self.assertEqual(item.customer, customer) + + self.assertFalse( + StockItemTracking.objects.filter( + item=item, tracking_type=StockHistoryCode.SPLIT_CHILD_ITEM.value + ).exists() + ) + self.assertTrue( + StockItemTracking.objects.filter( + item=item, + tracking_type=StockHistoryCode.SHIPPED_AGAINST_SALES_ORDER.value, + deltas__salesorder=so.pk, + ).exists() + ) + + # Check that the split (excess-quantity) stock items were correctly split + for pk in split_pks: + original_item = StockItem.objects.get(pk=pk) + original_quantity = original_quantities[pk] + + # The original item should retain the remainder, and *not* be shipped + self.assertEqual(original_item.quantity, original_quantity - 1) + self.assertIsNone(original_item.sales_order) + self.assertIsNone(original_item.customer) + + # Spot-check a single split item for the exact linkage between tracking entries + # (aggregate counts above already confirm this holds for every split item) + sample_pk = next(iter(split_pks)) + sample_original = StockItem.objects.get(pk=sample_pk) + sample_new_item = StockItem.objects.get( + part=part, quantity=1, sales_order=so, parent=sample_original + ) + + self.assertEqual(sample_new_item.quantity, 1) + self.assertTrue( + StockItemTracking.objects.filter( + item=sample_new_item, + tracking_type=StockHistoryCode.SPLIT_FROM_PARENT.value, + deltas__stockitem=sample_original.pk, + ).exists() + ) + self.assertTrue( + StockItemTracking.objects.filter( + item=sample_original, + tracking_type=StockHistoryCode.SPLIT_CHILD_ITEM.value, + ).exists() + ) + self.assertTrue( + StockItemTracking.objects.filter( + item=sample_new_item, + tracking_type=StockHistoryCode.SHIPPED_AGAINST_SALES_ORDER.value, + deltas__salesorder=so.pk, + ).exists() + ) + self.assertTrue( + any( + task.kwargs() + == {'id': sample_new_item.pk, 'parent': sample_original.pk} + for task in split_event_tasks + ) + ) + + # The allocation for the split item should now point at the new (split-off) item + allocation = SalesOrderAllocation.objects.get( + line=line, shipment=shipment, item=sample_new_item + ) + self.assertNotEqual(allocation.item_id, sample_pk) + def test_default_shipment(self): """Test sales order default shipment creation.""" # Default setting value should be False