mirror of
https://github.com/inventree/InvenTree.git
synced 2025-04-28 11:36:44 +00:00
[Refactor] Barcode scanning (#8658)
* Enhance SupplierPart barcode mixin - Provide richer response based on available data * Add comment * Add new fields to BarcodePOReceiveSerializer * Add "supplier" information to serializer * Add 'is_completed' method for PurchaseOrderLineItem * Refactor SupplierBarcodeMixin * Tweak error message * Logic fix * Fix response data * Improved error checking * Bump API version * Bump API version * Error handling - Improve get_supplier_part method - Error handling - Wrap calls to external plugins * Enhanced unit testing and exception handling * More exception handling * Fix circula imports * Improve unit tests * Adjust filtering * Remove outdated tests * Remove success code for matching item
This commit is contained in:
parent
169f4f8350
commit
d42435c841
@ -1,13 +1,16 @@
|
||||
"""InvenTree API version information."""
|
||||
|
||||
# InvenTree API version
|
||||
INVENTREE_API_VERSION = 292
|
||||
INVENTREE_API_VERSION = 293
|
||||
|
||||
"""Increment this API version number whenever there is a significant change to the API that any clients need to know about."""
|
||||
|
||||
|
||||
INVENTREE_API_TEXT = """
|
||||
|
||||
v293 - 2024-12-14 : https://github.com/inventree/InvenTree/pull/8658
|
||||
- Adds new fields to the supplier barcode API endpoints
|
||||
|
||||
v292 - 2024-12-03 : https://github.com/inventree/InvenTree/pull/8625
|
||||
- Add "on_order" and "in_stock" annotations to SupplierPart API
|
||||
- Enhanced filtering for the SupplierPart API
|
||||
|
@ -11,13 +11,10 @@ from django.core.exceptions import ValidationError as DjangoValidationError
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
|
||||
import rest_framework.views as drfviews
|
||||
from error_report.models import Error
|
||||
from rest_framework import serializers
|
||||
from rest_framework.exceptions import ValidationError as DRFValidationError
|
||||
from rest_framework.response import Response
|
||||
|
||||
import InvenTree.sentry
|
||||
|
||||
logger = logging.getLogger('inventree')
|
||||
|
||||
|
||||
@ -34,6 +31,8 @@ def log_error(path, error_name=None, error_info=None, error_data=None):
|
||||
error_info: The error information (optional, overrides 'info')
|
||||
error_data: The error data (optional, overrides 'data')
|
||||
"""
|
||||
from error_report.models import Error
|
||||
|
||||
kind, info, data = sys.exc_info()
|
||||
|
||||
# Check if the error is on the ignore list
|
||||
@ -75,6 +74,8 @@ def exception_handler(exc, context):
|
||||
|
||||
If sentry error reporting is enabled, we will also provide the original exception to sentry.io
|
||||
"""
|
||||
import InvenTree.sentry
|
||||
|
||||
response = None
|
||||
|
||||
# Pass exception to sentry.io handler
|
||||
|
@ -128,10 +128,18 @@ class PluginValidationMixin(DiffMixin):
|
||||
|
||||
Note: Each plugin may raise a ValidationError to prevent deletion.
|
||||
"""
|
||||
from InvenTree.exceptions import log_error
|
||||
from plugin.registry import registry
|
||||
|
||||
for plugin in registry.with_mixin('validation'):
|
||||
plugin.validate_model_deletion(self)
|
||||
try:
|
||||
plugin.validate_model_deletion(self)
|
||||
except ValidationError as e:
|
||||
# Plugin might raise a ValidationError to prevent deletion
|
||||
raise e
|
||||
except Exception:
|
||||
log_error('plugin.validate_model_deletion')
|
||||
continue
|
||||
|
||||
super().delete()
|
||||
|
||||
|
@ -71,12 +71,14 @@ def get_icon_packs():
|
||||
)
|
||||
]
|
||||
|
||||
from InvenTree.exceptions import log_error
|
||||
from plugin import registry
|
||||
|
||||
for plugin in registry.with_mixin('icon_pack', active=True):
|
||||
try:
|
||||
icon_packs.extend(plugin.icon_packs())
|
||||
except Exception as e:
|
||||
log_error('get_icon_packs')
|
||||
logger.warning('Error loading icon pack from plugin %s: %s', plugin, e)
|
||||
|
||||
_icon_packs = {pack.prefix: pack for pack in icon_packs}
|
||||
|
@ -558,6 +558,7 @@ class PurchaseOrder(TotalPriceMixin, Order):
|
||||
group: bool = True,
|
||||
reference: str = '',
|
||||
purchase_price=None,
|
||||
destination=None,
|
||||
):
|
||||
"""Add a new line item to this purchase order.
|
||||
|
||||
@ -565,12 +566,13 @@ class PurchaseOrder(TotalPriceMixin, Order):
|
||||
* The supplier part matches the supplier specified for this purchase order
|
||||
* The quantity is greater than zero
|
||||
|
||||
Args:
|
||||
Arguments:
|
||||
supplier_part: The supplier_part to add
|
||||
quantity : The number of items to add
|
||||
group (bool, optional): If True, this new quantity will be added to an existing line item for the same supplier_part (if it exists). Defaults to True.
|
||||
reference (str, optional): Reference to item. Defaults to ''.
|
||||
purchase_price (optional): Price of item. Defaults to None.
|
||||
destination (optional): Destination for item. Defaults to None.
|
||||
|
||||
Returns:
|
||||
The newly created PurchaseOrderLineItem instance
|
||||
@ -619,6 +621,7 @@ class PurchaseOrder(TotalPriceMixin, Order):
|
||||
quantity=quantity,
|
||||
reference=reference,
|
||||
purchase_price=purchase_price,
|
||||
destination=destination,
|
||||
)
|
||||
|
||||
line.save()
|
||||
@ -1608,6 +1611,10 @@ class PurchaseOrderLineItem(OrderLineItem):
|
||||
r = self.quantity - self.received
|
||||
return max(r, 0)
|
||||
|
||||
def is_completed(self) -> bool:
|
||||
"""Determine if this lien item has been fully received."""
|
||||
return self.received >= self.quantity
|
||||
|
||||
def update_pricing(self):
|
||||
"""Update pricing information based on the supplier part data."""
|
||||
if self.part:
|
||||
|
@ -6,6 +6,7 @@ from rest_framework import permissions, serializers
|
||||
from rest_framework.generics import GenericAPIView
|
||||
from rest_framework.response import Response
|
||||
|
||||
from InvenTree.exceptions import log_error
|
||||
from plugin import registry
|
||||
|
||||
|
||||
@ -33,9 +34,12 @@ class ActionPluginView(GenericAPIView):
|
||||
|
||||
action_plugins = registry.with_mixin('action')
|
||||
for plugin in action_plugins:
|
||||
if plugin.action_name() == action:
|
||||
plugin.perform_action(request.user, data=data)
|
||||
return Response(plugin.get_response(request.user, data=data))
|
||||
try:
|
||||
if plugin.action_name() == action:
|
||||
plugin.perform_action(request.user, data=data)
|
||||
return Response(plugin.get_response(request.user, data=data))
|
||||
except Exception:
|
||||
log_error('action_plugin')
|
||||
|
||||
# If we got to here, no matching action was found
|
||||
return Response({'error': _('No matching action found'), 'action': action})
|
||||
|
@ -151,11 +151,18 @@ class BarcodeView(CreateAPIView):
|
||||
response = {}
|
||||
|
||||
for current_plugin in plugins:
|
||||
result = current_plugin.scan(barcode)
|
||||
try:
|
||||
result = current_plugin.scan(barcode)
|
||||
except Exception:
|
||||
log_error('BarcodeView.scan_barcode')
|
||||
continue
|
||||
|
||||
if result is None:
|
||||
continue
|
||||
|
||||
if len(result) == 0:
|
||||
continue
|
||||
|
||||
if 'error' in result:
|
||||
logger.info(
|
||||
'%s.scan(...) returned an error: %s',
|
||||
@ -166,6 +173,7 @@ class BarcodeView(CreateAPIView):
|
||||
plugin = current_plugin
|
||||
response = result
|
||||
else:
|
||||
# Return the first successful match
|
||||
plugin = current_plugin
|
||||
response = result
|
||||
break
|
||||
@ -280,6 +288,8 @@ class BarcodeAssign(BarcodeView):
|
||||
result['plugin'] = inventree_barcode_plugin.name
|
||||
result['barcode_data'] = barcode
|
||||
|
||||
result.pop('success', None)
|
||||
|
||||
raise ValidationError(result)
|
||||
|
||||
barcode_hash = hash_barcode(barcode)
|
||||
@ -497,8 +507,11 @@ class BarcodePOReceive(BarcodeView):
|
||||
logger.debug("BarcodePOReceive: scanned barcode - '%s'", barcode)
|
||||
|
||||
# Extract optional fields from the dataset
|
||||
supplier = kwargs.get('supplier')
|
||||
purchase_order = kwargs.get('purchase_order')
|
||||
location = kwargs.get('location')
|
||||
line_item = kwargs.get('line_item')
|
||||
auto_allocate = kwargs.get('auto_allocate', True)
|
||||
|
||||
# Extract location from PurchaseOrder, if available
|
||||
if not location and purchase_order:
|
||||
@ -532,9 +545,19 @@ class BarcodePOReceive(BarcodeView):
|
||||
plugin_response = None
|
||||
|
||||
for current_plugin in plugins:
|
||||
result = current_plugin.scan_receive_item(
|
||||
barcode, request.user, purchase_order=purchase_order, location=location
|
||||
)
|
||||
try:
|
||||
result = current_plugin.scan_receive_item(
|
||||
barcode,
|
||||
request.user,
|
||||
supplier=supplier,
|
||||
purchase_order=purchase_order,
|
||||
location=location,
|
||||
line_item=line_item,
|
||||
auto_allocate=auto_allocate,
|
||||
)
|
||||
except Exception:
|
||||
log_error('BarcodePOReceive.handle_barcode')
|
||||
continue
|
||||
|
||||
if result is None:
|
||||
continue
|
||||
@ -560,7 +583,7 @@ class BarcodePOReceive(BarcodeView):
|
||||
|
||||
# A plugin has not been found!
|
||||
if plugin is None:
|
||||
response['error'] = _('No match for supplier barcode')
|
||||
response['error'] = _('No plugin match for supplier barcode')
|
||||
|
||||
self.log_scan(request, response, 'success' in response)
|
||||
|
||||
|
@ -3,17 +3,17 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from decimal import Decimal, InvalidOperation
|
||||
|
||||
from django.contrib.auth.models import User
|
||||
from django.db.models import F, Q
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.db.models import Q
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
|
||||
from company.models import Company, SupplierPart
|
||||
from company.models import Company, ManufacturerPart, SupplierPart
|
||||
from InvenTree.exceptions import log_error
|
||||
from InvenTree.models import InvenTreeBarcodeMixin
|
||||
from order.models import PurchaseOrder, PurchaseOrderStatus
|
||||
from order.models import PurchaseOrder
|
||||
from part.models import Part
|
||||
from plugin.base.integration.SettingsMixin import SettingsMixin
|
||||
from stock.models import StockLocation
|
||||
|
||||
logger = logging.getLogger('inventree')
|
||||
|
||||
@ -112,6 +112,11 @@ class SupplierBarcodeMixin(BarcodeMixin):
|
||||
|
||||
return fields.get(key, backup_value)
|
||||
|
||||
def get_part(self) -> Part | None:
|
||||
"""Extract the Part object from the barcode fields."""
|
||||
# TODO: Implement this
|
||||
return None
|
||||
|
||||
@property
|
||||
def quantity(self):
|
||||
"""Return the quantity from the barcode fields."""
|
||||
@ -122,11 +127,81 @@ class SupplierBarcodeMixin(BarcodeMixin):
|
||||
"""Return the supplier part number from the barcode fields."""
|
||||
return self.get_field_value(self.SUPPLIER_PART_NUMBER)
|
||||
|
||||
def get_supplier_part(self) -> SupplierPart | None:
|
||||
"""Return the SupplierPart object for the scanned barcode.
|
||||
|
||||
Returns:
|
||||
SupplierPart object or None
|
||||
|
||||
- Filter by the Supplier ID associated with the plugin
|
||||
- Filter by SKU (if available)
|
||||
- If more than one match is found, filter by MPN (if available)
|
||||
|
||||
"""
|
||||
sku = self.supplier_part_number
|
||||
mpn = self.manufacturer_part_number
|
||||
|
||||
# Require at least SKU or MPN for lookup
|
||||
if not sku and not mpn:
|
||||
return None
|
||||
|
||||
supplier_parts = SupplierPart.objects.all()
|
||||
|
||||
# Filter by supplier
|
||||
if supplier := self.get_supplier(cache=True):
|
||||
supplier_parts = supplier_parts.filter(supplier=supplier)
|
||||
|
||||
if sku:
|
||||
supplier_parts = supplier_parts.filter(SKU=sku)
|
||||
|
||||
# Attempt additional filtering by MPN if multiple matches are found
|
||||
if mpn and supplier_parts.count() > 1:
|
||||
manufacturer_parts = ManufacturerPart.objects.filter(MPN=mpn)
|
||||
if manufacturer_parts.count() > 0:
|
||||
supplier_parts = supplier_parts.filter(
|
||||
manufacturer_part__in=manufacturer_parts
|
||||
)
|
||||
|
||||
# Requires a unique match
|
||||
if len(supplier_parts) == 1:
|
||||
return supplier_parts.first()
|
||||
|
||||
@property
|
||||
def manufacturer_part_number(self):
|
||||
"""Return the manufacturer part number from the barcode fields."""
|
||||
return self.get_field_value(self.MANUFACTURER_PART_NUMBER)
|
||||
|
||||
def get_manufacturer_part(self) -> ManufacturerPart | None:
|
||||
"""Return the ManufacturerPart object for the scanned barcode.
|
||||
|
||||
Returns:
|
||||
ManufacturerPart object or None
|
||||
"""
|
||||
mpn = self.manufacturer_part_number
|
||||
|
||||
if not mpn:
|
||||
return None
|
||||
|
||||
parts = ManufacturerPart.objects.filter(MPN=mpn)
|
||||
|
||||
if supplier := self.get_supplier(cache=True):
|
||||
# Manufacturer part must be associated with the supplier
|
||||
# Case 1: Manufactured by this supplier
|
||||
q1 = Q(manufacturer=supplier)
|
||||
# Case 2: Supplied by this supplier
|
||||
m = (
|
||||
SupplierPart.objects.filter(supplier=supplier)
|
||||
.values_list('manufacturer_part', flat=True)
|
||||
.distinct()
|
||||
)
|
||||
q2 = Q(pk__in=m)
|
||||
|
||||
parts = parts.filter(q1 | q2).distinct()
|
||||
|
||||
# Requires a unique match
|
||||
if len(parts) == 1:
|
||||
return parts.first()
|
||||
|
||||
@property
|
||||
def customer_order_number(self):
|
||||
"""Return the customer order number from the barcode fields."""
|
||||
@ -137,7 +212,38 @@ class SupplierBarcodeMixin(BarcodeMixin):
|
||||
"""Return the supplier order number from the barcode fields."""
|
||||
return self.get_field_value(self.SUPPLIER_ORDER_NUMBER)
|
||||
|
||||
def extract_barcode_fields(self, barcode_data) -> dict[str, str]:
|
||||
def get_purchase_order(self) -> PurchaseOrder | None:
|
||||
"""Extract the PurchaseOrder object from the barcode fields.
|
||||
|
||||
Inspect the customer_order_number and supplier_order_number fields,
|
||||
and try to find a matching PurchaseOrder object.
|
||||
|
||||
Returns:
|
||||
PurchaseOrder object or None
|
||||
"""
|
||||
customer_order_number = self.customer_order_number
|
||||
supplier_order_number = self.supplier_order_number
|
||||
|
||||
if not (customer_order_number or supplier_order_number):
|
||||
return None
|
||||
|
||||
# First, attempt lookup based on the customer_order_number
|
||||
|
||||
if customer_order_number:
|
||||
orders = PurchaseOrder.objects.filter(reference=customer_order_number)
|
||||
elif supplier_order_number:
|
||||
orders = PurchaseOrder.objects.filter(
|
||||
supplier_reference=supplier_order_number
|
||||
)
|
||||
|
||||
if supplier := self.get_supplier(cache=True):
|
||||
orders = orders.filter(supplier=supplier)
|
||||
|
||||
# Requires a unique match
|
||||
if len(orders) == 1:
|
||||
return orders.first()
|
||||
|
||||
def extract_barcode_fields(self, barcode_data: str) -> dict[str, str]:
|
||||
"""Method to extract barcode fields from barcode data.
|
||||
|
||||
This method should return a dict object where the keys are the field names,
|
||||
@ -153,93 +259,177 @@ class SupplierBarcodeMixin(BarcodeMixin):
|
||||
'extract_barcode_fields must be implemented by each plugin'
|
||||
)
|
||||
|
||||
def scan(self, barcode_data):
|
||||
"""Try to match a supplier barcode to a supplier part."""
|
||||
def scan(self, barcode_data: str) -> dict:
|
||||
"""Perform a generic 'scan' operation on a supplier barcode.
|
||||
|
||||
The supplier barcode may provide sufficient information to match against
|
||||
one of the following model types:
|
||||
|
||||
- SupplierPart
|
||||
- ManufacturerPart
|
||||
- PurchaseOrder
|
||||
- PurchaseOrderLineItem (todo)
|
||||
- StockItem (todo)
|
||||
- Part (todo)
|
||||
|
||||
If any matches are made, return a dict object containing the relevant information.
|
||||
"""
|
||||
barcode_data = str(barcode_data).strip()
|
||||
|
||||
self.barcode_fields = self.extract_barcode_fields(barcode_data)
|
||||
|
||||
if self.supplier_part_number is None and self.manufacturer_part_number is None:
|
||||
return None
|
||||
|
||||
supplier_parts = self.get_supplier_parts(
|
||||
sku=self.supplier_part_number,
|
||||
mpn=self.manufacturer_part_number,
|
||||
supplier=self.get_supplier(),
|
||||
)
|
||||
|
||||
if len(supplier_parts) > 1:
|
||||
return {'error': _('Found multiple matching supplier parts for barcode')}
|
||||
elif not supplier_parts:
|
||||
return None
|
||||
|
||||
supplier_part = supplier_parts[0]
|
||||
|
||||
data = {
|
||||
'pk': supplier_part.pk,
|
||||
'api_url': f'{SupplierPart.get_api_url()}{supplier_part.pk}/',
|
||||
'web_url': supplier_part.get_absolute_url(),
|
||||
# Generate possible matches for this barcode
|
||||
# Note: Each of these functions can be overridden by the plugin (if necessary)
|
||||
matches = {
|
||||
Part.barcode_model_type(): self.get_part(),
|
||||
PurchaseOrder.barcode_model_type(): self.get_purchase_order(),
|
||||
SupplierPart.barcode_model_type(): self.get_supplier_part(),
|
||||
ManufacturerPart.barcode_model_type(): self.get_manufacturer_part(),
|
||||
}
|
||||
|
||||
return {SupplierPart.barcode_model_type(): data}
|
||||
data = {}
|
||||
|
||||
def scan_receive_item(self, barcode_data, user, purchase_order=None, location=None):
|
||||
"""Try to scan a supplier barcode to receive a purchase order item."""
|
||||
# At least one matching item was found
|
||||
has_match = False
|
||||
|
||||
for k, v in matches.items():
|
||||
if v and hasattr(v, 'pk'):
|
||||
has_match = True
|
||||
data[k] = v.format_matched_response()
|
||||
|
||||
if not has_match:
|
||||
return None
|
||||
|
||||
# Add in supplier information (if available)
|
||||
if supplier := self.get_supplier():
|
||||
data['company'] = {'pk': supplier.pk}
|
||||
|
||||
data['success'] = _('Found matching item')
|
||||
|
||||
return data
|
||||
|
||||
def scan_receive_item(
|
||||
self,
|
||||
barcode_data: str,
|
||||
user,
|
||||
supplier=None,
|
||||
line_item=None,
|
||||
purchase_order=None,
|
||||
location=None,
|
||||
auto_allocate: bool = True,
|
||||
**kwargs,
|
||||
) -> dict | None:
|
||||
"""Attempt to receive an item against a PurchaseOrder via barcode scanning.
|
||||
|
||||
Arguments:
|
||||
barcode_data: The raw barcode data
|
||||
user: The User performing the action
|
||||
supplier: The Company object to receive against (or None)
|
||||
purchase_order: The PurchaseOrder object to receive against (or None)
|
||||
line_item: The PurchaseOrderLineItem object to receive against (or None)
|
||||
location: The StockLocation object to receive into (or None)
|
||||
auto_allocate: If True, automatically receive the item (if possible)
|
||||
|
||||
Returns:
|
||||
A dict object containing the result of the action.
|
||||
|
||||
The more "context" data that can be provided, the better the chances of a successful match.
|
||||
"""
|
||||
barcode_data = str(barcode_data).strip()
|
||||
|
||||
self.barcode_fields = self.extract_barcode_fields(barcode_data)
|
||||
|
||||
if self.supplier_part_number is None and self.manufacturer_part_number is None:
|
||||
# Extract supplier information
|
||||
supplier = supplier or self.get_supplier(cache=True)
|
||||
|
||||
if not supplier:
|
||||
# No supplier information available
|
||||
return None
|
||||
|
||||
supplier = self.get_supplier()
|
||||
# Extract purchase order information
|
||||
purchase_order = purchase_order or self.get_purchase_order()
|
||||
|
||||
supplier_parts = self.get_supplier_parts(
|
||||
sku=self.supplier_part_number,
|
||||
mpn=self.manufacturer_part_number,
|
||||
supplier=supplier,
|
||||
if not purchase_order or purchase_order.supplier != supplier:
|
||||
# Purchase order does not match supplier
|
||||
return None
|
||||
|
||||
supplier_part = self.get_supplier_part()
|
||||
|
||||
if not supplier_part:
|
||||
# No supplier part information available
|
||||
return None
|
||||
|
||||
# Attempt to find matching line item
|
||||
if not line_item:
|
||||
line_items = purchase_order.lines.filter(part=supplier_part)
|
||||
if line_items.count() == 1:
|
||||
line_item = line_items.first()
|
||||
|
||||
if not line_item:
|
||||
# No line item information available
|
||||
return None
|
||||
|
||||
if line_item.part != supplier_part:
|
||||
return {'error': _('Supplier part does not match line item')}
|
||||
|
||||
if line_item.is_completed():
|
||||
return {'error': _('Line item is already completed')}
|
||||
|
||||
# Extract location information for the line item
|
||||
location = (
|
||||
location
|
||||
or line_item.destination
|
||||
or purchase_order.destination
|
||||
or line_item.part.part.get_default_location()
|
||||
)
|
||||
|
||||
if len(supplier_parts) > 1:
|
||||
return {'error': _('Found multiple matching supplier parts for barcode')}
|
||||
elif not supplier_parts:
|
||||
return None
|
||||
# Extract quantity information
|
||||
quantity = self.quantity
|
||||
|
||||
supplier_part = supplier_parts[0]
|
||||
# At this stage, we *should* have enough information to attempt to receive the item
|
||||
# If auto_allocate is True, attempt to receive the item automatically
|
||||
# Otherwise, return the required information to the client
|
||||
action_required = not auto_allocate or location is None or quantity is None
|
||||
|
||||
# If a purchase order is not provided, extract it from the provided data
|
||||
if not purchase_order:
|
||||
matching_orders = self.get_purchase_orders(
|
||||
self.customer_order_number,
|
||||
self.supplier_order_number,
|
||||
supplier=supplier,
|
||||
if quantity is None:
|
||||
quantity = line_item.remaining()
|
||||
|
||||
quantity = float(quantity)
|
||||
|
||||
# Construct a response object
|
||||
response = {
|
||||
'lineitem': {
|
||||
'pk': line_item.pk,
|
||||
'quantity': quantity,
|
||||
'supplier_part': supplier_part.pk,
|
||||
'purchase_order': purchase_order.pk,
|
||||
'location': location.pk if location else None,
|
||||
}
|
||||
}
|
||||
|
||||
if action_required:
|
||||
# Further information is required to receive the item
|
||||
response['action_required'] = _(
|
||||
'Further information required to receive line item'
|
||||
)
|
||||
else:
|
||||
# Use the information we have to attempt to receive the item into stock
|
||||
try:
|
||||
purchase_order.receive_line_item(
|
||||
line_item, location, quantity, user, barcode=barcode_data
|
||||
)
|
||||
response['success'] = _('Received purchase order line item')
|
||||
except ValidationError as e:
|
||||
# Pass a ValidationError back to the client
|
||||
response['error'] = e.message
|
||||
except Exception:
|
||||
# Handle any other exceptions
|
||||
log_error('scan_receive_item')
|
||||
response['error'] = _('Failed to receive line item')
|
||||
|
||||
order = self.customer_order_number or self.supplier_order_number
|
||||
return response
|
||||
|
||||
if len(matching_orders) > 1:
|
||||
return {
|
||||
'error': _(f"Found multiple purchase orders matching '{order}'")
|
||||
}
|
||||
|
||||
if len(matching_orders) == 0:
|
||||
return {'error': _(f"No matching purchase order for '{order}'")}
|
||||
|
||||
purchase_order = matching_orders.first()
|
||||
|
||||
if supplier and purchase_order and purchase_order.supplier != supplier:
|
||||
return {'error': _('Purchase order does not match supplier')}
|
||||
|
||||
return self.receive_purchase_order_item(
|
||||
supplier_part,
|
||||
user,
|
||||
quantity=self.quantity,
|
||||
purchase_order=purchase_order,
|
||||
location=location,
|
||||
barcode=barcode_data,
|
||||
)
|
||||
|
||||
def get_supplier(self) -> Company | None:
|
||||
def get_supplier(self, cache: bool = False) -> Company | None:
|
||||
"""Get the supplier for the SUPPLIER_ID set in the plugin settings.
|
||||
|
||||
If it's not defined, try to guess it and set it if possible.
|
||||
@ -247,29 +437,32 @@ class SupplierBarcodeMixin(BarcodeMixin):
|
||||
if not isinstance(self, SettingsMixin):
|
||||
return None
|
||||
|
||||
def _cache_supplier(supplier):
|
||||
"""Cache and return the supplier object."""
|
||||
if cache:
|
||||
self._supplier = supplier
|
||||
return supplier
|
||||
|
||||
# Cache the supplier object, so we don't have to look it up every time
|
||||
if cache and hasattr(self, '_supplier'):
|
||||
return self._supplier
|
||||
|
||||
if supplier_pk := self.get_setting('SUPPLIER_ID'):
|
||||
try:
|
||||
return Company.objects.get(pk=supplier_pk)
|
||||
except Company.DoesNotExist:
|
||||
logger.error(
|
||||
'No company with pk %d (set "SUPPLIER_ID" setting to a valid value)',
|
||||
supplier_pk,
|
||||
)
|
||||
return None
|
||||
return _cache_supplier(Company.objects.filter(pk=supplier_pk).first())
|
||||
|
||||
if not (supplier_name := getattr(self, 'DEFAULT_SUPPLIER_NAME', None)):
|
||||
return None
|
||||
return _cache_supplier(None)
|
||||
|
||||
suppliers = Company.objects.filter(
|
||||
name__icontains=supplier_name, is_supplier=True
|
||||
)
|
||||
|
||||
if len(suppliers) != 1:
|
||||
return None
|
||||
return _cache_supplier(None)
|
||||
|
||||
self.set_setting('SUPPLIER_ID', suppliers.first().pk)
|
||||
|
||||
return suppliers.first()
|
||||
return _cache_supplier(suppliers.first())
|
||||
|
||||
@classmethod
|
||||
def ecia_field_map(cls):
|
||||
@ -358,150 +551,3 @@ class SupplierBarcodeMixin(BarcodeMixin):
|
||||
return SupplierBarcodeMixin.split_fields(
|
||||
barcode_data, delimiter=DELIMITER, header=HEADER, trailer=TRAILER
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_purchase_orders(
|
||||
customer_order_number, supplier_order_number, supplier: Company = None
|
||||
):
|
||||
"""Attempt to find a purchase order from the extracted customer and supplier order numbers."""
|
||||
orders = PurchaseOrder.objects.filter(status=PurchaseOrderStatus.PLACED.value)
|
||||
|
||||
if supplier:
|
||||
orders = orders.filter(supplier=supplier)
|
||||
|
||||
# this works because reference and supplier_reference are not nullable, so if
|
||||
# customer_order_number or supplier_order_number is None, the query won't return anything
|
||||
reference_filter = Q(reference__iexact=customer_order_number)
|
||||
supplier_reference_filter = Q(supplier_reference__iexact=supplier_order_number)
|
||||
|
||||
orders_union = orders.filter(reference_filter | supplier_reference_filter)
|
||||
if orders_union.count() == 1:
|
||||
return orders_union
|
||||
else:
|
||||
orders_intersection = orders.filter(
|
||||
reference_filter & supplier_reference_filter
|
||||
)
|
||||
return orders_intersection if orders_intersection else orders_union
|
||||
|
||||
@staticmethod
|
||||
def get_supplier_parts(
|
||||
sku: str | None = None, supplier: Company = None, mpn: str | None = None
|
||||
):
|
||||
"""Get a supplier part from SKU or by supplier and MPN."""
|
||||
if not (sku or supplier or mpn):
|
||||
return SupplierPart.objects.none()
|
||||
|
||||
supplier_parts = SupplierPart.objects.all()
|
||||
|
||||
if sku:
|
||||
supplier_parts = supplier_parts.filter(SKU__iexact=sku)
|
||||
if len(supplier_parts) == 1:
|
||||
return supplier_parts
|
||||
|
||||
if supplier:
|
||||
supplier_parts = supplier_parts.filter(supplier=supplier.pk)
|
||||
if len(supplier_parts) == 1:
|
||||
return supplier_parts
|
||||
|
||||
if mpn:
|
||||
supplier_parts = supplier_parts.filter(manufacturer_part__MPN__iexact=mpn)
|
||||
if len(supplier_parts) == 1:
|
||||
return supplier_parts
|
||||
|
||||
logger.warning(
|
||||
"Found %d supplier parts for SKU '%s', supplier '%s', MPN '%s'",
|
||||
supplier_parts.count(),
|
||||
sku,
|
||||
supplier.name if supplier else None,
|
||||
mpn,
|
||||
)
|
||||
|
||||
return supplier_parts
|
||||
|
||||
@staticmethod
|
||||
def receive_purchase_order_item(
|
||||
supplier_part: SupplierPart,
|
||||
user: User,
|
||||
quantity: Decimal | str | None = None,
|
||||
purchase_order: PurchaseOrder = None,
|
||||
location: StockLocation = None,
|
||||
barcode: str | None = None,
|
||||
) -> dict:
|
||||
"""Try to receive a purchase order item.
|
||||
|
||||
Returns:
|
||||
A dict object containing:
|
||||
- on success: a "success" message
|
||||
- on partial success: the "lineitem" with quantity and location (both can be None)
|
||||
- on failure: an "error" message
|
||||
"""
|
||||
if quantity:
|
||||
try:
|
||||
quantity = Decimal(quantity)
|
||||
except InvalidOperation:
|
||||
logger.warning("Failed to parse quantity '%s'", quantity)
|
||||
quantity = None
|
||||
|
||||
# find incomplete line_items that match the supplier_part
|
||||
line_items = purchase_order.lines.filter(
|
||||
part=supplier_part.pk, quantity__gt=F('received')
|
||||
)
|
||||
if len(line_items) == 1 or not quantity:
|
||||
line_item = line_items[0]
|
||||
else:
|
||||
# if there are multiple line items and the barcode contains a quantity:
|
||||
# 1. return the first line_item where line_item.quantity == quantity
|
||||
# 2. return the first line_item where line_item.quantity > quantity
|
||||
# 3. return the first line_item
|
||||
for line_item in line_items:
|
||||
if line_item.quantity == quantity:
|
||||
break
|
||||
else:
|
||||
for line_item in line_items:
|
||||
if line_item.quantity > quantity:
|
||||
break
|
||||
else:
|
||||
line_item = line_items.first()
|
||||
|
||||
if not line_item:
|
||||
return {'error': _('Failed to find pending line item for supplier part')}
|
||||
|
||||
no_stock_locations = False
|
||||
if not location:
|
||||
# try to guess the destination were the stock_part should go
|
||||
# 1. check if it's defined on the line_item
|
||||
# 2. check if it's defined on the part
|
||||
# 3. check if there's 1 or 0 stock locations defined in InvenTree
|
||||
# -> assume all stock is going into that location (or no location)
|
||||
if (location := line_item.destination) or (
|
||||
location := supplier_part.part.get_default_location()
|
||||
):
|
||||
pass
|
||||
elif StockLocation.objects.count() <= 1:
|
||||
if not (location := StockLocation.objects.first()):
|
||||
no_stock_locations = True
|
||||
|
||||
response = {
|
||||
'lineitem': {'pk': line_item.pk, 'purchase_order': purchase_order.pk}
|
||||
}
|
||||
|
||||
if quantity:
|
||||
response['lineitem']['quantity'] = quantity
|
||||
if location:
|
||||
response['lineitem']['location'] = location.pk
|
||||
|
||||
# if either the quantity is missing or no location is defined/found
|
||||
# -> return the line_item found, so the client can gather the missing
|
||||
# information and complete the action with an 'api-po-receive' call
|
||||
if not quantity or (not location and not no_stock_locations):
|
||||
response['action_required'] = _(
|
||||
'Further information required to receive line item'
|
||||
)
|
||||
return response
|
||||
|
||||
purchase_order.receive_line_item(
|
||||
line_item, location, quantity, user, barcode=barcode
|
||||
)
|
||||
|
||||
response['success'] = _('Received purchase order line item')
|
||||
return response
|
||||
|
@ -6,6 +6,7 @@ from django.utils.translation import gettext_lazy as _
|
||||
from rest_framework import serializers
|
||||
|
||||
import common.models
|
||||
import company.models
|
||||
import order.models
|
||||
import plugin.base.barcodes.helper
|
||||
import stock.models
|
||||
@ -149,6 +150,13 @@ class BarcodePOReceiveSerializer(BarcodeSerializer):
|
||||
- location: Location to receive items into
|
||||
"""
|
||||
|
||||
supplier = serializers.PrimaryKeyRelatedField(
|
||||
queryset=company.models.Company.objects.all(),
|
||||
required=False,
|
||||
allow_null=True,
|
||||
help_text=_('Supplier to receive items from'),
|
||||
)
|
||||
|
||||
purchase_order = serializers.PrimaryKeyRelatedField(
|
||||
queryset=order.models.PurchaseOrder.objects.all(),
|
||||
required=False,
|
||||
@ -177,6 +185,19 @@ class BarcodePOReceiveSerializer(BarcodeSerializer):
|
||||
|
||||
return location
|
||||
|
||||
line_item = serializers.PrimaryKeyRelatedField(
|
||||
queryset=order.models.PurchaseOrderLineItem.objects.all(),
|
||||
required=False,
|
||||
allow_null=True,
|
||||
help_text=_('Purchase order line item to receive items against'),
|
||||
)
|
||||
|
||||
auto_allocate = serializers.BooleanField(
|
||||
required=False,
|
||||
default=True,
|
||||
help_text=_('Automatically allocate stock items to the purchase order'),
|
||||
)
|
||||
|
||||
|
||||
class BarcodeSOAllocateSerializer(BarcodeSerializer):
|
||||
"""Serializr for allocating stock items to a sales order.
|
||||
|
@ -10,8 +10,6 @@ from InvenTree.unit_test import InvenTreeAPITestCase
|
||||
from part.models import Part
|
||||
from stock.models import StockItem
|
||||
|
||||
from .mixins import SupplierBarcodeMixin
|
||||
|
||||
|
||||
class BarcodeAPITest(InvenTreeAPITestCase):
|
||||
"""Tests for barcode api."""
|
||||
@ -390,142 +388,3 @@ class SOAllocateTest(InvenTreeAPITestCase):
|
||||
self.line_item.refresh_from_db()
|
||||
self.assertEqual(self.line_item.allocated_quantity(), 10)
|
||||
self.assertTrue(self.line_item.is_fully_allocated())
|
||||
|
||||
|
||||
class SupplierBarcodeMixinTest(InvenTreeAPITestCase):
|
||||
"""Unit tests for the SupplierBarcodeMixin class."""
|
||||
|
||||
@classmethod
|
||||
def setUpTestData(cls):
|
||||
"""Setup for all tests."""
|
||||
super().setUpTestData()
|
||||
|
||||
cls.supplier = company.models.Company.objects.create(
|
||||
name='Supplier Barcode Mixin Test Company', is_supplier=True
|
||||
)
|
||||
|
||||
cls.supplier_other = company.models.Company.objects.create(
|
||||
name='Other Supplier Barcode Mixin Test Company', is_supplier=True
|
||||
)
|
||||
|
||||
cls.supplier_no_orders = company.models.Company.objects.create(
|
||||
name='Supplier Barcode Mixin Test Company with no Orders', is_supplier=True
|
||||
)
|
||||
|
||||
cls.purchase_order_pending = order.models.PurchaseOrder.objects.create(
|
||||
status=order.models.PurchaseOrderStatus.PENDING.value,
|
||||
supplier=cls.supplier,
|
||||
supplier_reference='ORDER#1337',
|
||||
)
|
||||
|
||||
cls.purchase_order_1 = order.models.PurchaseOrder.objects.create(
|
||||
status=order.models.PurchaseOrderStatus.PLACED.value,
|
||||
supplier=cls.supplier,
|
||||
supplier_reference='ORDER#1338',
|
||||
)
|
||||
|
||||
cls.purchase_order_duplicate_1 = order.models.PurchaseOrder.objects.create(
|
||||
status=order.models.PurchaseOrderStatus.PLACED.value,
|
||||
supplier=cls.supplier,
|
||||
supplier_reference='ORDER#1339',
|
||||
)
|
||||
|
||||
cls.purchase_order_duplicate_2 = order.models.PurchaseOrder.objects.create(
|
||||
status=order.models.PurchaseOrderStatus.PLACED.value,
|
||||
supplier=cls.supplier_other,
|
||||
supplier_reference='ORDER#1339',
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
"""Setup method for each test."""
|
||||
super().setUp()
|
||||
|
||||
def test_order_not_placed(self):
|
||||
"""Check that purchase order which has not been placed doesn't get returned."""
|
||||
purchase_orders = SupplierBarcodeMixin.get_purchase_orders(
|
||||
self.purchase_order_pending.reference, None
|
||||
)
|
||||
self.assertIsNone(purchase_orders.first())
|
||||
|
||||
purchase_orders = SupplierBarcodeMixin.get_purchase_orders(
|
||||
None, self.purchase_order_pending.supplier_reference
|
||||
)
|
||||
self.assertIsNone(purchase_orders.first())
|
||||
|
||||
def test_order_simple(self):
|
||||
"""Check that we can get a purchase order by either reference, supplier_reference or both."""
|
||||
purchase_orders = SupplierBarcodeMixin.get_purchase_orders(
|
||||
self.purchase_order_1.reference, None
|
||||
)
|
||||
self.assertEqual(purchase_orders.count(), 1)
|
||||
self.assertEqual(purchase_orders.first(), self.purchase_order_1)
|
||||
|
||||
purchase_orders = SupplierBarcodeMixin.get_purchase_orders(
|
||||
None, self.purchase_order_1.supplier_reference
|
||||
)
|
||||
self.assertEqual(purchase_orders.count(), 1)
|
||||
self.assertEqual(purchase_orders.first(), self.purchase_order_1)
|
||||
|
||||
purchase_orders = SupplierBarcodeMixin.get_purchase_orders(
|
||||
self.purchase_order_1.reference, self.purchase_order_1.supplier_reference
|
||||
)
|
||||
self.assertEqual(purchase_orders.count(), 1)
|
||||
self.assertEqual(purchase_orders.first(), self.purchase_order_1)
|
||||
|
||||
purchase_orders = SupplierBarcodeMixin.get_purchase_orders(
|
||||
self.purchase_order_1.reference,
|
||||
self.purchase_order_1.supplier_reference,
|
||||
supplier=self.supplier,
|
||||
)
|
||||
self.assertEqual(purchase_orders.count(), 1)
|
||||
self.assertEqual(purchase_orders.first(), self.purchase_order_1)
|
||||
|
||||
def test_wrong_supplier_order(self):
|
||||
"""Check that no orders get returned if the wrong supplier is specified."""
|
||||
purchase_orders = SupplierBarcodeMixin.get_purchase_orders(
|
||||
self.purchase_order_1.reference, None, supplier=self.supplier_no_orders
|
||||
)
|
||||
self.assertIsNone(purchase_orders.first())
|
||||
|
||||
purchase_orders = SupplierBarcodeMixin.get_purchase_orders(
|
||||
None,
|
||||
self.purchase_order_1.supplier_reference,
|
||||
supplier=self.supplier_no_orders,
|
||||
)
|
||||
self.assertIsNone(purchase_orders.first())
|
||||
|
||||
def test_supplier_order_duplicate(self):
|
||||
"""Test getting purchase_orders with the same supplier_reference works correctly."""
|
||||
purchase_orders = SupplierBarcodeMixin.get_purchase_orders(
|
||||
None, self.purchase_order_duplicate_1.supplier_reference
|
||||
)
|
||||
self.assertEqual(purchase_orders.count(), 2)
|
||||
self.assertEqual(
|
||||
set(purchase_orders),
|
||||
{self.purchase_order_duplicate_1, self.purchase_order_duplicate_2},
|
||||
)
|
||||
|
||||
purchase_orders = SupplierBarcodeMixin.get_purchase_orders(
|
||||
self.purchase_order_duplicate_1.reference,
|
||||
self.purchase_order_duplicate_1.supplier_reference,
|
||||
)
|
||||
self.assertEqual(purchase_orders.count(), 1)
|
||||
self.assertEqual(purchase_orders.first(), self.purchase_order_duplicate_1)
|
||||
|
||||
# check that mixing the reference and supplier_reference doesn't work
|
||||
|
||||
purchase_orders = SupplierBarcodeMixin.get_purchase_orders(
|
||||
self.purchase_order_duplicate_1.supplier_reference,
|
||||
self.purchase_order_duplicate_1.reference,
|
||||
)
|
||||
self.assertIsNone(purchase_orders.first())
|
||||
|
||||
# check that specifying the right supplier works
|
||||
|
||||
purchase_orders = SupplierBarcodeMixin.get_purchase_orders(
|
||||
None,
|
||||
self.purchase_order_duplicate_1.supplier_reference,
|
||||
supplier=self.supplier_other,
|
||||
)
|
||||
self.assertEqual(purchase_orders.count(), 1)
|
||||
self.assertEqual(purchase_orders.first(), self.purchase_order_duplicate_2)
|
||||
|
@ -1,10 +1,11 @@
|
||||
"""API for location plugins."""
|
||||
|
||||
from rest_framework import permissions, serializers
|
||||
from rest_framework.exceptions import NotFound, ParseError
|
||||
from rest_framework.exceptions import NotFound, ParseError, ValidationError
|
||||
from rest_framework.generics import GenericAPIView
|
||||
from rest_framework.response import Response
|
||||
|
||||
from InvenTree.exceptions import log_error
|
||||
from InvenTree.tasks import offload_task
|
||||
from plugin.registry import call_plugin_function, registry
|
||||
from stock.models import StockItem, StockLocation
|
||||
@ -72,6 +73,9 @@ class LocatePluginView(GenericAPIView):
|
||||
|
||||
except (ValueError, StockItem.DoesNotExist):
|
||||
raise NotFound(f"StockItem matching PK '{item_pk}' not found")
|
||||
except Exception:
|
||||
log_error('locate_stock_item')
|
||||
return ValidationError('Error locating stock item')
|
||||
|
||||
elif location_pk:
|
||||
try:
|
||||
@ -91,6 +95,8 @@ class LocatePluginView(GenericAPIView):
|
||||
|
||||
except (ValueError, StockLocation.DoesNotExist):
|
||||
raise NotFound(f"StockLocation matching PK '{location_pk}' not found")
|
||||
|
||||
except Exception:
|
||||
log_error('locate_stock_location')
|
||||
return ValidationError('Error locating stock location')
|
||||
else:
|
||||
raise ParseError("Must supply either 'item' or 'location' parameter")
|
||||
|
@ -98,6 +98,8 @@ class InvenTreeInternalBarcodePlugin(SettingsMixin, BarcodeMixin, InvenTreePlugi
|
||||
|
||||
supported_models = plugin.base.barcodes.helper.get_supported_barcode_models()
|
||||
|
||||
succcess_message = _('Found matching item')
|
||||
|
||||
if barcode_dict is not None and type(barcode_dict) is dict:
|
||||
# Look for various matches. First good match will be returned
|
||||
for model in supported_models:
|
||||
@ -107,7 +109,11 @@ class InvenTreeInternalBarcodePlugin(SettingsMixin, BarcodeMixin, InvenTreePlugi
|
||||
try:
|
||||
pk = int(barcode_dict[label])
|
||||
instance = model.objects.get(pk=pk)
|
||||
return self.format_matched_response(label, model, instance)
|
||||
|
||||
return {
|
||||
**self.format_matched_response(label, model, instance),
|
||||
'success': succcess_message,
|
||||
}
|
||||
except (ValueError, model.DoesNotExist):
|
||||
pass
|
||||
|
||||
@ -122,7 +128,10 @@ class InvenTreeInternalBarcodePlugin(SettingsMixin, BarcodeMixin, InvenTreePlugi
|
||||
instance = model.lookup_barcode(barcode_hash)
|
||||
|
||||
if instance is not None:
|
||||
return self.format_matched_response(label, model, instance)
|
||||
return {
|
||||
**self.format_matched_response(label, model, instance),
|
||||
'success': succcess_message,
|
||||
}
|
||||
|
||||
def generate(self, model_instance: InvenTreeBarcodeMixin):
|
||||
"""Generate a barcode for a given model instance."""
|
||||
|
@ -6,6 +6,7 @@ from company.models import Company, ManufacturerPart, SupplierPart
|
||||
from InvenTree.unit_test import InvenTreeAPITestCase
|
||||
from order.models import PurchaseOrder, PurchaseOrderLineItem
|
||||
from part.models import Part
|
||||
from plugin.registry import registry
|
||||
from stock.models import StockItem, StockLocation
|
||||
|
||||
|
||||
@ -32,19 +33,33 @@ class SupplierBarcodeTests(InvenTreeAPITestCase):
|
||||
part=part, manufacturer=manufacturer, MPN='LDK320ADU33R'
|
||||
)
|
||||
|
||||
supplier = Company.objects.create(name='Supplier', is_supplier=True)
|
||||
mouser = Company.objects.create(name='Mouser Test', is_supplier=True)
|
||||
digikey_supplier = Company.objects.create(name='Supplier', is_supplier=True)
|
||||
mouser_supplier = Company.objects.create(name='Mouser Test', is_supplier=True)
|
||||
|
||||
supplier_parts = [
|
||||
SupplierPart(SKU='296-LM358BIDDFRCT-ND', part=part, supplier=supplier),
|
||||
SupplierPart(SKU='1', part=part, manufacturer_part=mpart1, supplier=mouser),
|
||||
SupplierPart(SKU='2', part=part, manufacturer_part=mpart2, supplier=mouser),
|
||||
SupplierPart(SKU='C312270', part=part, supplier=supplier),
|
||||
SupplierPart(SKU='WBP-302', part=part, supplier=supplier),
|
||||
SupplierPart(
|
||||
SKU='296-LM358BIDDFRCT-ND', part=part, supplier=digikey_supplier
|
||||
),
|
||||
SupplierPart(SKU='C312270', part=part, supplier=digikey_supplier),
|
||||
SupplierPart(SKU='WBP-302', part=part, supplier=digikey_supplier),
|
||||
SupplierPart(
|
||||
SKU='1', part=part, manufacturer_part=mpart1, supplier=mouser_supplier
|
||||
),
|
||||
SupplierPart(
|
||||
SKU='2', part=part, manufacturer_part=mpart2, supplier=mouser_supplier
|
||||
),
|
||||
]
|
||||
|
||||
SupplierPart.objects.bulk_create(supplier_parts)
|
||||
|
||||
# Assign supplier information to the plugins
|
||||
# Add supplier information to each custom plugin
|
||||
digikey_plugin = registry.get_plugin('digikeyplugin')
|
||||
digikey_plugin.set_setting('SUPPLIER_ID', digikey_supplier.pk)
|
||||
|
||||
mouser_plugin = registry.get_plugin('mouserplugin')
|
||||
mouser_plugin.set_setting('SUPPLIER_ID', mouser_supplier.pk)
|
||||
|
||||
def test_digikey_barcode(self):
|
||||
"""Test digikey barcode."""
|
||||
result = self.post(
|
||||
@ -63,6 +78,7 @@ class SupplierBarcodeTests(InvenTreeAPITestCase):
|
||||
result = self.post(
|
||||
self.SCAN_URL, data={'barcode': DIGIKEY_BARCODE_2}, expected_code=200
|
||||
)
|
||||
|
||||
self.assertEqual(result.data['plugin'], 'DigiKeyPlugin')
|
||||
|
||||
supplier_part_data = result.data.get('supplierpart')
|
||||
@ -147,8 +163,11 @@ class SupplierBarcodePOReceiveTests(InvenTreeAPITestCase):
|
||||
"""Create supplier part and purchase_order."""
|
||||
super().setUp()
|
||||
|
||||
self.loc_1 = StockLocation.objects.create(name='Location 1')
|
||||
self.loc_2 = StockLocation.objects.create(name='Location 2')
|
||||
|
||||
part = Part.objects.create(name='Test Part', description='Test Part')
|
||||
supplier = Company.objects.create(name='Supplier', is_supplier=True)
|
||||
digikey_supplier = Company.objects.create(name='Supplier', is_supplier=True)
|
||||
manufacturer = Company.objects.create(
|
||||
name='Test Manufacturer', is_manufacturer=True
|
||||
)
|
||||
@ -159,23 +178,29 @@ class SupplierBarcodePOReceiveTests(InvenTreeAPITestCase):
|
||||
)
|
||||
|
||||
self.purchase_order1 = PurchaseOrder.objects.create(
|
||||
supplier_reference='72991337', supplier=supplier
|
||||
supplier_reference='72991337',
|
||||
supplier=digikey_supplier,
|
||||
destination=self.loc_1,
|
||||
)
|
||||
|
||||
supplier_parts1 = [
|
||||
SupplierPart(SKU=f'1_{i}', part=part, supplier=supplier) for i in range(6)
|
||||
SupplierPart(SKU=f'1_{i}', part=part, supplier=digikey_supplier)
|
||||
for i in range(6)
|
||||
]
|
||||
|
||||
supplier_parts1.insert(
|
||||
2, SupplierPart(SKU='296-LM358BIDDFRCT-ND', part=part, supplier=supplier)
|
||||
2,
|
||||
SupplierPart(
|
||||
SKU='296-LM358BIDDFRCT-ND', part=part, supplier=digikey_supplier
|
||||
),
|
||||
)
|
||||
|
||||
for supplier_part in supplier_parts1:
|
||||
supplier_part.save()
|
||||
self.purchase_order1.add_line_item(supplier_part, 8)
|
||||
self.purchase_order1.add_line_item(supplier_part, 8, destination=self.loc_2)
|
||||
|
||||
self.purchase_order2 = PurchaseOrder.objects.create(
|
||||
reference='P0-1337', supplier=mouser
|
||||
reference='P0-1337', supplier=mouser, destination=self.loc_1
|
||||
)
|
||||
|
||||
self.purchase_order2.place_order()
|
||||
@ -190,42 +215,63 @@ class SupplierBarcodePOReceiveTests(InvenTreeAPITestCase):
|
||||
|
||||
for supplier_part in supplier_parts2:
|
||||
supplier_part.save()
|
||||
self.purchase_order2.add_line_item(supplier_part, 5)
|
||||
self.purchase_order2.add_line_item(supplier_part, 5, destination=self.loc_2)
|
||||
|
||||
# Add supplier information to each custom plugin
|
||||
digikey_plugin = registry.get_plugin('digikeyplugin')
|
||||
digikey_plugin.set_setting('SUPPLIER_ID', digikey_supplier.pk)
|
||||
|
||||
mouser_plugin = registry.get_plugin('mouserplugin')
|
||||
mouser_plugin.set_setting('SUPPLIER_ID', mouser.pk)
|
||||
|
||||
def test_receive(self):
|
||||
"""Test receiving an item from a barcode."""
|
||||
url = reverse('api-barcode-po-receive')
|
||||
|
||||
# First attempt - PO is not yet "placed"
|
||||
result1 = self.post(url, data={'barcode': DIGIKEY_BARCODE}, expected_code=400)
|
||||
|
||||
self.assertTrue(result1.data['error'].startswith('No matching purchase order'))
|
||||
self.assertIn('received against an order marked as', result1.data['error'])
|
||||
|
||||
# Next, place the order - receipt should then work
|
||||
self.purchase_order1.place_order()
|
||||
|
||||
result2 = self.post(url, data={'barcode': DIGIKEY_BARCODE}, expected_code=200)
|
||||
self.assertIn('success', result2.data)
|
||||
|
||||
# Attempt to receive the same item again
|
||||
# Already received - failure expected
|
||||
result3 = self.post(url, data={'barcode': DIGIKEY_BARCODE}, expected_code=400)
|
||||
self.assertEqual(result3.data['error'], 'Item has already been received')
|
||||
|
||||
result4 = self.post(
|
||||
url, data={'barcode': DIGIKEY_BARCODE[:-1]}, expected_code=400
|
||||
)
|
||||
self.assertTrue(
|
||||
result4.data['error'].startswith(
|
||||
'Failed to find pending line item for supplier part'
|
||||
)
|
||||
)
|
||||
|
||||
result5 = self.post(
|
||||
reverse('api-barcode-scan'),
|
||||
data={'barcode': DIGIKEY_BARCODE},
|
||||
expected_code=200,
|
||||
)
|
||||
|
||||
stock_item = StockItem.objects.get(pk=result5.data['stockitem']['pk'])
|
||||
self.assertEqual(stock_item.supplier_part.SKU, '296-LM358BIDDFRCT-ND')
|
||||
self.assertEqual(stock_item.quantity, 10)
|
||||
self.assertEqual(stock_item.location, None)
|
||||
self.assertEqual(stock_item.location, self.loc_2)
|
||||
|
||||
def test_no_auto_allocate(self):
|
||||
"""Test with auto_allocate explicitly disabled."""
|
||||
url = reverse('api-barcode-po-receive')
|
||||
self.purchase_order1.place_order()
|
||||
|
||||
response = self.post(
|
||||
url,
|
||||
data={'barcode': DIGIKEY_BARCODE, 'auto_allocate': False},
|
||||
expected_code=200,
|
||||
)
|
||||
|
||||
self.assertEqual(response.data['plugin'], 'DigiKeyPlugin')
|
||||
self.assertIn('action_required', response.data)
|
||||
item = response.data['lineitem']
|
||||
self.assertEqual(item['quantity'], 10.0)
|
||||
self.assertEqual(item['purchase_order'], self.purchase_order1.pk)
|
||||
self.assertEqual(item['location'], self.loc_2.pk)
|
||||
|
||||
def test_receive_custom_order_number(self):
|
||||
"""Test receiving an item from a barcode with a custom order number."""
|
||||
@ -233,23 +279,32 @@ class SupplierBarcodePOReceiveTests(InvenTreeAPITestCase):
|
||||
result1 = self.post(url, data={'barcode': MOUSER_BARCODE}, expected_code=200)
|
||||
self.assertIn('success', result1.data)
|
||||
|
||||
# Scan the same barcode again - should be resolved to the created item
|
||||
result2 = self.post(
|
||||
reverse('api-barcode-scan'),
|
||||
data={'barcode': MOUSER_BARCODE},
|
||||
expected_code=200,
|
||||
)
|
||||
stock_item = StockItem.objects.get(pk=result2.data['stockitem']['pk'])
|
||||
|
||||
self.assertEqual(stock_item.supplier_part.SKU, '42')
|
||||
self.assertEqual(stock_item.supplier_part.manufacturer_part.MPN, 'MC34063ADR')
|
||||
self.assertEqual(stock_item.quantity, 3)
|
||||
self.assertEqual(stock_item.location, None)
|
||||
self.assertEqual(stock_item.location, self.loc_2)
|
||||
self.assertEqual(stock_item.barcode_data, MOUSER_BARCODE)
|
||||
|
||||
def test_receive_one_stock_location(self):
|
||||
"""Test receiving an item when only one stock location exists."""
|
||||
def test_receive_stock_location(self):
|
||||
"""Test receiving an item when the location is provided."""
|
||||
stock_location = StockLocation.objects.create(name='Test Location')
|
||||
|
||||
url = reverse('api-barcode-po-receive')
|
||||
result1 = self.post(url, data={'barcode': MOUSER_BARCODE}, expected_code=200)
|
||||
|
||||
result1 = self.post(
|
||||
url,
|
||||
data={'barcode': MOUSER_BARCODE, 'location': stock_location.pk},
|
||||
expected_code=200,
|
||||
)
|
||||
|
||||
self.assertIn('success', result1.data)
|
||||
|
||||
result2 = self.post(
|
||||
@ -257,6 +312,7 @@ class SupplierBarcodePOReceiveTests(InvenTreeAPITestCase):
|
||||
data={'barcode': MOUSER_BARCODE},
|
||||
expected_code=200,
|
||||
)
|
||||
|
||||
stock_item = StockItem.objects.get(pk=result2.data['stockitem']['pk'])
|
||||
self.assertEqual(stock_item.location, stock_location)
|
||||
|
||||
@ -286,6 +342,15 @@ class SupplierBarcodePOReceiveTests(InvenTreeAPITestCase):
|
||||
StockLocation.objects.create(name='Test Location 1')
|
||||
stock_location2 = StockLocation.objects.create(name='Test Location 2')
|
||||
|
||||
# Ensure no other fallback locations are set
|
||||
# This is to ensure that the part location is used instead
|
||||
self.purchase_order2.destination = None
|
||||
self.purchase_order2.save()
|
||||
|
||||
for line in self.purchase_order2.lines.all():
|
||||
line.destination = None
|
||||
line.save()
|
||||
|
||||
part = Part.objects.all()[0]
|
||||
part.default_location = stock_location2
|
||||
part.save()
|
||||
@ -332,8 +397,12 @@ class SupplierBarcodePOReceiveTests(InvenTreeAPITestCase):
|
||||
barcode = MOUSER_BARCODE.replace('\x1dQ3', '')
|
||||
response = self.post(url, data={'barcode': barcode}, expected_code=200)
|
||||
|
||||
self.assertIn('action_required', response.data)
|
||||
|
||||
self.assertIn('lineitem', response.data)
|
||||
self.assertNotIn('quantity', response.data['lineitem'])
|
||||
|
||||
# Quantity should be pre-filled with the remaining quantity
|
||||
self.assertEqual(5, response.data['lineitem']['quantity'])
|
||||
|
||||
|
||||
DIGIKEY_BARCODE = (
|
||||
|
@ -416,7 +416,12 @@ class LabelTemplate(TemplateUploadMixin, ReportTemplateBase):
|
||||
|
||||
for plugin in plugins:
|
||||
# Let each plugin add its own context data
|
||||
plugin.add_label_context(self, instance, request, context)
|
||||
try:
|
||||
plugin.add_label_context(self, instance, request, context)
|
||||
except Exception:
|
||||
InvenTree.exceptions.log_error(
|
||||
f'plugins.{plugin.slug}.add_label_context'
|
||||
)
|
||||
|
||||
return context
|
||||
|
||||
|
@ -533,7 +533,13 @@ class StockItem(
|
||||
# If a non-null value is returned (by any plugin) we will use that
|
||||
|
||||
for plugin in registry.with_mixin('validation'):
|
||||
serial_int = plugin.convert_serial_to_int(serial)
|
||||
try:
|
||||
serial_int = plugin.convert_serial_to_int(serial)
|
||||
except Exception:
|
||||
InvenTree.exceptions.log_error(
|
||||
f'plugin.{plugin.slug}.convert_serial_to_int'
|
||||
)
|
||||
serial_int = None
|
||||
|
||||
# Save the first returned result
|
||||
if serial_int is not None:
|
||||
|
Loading…
x
Reference in New Issue
Block a user