mirror of
https://github.com/inventree/InvenTree.git
synced 2026-07-18 04:33:48 +00:00
[feature] Break apart assemblies (#12310)
* Implement API endpoint for stock disassembly * Add basic frontend implementation * Adjust required user permission * Add preFormContent * Read-only if serialized * Adjust location and status of each subcomponent * Handle null value * display installed items in frontend form * More unit tests * Traceability * Add docs * Exclude virtual / consumable stock * Additional docs * Added bundled items docs * More docs * more docs tweaks * Updated part docs * Add playwright tests * Robustify test * suppress certain warnings in schema * Adjust playwright tests * bug fix * Tweak playwright tests
This commit is contained in:
Oliver
@@ -1,11 +1,15 @@
|
||||
"""InvenTree API version information."""
|
||||
|
||||
# InvenTree API version
|
||||
INVENTREE_API_VERSION = 519
|
||||
INVENTREE_API_VERSION = 520
|
||||
"""Increment this API version number whenever there is a significant change to the API that any clients need to know about."""
|
||||
|
||||
INVENTREE_API_TEXT = """
|
||||
|
||||
v520 -> 2026-07-11 : https://github.com/inventree/InvenTree/pull/xxxx
|
||||
- Adds new "disassemble" API endpoint for stock items
|
||||
- Allows a stock item to be broken down into component parts, based on its Bill of Materials
|
||||
|
||||
v519 -> 2026-07-09 : https://github.com/inventree/InvenTree/pull/TODO
|
||||
- Adds optional "roles" and "permissions" fields to the /user/me/ API endpoint, via the "?roles=true" query parameter
|
||||
|
||||
|
||||
@@ -223,12 +223,18 @@ def postprocess_schema_enums(result, generator, **kwargs):
|
||||
"""Custom patch to ignore some drf-spectacular warnings.
|
||||
|
||||
- Some warnings are unavoidable due to the way that InvenTree implements generic relationships (via ContentType).
|
||||
- Some warnings are unavoidable due to the way that InvenTree implements custom (database-editable) status codes:
|
||||
multiple serializers legitimately expose a 'status' field backed by the same dynamic StockStatus choice set
|
||||
(e.g. stock adjustment, receiving a purchase order line, disassembling a stock item), and drf-spectacular
|
||||
cannot settle on a single stable name for the shared, runtime-dependent choice set.
|
||||
- The cleanest way to handle this appears to be to override the 'warn' function from drf-spectacular.
|
||||
|
||||
Ref: https://github.com/inventree/InvenTree/pull/10699
|
||||
"""
|
||||
ignore_patterns = [
|
||||
'enum naming encountered a non-optimally resolvable collision for fields named "model_type"'
|
||||
'enum naming encountered a non-optimally resolvable collision for fields named "model_type"',
|
||||
'enum naming encountered a non-optimally resolvable collision for fields named "status"',
|
||||
'encountered multiple names for the same choice set (StatusCustomKeyEnum)',
|
||||
]
|
||||
|
||||
if any(pattern in msg for pattern in ignore_patterns):
|
||||
|
||||
@@ -173,6 +173,36 @@ class StockItemConvert(StockItemContextMixin, CreateAPI):
|
||||
serializer_class = StockSerializers.ConvertStockItemSerializer
|
||||
|
||||
|
||||
@extend_schema(responses={201: StockSerializers.StockItemSerializer(many=True)})
|
||||
class StockItemDisassemble(StockItemContextMixin, CreateAPI):
|
||||
"""API endpoint for disassembling a stock item into its component parts.
|
||||
|
||||
The components are generated based on the BOM lines provided in the request,
|
||||
which must be valid BOM lines for the part associated with the stock item.
|
||||
"""
|
||||
|
||||
serializer_class = StockSerializers.DisassembleStockItemSerializer
|
||||
pagination_class = None
|
||||
|
||||
def create(self, request, *args, **kwargs):
|
||||
"""Disassemble the provided StockItem."""
|
||||
serializer = self.get_serializer(data=request.data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
|
||||
# Perform the actual disassembly step
|
||||
items = serializer.save()
|
||||
|
||||
queryset = StockSerializers.StockItemSerializer.annotate_queryset(
|
||||
StockItem.objects.filter(pk__in=[item.pk for item in items])
|
||||
)
|
||||
|
||||
response = StockSerializers.StockItemSerializer(
|
||||
queryset, many=True, context=self.get_serializer_context()
|
||||
)
|
||||
|
||||
return Response(response.data, status=status.HTTP_201_CREATED)
|
||||
|
||||
|
||||
class StockAdjustView(CreateAPI):
|
||||
"""A generic class for handling stocktake actions.
|
||||
|
||||
@@ -1771,6 +1801,11 @@ stock_api_urls = [
|
||||
'<int:pk>/',
|
||||
include([
|
||||
path('convert/', StockItemConvert.as_view(), name='api-stock-item-convert'),
|
||||
path(
|
||||
'disassemble/',
|
||||
StockItemDisassemble.as_view(),
|
||||
name='api-stock-item-disassemble',
|
||||
),
|
||||
path('install/', StockItemInstall.as_view(), name='api-stock-item-install'),
|
||||
meta_path(StockItem),
|
||||
path(
|
||||
|
||||
@@ -15,5 +15,6 @@ class StockEvents(BaseEventEnum):
|
||||
ITEM_COUNTED = 'stockitem.counted'
|
||||
ITEM_QUANTITY_UPDATED = 'stockitem.quantityupdated'
|
||||
ITEM_INSTALLED_INTO_ASSEMBLY = 'stockitem.installed'
|
||||
ITEM_DISASSEMBLED = 'stockitem.disassembled'
|
||||
|
||||
ITEMS_CREATED = 'stockitem.created_items'
|
||||
|
||||
@@ -1813,6 +1813,276 @@ class StockItem(
|
||||
|
||||
self.save()
|
||||
|
||||
def allocate_disassembly_pricing(self, quantity, lines: list) -> list:
|
||||
"""Allocate the purchase price of this stock item across disassembly lines.
|
||||
|
||||
Automatic cost allocation is only performed if:
|
||||
- This stock item has a recorded purchase price
|
||||
- None of the provided lines specify an explicit purchase price
|
||||
|
||||
The total cost is apportioned across the lines pro-rata,
|
||||
weighted by existing pricing data for each component part.
|
||||
If pricing data is not available for *every* line,
|
||||
the cost is split evenly across the generated units.
|
||||
|
||||
Args:
|
||||
quantity: The number of units of this stock item being disassembled
|
||||
lines: A list of dict objects (see the disassemble method)
|
||||
|
||||
Returns:
|
||||
The list of lines, with 'purchase_price' values filled in where possible
|
||||
"""
|
||||
if not self.purchase_price:
|
||||
return lines
|
||||
|
||||
if any(line.get('purchase_price') is not None for line in lines):
|
||||
# Explicit pricing has been provided - do not override
|
||||
return lines
|
||||
|
||||
# Calculate a relative pricing "weight" for each line
|
||||
weights = []
|
||||
|
||||
for line in lines:
|
||||
pricing = line['bom_item'].sub_part.pricing
|
||||
|
||||
price_range = [
|
||||
price
|
||||
for price in [pricing.overall_min, pricing.overall_max]
|
||||
if price is not None
|
||||
]
|
||||
|
||||
weight = None
|
||||
|
||||
if price_range:
|
||||
mid = sum(price_range[1:], price_range[0]) / len(price_range)
|
||||
|
||||
if mid.amount > 0:
|
||||
weight = Decimal(mid.amount) * Decimal(line['quantity'])
|
||||
|
||||
weights.append(weight)
|
||||
|
||||
if any(weight is None for weight in weights):
|
||||
# Insufficient pricing data - fall back to an even (per unit) split
|
||||
weights = [Decimal(line['quantity']) for line in lines]
|
||||
|
||||
total_weight = sum(weights)
|
||||
|
||||
if total_weight <= 0:
|
||||
return lines
|
||||
|
||||
# Total cost of the disassembled units
|
||||
total_cost = self.purchase_price * quantity
|
||||
|
||||
for line, weight in zip(lines, weights, strict=True):
|
||||
line_quantity = Decimal(line['quantity'])
|
||||
|
||||
if line_quantity > 0:
|
||||
if weight is None:
|
||||
# Fallback to an even split
|
||||
line['purchase_price'] = total_cost / len(lines) / line_quantity
|
||||
else:
|
||||
line['purchase_price'] = (
|
||||
total_cost * (weight / total_weight) / line_quantity
|
||||
)
|
||||
|
||||
return lines
|
||||
|
||||
@transaction.atomic
|
||||
def disassemble(
|
||||
self, quantity, lines: list, user, location=None, notes: str = ''
|
||||
) -> list[StockItem]:
|
||||
"""Disassemble this stock item into its component parts.
|
||||
|
||||
The provided BOM lines determine which component parts are "broken out"
|
||||
of this stock item. A new stock item is created for each line,
|
||||
and the quantity of this item is reduced accordingly.
|
||||
|
||||
Args:
|
||||
quantity: The number of units of this stock item to disassemble
|
||||
lines: A list of dict objects, each containing:
|
||||
bom_item: The BomItem instance defining the component to break out
|
||||
quantity: The total quantity of the component to create
|
||||
location: Optional destination StockLocation for the component
|
||||
status: Optional StockStatus code for the component (default: OK)
|
||||
purchase_price: Optional unit purchase price for the component
|
||||
user: The user performing the operation
|
||||
location: Default destination location for the generated components
|
||||
notes: Optional transaction notes
|
||||
|
||||
Returns:
|
||||
A list of the component StockItem objects (created or uninstalled)
|
||||
|
||||
Note:
|
||||
This stock item is *retained* (with reduced quantity) even if the
|
||||
quantity reaches zero, to preserve traceability of the disassembly.
|
||||
|
||||
Note:
|
||||
Any stock items which are *installed* within this stock item are
|
||||
broken out (uninstalled) as part of the disassembly operation,
|
||||
rather than creating new stock items for them.
|
||||
|
||||
Note:
|
||||
Traceability data is passed down from this stock item to the
|
||||
generated components: the batch code and source purchase order
|
||||
are copied across, while any source build order is recorded
|
||||
in the stock tracking history of each component.
|
||||
"""
|
||||
try:
|
||||
quantity = Decimal(quantity)
|
||||
except (InvalidOperation, ValueError):
|
||||
raise ValidationError({'quantity': _('Invalid quantity provided')})
|
||||
|
||||
if quantity <= 0:
|
||||
raise ValidationError({'quantity': _('Quantity must be greater than zero')})
|
||||
|
||||
if quantity > self.quantity:
|
||||
raise ValidationError({
|
||||
'quantity': _('Quantity must not exceed available stock quantity')
|
||||
})
|
||||
|
||||
if self.serialized and quantity != self.quantity:
|
||||
raise ValidationError({
|
||||
'quantity': _(
|
||||
'Serialized stock item must be disassembled in its entirety'
|
||||
)
|
||||
})
|
||||
|
||||
if len(lines) == 0:
|
||||
raise ValidationError(_('No BOM lines provided'))
|
||||
|
||||
# Any items installed within this stock item *must* be broken out,
|
||||
# which requires that the entire quantity is disassembled
|
||||
installed_items = list(self.installed_parts.all())
|
||||
|
||||
if len(installed_items) > 0 and quantity != self.quantity:
|
||||
raise ValidationError({
|
||||
'quantity': _(
|
||||
'Stock item with installed items must be disassembled in its entirety'
|
||||
)
|
||||
})
|
||||
|
||||
# Cache the custom status options for the StockItem model
|
||||
custom_status_values = StockItem.STATUS_CLASS.custom_values()
|
||||
|
||||
items = []
|
||||
|
||||
# Match installed items against the provided BOM lines,
|
||||
# so that the correct location and status values can be applied
|
||||
line_map = {}
|
||||
|
||||
for line in lines:
|
||||
for part in line['bom_item'].get_valid_parts_for_allocation():
|
||||
line_map.setdefault(part.pk, line)
|
||||
|
||||
for installed_item in installed_items:
|
||||
line = line_map.get(installed_item.part.pk)
|
||||
|
||||
destination = (
|
||||
(line.get('location') if line else None) or location or self.location
|
||||
)
|
||||
|
||||
if line:
|
||||
# The uninstalled quantity reduces the quantity of new items
|
||||
# to be created against the matching BOM line
|
||||
line['quantity'] = max(
|
||||
Decimal(0), Decimal(line['quantity']) - installed_item.quantity
|
||||
)
|
||||
|
||||
if status := line.get('status'):
|
||||
installed_item.set_status(
|
||||
status, custom_values=custom_status_values
|
||||
)
|
||||
|
||||
# Break the installed item out into the destination location
|
||||
# Note: this also saves any status change applied above
|
||||
installed_item.uninstall_into_location(destination, user, notes)
|
||||
|
||||
items.append(installed_item)
|
||||
|
||||
# Allocate pricing data across the generated components.
|
||||
# Note: Deliberately performed *after* the installed items are uninstalled,
|
||||
# so that the cost is only spread across newly created stock items
|
||||
lines = self.allocate_disassembly_pricing(quantity, lines)
|
||||
|
||||
for line in lines:
|
||||
bom_item = line['bom_item']
|
||||
line_quantity = Decimal(line['quantity'])
|
||||
|
||||
if line_quantity <= 0:
|
||||
continue
|
||||
|
||||
destination = line.get('location') or location or self.location
|
||||
|
||||
new_item = StockItem(
|
||||
part=bom_item.sub_part,
|
||||
quantity=line_quantity,
|
||||
location=destination,
|
||||
parent=self,
|
||||
batch=self.batch,
|
||||
purchase_order=self.purchase_order,
|
||||
purchase_price=line.get('purchase_price', None),
|
||||
)
|
||||
|
||||
if status := line.get('status'):
|
||||
new_item.set_status(status, custom_values=custom_status_values)
|
||||
|
||||
# Ensure the tree structure is observed
|
||||
new_item.tree_id = None
|
||||
new_item.save(add_note=False)
|
||||
|
||||
deltas = {'stockitem': self.pk, 'quantity': float(line_quantity)}
|
||||
|
||||
# Record traceability links against the disassembled assembly
|
||||
if self.build:
|
||||
deltas['buildorder'] = self.build.pk
|
||||
|
||||
if self.purchase_order:
|
||||
deltas['purchaseorder'] = self.purchase_order.pk
|
||||
|
||||
new_item.add_tracking_entry(
|
||||
StockHistoryCode.CREATED_FROM_DISASSEMBLY,
|
||||
user,
|
||||
notes=notes,
|
||||
deltas=deltas,
|
||||
location=destination,
|
||||
)
|
||||
|
||||
items.append(new_item)
|
||||
|
||||
# Consume this stock item.
|
||||
# Note: the item is deliberately retained (not deleted) at zero quantity,
|
||||
# to preserve the disassembly history
|
||||
if self.serialized:
|
||||
# A serialized item cannot be reduced to zero quantity,
|
||||
# so instead it is marked as "destroyed"
|
||||
deltas = {
|
||||
'removed': float(quantity),
|
||||
'status': StockStatus.DESTROYED.value,
|
||||
'old_status': self.status,
|
||||
}
|
||||
|
||||
self.set_status(StockStatus.DESTROYED.value)
|
||||
else:
|
||||
deltas = {
|
||||
'removed': float(quantity),
|
||||
'quantity': float(self.quantity - quantity),
|
||||
}
|
||||
|
||||
self.quantity = self.quantity - quantity
|
||||
|
||||
self.save(add_note=False)
|
||||
|
||||
self.add_tracking_entry(
|
||||
StockHistoryCode.DISASSEMBLED, user, notes=notes, deltas=deltas
|
||||
)
|
||||
|
||||
# Rebuild the stock tree for this item
|
||||
stock.tasks.rebuild_stock_item_tree(self.tree_id)
|
||||
|
||||
trigger_event(StockEvents.ITEM_DISASSEMBLED, id=self.pk)
|
||||
|
||||
return items
|
||||
|
||||
@property
|
||||
def children(self):
|
||||
"""Return a list of the child items which have been split from this stock item."""
|
||||
@@ -1922,7 +2192,7 @@ class StockItem(
|
||||
return None
|
||||
|
||||
# Extract any additional information from the kwargs
|
||||
self._apply_model_reference_fields(deltas, kwargs)
|
||||
self._apply_model_reference_fields(kwargs, deltas)
|
||||
|
||||
# Quantity specified?
|
||||
quantity = kwargs.get('quantity')
|
||||
|
||||
@@ -945,6 +945,227 @@ class UninstallStockItemSerializer(serializers.Serializer):
|
||||
item.uninstall_into_location(location, request.user, note)
|
||||
|
||||
|
||||
class StockStatusCustomSerializer(serializers.ChoiceField):
|
||||
"""Serializer to allow annotating the schema to use int where custom values may be entered."""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
"""Initialize the status selector."""
|
||||
if 'choices' not in kwargs:
|
||||
kwargs['choices'] = stock.status_codes.StockStatus.items(custom=True)
|
||||
|
||||
if 'label' not in kwargs:
|
||||
kwargs['label'] = _('Status')
|
||||
|
||||
if 'help_text' not in kwargs:
|
||||
kwargs['help_text'] = _('Stock item status code')
|
||||
|
||||
if InvenTree.ready.isGeneratingSchema():
|
||||
kwargs['help_text'] = (
|
||||
kwargs['help_text']
|
||||
+ '\n\n'
|
||||
+ '\n'.join(
|
||||
f'* `{value}` - {label}' for value, label in kwargs['choices']
|
||||
)
|
||||
+ "\n\nAdditional custom status keys may be retrieved from the 'stock_status_retrieve' call."
|
||||
)
|
||||
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
|
||||
class DisassemblyLineSerializer(serializers.Serializer):
|
||||
"""Serializer for a single component line in a stock disassembly operation."""
|
||||
|
||||
class Meta:
|
||||
"""Metaclass options."""
|
||||
|
||||
fields = [
|
||||
'bom_item',
|
||||
'quantity',
|
||||
'location',
|
||||
'status',
|
||||
'purchase_price',
|
||||
'purchase_price_currency',
|
||||
]
|
||||
|
||||
bom_item = serializers.PrimaryKeyRelatedField(
|
||||
queryset=part_models.BomItem.objects.all(),
|
||||
many=False,
|
||||
required=True,
|
||||
allow_null=False,
|
||||
label=_('BOM Item'),
|
||||
help_text=_('BOM line which defines the component part to break out'),
|
||||
)
|
||||
|
||||
quantity = serializers.DecimalField(
|
||||
max_digits=15,
|
||||
decimal_places=5,
|
||||
min_value=Decimal(0),
|
||||
required=True,
|
||||
label=_('Quantity'),
|
||||
help_text=_('Quantity of the component part to create'),
|
||||
)
|
||||
|
||||
def validate_quantity(self, quantity):
|
||||
"""Validation for the 'quantity' field."""
|
||||
if quantity <= 0:
|
||||
raise ValidationError(_('Quantity must be greater than zero'))
|
||||
|
||||
return quantity
|
||||
|
||||
location = serializers.PrimaryKeyRelatedField(
|
||||
queryset=StockLocation.objects.all(),
|
||||
many=False,
|
||||
allow_null=True,
|
||||
required=False,
|
||||
label=_('Location'),
|
||||
help_text=_('Destination location for the component items'),
|
||||
)
|
||||
|
||||
status = StockStatusCustomSerializer(
|
||||
required=False, help_text=_('Status code for the component items')
|
||||
)
|
||||
|
||||
purchase_price = InvenTree.serializers.InvenTreeMoneySerializer(
|
||||
label=_('Purchase Price'),
|
||||
required=False,
|
||||
allow_null=True,
|
||||
default=None,
|
||||
help_text=_(
|
||||
'Unit purchase price for the component items (leave blank for automatic cost allocation)'
|
||||
),
|
||||
)
|
||||
|
||||
purchase_price_currency = InvenTreeCurrencySerializer(
|
||||
required=False, help_text=_('Purchase price currency')
|
||||
)
|
||||
|
||||
|
||||
class DisassembleStockItemSerializer(serializers.Serializer):
|
||||
"""DRF serializer class for disassembling a StockItem into component parts.
|
||||
|
||||
Note: The stock item being disassembled is provided via the serializer context
|
||||
"""
|
||||
|
||||
class Meta:
|
||||
"""Metaclass options."""
|
||||
|
||||
fields = ['items', 'quantity', 'location', 'notes']
|
||||
|
||||
items = DisassemblyLineSerializer(many=True)
|
||||
|
||||
quantity = serializers.DecimalField(
|
||||
max_digits=15,
|
||||
decimal_places=5,
|
||||
min_value=Decimal(0),
|
||||
required=True,
|
||||
label=_('Quantity'),
|
||||
help_text=_('Number of assemblies to disassemble'),
|
||||
)
|
||||
|
||||
def validate_quantity(self, quantity):
|
||||
"""Validation for the 'quantity' field."""
|
||||
item = self.context.get('item')
|
||||
|
||||
if not item:
|
||||
raise ValidationError(_('No stock item provided'))
|
||||
|
||||
if quantity <= 0:
|
||||
raise ValidationError(_('Quantity must be greater than zero'))
|
||||
|
||||
if quantity > item.quantity:
|
||||
raise ValidationError(
|
||||
_('Quantity must not exceed available stock quantity')
|
||||
+ f' ({item.quantity})'
|
||||
)
|
||||
|
||||
return quantity
|
||||
|
||||
location = serializers.PrimaryKeyRelatedField(
|
||||
queryset=StockLocation.objects.all(),
|
||||
many=False,
|
||||
allow_null=True,
|
||||
required=False,
|
||||
label=_('Location'),
|
||||
help_text=_('Default destination location for the component items'),
|
||||
)
|
||||
|
||||
notes = serializers.CharField(
|
||||
required=False,
|
||||
allow_blank=True,
|
||||
label=_('Notes'),
|
||||
help_text=_('Optional note field'),
|
||||
)
|
||||
|
||||
def validate(self, data):
|
||||
"""Validate the disassembly operation.
|
||||
|
||||
- The stock item must be "in stock"
|
||||
- Each line must reference a valid BOM item for the part
|
||||
- Each BOM item may only be referenced once
|
||||
"""
|
||||
data = super().validate(data)
|
||||
|
||||
item = self.context.get('item')
|
||||
|
||||
if not item:
|
||||
raise ValidationError(_('No stock item provided'))
|
||||
|
||||
if not item.in_stock:
|
||||
raise ValidationError(_('Stock item is unavailable'))
|
||||
|
||||
items = data.get('items', [])
|
||||
|
||||
if len(items) == 0:
|
||||
raise ValidationError(_('Line items must be provided'))
|
||||
|
||||
# The set of valid BOM items for this part.
|
||||
# Consumable BOM lines (and lines pointing to a virtual part) are excluded,
|
||||
# as these components are not expected to be tracked as physical stock
|
||||
bom_items = item.part.get_bom_items(include_virtual=False).filter(
|
||||
part_models.BomItem.consumable_filter(consumable=False)
|
||||
)
|
||||
|
||||
bom_item_pks = set()
|
||||
|
||||
for line in items:
|
||||
bom_item = line['bom_item']
|
||||
|
||||
if not bom_items.filter(pk=bom_item.pk).exists():
|
||||
raise ValidationError(
|
||||
_('BOM item is not valid for the selected stock item')
|
||||
)
|
||||
|
||||
if bom_item.pk in bom_item_pks:
|
||||
raise ValidationError(_('Duplicate BOM items provided'))
|
||||
|
||||
bom_item_pks.add(bom_item.pk)
|
||||
|
||||
return data
|
||||
|
||||
def save(self) -> list[StockItem]:
|
||||
"""Disassemble the provided StockItem into its component parts.
|
||||
|
||||
Returns:
|
||||
A list of StockItem objects created by the disassembly operation.
|
||||
"""
|
||||
item = self.context['item']
|
||||
request = self.context.get('request')
|
||||
|
||||
data = self.validated_data
|
||||
|
||||
try:
|
||||
return item.disassemble(
|
||||
data['quantity'],
|
||||
data['items'],
|
||||
request.user if request else None,
|
||||
location=data.get('location', None),
|
||||
notes=data.get('notes', ''),
|
||||
)
|
||||
except DjangoValidationError as exc:
|
||||
# Catch model errors and re-throw as DRF errors
|
||||
raise ValidationError(detail=serializers.as_serializer_error(exc))
|
||||
|
||||
|
||||
class ConvertStockItemSerializer(serializers.Serializer):
|
||||
"""DRF serializer class for converting a StockItem to a valid variant part."""
|
||||
|
||||
@@ -1008,33 +1229,6 @@ class ConvertStockItemSerializer(serializers.Serializer):
|
||||
help_text='Status key, chosen from the list of StockStatus keys'
|
||||
)
|
||||
)
|
||||
class StockStatusCustomSerializer(serializers.ChoiceField):
|
||||
"""Serializer to allow annotating the schema to use int where custom values may be entered."""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
"""Initialize the status selector."""
|
||||
if 'choices' not in kwargs:
|
||||
kwargs['choices'] = stock.status_codes.StockStatus.items(custom=True)
|
||||
|
||||
if 'label' not in kwargs:
|
||||
kwargs['label'] = _('Status')
|
||||
|
||||
if 'help_text' not in kwargs:
|
||||
kwargs['help_text'] = _('Stock item status code')
|
||||
|
||||
if InvenTree.ready.isGeneratingSchema():
|
||||
kwargs['help_text'] = (
|
||||
kwargs['help_text']
|
||||
+ '\n\n'
|
||||
+ '\n'.join(
|
||||
f'* `{value}` - {label}' for value, label in kwargs['choices']
|
||||
)
|
||||
+ "\n\nAdditional custom status keys may be retrieved from the 'stock_status_retrieve' call."
|
||||
)
|
||||
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
|
||||
class StockChangeStatusSerializer(serializers.Serializer):
|
||||
"""Serializer for changing status of multiple StockItem objects."""
|
||||
|
||||
|
||||
@@ -75,6 +75,10 @@ class StockHistoryCode(StatusCode):
|
||||
# Stock merging operations
|
||||
MERGED_STOCK_ITEMS = 45, _('Merged stock items')
|
||||
|
||||
# Stock disassembly operations
|
||||
DISASSEMBLED = 46, _('Disassembled into components')
|
||||
CREATED_FROM_DISASSEMBLY = 47, _('Created from disassembly')
|
||||
|
||||
# Convert stock item to variant
|
||||
CONVERTED_TO_VARIANT = 48, _('Converted to variant')
|
||||
|
||||
|
||||
@@ -14,6 +14,7 @@ from rest_framework import status
|
||||
|
||||
import build.models
|
||||
import company.models
|
||||
import order.models
|
||||
import part.models
|
||||
from common.models import InvenTreeCustomUserStateModel, InvenTreeSetting
|
||||
from common.settings import set_global_setting
|
||||
@@ -2215,6 +2216,487 @@ class StockItemTest(StockAPITestCase):
|
||||
self.assertEqual(item.batch, 'NEW-BATCH-CODE')
|
||||
|
||||
|
||||
class StockItemDisassembleTest(StockAPITestCase):
|
||||
"""Series of API tests for the StockItem disassembly endpoint."""
|
||||
|
||||
def setUp(self):
|
||||
"""Create a stock item of an assembly part, ready for disassembly."""
|
||||
super().setUp()
|
||||
|
||||
# Part 100 ("Bob") is an assembly with 4 BOM lines
|
||||
self.assembly = part.models.Part.objects.get(pk=100)
|
||||
|
||||
self.item = StockItem.objects.create(
|
||||
part=self.assembly,
|
||||
quantity=10,
|
||||
location=StockLocation.objects.get(pk=1),
|
||||
purchase_price=Money(100, 'USD'),
|
||||
)
|
||||
|
||||
self.url = reverse('api-stock-item-disassemble', kwargs={'pk': self.item.pk})
|
||||
|
||||
def test_validation(self):
|
||||
"""Test validation checks for the disassembly endpoint."""
|
||||
# Empty request
|
||||
response = self.post(self.url, {}, expected_code=400)
|
||||
|
||||
self.assertIn('This field is required', str(response.data['items']))
|
||||
self.assertIn('This field is required', str(response.data['quantity']))
|
||||
|
||||
# No line items provided
|
||||
response = self.post(self.url, {'items': [], 'quantity': 1}, expected_code=400)
|
||||
|
||||
self.assertIn('Line items must be provided', str(response.data))
|
||||
|
||||
# Quantity exceeds available stock
|
||||
response = self.post(
|
||||
self.url,
|
||||
{'items': [{'bom_item': 1, 'quantity': 10}], 'quantity': 100},
|
||||
expected_code=400,
|
||||
)
|
||||
|
||||
self.assertIn(
|
||||
'Quantity must not exceed available stock quantity', str(response.data)
|
||||
)
|
||||
|
||||
# BOM item which points to a different part (BomItem pk=5 -> part 1)
|
||||
response = self.post(
|
||||
self.url,
|
||||
{'items': [{'bom_item': 5, 'quantity': 1}], 'quantity': 1},
|
||||
expected_code=400,
|
||||
)
|
||||
|
||||
self.assertIn(
|
||||
'BOM item is not valid for the selected stock item', str(response.data)
|
||||
)
|
||||
|
||||
# Duplicated BOM items
|
||||
response = self.post(
|
||||
self.url,
|
||||
{
|
||||
'items': [
|
||||
{'bom_item': 1, 'quantity': 10},
|
||||
{'bom_item': 1, 'quantity': 5},
|
||||
],
|
||||
'quantity': 1,
|
||||
},
|
||||
expected_code=400,
|
||||
)
|
||||
|
||||
self.assertIn('Duplicate BOM items provided', str(response.data))
|
||||
|
||||
def test_consumable_and_virtual_excluded(self):
|
||||
"""Consumable BOM lines, and lines pointing to a virtual part, cannot be disassembled."""
|
||||
# BOM line marked as consumable directly
|
||||
consumable_line = part.models.BomItem.objects.create(
|
||||
part=self.assembly,
|
||||
sub_part=part.models.Part.objects.get(pk=2),
|
||||
quantity=1,
|
||||
consumable=True,
|
||||
)
|
||||
|
||||
# BOM line whose sub_part is marked as consumable
|
||||
consumable_part = part.models.Part.objects.get(pk=3)
|
||||
consumable_part.consumable = True
|
||||
consumable_part.save()
|
||||
|
||||
# BOM line whose sub_part is marked as virtual
|
||||
virtual_part = part.models.Part.objects.get(pk=5)
|
||||
virtual_part.virtual = True
|
||||
virtual_part.save()
|
||||
|
||||
for bom_item_pk in [consumable_line.pk, 2, 3]:
|
||||
response = self.post(
|
||||
self.url,
|
||||
{'items': [{'bom_item': bom_item_pk, 'quantity': 1}], 'quantity': 1},
|
||||
expected_code=400,
|
||||
)
|
||||
|
||||
self.assertIn(
|
||||
'BOM item is not valid for the selected stock item', str(response.data)
|
||||
)
|
||||
|
||||
def test_disassemble(self):
|
||||
"""Test a valid disassembly operation, using a subset of the BOM lines."""
|
||||
n = StockItem.objects.count()
|
||||
|
||||
# Disassemble 4 assemblies, but only break out 2 of the 4 BOM lines
|
||||
# The second line specifies a custom location and status
|
||||
response = self.post(
|
||||
self.url,
|
||||
{
|
||||
'items': [
|
||||
{'bom_item': 1, 'quantity': 40},
|
||||
{
|
||||
'bom_item': 4,
|
||||
'quantity': 12,
|
||||
'location': 2,
|
||||
'status': StockStatus.DAMAGED.value,
|
||||
},
|
||||
],
|
||||
'quantity': 4,
|
||||
'notes': 'Breaking apart',
|
||||
},
|
||||
expected_code=201,
|
||||
)
|
||||
|
||||
self.assertEqual(len(response.data), 2)
|
||||
self.assertEqual(StockItem.objects.count(), n + 2)
|
||||
|
||||
self.item.refresh_from_db()
|
||||
self.assertEqual(self.item.quantity, 6)
|
||||
|
||||
items = {entry['part']: entry for entry in response.data}
|
||||
|
||||
# Check the generated component items
|
||||
item_1 = StockItem.objects.get(pk=items[1]['pk'])
|
||||
item_50 = StockItem.objects.get(pk=items[50]['pk'])
|
||||
|
||||
self.assertEqual(item_1.quantity, 40)
|
||||
self.assertEqual(item_50.quantity, 12)
|
||||
|
||||
# Components inherit the location of the disassembled item by default
|
||||
self.assertEqual(item_1.location, self.item.location)
|
||||
self.assertEqual(item_1.status, StockStatus.OK.value)
|
||||
|
||||
# A custom location and status can be specified per line
|
||||
self.assertEqual(item_50.location.pk, 2)
|
||||
self.assertEqual(item_50.status, StockStatus.DAMAGED.value)
|
||||
|
||||
# Components are linked to the disassembled item
|
||||
self.assertEqual(item_1.parent.pk, self.item.pk)
|
||||
self.assertEqual(item_50.parent.pk, self.item.pk)
|
||||
|
||||
# Total cost (4 x 100 USD) is allocated evenly across 52 units
|
||||
self.assertAlmostEqual(float(item_1.purchase_price.amount), 400 / 52, places=4)
|
||||
self.assertAlmostEqual(float(item_50.purchase_price.amount), 400 / 52, places=4)
|
||||
self.assertEqual(str(item_1.purchase_price.currency), 'USD')
|
||||
|
||||
# Check stock tracking entries have been created
|
||||
self.assertTrue(
|
||||
self.item.tracking_info.filter(
|
||||
tracking_type=StockHistoryCode.DISASSEMBLED.value
|
||||
).exists()
|
||||
)
|
||||
|
||||
for item in [item_1, item_50]:
|
||||
entry = item.tracking_info.filter(
|
||||
tracking_type=StockHistoryCode.CREATED_FROM_DISASSEMBLY.value
|
||||
).first()
|
||||
|
||||
self.assertIsNotNone(entry)
|
||||
self.assertEqual(entry.deltas['stockitem'], self.item.pk)
|
||||
|
||||
def test_full_disassembly(self):
|
||||
"""Test that a fully disassembled item is retained at zero quantity."""
|
||||
self.post(
|
||||
self.url,
|
||||
{'items': [{'bom_item': 2, 'quantity': 400}], 'quantity': 10},
|
||||
expected_code=201,
|
||||
)
|
||||
|
||||
self.item.refresh_from_db()
|
||||
|
||||
self.assertEqual(self.item.quantity, 0)
|
||||
self.assertFalse(self.item.in_stock)
|
||||
|
||||
def test_traceability_inheritance(self):
|
||||
"""Test that traceability data is passed down to the generated components."""
|
||||
po = order.models.PurchaseOrder.objects.create(
|
||||
supplier=company.models.Company.objects.get(pk=1), reference='PO-9999'
|
||||
)
|
||||
|
||||
bo = build.models.Build.objects.create(
|
||||
part=self.assembly, quantity=10, reference='BO-9999'
|
||||
)
|
||||
|
||||
self.item.batch = 'ABC-123'
|
||||
self.item.purchase_order = po
|
||||
self.item.build = bo
|
||||
self.item.save()
|
||||
|
||||
response = self.post(
|
||||
self.url,
|
||||
{'items': [{'bom_item': 1, 'quantity': 10}], 'quantity': 1},
|
||||
expected_code=201,
|
||||
)
|
||||
|
||||
component = StockItem.objects.get(pk=response.data[0]['pk'])
|
||||
|
||||
# Batch code and source purchase order are inherited directly
|
||||
self.assertEqual(component.batch, 'ABC-123')
|
||||
self.assertEqual(component.purchase_order, po)
|
||||
|
||||
# The source build order cannot be copied across (the component is not
|
||||
# an output of the build) - instead it is recorded in the stock history
|
||||
self.assertIsNone(component.build)
|
||||
|
||||
entry = component.tracking_info.filter(
|
||||
tracking_type=StockHistoryCode.CREATED_FROM_DISASSEMBLY.value
|
||||
).first()
|
||||
|
||||
self.assertIsNotNone(entry)
|
||||
self.assertEqual(entry.deltas['buildorder'], bo.pk)
|
||||
self.assertEqual(entry.deltas['purchaseorder'], po.pk)
|
||||
|
||||
def test_explicit_pricing(self):
|
||||
"""Test that automatic cost allocation is skipped if explicit pricing is provided."""
|
||||
response = self.post(
|
||||
self.url,
|
||||
{
|
||||
'items': [
|
||||
{
|
||||
'bom_item': 1,
|
||||
'quantity': 10,
|
||||
'purchase_price': 5,
|
||||
'purchase_price_currency': 'NZD',
|
||||
},
|
||||
{'bom_item': 4, 'quantity': 3},
|
||||
],
|
||||
'quantity': 1,
|
||||
},
|
||||
expected_code=201,
|
||||
)
|
||||
|
||||
prices = {entry['part']: entry['purchase_price'] for entry in response.data}
|
||||
|
||||
self.assertAlmostEqual(prices[1], 5, places=3)
|
||||
self.assertIsNone(prices[50])
|
||||
|
||||
def install_item(self, part_pk: int, quantity, **kwargs) -> StockItem:
|
||||
"""Install a new stock item into the assembly under test."""
|
||||
return StockItem.objects.create(
|
||||
part=part.models.Part.objects.get(pk=part_pk),
|
||||
quantity=quantity,
|
||||
belongs_to=self.item,
|
||||
location=None,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
def test_installed_items_partial(self):
|
||||
"""Installed items are uninstalled, and only the remainder is created afresh."""
|
||||
sub_1 = self.install_item(1, 12)
|
||||
sub_2 = self.install_item(1, 8)
|
||||
|
||||
n = StockItem.objects.count()
|
||||
|
||||
# Disassemble all 10 assemblies - the line requires 100 units in total
|
||||
response = self.post(
|
||||
self.url,
|
||||
{'items': [{'bom_item': 1, 'quantity': 100}], 'quantity': 10},
|
||||
expected_code=201,
|
||||
)
|
||||
|
||||
# Two uninstalled items, plus a single new item for the remainder
|
||||
self.assertEqual(len(response.data), 3)
|
||||
self.assertEqual(StockItem.objects.count(), n + 1)
|
||||
|
||||
pks = {entry['pk'] for entry in response.data}
|
||||
self.assertIn(sub_1.pk, pks)
|
||||
self.assertIn(sub_2.pk, pks)
|
||||
|
||||
for sub in [sub_1, sub_2]:
|
||||
sub.refresh_from_db()
|
||||
self.assertIsNone(sub.belongs_to)
|
||||
self.assertEqual(sub.location, self.item.location)
|
||||
|
||||
new_item = StockItem.objects.get(pk=(pks - {sub_1.pk, sub_2.pk}).pop())
|
||||
self.assertEqual(new_item.part.pk, 1)
|
||||
self.assertEqual(new_item.quantity, 80)
|
||||
|
||||
# Uninstalled items are updated, not created afresh
|
||||
self.assertTrue(
|
||||
sub_1.tracking_info.filter(
|
||||
tracking_type=StockHistoryCode.REMOVED_FROM_ASSEMBLY.value
|
||||
).exists()
|
||||
)
|
||||
self.assertFalse(
|
||||
sub_1.tracking_info.filter(
|
||||
tracking_type=StockHistoryCode.CREATED_FROM_DISASSEMBLY.value
|
||||
).exists()
|
||||
)
|
||||
self.assertTrue(
|
||||
self.item.tracking_info.filter(
|
||||
tracking_type=StockHistoryCode.REMOVED_CHILD_ITEM.value
|
||||
).exists()
|
||||
)
|
||||
self.assertTrue(
|
||||
new_item.tracking_info.filter(
|
||||
tracking_type=StockHistoryCode.CREATED_FROM_DISASSEMBLY.value
|
||||
).exists()
|
||||
)
|
||||
|
||||
def test_installed_items_full_coverage(self):
|
||||
"""No new stock item is created for a line fully covered by installed items."""
|
||||
sub = self.install_item(1, 120, purchase_price=Money(3, 'USD'))
|
||||
|
||||
n = StockItem.objects.count()
|
||||
|
||||
response = self.post(
|
||||
self.url,
|
||||
{
|
||||
'items': [
|
||||
{'bom_item': 1, 'quantity': 100},
|
||||
{'bom_item': 4, 'quantity': 30},
|
||||
],
|
||||
'quantity': 10,
|
||||
},
|
||||
expected_code=201,
|
||||
)
|
||||
|
||||
# The first line is fully covered by the uninstalled item
|
||||
self.assertEqual(len(response.data), 2)
|
||||
self.assertEqual(StockItem.objects.count(), n + 1)
|
||||
|
||||
sub.refresh_from_db()
|
||||
self.assertIsNone(sub.belongs_to)
|
||||
|
||||
# The uninstalled item retains its own purchase price
|
||||
self.assertEqual(sub.purchase_price, Money(3, 'USD'))
|
||||
|
||||
# The total cost (10 x 100 USD) is spread only across newly created units
|
||||
new_item = StockItem.objects.get(part=50, parent=self.item)
|
||||
self.assertEqual(new_item.quantity, 30)
|
||||
self.assertAlmostEqual(
|
||||
float(new_item.purchase_price.amount), 1000 / 30, places=4
|
||||
)
|
||||
|
||||
def test_installed_items_leftover(self):
|
||||
"""Installed items which do not match a BOM line are still uninstalled."""
|
||||
# Part 2 is not a component of the assembly
|
||||
sub = self.install_item(2, 5, status=StockStatus.ATTENTION.value)
|
||||
|
||||
response = self.post(
|
||||
self.url,
|
||||
{
|
||||
'items': [{'bom_item': 2, 'quantity': 400}],
|
||||
'quantity': 10,
|
||||
'location': 2,
|
||||
},
|
||||
expected_code=201,
|
||||
)
|
||||
|
||||
self.assertEqual(len(response.data), 2)
|
||||
|
||||
sub.refresh_from_db()
|
||||
self.assertIsNone(sub.belongs_to)
|
||||
|
||||
# Uninstalled to the top-level destination, retaining its own status
|
||||
self.assertEqual(sub.location.pk, 2)
|
||||
self.assertEqual(sub.status, StockStatus.ATTENTION.value)
|
||||
|
||||
# The BOM line quantity is not reduced by the unmatched item
|
||||
new_item = StockItem.objects.get(part=3, parent=self.item)
|
||||
self.assertEqual(new_item.quantity, 400)
|
||||
|
||||
def test_installed_items_require_full_disassembly(self):
|
||||
"""Partial disassembly is rejected while items are installed."""
|
||||
self.install_item(1, 1)
|
||||
|
||||
response = self.post(
|
||||
self.url,
|
||||
{'items': [{'bom_item': 1, 'quantity': 50}], 'quantity': 5},
|
||||
expected_code=400,
|
||||
)
|
||||
|
||||
self.assertIn('must be disassembled in its entirety', str(response.data))
|
||||
|
||||
def test_installed_items_status_and_location(self):
|
||||
"""Explicit per-line status and location values are applied to uninstalled items."""
|
||||
sub_1 = self.install_item(1, 10)
|
||||
sub_2 = self.install_item(50, 4, status=StockStatus.ATTENTION.value)
|
||||
|
||||
self.post(
|
||||
self.url,
|
||||
{
|
||||
'items': [
|
||||
{
|
||||
'bom_item': 1,
|
||||
'quantity': 100,
|
||||
'location': 2,
|
||||
'status': StockStatus.DAMAGED.value,
|
||||
},
|
||||
{'bom_item': 4, 'quantity': 30},
|
||||
],
|
||||
'quantity': 10,
|
||||
},
|
||||
expected_code=201,
|
||||
)
|
||||
|
||||
sub_1.refresh_from_db()
|
||||
self.assertEqual(sub_1.location.pk, 2)
|
||||
self.assertEqual(sub_1.status, StockStatus.DAMAGED.value)
|
||||
|
||||
# No explicit status provided for the second line - item status is retained
|
||||
sub_2.refresh_from_db()
|
||||
self.assertEqual(sub_2.location, self.item.location)
|
||||
self.assertEqual(sub_2.status, StockStatus.ATTENTION.value)
|
||||
|
||||
def test_installed_items_variants_and_substitutes(self):
|
||||
"""Installed variant and substitute parts are matched against BOM lines."""
|
||||
# Designate part 2 as a substitute for BOM line 4
|
||||
part.models.BomItemSubstitute.objects.create(
|
||||
bom_item=part.models.BomItem.objects.get(pk=4),
|
||||
part=part.models.Part.objects.get(pk=2),
|
||||
)
|
||||
|
||||
# New BOM line against a template part, allowing variants
|
||||
bom_line = part.models.BomItem.objects.create(
|
||||
part=self.assembly,
|
||||
sub_part=part.models.Part.objects.get(pk=10000),
|
||||
quantity=5,
|
||||
allow_variants=True,
|
||||
)
|
||||
|
||||
substitute_item = self.install_item(2, 10)
|
||||
variant_item = self.install_item(10001, 20)
|
||||
|
||||
response = self.post(
|
||||
self.url,
|
||||
{
|
||||
'items': [
|
||||
{'bom_item': 4, 'quantity': 30},
|
||||
{'bom_item': bom_line.pk, 'quantity': 50},
|
||||
],
|
||||
'quantity': 10,
|
||||
},
|
||||
expected_code=201,
|
||||
)
|
||||
|
||||
self.assertEqual(len(response.data), 4)
|
||||
|
||||
for sub in [substitute_item, variant_item]:
|
||||
sub.refresh_from_db()
|
||||
self.assertIsNone(sub.belongs_to)
|
||||
|
||||
# Line quantities are reduced by the matched installed items
|
||||
new_50 = StockItem.objects.get(part=50, parent=self.item)
|
||||
self.assertEqual(new_50.quantity, 20)
|
||||
|
||||
new_chair = StockItem.objects.get(part=10000, parent=self.item)
|
||||
self.assertEqual(new_chair.quantity, 30)
|
||||
|
||||
def test_serialized(self):
|
||||
"""Test disassembly of a serialized stock item."""
|
||||
item = StockItem.objects.create(
|
||||
part=self.assembly, quantity=1, serial='SN-DISASSEMBLE-1'
|
||||
)
|
||||
|
||||
url = reverse('api-stock-item-disassemble', kwargs={'pk': item.pk})
|
||||
|
||||
self.post(
|
||||
url,
|
||||
{'items': [{'bom_item': 1, 'quantity': 10}], 'quantity': 1},
|
||||
expected_code=201,
|
||||
)
|
||||
|
||||
item.refresh_from_db()
|
||||
|
||||
# A serialized item cannot be "depleted" - it is marked as destroyed instead
|
||||
self.assertEqual(item.status, StockStatus.DESTROYED.value)
|
||||
self.assertFalse(item.in_stock)
|
||||
|
||||
|
||||
class StocktakeTest(StockAPITestCase):
|
||||
"""Series of tests for the Stocktake API."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user