[bug] Fix for merging stock items (#13091)

- Use *target* stock status if supplied
- Closes https://github.com/inventree/InvenTree/issues/13086
This commit is contained in:
Oliver authored and GitHub committed 2026-10-24 19:00:18 +11:00
1 parent ff697674d7
commit cdb3b1ecf7
3 files changed
+99 -4

No files matched your search

+19 -3
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import contextvars
import copy
import os
from contextlib import contextmanager
from datetime import timedelta
@@ -2671,8 +2672,14 @@ class StockItem(
return True
def find_merge_target(self, location):
"""Find an existing stock item at location that can absorb this item."""
def find_merge_target(self, location, status=None):
"""Find an existing stock item at location that can absorb this item.
Arguments:
location: The destination StockLocation
status: Optional status which will be applied to this item as part of the move.
If provided, the merge target must match this status (rather than the current status)
"""
if location is None:
return None
@@ -2681,6 +2688,15 @@ class StockItem(
if not self.can_merge():
return None
# Compare candidates against the status this item will have *after* the move
incoming = self
if status and not self.compare_status(status):
incoming = copy.copy(self)
if not incoming.set_status(status):
return None
candidates = list(
StockItem.objects
.filter(part=self.part, location=location)
@@ -2700,7 +2716,7 @@ class StockItem(
search_order = candidates
for target in search_order:
if target.can_merge(other=self, raise_error=False):
if target.can_merge(other=incoming, raise_error=False):
return target
return None
+8 -1
View File
@@ -2219,7 +2219,9 @@ class StockTransferSerializer(StockAdjustmentSerializer):
kwargs[field_name] = field_value
if merge:
target = stock_item.find_merge_target(location)
target = stock_item.find_merge_target(
location, status=kwargs.get('status')
)
if target:
merge_kwargs = {
@@ -2258,6 +2260,11 @@ class StockTransferSerializer(StockAdjustmentSerializer):
if field_name in kwargs:
transfer_deltas[field_name] = kwargs[field_name]
# Apply the new status (in memory only, as the item is about to be merged)
# so that the merge checks compare against the requested status
if status := kwargs.get('status'):
stock_item.set_status(status)
merge_kwargs['transfer_deltas'] = transfer_deltas
target.merge_stock_items([stock_item], **merge_kwargs)
+72
View File
@@ -3610,6 +3610,78 @@ class StockTransferMergeTest(StockAPITestCase):
existing.refresh_from_db()
self.assertEqual(existing.quantity, 100)
def test_transfer_merge_with_status_change(self):
"""Merge-on-transfer must match against the *requested* status.
Regression test for https://github.com/inventree/InvenTree/issues/13086
The merge target used to be selected using the current status of the item,
so a full transfer with a new status was merged into stock with the old status.
"""
damaged = StockStatus.DAMAGED.value
# Full quantity transfer, with no matching target at the destination
existing = StockItem.objects.create(
part=self.part, location=self.dest, quantity=10
)
incoming = StockItem.objects.create(
part=self.part, location=self.source_loc, quantity=5
)
self.post(
self.url,
{
'items': [
{'pk': incoming.pk, 'quantity': 5, 'status': damaged, 'merge': True}
],
'location': self.dest.pk,
},
expected_code=201,
)
existing.refresh_from_db()
self.assertEqual(existing.quantity, 10)
self.assertEqual(existing.status, StockStatus.OK.value)
incoming.refresh_from_db()
self.assertEqual(incoming.location, self.dest)
self.assertEqual(incoming.quantity, 5)
self.assertEqual(incoming.status, damaged)
# Subsequent damaged stock should merge into the damaged item (not the OK item)
# Check both a full quantity (3 of 3) and a partial (2 of 5) transfer
for quantity, available in [(3, 3), (2, 5)]:
source = StockItem.objects.create(
part=self.part, location=self.source_loc, quantity=available
)
self.post(
self.url,
{
'items': [
{
'pk': source.pk,
'quantity': quantity,
'status': damaged,
'merge': True,
}
],
'location': self.dest.pk,
},
expected_code=201,
)
existing.refresh_from_db()
self.assertEqual(existing.quantity, 10)
self.assertEqual(existing.status, StockStatus.OK.value)
incoming.refresh_from_db()
self.assertEqual(incoming.quantity, 10)
self.assertEqual(incoming.status, damaged)
self.assertEqual(
StockItem.objects.filter(part=self.part, location=self.dest).count(), 2
)
class StockItemDeletionTest(StockAPITestCase):
"""Tests for stock item deletion via the API."""