2
0
mirror of https://github.com/inventree/InvenTree.git synced 2025-07-01 11:10:54 +00:00

Merge remote-tracking branch 'inventree/master' into locate-mixin

# Conflicts:
#	InvenTree/InvenTree/api_version.py
#	InvenTree/InvenTree/urls.py
This commit is contained in:
Oliver Walters
2022-05-15 23:44:07 +10:00
66 changed files with 17711 additions and 17204 deletions

245
InvenTree/plugin/barcode.py Normal file
View File

@ -0,0 +1,245 @@
# -*- coding: utf-8 -*-
from django.urls import reverse, path, re_path
from django.utils.translation import gettext_lazy as _
from rest_framework.exceptions import ValidationError
from rest_framework import permissions
from rest_framework.response import Response
from rest_framework.views import APIView
from stock.models import StockItem
from stock.serializers import StockItemSerializer
from plugin.builtin.barcodes.inventree_barcode import InvenTreeBarcodePlugin
from plugin.builtin.barcodes.mixins import hash_barcode
from plugin import registry
class BarcodeScan(APIView):
"""
Endpoint for handling generic barcode scan requests.
Barcode data are decoded by the client application,
and sent to this endpoint (as a JSON object) for validation.
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 post(self, request, *args, **kwargs):
"""
Respond to a barcode POST request
"""
data = request.data
if 'barcode' not in data:
raise ValidationError({'barcode': _('Must provide barcode_data parameter')})
plugins = registry.with_mixin('barcode')
barcode_data = data.get('barcode')
# Ensure that the default barcode handler is installed
plugins.append(InvenTreeBarcodePlugin())
# Look for a barcode plugin which knows how to deal with this barcode
plugin = None
for current_plugin in plugins:
# TODO @matmair make simpler after InvenTree 0.7.0 release
current_plugin.init(barcode_data)
if current_plugin.validate():
plugin = current_plugin
break
match_found = False
response = {}
response['barcode_data'] = barcode_data
# A plugin has been found!
if plugin is not None:
# Try to associate with a stock item
item = plugin.getStockItem()
if item is None:
item = plugin.getStockItemByHash()
if item is not None:
response['stockitem'] = plugin.renderStockItem(item)
response['url'] = reverse('stock-item-detail', kwargs={'pk': item.id})
match_found = True
# Try to associate with a stock location
loc = plugin.getStockLocation()
if loc is not None:
response['stocklocation'] = plugin.renderStockLocation(loc)
response['url'] = reverse('stock-location-detail', kwargs={'pk': loc.id})
match_found = True
# Try to associate with a part
part = plugin.getPart()
if part is not None:
response['part'] = plugin.renderPart(part)
response['url'] = reverse('part-detail', kwargs={'pk': part.id})
match_found = True
response['hash'] = plugin.hash()
response['plugin'] = plugin.name
# No plugin is found!
# However, the hash of the barcode may still be associated with a StockItem!
else:
result_hash = hash_barcode(barcode_data)
response['hash'] = result_hash
response['plugin'] = None
# Try to look for a matching StockItem
try:
item = StockItem.objects.get(uid=result_hash)
serializer = StockItemSerializer(item, part_detail=True, location_detail=True, supplier_part_detail=True)
response['stockitem'] = serializer.data
response['url'] = reverse('stock-item-detail', kwargs={'pk': item.id})
match_found = True
except StockItem.DoesNotExist:
pass
if not match_found:
response['error'] = _('No match found for barcode data')
else:
response['success'] = _('Match found for barcode data')
return Response(response)
class BarcodeAssign(APIView):
"""
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
]
def post(self, request, *args, **kwargs):
data = request.data
if 'barcode' not in data:
raise ValidationError({'barcode': _('Must provide barcode_data parameter')})
if 'stockitem' not in data:
raise ValidationError({'stockitem': _('Must provide stockitem parameter')})
barcode_data = data['barcode']
try:
item = StockItem.objects.get(pk=data['stockitem'])
except (ValueError, StockItem.DoesNotExist):
raise ValidationError({'stockitem': _('No matching stock item found')})
plugins = registry.with_mixin('barcode')
plugin = None
for current_plugin in plugins:
# TODO @matmair make simpler after InvenTree 0.7.0 release
current_plugin.init(barcode_data)
if current_plugin.validate():
plugin = current_plugin
break
match_found = False
response = {}
response['barcode_data'] = barcode_data
# Matching plugin was found
if plugin is not None:
result_hash = plugin.hash()
response['hash'] = result_hash
response['plugin'] = plugin.name
# Ensure that the barcode does not already match a database entry
if plugin.getStockItem() is not None:
match_found = True
response['error'] = _('Barcode already matches Stock Item')
if plugin.getStockLocation() is not None:
match_found = True
response['error'] = _('Barcode already matches Stock Location')
if plugin.getPart() is not None:
match_found = True
response['error'] = _('Barcode already matches Part')
if not match_found:
item = plugin.getStockItemByHash()
if item is not None:
response['error'] = _('Barcode hash already matches Stock Item')
match_found = True
else:
result_hash = hash_barcode(barcode_data)
response['hash'] = result_hash
response['plugin'] = None
# Lookup stock item by hash
try:
item = StockItem.objects.get(uid=result_hash)
response['error'] = _('Barcode hash already matches Stock Item')
match_found = True
except StockItem.DoesNotExist:
pass
if not match_found:
response['success'] = _('Barcode associated with Stock Item')
# Save the barcode hash
item.uid = response['hash']
item.save()
serializer = StockItemSerializer(item, part_detail=True, location_detail=True, supplier_part_detail=True)
response['stockitem'] = serializer.data
return Response(response)
barcode_api_urls = [
path('link/', BarcodeAssign.as_view(), name='api-barcode-link'),
# Catch-all performs barcode 'scan'
re_path(r'^.*$', BarcodeScan.as_view(), name='api-barcode-scan'),
]

View File

@ -0,0 +1,144 @@
"""
The InvenTreeBarcodePlugin validates barcodes generated by InvenTree itself.
It can be used as a template for developing third-party barcode plugins.
The data format is very simple, and maps directly to database objects,
via the "id" parameter.
Parsing an InvenTree barcode simply involves validating that the
references model objects actually exist in the database.
"""
# -*- coding: utf-8 -*-
import json
from plugin import IntegrationPluginBase
from plugin.mixins import BarcodeMixin
from stock.models import StockItem, StockLocation
from part.models import Part
from rest_framework.exceptions import ValidationError
class InvenTreeBarcodePlugin(BarcodeMixin, IntegrationPluginBase):
PLUGIN_NAME = "InvenTreeBarcode"
def validate(self):
"""
An "InvenTree" barcode must be a jsonnable-dict with the following tags:
{
'tool': 'InvenTree',
'version': <anything>
}
"""
# The data must either be dict or be able to dictified
if type(self.data) is dict:
pass
elif type(self.data) is str:
try:
self.data = json.loads(self.data)
if type(self.data) is not dict:
return False
except json.JSONDecodeError:
return False
else:
return False # pragma: no cover
# If any of the following keys are in the JSON data,
# let's go ahead and assume that the code is a valid InvenTree one...
for key in ['tool', 'version', 'InvenTree', 'stockitem', 'stocklocation', 'part']:
if key in self.data.keys():
return True
return True
def getStockItem(self):
for k in self.data.keys():
if k.lower() == 'stockitem':
data = self.data[k]
pk = None
# Initially try casting to an integer
try:
pk = int(data)
except (TypeError, ValueError): # pragma: no cover
pk = None
if pk is None: # pragma: no cover
try:
pk = self.data[k]['id']
except (AttributeError, KeyError):
raise ValidationError({k: "id parameter not supplied"})
try:
item = StockItem.objects.get(pk=pk)
return item
except (ValueError, StockItem.DoesNotExist): # pragma: no cover
raise ValidationError({k: "Stock item does not exist"})
return None
def getStockLocation(self):
for k in self.data.keys():
if k.lower() == 'stocklocation':
pk = None
# First try simple integer lookup
try:
pk = int(self.data[k])
except (TypeError, ValueError): # pragma: no cover
pk = None
if pk is None: # pragma: no cover
# Lookup by 'id' field
try:
pk = self.data[k]['id']
except (AttributeError, KeyError):
raise ValidationError({k: "id parameter not supplied"})
try:
loc = StockLocation.objects.get(pk=pk)
return loc
except (ValueError, StockLocation.DoesNotExist): # pragma: no cover
raise ValidationError({k: "Stock location does not exist"})
return None
def getPart(self):
for k in self.data.keys():
if k.lower() == 'part':
pk = None
# Try integer lookup first
try:
pk = int(self.data[k])
except (TypeError, ValueError): # pragma: no cover
pk = None
if pk is None: # pragma: no cover
try:
pk = self.data[k]['id']
except (AttributeError, KeyError):
raise ValidationError({k: 'id parameter not supplied'})
try:
part = Part.objects.get(pk=pk)
return part
except (ValueError, Part.DoesNotExist): # pragma: no cover
raise ValidationError({k: 'Part does not exist'})
return None

View File

@ -0,0 +1,62 @@
# -*- coding: utf-8 -*-
"""Unit tests for InvenTreeBarcodePlugin"""
from django.contrib.auth import get_user_model
from django.urls import reverse
from rest_framework.test import APITestCase
from rest_framework import status
class TestInvenTreeBarcode(APITestCase):
fixtures = [
'category',
'part',
'location',
'stock'
]
def setUp(self):
# Create a user for auth
user = get_user_model()
user.objects.create_user('testuser', 'test@testing.com', 'password')
self.client.login(username='testuser', password='password')
def test_errors(self):
"""
Test all possible error cases for assigment action
"""
def test_assert_error(barcode_data):
response = self.client.post(
reverse('api-barcode-link'), format='json',
data={
'barcode': barcode_data,
'stockitem': 521
}
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('error', response.data)
# test with already existing stock
test_assert_error('{"stockitem": 521}')
# test with already existing stock location
test_assert_error('{"stocklocation": 7}')
# test with already existing part location
test_assert_error('{"part": 10004}')
# test with hash
test_assert_error('{"blbla": 10004}')
def test_scan(self):
"""
Test that a barcode can be scanned
"""
response = self.client.post(reverse('api-barcode-scan'), format='json', data={'barcode': 'blbla=10004'})
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('success', response.data)

View File

@ -17,7 +17,7 @@ from django.dispatch.dispatcher import receiver
from common.models import InvenTreeSetting
import common.notifications
from InvenTree.ready import canAppAccessDatabase
from InvenTree.ready import canAppAccessDatabase, isImportingData
from InvenTree.tasks import offload_task
from plugin.registry import registry
@ -113,6 +113,10 @@ def allow_table_event(table_name):
We *do not* want events to be fired for some tables!
"""
if isImportingData():
# Prevent table events during the data import process
return False
table_name = table_name.lower().strip()
# Ignore any tables which start with these prefixes

View File

@ -18,7 +18,7 @@ from ..builtin.integration.mixins import (
from common.notifications import SingleNotificationMethod, BulkNotificationMethod
from ..builtin.action.mixins import ActionMixin
from ..builtin.barcode.mixins import BarcodeMixin
from ..builtin.barcodes.mixins import BarcodeMixin
__all__ = [
'ActionMixin',

View File

@ -102,7 +102,7 @@ class PluginConfig(models.Model):
return ret
class PluginSetting(common.models.GenericReferencedSettingClass, common.models.BaseInvenTreeSetting):
class PluginSetting(common.models.BaseInvenTreeSetting):
"""
This model represents settings for individual plugins
"""
@ -112,7 +112,13 @@ class PluginSetting(common.models.GenericReferencedSettingClass, common.models.B
('plugin', 'key'),
]
REFERENCE_NAME = 'plugin'
plugin = models.ForeignKey(
PluginConfig,
related_name='settings',
null=False,
verbose_name=_('Plugin'),
on_delete=models.CASCADE,
)
@classmethod
def get_setting_definition(cls, key, **kwargs):
@ -142,16 +148,18 @@ class PluginSetting(common.models.GenericReferencedSettingClass, common.models.B
return super().get_setting_definition(key, **kwargs)
plugin = models.ForeignKey(
PluginConfig,
related_name='settings',
null=False,
verbose_name=_('Plugin'),
on_delete=models.CASCADE,
)
def get_kwargs(self):
"""
Explicit kwargs required to uniquely identify a particular setting object,
in addition to the 'key' parameter
"""
return {
'plugin': self.plugin,
}
class NotificationUserSetting(common.models.GenericReferencedSettingClass, common.models.BaseInvenTreeSetting):
class NotificationUserSetting(common.models.BaseInvenTreeSetting):
"""
This model represents notification settings for a user
"""
@ -161,8 +169,6 @@ class NotificationUserSetting(common.models.GenericReferencedSettingClass, commo
('method', 'user', 'key'),
]
REFERENCE_NAME = 'method'
@classmethod
def get_setting_definition(cls, key, **kwargs):
from common.notifications import storage
@ -171,6 +177,17 @@ class NotificationUserSetting(common.models.GenericReferencedSettingClass, commo
return super().get_setting_definition(key, **kwargs)
def get_kwargs(self):
"""
Explicit kwargs required to uniquely identify a particular setting object,
in addition to the 'key' parameter
"""
return {
'method': self.method,
'user': self.user,
}
method = models.CharField(
max_length=255,
verbose_name=_('Method'),

View File

@ -65,6 +65,16 @@ class SampleIntegrationPlugin(AppMixin, SettingsMixin, UrlsMixin, NavigationMixi
],
'default': 'A',
},
'SELECT_COMPANY': {
'name': 'Company',
'description': 'Select a company object from the database',
'model': 'company.company',
},
'SELECT_PART': {
'name': 'Part',
'description': 'Select a part object from the database',
'model': 'part.part',
},
}
NAVIGATION = [

View File

@ -6,6 +6,6 @@
<em>This location has no sublocations!</em>
<ul>
<li><b>Location Name</b>: {{ location.name }}</li>
<li><b>Location Path</b>: {{ location.pathstring }}</li>
<li><strong>Location Name</strong>: {{ location.name }}</li>
<li><strong>Location Path</strong>: {{ location.pathstring }}</li>
</ul>

View File

@ -0,0 +1,266 @@
# -*- coding: utf-8 -*-
"""
Unit tests for Barcode endpoints
"""
from django.contrib.auth import get_user_model
from django.urls import reverse
from rest_framework.test import APITestCase
from rest_framework import status
from stock.models import StockItem
class BarcodeAPITest(APITestCase):
fixtures = [
'category',
'part',
'location',
'stock'
]
def setUp(self):
# Create a user for auth
user = get_user_model()
user.objects.create_user('testuser', 'test@testing.com', 'password')
self.client.login(username='testuser', password='password')
self.scan_url = reverse('api-barcode-scan')
self.assign_url = reverse('api-barcode-link')
def postBarcode(self, url, barcode):
return self.client.post(url, format='json', data={'barcode': str(barcode)})
def test_invalid(self):
# test scan url
response = self.client.post(self.scan_url, format='json', data={})
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
# 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)
def test_empty(self):
response = self.postBarcode(self.scan_url, '')
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.data
self.assertIn('error', data)
self.assertIn('barcode_data', data)
self.assertIn('hash', data)
self.assertIn('plugin', data)
self.assertIsNone(data['plugin'])
def test_find_part(self):
"""
Test that we can lookup a part based on ID
"""
response = self.client.post(
self.scan_url,
{
'barcode': {
'part': 1,
},
},
format='json',
)
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)
def test_invalid_part(self):
"""Test response for invalid part"""
response = self.client.post(
self.scan_url,
{
'barcode': {
'part': 999999999,
}
},
format='json'
)
self.assertEqual(response.status_code, 400)
self.assertEqual(response.data['part'], 'Part does not exist')
def test_find_stock_item(self):
"""
Test that we can lookup a stock item based on ID
"""
response = self.client.post(
self.scan_url,
{
'barcode': {
'stockitem': 1,
}
},
format='json',
)
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)
def test_invalid_item(self):
"""Test response for invalid stock item"""
response = self.client.post(
self.scan_url,
{
'barcode': {
'stockitem': 999999999,
}
},
format='json'
)
self.assertEqual(response.status_code, 400)
self.assertEqual(response.data['stockitem'], 'Stock item does not exist')
def test_find_location(self):
"""
Test that we can lookup a stock location based on ID
"""
response = self.client.post(
self.scan_url,
{
'barcode': {
'stocklocation': 1,
},
},
format='json'
)
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(
self.scan_url,
{
'barcode': {
'stocklocation': 999999999,
}
},
format='json'
)
self.assertEqual(response.status_code, 400)
self.assertEqual(response.data['stocklocation'], 'Stock location does not exist')
def test_integer_barcode(self):
response = self.postBarcode(self.scan_url, '123456789')
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.data
self.assertIn('error', data)
self.assertIn('barcode_data', data)
self.assertIn('hash', data)
self.assertIn('plugin', data)
self.assertIsNone(data['plugin'])
def test_array_barcode(self):
response = self.postBarcode(self.scan_url, "['foo', 'bar']")
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.data
self.assertIn('error', data)
self.assertIn('barcode_data', data)
self.assertIn('hash', data)
self.assertIn('plugin', data)
self.assertIsNone(data['plugin'])
def test_barcode_generation(self):
item = StockItem.objects.get(pk=522)
response = self.postBarcode(self.scan_url, item.format_barcode())
data = response.data
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('stockitem', data)
pk = data['stockitem']['pk']
self.assertEqual(pk, item.pk)
def test_association(self):
"""
Test that a barcode can be associated with a StockItem
"""
item = StockItem.objects.get(pk=522)
self.assertEqual(len(item.uid), 0)
barcode_data = 'A-TEST-BARCODE-STRING'
response = self.client.post(
self.assign_url, format='json',
data={
'barcode': barcode_data,
'stockitem': item.pk
}
)
data = response.data
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('success', data)
result_hash = data['hash']
# Read the item out from the database again
item = StockItem.objects.get(pk=522)
self.assertEqual(result_hash, item.uid)
# Ensure that the same UID cannot be assigned to a different stock item!
response = self.client.post(
self.assign_url, format='json',
data={
'barcode': barcode_data,
'stockitem': 521
}
)
data = response.data
self.assertIn('error', data)
self.assertNotIn('success', data)