mirror of
https://github.com/inventree/InvenTree.git
synced 2025-04-28 19:46:46 +00:00
Refactor existing barcode API endpoints (#5937)
* Refactor existing barcode API endpoints - Expose fields using proper DRF serializers - API endpoints are now self documenting - Validation is handled by serializer models - Serializers and endpoints are extensible - Extended existing unit tests * Catch errors if db not yet loaded * Tweak unit tests
This commit is contained in:
parent
8cb2ed3bd6
commit
52b01b09bf
@ -349,7 +349,7 @@ def MakeBarcode(cls_name, object_pk: int, object_data=None, **kwargs):
|
||||
object_data['id'] = object_pk
|
||||
data[cls_name] = object_data
|
||||
|
||||
return json.dumps(data, sort_keys=True)
|
||||
return str(json.dumps(data, sort_keys=True))
|
||||
|
||||
|
||||
def GetExportFormats():
|
||||
|
@ -228,7 +228,11 @@ def getModelsWithMixin(mixin_class) -> list:
|
||||
"""
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
|
||||
db_models = [x.model_class() for x in ContentType.objects.all() if x is not None]
|
||||
try:
|
||||
db_models = [x.model_class() for x in ContentType.objects.all() if x is not None]
|
||||
except (OperationalError, ProgrammingError):
|
||||
# Database is likely not yet ready
|
||||
db_models = []
|
||||
|
||||
return [x for x in db_models if x is not None and issubclass(x, mixin_class)]
|
||||
|
||||
|
@ -522,7 +522,7 @@ class PurchaseOrder(TotalPriceMixin, Order):
|
||||
@property
|
||||
def is_pending(self):
|
||||
"""Return True if the PurchaseOrder is 'pending'"""
|
||||
return self.status == PurchaseOrderStatus.PENDING
|
||||
return self.status == PurchaseOrderStatus.PENDING.value
|
||||
|
||||
@property
|
||||
def is_open(self):
|
||||
@ -536,8 +536,8 @@ class PurchaseOrder(TotalPriceMixin, Order):
|
||||
- Status is PENDING
|
||||
"""
|
||||
return self.status in [
|
||||
PurchaseOrderStatus.PLACED,
|
||||
PurchaseOrderStatus.PENDING
|
||||
PurchaseOrderStatus.PLACED.value,
|
||||
PurchaseOrderStatus.PENDING.value
|
||||
]
|
||||
|
||||
@transaction.atomic
|
||||
|
@ -7,21 +7,60 @@ from django.utils.translation import gettext_lazy as _
|
||||
|
||||
from rest_framework import permissions
|
||||
from rest_framework.exceptions import PermissionDenied, ValidationError
|
||||
from rest_framework.generics import CreateAPIView
|
||||
from rest_framework.response import Response
|
||||
from rest_framework.views import APIView
|
||||
|
||||
from InvenTree.helpers import hash_barcode
|
||||
from order.models import PurchaseOrder
|
||||
from plugin import registry
|
||||
from plugin.builtin.barcodes.inventree_barcode import \
|
||||
InvenTreeInternalBarcodePlugin
|
||||
from stock.models import StockLocation
|
||||
from users.models import RuleSet
|
||||
|
||||
from . import serializers as barcode_serializers
|
||||
|
||||
logger = logging.getLogger('inventree')
|
||||
|
||||
|
||||
class BarcodeScan(APIView):
|
||||
class BarcodeView(CreateAPIView):
|
||||
"""Custom view class for handling a barcode scan"""
|
||||
|
||||
# Default serializer class (can be overridden)
|
||||
serializer_class = barcode_serializers.BarcodeSerializer
|
||||
|
||||
def queryset(self):
|
||||
"""This API view does not have a queryset"""
|
||||
return None
|
||||
|
||||
# Default permission classes (can be overridden)
|
||||
permission_classes = [
|
||||
permissions.IsAuthenticated,
|
||||
]
|
||||
|
||||
def create(self, request, *args, **kwargs):
|
||||
"""Handle create method - override default create"""
|
||||
|
||||
serializer = self.get_serializer(data=request.data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
data = serializer.validated_data
|
||||
|
||||
barcode = str(data.pop('barcode')).strip()
|
||||
|
||||
return self.handle_barcode(barcode, request, **data)
|
||||
|
||||
def handle_barcode(self, barcode: str, request, **kwargs):
|
||||
"""Handle barcode scan.
|
||||
|
||||
Arguments:
|
||||
barcode: Raw barcode value
|
||||
request: HTTP request object
|
||||
|
||||
kwargs:
|
||||
Any custom fields passed by the specific serializer
|
||||
"""
|
||||
raise NotImplementedError(f"handle_barcode not implemented for {self.__class__}")
|
||||
|
||||
|
||||
class BarcodeScan(BarcodeView):
|
||||
"""Endpoint for handling generic barcode scan requests.
|
||||
|
||||
Barcode data are decoded by the client application,
|
||||
@ -29,47 +68,29 @@ class BarcodeScan(APIView):
|
||||
|
||||
A barcode could follow the internal InvenTree barcode format,
|
||||
or it could match to a third-party barcode format (e.g. Digikey).
|
||||
|
||||
When a barcode is sent to the server, the following parameters must be provided:
|
||||
|
||||
- barcode: The raw barcode data
|
||||
|
||||
plugins:
|
||||
Third-party barcode formats may be supported using 'plugins'
|
||||
(more information to follow)
|
||||
|
||||
hashing:
|
||||
Barcode hashes are calculated using MD5
|
||||
"""
|
||||
|
||||
permission_classes = [
|
||||
permissions.IsAuthenticated,
|
||||
]
|
||||
def handle_barcode(self, barcode: str, request, **kwargs):
|
||||
"""Perform barcode scan action
|
||||
|
||||
def post(self, request, *args, **kwargs):
|
||||
"""Respond to a barcode POST request.
|
||||
Arguments:
|
||||
barcode: Raw barcode value
|
||||
request: HTTP request object
|
||||
|
||||
Check if required info was provided and then run though the plugin steps or try to match up-
|
||||
kwargs:
|
||||
Any custom fields passed by the specific serializer
|
||||
"""
|
||||
data = request.data
|
||||
|
||||
barcode_data = data.get('barcode', None)
|
||||
|
||||
if not barcode_data:
|
||||
raise ValidationError({'barcode': _('Missing barcode data')})
|
||||
|
||||
# Note: the default barcode handlers are loaded (and thus run) first
|
||||
plugins = registry.with_mixin('barcode')
|
||||
|
||||
barcode_hash = hash_barcode(barcode_data)
|
||||
|
||||
# Look for a barcode plugin which knows how to deal with this barcode
|
||||
plugin = None
|
||||
response = {}
|
||||
|
||||
for current_plugin in plugins:
|
||||
|
||||
result = current_plugin.scan(barcode_data)
|
||||
result = current_plugin.scan(barcode)
|
||||
|
||||
if result is None:
|
||||
continue
|
||||
@ -86,8 +107,8 @@ class BarcodeScan(APIView):
|
||||
break
|
||||
|
||||
response['plugin'] = plugin.name if plugin else None
|
||||
response['barcode_data'] = barcode_data
|
||||
response['barcode_hash'] = barcode_hash
|
||||
response['barcode_data'] = barcode
|
||||
response['barcode_hash'] = hash_barcode(barcode)
|
||||
|
||||
# A plugin has not been found!
|
||||
if plugin is None:
|
||||
@ -99,44 +120,36 @@ class BarcodeScan(APIView):
|
||||
return Response(response)
|
||||
|
||||
|
||||
class BarcodeAssign(APIView):
|
||||
class BarcodeAssign(BarcodeView):
|
||||
"""Endpoint for assigning a barcode to a stock item.
|
||||
|
||||
- This only works if the barcode is not already associated with an object in the database
|
||||
- If the barcode does not match an object, then the barcode hash is assigned to the StockItem
|
||||
"""
|
||||
|
||||
permission_classes = [
|
||||
permissions.IsAuthenticated
|
||||
]
|
||||
serializer_class = barcode_serializers.BarcodeAssignSerializer
|
||||
|
||||
def post(self, request, *args, **kwargs):
|
||||
"""Respond to a barcode assign POST request.
|
||||
def handle_barcode(self, barcode: str, request, **kwargs):
|
||||
"""Respond to a barcode assign request.
|
||||
|
||||
Checks inputs and assign barcode (hash) to StockItem.
|
||||
"""
|
||||
data = request.data
|
||||
|
||||
barcode_data = data.get('barcode', None)
|
||||
|
||||
if not barcode_data:
|
||||
raise ValidationError({'barcode': _('Missing barcode data')})
|
||||
|
||||
# Here we only check against 'InvenTree' plugins
|
||||
plugins = registry.with_mixin('barcode', builtin=True)
|
||||
|
||||
# First check if the provided barcode matches an existing database entry
|
||||
for plugin in plugins:
|
||||
result = plugin.scan(barcode_data)
|
||||
result = plugin.scan(barcode)
|
||||
|
||||
if result is not None:
|
||||
result["error"] = _("Barcode matches existing item")
|
||||
result["plugin"] = plugin.name
|
||||
result["barcode_data"] = barcode_data
|
||||
result["barcode_data"] = barcode
|
||||
|
||||
raise ValidationError(result)
|
||||
|
||||
barcode_hash = hash_barcode(barcode_data)
|
||||
barcode_hash = hash_barcode(barcode)
|
||||
|
||||
valid_labels = []
|
||||
|
||||
@ -144,39 +157,32 @@ class BarcodeAssign(APIView):
|
||||
label = model.barcode_model_type()
|
||||
valid_labels.append(label)
|
||||
|
||||
if label in data:
|
||||
try:
|
||||
instance = model.objects.get(pk=data[label])
|
||||
if instance := kwargs.get(label, None):
|
||||
|
||||
# Check that the user has the required permission
|
||||
app_label = model._meta.app_label
|
||||
model_name = model._meta.model_name
|
||||
# Check that the user has the required permission
|
||||
app_label = model._meta.app_label
|
||||
model_name = model._meta.model_name
|
||||
|
||||
table = f"{app_label}_{model_name}"
|
||||
table = f"{app_label}_{model_name}"
|
||||
|
||||
if not RuleSet.check_table_permission(request.user, table, "change"):
|
||||
raise PermissionDenied({
|
||||
"error": f"You do not have the required permissions for {table}"
|
||||
})
|
||||
|
||||
instance.assign_barcode(
|
||||
barcode_data=barcode_data,
|
||||
barcode_hash=barcode_hash,
|
||||
)
|
||||
|
||||
return Response({
|
||||
'success': f"Assigned barcode to {label} instance",
|
||||
label: {
|
||||
'pk': instance.pk,
|
||||
},
|
||||
"barcode_data": barcode_data,
|
||||
"barcode_hash": barcode_hash,
|
||||
if not RuleSet.check_table_permission(request.user, table, "change"):
|
||||
raise PermissionDenied({
|
||||
"error": f"You do not have the required permissions for {table}"
|
||||
})
|
||||
|
||||
except (ValueError, model.DoesNotExist):
|
||||
raise ValidationError({
|
||||
'error': f"No matching {label} instance found in database",
|
||||
})
|
||||
instance.assign_barcode(
|
||||
barcode_data=barcode,
|
||||
barcode_hash=barcode_hash,
|
||||
)
|
||||
|
||||
return Response({
|
||||
'success': f"Assigned barcode to {label} instance",
|
||||
label: {
|
||||
'pk': instance.pk,
|
||||
},
|
||||
"barcode_data": barcode,
|
||||
"barcode_hash": barcode_hash,
|
||||
})
|
||||
|
||||
# If we got here, it means that no valid model types were provided
|
||||
raise ValidationError({
|
||||
@ -184,23 +190,23 @@ class BarcodeAssign(APIView):
|
||||
})
|
||||
|
||||
|
||||
class BarcodeUnassign(APIView):
|
||||
class BarcodeUnassign(BarcodeView):
|
||||
"""Endpoint for unlinking / unassigning a custom barcode from a database object"""
|
||||
|
||||
permission_classes = [
|
||||
permissions.IsAuthenticated,
|
||||
]
|
||||
serializer_class = barcode_serializers.BarcodeUnassignSerializer
|
||||
|
||||
def create(self, request, *args, **kwargs):
|
||||
"""Respond to a barcode unassign request."""
|
||||
|
||||
serializer = self.get_serializer(data=request.data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
data = serializer.validated_data
|
||||
|
||||
def post(self, request, *args, **kwargs):
|
||||
"""Respond to a barcode unassign POST request"""
|
||||
# The following database models support assignment of third-party barcodes
|
||||
supported_models = InvenTreeInternalBarcodePlugin.get_supported_barcode_models()
|
||||
|
||||
supported_labels = [model.barcode_model_type() for model in supported_models]
|
||||
model_names = ', '.join(supported_labels)
|
||||
|
||||
data = request.data
|
||||
|
||||
matched_labels = []
|
||||
|
||||
for label in supported_labels:
|
||||
@ -219,15 +225,10 @@ class BarcodeUnassign(APIView):
|
||||
|
||||
# At this stage, we know that we have received a single valid field
|
||||
for model in supported_models:
|
||||
|
||||
label = model.barcode_model_type()
|
||||
|
||||
if label in data:
|
||||
try:
|
||||
instance = model.objects.get(pk=data[label])
|
||||
except (ValueError, model.DoesNotExist):
|
||||
raise ValidationError({
|
||||
label: _('No match found for provided value')
|
||||
})
|
||||
if instance := data.get(label, None):
|
||||
|
||||
# Check that the user has the required permission
|
||||
app_label = model._meta.app_label
|
||||
@ -253,7 +254,7 @@ class BarcodeUnassign(APIView):
|
||||
})
|
||||
|
||||
|
||||
class BarcodePOReceive(APIView):
|
||||
class BarcodePOReceive(BarcodeView):
|
||||
"""Endpoint for handling receiving parts by scanning their barcode.
|
||||
|
||||
Barcode data are decoded by the client application,
|
||||
@ -269,32 +270,16 @@ class BarcodePOReceive(APIView):
|
||||
- location: The destination location for the received item (optional)
|
||||
"""
|
||||
|
||||
permission_classes = [
|
||||
permissions.IsAuthenticated,
|
||||
]
|
||||
serializer_class = barcode_serializers.BarcodePOReceiveSerializer
|
||||
|
||||
def post(self, request, *args, **kwargs):
|
||||
"""Respond to a barcode POST request."""
|
||||
def handle_barcode(self, barcode: str, request, **kwargs):
|
||||
"""Handle a barcode scan for a purchase order item."""
|
||||
|
||||
data = request.data
|
||||
logger.debug("BarcodePOReceive: scanned barcode - '%s'", barcode)
|
||||
|
||||
if not (barcode_data := data.get("barcode")):
|
||||
raise ValidationError({"barcode": _("Missing barcode data")})
|
||||
|
||||
logger.debug("BarcodePOReceive: scanned barcode - '%s'", barcode_data)
|
||||
|
||||
purchase_order = None
|
||||
|
||||
if purchase_order_pk := data.get("purchase_order"):
|
||||
purchase_order = PurchaseOrder.objects.filter(pk=purchase_order_pk).first()
|
||||
if not purchase_order:
|
||||
raise ValidationError({"purchase_order": _("Invalid purchase order")})
|
||||
|
||||
location = None
|
||||
if (location_pk := data.get("location")):
|
||||
location = StockLocation.objects.get(pk=location_pk)
|
||||
if not location:
|
||||
raise ValidationError({"location": _("Invalid stock location")})
|
||||
# Extract optional fields from the dataset
|
||||
purchase_order = kwargs.get('purchase_order', None)
|
||||
location = kwargs.get('location', None)
|
||||
|
||||
plugins = registry.with_mixin("barcode")
|
||||
|
||||
@ -303,8 +288,10 @@ class BarcodePOReceive(APIView):
|
||||
response = {}
|
||||
|
||||
internal_barcode_plugin = next(filter(
|
||||
lambda plugin: plugin.name == "InvenTreeBarcode", plugins))
|
||||
if internal_barcode_plugin.scan(barcode_data):
|
||||
lambda plugin: plugin.name == "InvenTreeBarcode", plugins
|
||||
))
|
||||
|
||||
if internal_barcode_plugin.scan(barcode):
|
||||
response["error"] = _("Item has already been received")
|
||||
raise ValidationError(response)
|
||||
|
||||
@ -314,7 +301,7 @@ class BarcodePOReceive(APIView):
|
||||
for current_plugin in plugins:
|
||||
|
||||
result = current_plugin.scan_receive_item(
|
||||
barcode_data,
|
||||
barcode,
|
||||
request.user,
|
||||
purchase_order=purchase_order,
|
||||
location=location,
|
||||
@ -335,8 +322,8 @@ class BarcodePOReceive(APIView):
|
||||
break
|
||||
|
||||
response["plugin"] = plugin.name if plugin else None
|
||||
response["barcode_data"] = barcode_data
|
||||
response["barcode_hash"] = hash_barcode(barcode_data)
|
||||
response["barcode_data"] = barcode
|
||||
response["barcode_hash"] = hash_barcode(barcode)
|
||||
|
||||
# A plugin has not been found!
|
||||
if plugin is None:
|
||||
|
107
InvenTree/plugin/base/barcodes/serializers.py
Normal file
107
InvenTree/plugin/base/barcodes/serializers.py
Normal file
@ -0,0 +1,107 @@
|
||||
"""DRF serializers for barcode scanning API"""
|
||||
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
|
||||
from rest_framework import serializers
|
||||
|
||||
import order.models
|
||||
import stock.models
|
||||
from InvenTree.status_codes import PurchaseOrderStatus
|
||||
from plugin.builtin.barcodes.inventree_barcode import \
|
||||
InvenTreeInternalBarcodePlugin
|
||||
|
||||
|
||||
class BarcodeSerializer(serializers.Serializer):
|
||||
"""Generic serializer for receiving barcode data"""
|
||||
|
||||
MAX_BARCODE_LENGTH = 4095
|
||||
|
||||
barcode = serializers.CharField(
|
||||
required=True, help_text=_('Scanned barcode data'),
|
||||
max_length=MAX_BARCODE_LENGTH,
|
||||
)
|
||||
|
||||
|
||||
class BarcodeAssignMixin(serializers.Serializer):
|
||||
"""Serializer for linking and unlinking barcode to an internal class"""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
"""Generate serializer fields for each supported model type"""
|
||||
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
for model in InvenTreeInternalBarcodePlugin.get_supported_barcode_models():
|
||||
self.fields[model.barcode_model_type()] = serializers.PrimaryKeyRelatedField(
|
||||
queryset=model.objects.all(),
|
||||
required=False, allow_null=True,
|
||||
label=model._meta.verbose_name,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_model_fields():
|
||||
"""Return a list of model fields"""
|
||||
fields = [
|
||||
model.barcode_model_type() for model in InvenTreeInternalBarcodePlugin.get_supported_barcode_models()
|
||||
]
|
||||
|
||||
return fields
|
||||
|
||||
|
||||
class BarcodeAssignSerializer(BarcodeAssignMixin, BarcodeSerializer):
|
||||
"""Serializer class for linking a barcode to an internal model"""
|
||||
|
||||
class Meta:
|
||||
"""Meta class for BarcodeAssignSerializer"""
|
||||
|
||||
fields = [
|
||||
'barcode',
|
||||
*BarcodeAssignMixin.get_model_fields()
|
||||
]
|
||||
|
||||
|
||||
class BarcodeUnassignSerializer(BarcodeAssignMixin):
|
||||
"""Serializer class for unlinking a barcode from an internal model"""
|
||||
|
||||
class Meta:
|
||||
"""Meta class for BarcodeUnlinkSerializer"""
|
||||
|
||||
fields = BarcodeAssignMixin.get_model_fields()
|
||||
|
||||
|
||||
class BarcodePOReceiveSerializer(BarcodeSerializer):
|
||||
"""Serializer for receiving items against a purchase order.
|
||||
|
||||
The following additional fields may be specified:
|
||||
|
||||
- purchase_order: PurchaseOrder object to receive items against
|
||||
- location: Location to receive items into
|
||||
"""
|
||||
|
||||
purchase_order = serializers.PrimaryKeyRelatedField(
|
||||
queryset=order.models.PurchaseOrder.objects.all(),
|
||||
required=False,
|
||||
help_text=_('PurchaseOrder to receive items against'),
|
||||
)
|
||||
|
||||
def validate_purchase_order(self, order: order.models.PurchaseOrder):
|
||||
"""Validate the provided order"""
|
||||
|
||||
if order.status != PurchaseOrderStatus.PLACED.value:
|
||||
raise ValidationError(_("Purchase order has not been placed"))
|
||||
|
||||
return order
|
||||
|
||||
location = serializers.PrimaryKeyRelatedField(
|
||||
queryset=stock.models.StockLocation.objects.all(),
|
||||
required=False,
|
||||
help_text=_('Location to receive items into'),
|
||||
)
|
||||
|
||||
def validate_location(self, location: stock.models.StockLocation):
|
||||
"""Validate the provided location"""
|
||||
|
||||
if location.structural:
|
||||
raise ValidationError(_("Cannot select a structural location"))
|
||||
|
||||
return location
|
@ -2,9 +2,8 @@
|
||||
|
||||
from django.urls import reverse
|
||||
|
||||
from rest_framework import status
|
||||
|
||||
from InvenTree.unit_test import InvenTreeAPITestCase
|
||||
from part.models import Part
|
||||
from stock.models import StockItem
|
||||
|
||||
|
||||
@ -24,150 +23,128 @@ class BarcodeAPITest(InvenTreeAPITestCase):
|
||||
|
||||
self.scan_url = reverse('api-barcode-scan')
|
||||
self.assign_url = reverse('api-barcode-link')
|
||||
self.unassign_url = reverse('api-barcode-unlink')
|
||||
|
||||
def postBarcode(self, url, barcode):
|
||||
def postBarcode(self, url, barcode, expected_code=None):
|
||||
"""Post barcode and return results."""
|
||||
return self.client.post(url, format='json', data={'barcode': str(barcode)})
|
||||
return self.post(url, format='json', data={'barcode': str(barcode)}, expected_code=expected_code)
|
||||
|
||||
def test_invalid(self):
|
||||
"""Test that invalid requests fail."""
|
||||
# test scan url
|
||||
response = self.client.post(self.scan_url, format='json', data={})
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
self.post(self.scan_url, format='json', data={}, expected_code=400)
|
||||
|
||||
# test wrong assign urls
|
||||
response = self.client.post(self.assign_url, format='json', data={})
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
response = self.client.post(self.assign_url, format='json', data={'barcode': '123'})
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
response = self.client.post(self.assign_url, format='json', data={'barcode': '123', 'stockitem': '123'})
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
self.post(self.assign_url, format='json', data={}, expected_code=400)
|
||||
self.post(self.assign_url, format='json', data={'barcode': '123'}, expected_code=400)
|
||||
self.post(self.assign_url, format='json', data={'barcode': '123', 'stockitem': '123'}, expected_code=400)
|
||||
|
||||
def test_empty(self):
|
||||
"""Test an empty barcode scan.
|
||||
|
||||
Ensure that all required data is in the respomse.
|
||||
Ensure that all required data is in the response.
|
||||
"""
|
||||
response = self.postBarcode(self.scan_url, '')
|
||||
|
||||
self.assertEqual(response.status_code, 400)
|
||||
response = self.postBarcode(self.scan_url, '', expected_code=400)
|
||||
|
||||
data = response.data
|
||||
self.assertIn('barcode', data)
|
||||
self.assertIn('Missing barcode data', str(response.data['barcode']))
|
||||
|
||||
self.assertIn('This field may not be blank', str(response.data['barcode']))
|
||||
|
||||
def test_find_part(self):
|
||||
"""Test that we can lookup a part based on ID."""
|
||||
response = self.client.post(
|
||||
|
||||
part = Part.objects.first()
|
||||
|
||||
response = self.post(
|
||||
self.scan_url,
|
||||
{
|
||||
'barcode': {
|
||||
'part': 1,
|
||||
},
|
||||
'barcode': f'{{"part": {part.pk}}}',
|
||||
},
|
||||
format='json',
|
||||
expected_code=200
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertIn('part', response.data)
|
||||
self.assertIn('barcode_data', response.data)
|
||||
self.assertEqual(response.data['part']['pk'], 1)
|
||||
self.assertEqual(response.data['part']['pk'], part.pk)
|
||||
|
||||
def test_invalid_part(self):
|
||||
"""Test response for invalid part."""
|
||||
response = self.client.post(
|
||||
response = self.post(
|
||||
self.scan_url,
|
||||
{
|
||||
'barcode': {
|
||||
'part': 999999999,
|
||||
}
|
||||
'barcode': '{"part": 999999999}'
|
||||
},
|
||||
format='json'
|
||||
expected_code=400
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertIn('error', response.data)
|
||||
|
||||
def test_find_stock_item(self):
|
||||
"""Test that we can lookup a stock item based on ID."""
|
||||
response = self.client.post(
|
||||
|
||||
item = StockItem.objects.first()
|
||||
|
||||
response = self.post(
|
||||
self.scan_url,
|
||||
{
|
||||
'barcode': {
|
||||
'stockitem': 1,
|
||||
}
|
||||
'barcode': item.format_barcode(),
|
||||
},
|
||||
format='json',
|
||||
expected_code=200
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertIn('stockitem', response.data)
|
||||
self.assertIn('barcode_data', response.data)
|
||||
self.assertEqual(response.data['stockitem']['pk'], 1)
|
||||
self.assertEqual(response.data['stockitem']['pk'], item.pk)
|
||||
|
||||
def test_invalid_item(self):
|
||||
"""Test response for invalid stock item."""
|
||||
response = self.client.post(
|
||||
response = self.post(
|
||||
self.scan_url,
|
||||
{
|
||||
'barcode': {
|
||||
'stockitem': 999999999,
|
||||
}
|
||||
'barcode': '{"stockitem": 999999999}'
|
||||
},
|
||||
format='json'
|
||||
expected_code=400
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertIn('error', response.data)
|
||||
|
||||
def test_find_location(self):
|
||||
"""Test that we can lookup a stock location based on ID."""
|
||||
response = self.client.post(
|
||||
response = self.post(
|
||||
self.scan_url,
|
||||
{
|
||||
'barcode': {
|
||||
'stocklocation': 1,
|
||||
},
|
||||
'barcode': '{"stocklocation": 1}',
|
||||
},
|
||||
format='json'
|
||||
expected_code=200
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertIn('stocklocation', response.data)
|
||||
self.assertIn('barcode_data', response.data)
|
||||
self.assertEqual(response.data['stocklocation']['pk'], 1)
|
||||
|
||||
def test_invalid_location(self):
|
||||
"""Test response for an invalid location."""
|
||||
response = self.client.post(
|
||||
response = self.post(
|
||||
self.scan_url,
|
||||
{
|
||||
'barcode': {
|
||||
'stocklocation': 999999999,
|
||||
}
|
||||
'barcode': '{"stocklocation": 999999999}'
|
||||
},
|
||||
format='json'
|
||||
expected_code=400
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertIn('error', response.data)
|
||||
|
||||
def test_integer_barcode(self):
|
||||
"""Test scan of an integer barcode."""
|
||||
response = self.postBarcode(self.scan_url, '123456789')
|
||||
|
||||
self.assertEqual(response.status_code, 400)
|
||||
response = self.postBarcode(self.scan_url, '123456789', expected_code=400)
|
||||
|
||||
data = response.data
|
||||
self.assertIn('error', data)
|
||||
|
||||
def test_array_barcode(self):
|
||||
"""Test scan of barcode with string encoded array."""
|
||||
response = self.postBarcode(self.scan_url, "['foo', 'bar']")
|
||||
|
||||
self.assertEqual(response.status_code, 400)
|
||||
response = self.postBarcode(self.scan_url, "['foo', 'bar']", expected_code=400)
|
||||
|
||||
data = response.data
|
||||
self.assertIn('error', data)
|
||||
@ -176,11 +153,9 @@ class BarcodeAPITest(InvenTreeAPITestCase):
|
||||
"""Test that a barcode is generated with a scan."""
|
||||
item = StockItem.objects.get(pk=522)
|
||||
|
||||
response = self.postBarcode(self.scan_url, item.format_barcode())
|
||||
response = self.postBarcode(self.scan_url, item.format_barcode(), expected_code=200)
|
||||
data = response.data
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
self.assertIn('stockitem', data)
|
||||
|
||||
pk = data['stockitem']['pk']
|
||||
@ -197,37 +172,88 @@ class BarcodeAPITest(InvenTreeAPITestCase):
|
||||
|
||||
barcode_data = 'A-TEST-BARCODE-STRING'
|
||||
|
||||
response = self.client.post(
|
||||
response = self.post(
|
||||
self.assign_url, format='json',
|
||||
data={
|
||||
'barcode': barcode_data,
|
||||
'stockitem': item.pk
|
||||
}
|
||||
},
|
||||
expected_code=200
|
||||
)
|
||||
|
||||
data = response.data
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
self.assertIn('success', data)
|
||||
|
||||
result_hash = data['barcode_hash']
|
||||
|
||||
# Read the item out from the database again
|
||||
item = StockItem.objects.get(pk=522)
|
||||
item.refresh_from_db()
|
||||
|
||||
self.assertEqual(item.barcode_data, barcode_data)
|
||||
self.assertEqual(result_hash, item.barcode_hash)
|
||||
|
||||
# Ensure that the same barcode hash cannot be assigned to a different stock item!
|
||||
response = self.client.post(
|
||||
response = self.post(
|
||||
self.assign_url, format='json',
|
||||
data={
|
||||
'barcode': barcode_data,
|
||||
'stockitem': 521
|
||||
}
|
||||
},
|
||||
expected_code=400
|
||||
)
|
||||
|
||||
data = response.data
|
||||
self.assertIn('error', response.data)
|
||||
self.assertNotIn('success', response.data)
|
||||
|
||||
self.assertIn('error', data)
|
||||
self.assertNotIn('success', data)
|
||||
# Check that we can now unassign a barcode
|
||||
response = self.post(
|
||||
self.unassign_url,
|
||||
{
|
||||
'stockitem': item.pk,
|
||||
},
|
||||
expected_code=200
|
||||
)
|
||||
|
||||
item.refresh_from_db()
|
||||
self.assertEqual(item.barcode_data, '')
|
||||
|
||||
# Check that the 'unassign' endpoint fails if the stockitem is invalid
|
||||
response = self.post(
|
||||
self.unassign_url,
|
||||
{
|
||||
'stockitem': 999999999,
|
||||
},
|
||||
expected_code=400
|
||||
)
|
||||
|
||||
def test_unassign_endpoint(self):
|
||||
"""Test that the unassign endpoint works as expected"""
|
||||
|
||||
invalid_keys = ['cat', 'dog', 'fish']
|
||||
|
||||
# Invalid key should fail
|
||||
for k in invalid_keys:
|
||||
response = self.post(
|
||||
self.unassign_url,
|
||||
{
|
||||
k: 123
|
||||
},
|
||||
expected_code=400
|
||||
)
|
||||
|
||||
self.assertIn("Missing data: Provide one of", str(response.data['error']))
|
||||
|
||||
valid_keys = ['build', 'salesorder', 'part']
|
||||
|
||||
# Valid key but invalid pk should fail
|
||||
for k in valid_keys:
|
||||
response = self.post(
|
||||
self.unassign_url,
|
||||
{
|
||||
k: 999999999
|
||||
},
|
||||
expected_code=400
|
||||
)
|
||||
|
||||
self.assertIn("object does not exist", str(response.data[k]))
|
||||
|
@ -63,8 +63,6 @@ class InvenTreeInternalBarcodePlugin(BarcodeMixin, InvenTreePlugin):
|
||||
|
||||
Here we are looking for a dict object which contains a reference to a particular InvenTree database object
|
||||
"""
|
||||
# Create hash from raw barcode data
|
||||
barcode_hash = hash_barcode(barcode_data)
|
||||
|
||||
# Attempt to coerce the barcode data into a dict object
|
||||
# This is the internal barcode representation that InvenTree uses
|
||||
@ -78,9 +76,11 @@ class InvenTreeInternalBarcodePlugin(BarcodeMixin, InvenTreePlugin):
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
supported_models = self.get_supported_barcode_models()
|
||||
|
||||
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 self.get_supported_barcode_models():
|
||||
for model in supported_models:
|
||||
label = model.barcode_model_type()
|
||||
|
||||
if label in barcode_dict:
|
||||
@ -91,8 +91,11 @@ class InvenTreeInternalBarcodePlugin(BarcodeMixin, InvenTreePlugin):
|
||||
except (ValueError, model.DoesNotExist):
|
||||
pass
|
||||
|
||||
# Create hash from raw barcode data
|
||||
barcode_hash = hash_barcode(barcode_data)
|
||||
|
||||
# If no "direct" hits are found, look for assigned third-party barcodes
|
||||
for model in self.get_supported_barcode_models():
|
||||
for model in supported_models:
|
||||
label = model.barcode_model_type()
|
||||
|
||||
instance = model.lookup_barcode(barcode_hash)
|
||||
|
@ -80,8 +80,8 @@ class TestInvenTreeBarcode(InvenTreeAPITestCase):
|
||||
# Fail with too many fields provided
|
||||
response = self.unassign(
|
||||
{
|
||||
'stockitem': 'abcde',
|
||||
'part': 'abcde',
|
||||
'stockitem': stock.models.StockItem.objects.first().pk,
|
||||
'part': part.models.Part.objects.first().pk,
|
||||
},
|
||||
expected_code=400,
|
||||
)
|
||||
@ -96,17 +96,17 @@ class TestInvenTreeBarcode(InvenTreeAPITestCase):
|
||||
expected_code=400,
|
||||
)
|
||||
|
||||
self.assertIn('No match found', str(response.data['stockitem']))
|
||||
self.assertIn('Incorrect type', str(response.data['stockitem']))
|
||||
|
||||
# Fail with an invalid Part instance
|
||||
response = self.unassign(
|
||||
{
|
||||
'part': 'invalid',
|
||||
'part': 99999999999,
|
||||
},
|
||||
expected_code=400,
|
||||
)
|
||||
|
||||
self.assertIn('No match found', str(response.data['part']))
|
||||
self.assertIn('object does not exist', str(response.data['part']))
|
||||
|
||||
def test_assign_to_stock_item(self):
|
||||
"""Test that we can assign a unique barcode to a StockItem object"""
|
||||
@ -216,7 +216,7 @@ class TestInvenTreeBarcode(InvenTreeAPITestCase):
|
||||
expected_code=400,
|
||||
)
|
||||
|
||||
self.assertIn('No matching part instance found in database', str(response.data))
|
||||
self.assertIn('object does not exist', str(response.data['part']))
|
||||
|
||||
# Test assigning to a valid part (should pass)
|
||||
response = self.assign(
|
||||
|
Loading…
x
Reference in New Issue
Block a user