[Plugin] Allow custom plugins for running validation routines (#3776)

* Adds new plugin mixin for performing custom validation steps

* Adds simple test plugin for custom validation

* Run part name and IPN validators checks through loaded plugins

* Expose more validation functions to plugins:

- SalesOrder reference
- PurchaseOrder reference
- BuildOrder reference

* Remove custom validation of reference fields

- For now, this is too complex to consider given the current incrementing-reference implementation
- Might revisit this at a later stage.

* Custom validation of serial numbers:

- Replace "checkIfSerialNumberExists" method with "validate_serial_number"
- Pass serial number through to custom plugins
- General code / docstring improvements

* Update unit tests

* Update InvenTree/stock/tests.py

Co-authored-by: Matthias Mair <code@mjmair.com>

* Adds global setting to specify whether serial numbers must be unique globally

- Default is false to preserve behaviour

* Improved error message when attempting to create stock item with invalid serial numbers

* Add more detail to existing serial error message

* Add unit testing for serial number uniqueness

* Allow plugins to convert a serial number to an integer (for optimized sorting)

* Add placeholder plugin methods for incrementing and decrementing serial numbers

* Typo fix

* Add improved method for determining the "latest" serial number

* Remove calls to getLatestSerialNumber

* Update validate_serial_number method

- Add option to disable checking for duplicates
- Don't pass optional StockItem through to plugins

* Refactor serial number extraction methods

- Expose the "incrementing" portion to external plugins

* Bug fixes

* Update unit tests

* Fix for get_latest_serial_number

* Ensure custom serial integer values are clipped

* Adds a plugin for validating and generating hexadecimal serial numbers

* Update unit tests

* Add stub methods for batch code functionality

* remove "hex serials" plugin

- Was simply breaking unit tests

* Allow custom plugins to generate and validate batch codes

- Perform batch code validation when StockItem is saved
- Improve display of error message in modal forms

* Fix unit tests for stock app

* Log message if plugin has a duplicate slug

* Unit test fix

Co-authored-by: Matthias Mair <code@mjmair.com>
This commit is contained in:
Oliver 2022-10-18 23:50:07 +11:00 committed by GitHub
parent 269b269de3
commit 906ae218aa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
24 changed files with 755 additions and 289 deletions

View File

@ -342,7 +342,7 @@ def normalize(d):
return d.quantize(Decimal(1)) if d == d.to_integral() else d.normalize()
def increment(n):
def increment(value):
"""Attempt to increment an integer (or a string that looks like an integer).
e.g.
@ -351,12 +351,14 @@ def increment(n):
2 -> 3
AB01 -> AB02
QQQ -> QQQ
"""
value = str(n).strip()
value = str(value).strip()
# Ignore empty strings
if not value:
return value
if value in ['', None]:
# Provide a default value if provided with a null input
return '1'
pattern = r"(.*?)(\d+)?$"
@ -542,138 +544,211 @@ def DownloadFile(data, filename, content_type='application/text', inline=False)
return response
def extract_serial_numbers(serials, expected_quantity, next_number: int):
"""Attempt to extract serial numbers from an input string.
def increment_serial_number(serial: str):
"""Given a serial number, (attempt to) generate the *next* serial number.
Requirements:
- Serial numbers can be either strings, or integers
- Serial numbers can be split by whitespace / newline / commma chars
- Serial numbers can be supplied as an inclusive range using hyphen char e.g. 10-20
- Serial numbers can be defined as ~ for getting the next available serial number
- Serial numbers can be supplied as <start>+ for getting all expecteded numbers starting from <start>
- Serial numbers can be supplied as <start>+<length> for getting <length> numbers starting from <start>
Note: This method is exposed to custom plugins.
Args:
serials: input string with patterns
expected_quantity: The number of (unique) serial numbers we expect
next_number(int): the next possible serial number
Arguments:
serial: The serial number which should be incremented
Returns:
incremented value, or None if incrementing could not be performed.
"""
serials = serials.strip()
# fill in the next serial number into the serial
while '~' in serials:
serials = serials.replace('~', str(next_number), 1)
next_number += 1
from plugin.registry import registry
# Split input string by whitespace or comma (,) characters
groups = re.split(r"[\s,]+", serials)
# Ensure we start with a string value
if serial is not None:
serial = str(serial).strip()
numbers = []
errors = []
# First, let any plugins attempt to increment the serial number
for plugin in registry.with_mixin('validation'):
result = plugin.increment_serial_number(serial)
if result is not None:
return str(result)
# Helper function to check for duplicated numbers
def add_sn(sn):
# Attempt integer conversion first, so numerical strings are never stored
try:
sn = int(sn)
except ValueError:
pass
# If we get to here, no plugins were able to "increment" the provided serial value
# Attempt to perform increment according to some basic rules
return increment(serial)
if sn in numbers:
errors.append(_('Duplicate serial: {sn}').format(sn=sn))
else:
numbers.append(sn)
def extract_serial_numbers(input_string, expected_quantity: int, starting_value=None):
"""Extract a list of serial numbers from a provided input string.
The input string can be specified using the following concepts:
- Individual serials are separated by comma: 1, 2, 3, 6,22
- Sequential ranges with provided limits are separated by hyphens: 1-5, 20 - 40
- The "next" available serial number can be specified with the tilde (~) character
- Serial numbers can be supplied as <start>+ for getting all expecteded numbers starting from <start>
- Serial numbers can be supplied as <start>+<length> for getting <length> numbers starting from <start>
Actual generation of sequential serials is passed to the 'validation' plugin mixin,
allowing custom plugins to determine how serial values are incremented.
Arguments:
input_string: Input string with specified serial numbers (string, or integer)
expected_quantity: The number of (unique) serial numbers we expect
starting_value: Provide a starting value for the sequence (or None)
"""
if starting_value is None:
starting_value = increment_serial_number(None)
try:
expected_quantity = int(expected_quantity)
except ValueError:
raise ValidationError([_("Invalid quantity provided")])
if len(serials) == 0:
if input_string:
input_string = str(input_string).strip()
else:
input_string = ''
if len(input_string) == 0:
raise ValidationError([_("Empty serial number string")])
# If the user has supplied the correct number of serials, don't process them for groups
# just add them so any duplicates (or future validations) are checked
next_value = increment_serial_number(starting_value)
# Substitute ~ character with latest value
while '~' in input_string and next_value:
input_string = input_string.replace('~', str(next_value), 1)
next_value = increment_serial_number(next_value)
# Split input string by whitespace or comma (,) characters
groups = re.split(r"[\s,]+", input_string)
serials = []
errors = []
def add_error(error: str):
"""Helper function for adding an error message"""
if error not in errors:
errors.append(error)
def add_serial(serial):
"""Helper function to check for duplicated values"""
if serial in serials:
add_error(_("Duplicate serial") + f": {serial}")
else:
serials.append(serial)
# If the user has supplied the correct number of serials, do not split into groups
if len(groups) == expected_quantity:
for group in groups:
add_sn(group)
add_serial(group)
if len(errors) > 0:
raise ValidationError(errors)
return numbers
else:
return serials
for group in groups:
group = group.strip()
# Hyphen indicates a range of numbers
if '-' in group:
"""Hyphen indicates a range of values:
e.g. 10-20
"""
items = group.split('-')
if len(items) == 2 and all([i.isnumeric() for i in items]):
a = items[0].strip()
b = items[1].strip()
if len(items) == 2:
a = items[0]
b = items[1]
try:
a = int(a)
b = int(b)
if a < b:
for n in range(a, b + 1):
add_sn(n)
else:
errors.append(_("Invalid group range: {g}").format(g=group))
except ValueError:
errors.append(_("Invalid group: {g}").format(g=group))
if a == b:
# Invalid group
add_error(_("Invalid group range: {g}").format(g=group))
continue
else:
# More than 2 hyphens or non-numeric group so add without interpolating
add_sn(group)
# plus signals either
# 1: 'start+': expected number of serials, starting at start
# 2: 'start+number': number of serials, starting at start
group_items = []
count = 0
a_next = a
while a_next is not None and a_next not in group_items:
group_items.append(a_next)
count += 1
# Progress to the 'next' sequential value
a_next = str(increment_serial_number(a_next))
if a_next == b:
# Successfully got to the end of the range
group_items.append(b)
break
elif count > expected_quantity:
# More than the allowed number of items
break
elif a_next is None:
break
if len(group_items) > 0 and group_items[0] == a and group_items[-1] == b:
# In this case, the range extraction looks like it has worked
for item in group_items:
add_serial(item)
else:
add_serial(group)
# add_error(_("Invalid group range: {g}").format(g=group))
else:
# In the case of a different number of hyphens, simply add the entire group
add_serial(group)
elif '+' in group:
"""Plus character (+) indicates either:
- <start>+ - Expected number of serials, beginning at the specified 'start' character
- <start>+<num> - Specified number of serials, beginning at the specified 'start' character
"""
items = group.split('+')
# case 1, 2
if len(items) == 2:
start = int(items[0])
sequence_items = []
counter = 0
sequence_count = max(0, expected_quantity - len(serials))
# case 2
if bool(items[1]):
end = start + int(items[1]) + 1
if len(items) > 2 or len(items) == 0:
add_error(_("Invalid group sequence: {g}").format(g=group))
continue
elif len(items) == 2:
try:
if items[1] not in ['', None]:
sequence_count = int(items[1]) + 1
except ValueError:
add_error(_("Invalid group sequence: {g}").format(g=group))
continue
# case 1
else:
end = start + (expected_quantity - len(numbers))
value = items[0]
for n in range(start, end):
add_sn(n)
# no case
# Keep incrementing up to the specified quantity
while value is not None and value not in sequence_items and counter < sequence_count:
sequence_items.append(value)
value = increment_serial_number(value)
counter += 1
if len(sequence_items) == sequence_count:
for item in sequence_items:
add_serial(item)
else:
errors.append(_("Invalid group sequence: {g}").format(g=group))
add_error(_("Invalid group sequence: {g}").format(g=group))
# At this point, we assume that the "group" is just a single serial value
elif group:
add_sn(group)
# No valid input group detected
else:
raise ValidationError(_(f"Invalid/no group {group}"))
# At this point, we assume that the 'group' is just a single serial value
add_serial(group)
if len(errors) > 0:
raise ValidationError(errors)
if len(numbers) == 0:
if len(serials) == 0:
raise ValidationError([_("No serial numbers found")])
# The number of extracted serial numbers must match the expected quantity
if expected_quantity != len(numbers):
raise ValidationError([_("Number of unique serial numbers ({s}) must match quantity ({q})").format(s=len(numbers), q=expected_quantity)])
if len(serials) != expected_quantity:
raise ValidationError([_("Number of unique serial numbers ({s}) must match quantity ({q})").format(s=len(serials), q=expected_quantity)])
return numbers
return serials
def validateFilterString(value, model=None):

View File

@ -679,6 +679,10 @@ main {
color: #A94442;
}
.form-error-message {
display: block;
}
.modal input {
width: 100%;
}

View File

@ -39,8 +39,9 @@ class ValidatorTest(TestCase):
"""Test part name validator."""
validate_part_name('hello world')
# Validate with some strange chars
with self.assertRaises(django_exceptions.ValidationError):
validate_part_name('This | name is not } valid')
validate_part_name('### <> This | name is not } valid')
def test_overage(self):
"""Test overage validator."""
@ -309,7 +310,7 @@ class TestIncrement(TestCase):
def tests(self):
"""Test 'intelligent' incrementing function."""
tests = [
("", ""),
("", '1'),
(1, "2"),
("001", "002"),
("1001", "1002"),
@ -418,7 +419,11 @@ class TestMPTT(TestCase):
class TestSerialNumberExtraction(TestCase):
"""Tests for serial number extraction code."""
"""Tests for serial number extraction code.
Note that while serial number extraction is made available to custom plugins,
only simple integer-based extraction is tested here.
"""
def test_simple(self):
"""Test simple serial numbers."""
@ -427,7 +432,7 @@ class TestSerialNumberExtraction(TestCase):
sn = e("1-5", 5, 1)
self.assertEqual(len(sn), 5, 1)
for i in range(1, 6):
self.assertIn(i, sn)
self.assertIn(str(i), sn)
sn = e("1, 2, 3, 4, 5", 5, 1)
self.assertEqual(len(sn), 5)
@ -435,55 +440,55 @@ class TestSerialNumberExtraction(TestCase):
# Test partially specifying serials
sn = e("1, 2, 4+", 5, 1)
self.assertEqual(len(sn), 5)
self.assertEqual(sn, [1, 2, 4, 5, 6])
self.assertEqual(sn, ['1', '2', '4', '5', '6'])
# Test groups are not interpolated if enough serials are supplied
sn = e("1, 2, 3, AF5-69H, 5", 5, 1)
self.assertEqual(len(sn), 5)
self.assertEqual(sn, [1, 2, 3, "AF5-69H", 5])
self.assertEqual(sn, ['1', '2', '3', 'AF5-69H', '5'])
# Test groups are not interpolated with more than one hyphen in a word
sn = e("1, 2, TG-4SR-92, 4+", 5, 1)
self.assertEqual(len(sn), 5)
self.assertEqual(sn, [1, 2, "TG-4SR-92", 4, 5])
self.assertEqual(sn, ['1', '2', "TG-4SR-92", '4', '5'])
# Test groups are not interpolated with alpha characters
sn = e("1, A-2, 3+", 5, 1)
self.assertEqual(len(sn), 5)
self.assertEqual(sn, [1, "A-2", 3, 4, 5])
self.assertEqual(sn, ['1', "A-2", '3', '4', '5'])
# Test multiple placeholders
sn = e("1 2 ~ ~ ~", 5, 3)
sn = e("1 2 ~ ~ ~", 5, 2)
self.assertEqual(len(sn), 5)
self.assertEqual(sn, [1, 2, 3, 4, 5])
self.assertEqual(sn, ['1', '2', '3', '4', '5'])
sn = e("1-5, 10-15", 11, 1)
self.assertIn(3, sn)
self.assertIn(13, sn)
self.assertIn('3', sn)
self.assertIn('13', sn)
sn = e("1+", 10, 1)
self.assertEqual(len(sn), 10)
self.assertEqual(sn, [_ for _ in range(1, 11)])
self.assertEqual(sn, [str(_) for _ in range(1, 11)])
sn = e("4, 1+2", 4, 1)
self.assertEqual(len(sn), 4)
self.assertEqual(sn, [4, 1, 2, 3])
self.assertEqual(sn, ['4', '1', '2', '3'])
sn = e("~", 1, 1)
self.assertEqual(len(sn), 1)
self.assertEqual(sn, [1])
self.assertEqual(sn, ['2'])
sn = e("~", 1, 3)
self.assertEqual(len(sn), 1)
self.assertEqual(sn, [3])
self.assertEqual(sn, ['4'])
sn = e("~+", 2, 5)
sn = e("~+", 2, 4)
self.assertEqual(len(sn), 2)
self.assertEqual(sn, [5, 6])
self.assertEqual(sn, ['5', '6'])
sn = e("~+3", 4, 5)
sn = e("~+3", 4, 4)
self.assertEqual(len(sn), 4)
self.assertEqual(sn, [5, 6, 7, 8])
self.assertEqual(sn, ['5', '6', '7', '8'])
def test_failures(self):
"""Test wron serial numbers."""
@ -522,19 +527,19 @@ class TestSerialNumberExtraction(TestCase):
sn = e("1 3-5 9+2", 7, 1)
self.assertEqual(len(sn), 7)
self.assertEqual(sn, [1, 3, 4, 5, 9, 10, 11])
self.assertEqual(sn, ['1', '3', '4', '5', '9', '10', '11'])
sn = e("1,3-5,9+2", 7, 1)
self.assertEqual(len(sn), 7)
self.assertEqual(sn, [1, 3, 4, 5, 9, 10, 11])
self.assertEqual(sn, ['1', '3', '4', '5', '9', '10', '11'])
sn = e("~+2", 3, 14)
sn = e("~+2", 3, 13)
self.assertEqual(len(sn), 3)
self.assertEqual(sn, [14, 15, 16])
self.assertEqual(sn, ['14', '15', '16'])
sn = e("~+", 2, 14)
sn = e("~+", 2, 13)
self.assertEqual(len(sn), 2)
self.assertEqual(sn, [14, 15])
self.assertEqual(sn, ['14', '15'])
class TestVersionNumber(TestCase):

View File

@ -47,16 +47,40 @@ class AllowedURLValidator(validators.URLValidator):
def validate_part_name(value):
"""Prevent some illegal characters in part names."""
for c in ['|', '#', '$', '{', '}']:
if c in str(value):
raise ValidationError(
_('Invalid character in part name')
)
"""Validate the name field for a Part instance
This function is exposed to any Validation plugins, and thus can be customized.
"""
from plugin.registry import registry
for plugin in registry.with_mixin('validation'):
# Run the name through each custom validator
# If the plugin returns 'True' we will skip any subsequent validation
if plugin.validate_part_name(value):
return
def validate_part_ipn(value):
"""Validate the Part IPN against regex rule."""
"""Validate the IPN field for a Part instance.
This function is exposed to any Validation plugins, and thus can be customized.
If no validation errors are raised, the IPN is also validated against a configurable regex pattern.
"""
from plugin.registry import registry
plugins = registry.with_mixin('validation')
for plugin in plugins:
# Run the IPN through each custom validator
# If the plugin returns 'True' we will skip any subsequent validation
if plugin.validate_part_ipn(value):
return
# If we get to here, none of the plugins have raised an error
pattern = common.models.InvenTreeSetting.get_setting('PART_IPN_REGEX')
if pattern:
@ -68,28 +92,25 @@ def validate_part_ipn(value):
def validate_purchase_order_reference(value):
"""Validate the 'reference' field of a PurchaseOrder."""
pattern = common.models.InvenTreeSetting.get_setting('PURCHASEORDER_REFERENCE_REGEX')
if pattern:
match = re.search(pattern, value)
from order.models import PurchaseOrder
if match is None:
raise ValidationError(_('Reference must match pattern {pattern}').format(pattern=pattern))
# If we get to here, run the "default" validation routine
PurchaseOrder.validate_reference_field(value)
def validate_sales_order_reference(value):
"""Validate the 'reference' field of a SalesOrder."""
pattern = common.models.InvenTreeSetting.get_setting('SALESORDER_REFERENCE_REGEX')
if pattern:
match = re.search(pattern, value)
from order.models import SalesOrder
if match is None:
raise ValidationError(_('Reference must match pattern {pattern}').format(pattern=pattern))
# If we get to here, run the "default" validation routine
SalesOrder.validate_reference_field(value)
def validate_tree_name(value):
"""Placeholder for legacy function used in migrations."""
...
def validate_overage(value):

View File

@ -14,7 +14,6 @@ from InvenTree.serializers import InvenTreeModelSerializer, InvenTreeAttachmentS
from InvenTree.serializers import UserSerializer
import InvenTree.helpers
from InvenTree.helpers import extract_serial_numbers
from InvenTree.serializers import InvenTreeDecimalField
from InvenTree.status_codes import StockStatus
@ -260,7 +259,11 @@ class BuildOutputCreateSerializer(serializers.Serializer):
if serial_numbers:
try:
self.serials = extract_serial_numbers(serial_numbers, quantity, part.getLatestSerialNumberInt())
self.serials = InvenTree.helpers.extract_serial_numbers(
serial_numbers,
quantity,
part.get_latest_serial_number()
)
except DjangoValidationError as e:
raise ValidationError({
'serial_numbers': e.messages,
@ -270,12 +273,12 @@ class BuildOutputCreateSerializer(serializers.Serializer):
existing = []
for serial in self.serials:
if part.checkIfSerialNumberExists(serial):
if not part.validate_serial_number(serial):
existing.append(serial)
if len(existing) > 0:
msg = _("The following serial numbers already exist")
msg = _("The following serial numbers already exist or are invalid")
msg += " : "
msg += ",".join([str(e) for e in existing])

View File

@ -389,7 +389,7 @@ class BuildTest(BuildAPITest):
expected_code=400,
)
self.assertIn('The following serial numbers already exist : 1,2,3', str(response.data))
self.assertIn('The following serial numbers already exist or are invalid : 1,2,3', str(response.data))
# Double check no new outputs have been created
self.assertEqual(n_outputs + 5, bo.output_count)

View File

@ -18,8 +18,9 @@ def validate_build_order_reference_pattern(pattern):
def validate_build_order_reference(value):
"""Validate that the BuildOrder reference field matches the required pattern"""
"""Validate that the BuildOrder reference field matches the required pattern."""
from build.models import Build
# If we get to here, run the "default" validation routine
Build.validate_reference_field(value)

View File

@ -1139,6 +1139,13 @@ class InvenTreeSetting(BaseInvenTreeSetting):
'validator': bool,
},
'SERIAL_NUMBER_GLOBALLY_UNIQUE': {
'name': _('Globally Unique Serials'),
'description': _('Serial numbers for stock items must be globally unique'),
'default': False,
'validator': bool,
},
'STOCK_BATCH_CODE_TEMPLATE': {
'name': _('Batch Code Template'),
'description': _('Template for generating default batch codes for stock items'),

View File

@ -531,7 +531,11 @@ class PurchaseOrderLineItemReceiveSerializer(serializers.Serializer):
if serial_numbers:
try:
# Pass the serial numbers through to the parent serializer once validated
data['serials'] = extract_serial_numbers(serial_numbers, pack_quantity, base_part.getLatestSerialNumberInt())
data['serials'] = extract_serial_numbers(
serial_numbers,
pack_quantity,
base_part.get_latest_serial_number()
)
except DjangoValidationError as e:
raise ValidationError({
'serial_numbers': e.messages,
@ -1256,7 +1260,11 @@ class SalesOrderSerialAllocationSerializer(serializers.Serializer):
part = line_item.part
try:
data['serials'] = extract_serial_numbers(serial_numbers, quantity, part.getLatestSerialNumberInt())
data['serials'] = extract_serial_numbers(
serial_numbers,
quantity,
part.get_latest_serial_number()
)
except DjangoValidationError as e:
raise ValidationError({
'serial_numbers': e.messages,

View File

@ -25,8 +25,8 @@ from company.models import Company, ManufacturerPart, SupplierPart
from InvenTree.api import (APIDownloadMixin, AttachmentMixin,
ListCreateDestroyAPIView)
from InvenTree.filters import InvenTreeOrderingFilter
from InvenTree.helpers import (DownloadFile, increment, isNull, str2bool,
str2int)
from InvenTree.helpers import (DownloadFile, increment_serial_number, isNull,
str2bool, str2int)
from InvenTree.mixins import (CreateAPI, ListAPI, ListCreateAPI, RetrieveAPI,
RetrieveUpdateAPI, RetrieveUpdateDestroyAPI,
UpdateAPI)
@ -717,16 +717,16 @@ class PartSerialNumberDetail(RetrieveAPI):
part = self.get_object()
# Calculate the "latest" serial number
latest = part.getLatestSerialNumber()
latest = part.get_latest_serial_number()
data = {
'latest': latest,
}
if latest is not None:
next_serial = increment(latest)
next_serial = increment_serial_number(latest)
if next_serial != increment:
if next_serial != latest:
data['next'] = next_serial
return Response(data)

View File

@ -529,112 +529,126 @@ class Part(InvenTreeBarcodeMixin, MetadataMixin, MPTTModel):
return result
def checkIfSerialNumberExists(self, sn, exclude_self=False):
"""Check if a serial number exists for this Part.
def validate_serial_number(self, serial: str, stock_item=None, check_duplicates=True, raise_error=False):
"""Validate a serial number against this Part instance.
Note: Serial numbers must be unique across an entire Part "tree", so here we filter by the entire tree.
Note: This function is exposed to any Validation plugins, and thus can be customized.
Any plugins which implement the 'validate_serial_number' method have three possible outcomes:
- Decide the serial is objectionable and raise a django.core.exceptions.ValidationError
- Decide the serial is acceptable, and return None to proceed to other tests
- Decide the serial is acceptable, and return True to skip any further tests
Arguments:
serial: The proposed serial number
stock_item: (optional) A StockItem instance which has this serial number assigned (e.g. testing for duplicates)
raise_error: If False, and ValidationError(s) will be handled
Returns:
True if serial number is 'valid' else False
Raises:
ValidationError if serial number is invalid and raise_error = True
"""
parts = Part.objects.filter(tree_id=self.tree_id)
stock = StockModels.StockItem.objects.filter(part__in=parts, serial=sn)
serial = str(serial).strip()
if exclude_self:
stock = stock.exclude(pk=self.pk)
# First, throw the serial number against each of the loaded validation plugins
from plugin.registry import registry
return stock.exists()
try:
for plugin in registry.with_mixin('validation'):
# Run the serial number through each custom validator
# If the plugin returns 'True' we will skip any subsequent validation
if plugin.validate_serial_number(serial):
return True
except ValidationError as exc:
if raise_error:
# Re-throw the error
raise exc
else:
return False
def find_conflicting_serial_numbers(self, serials):
"""
If we are here, none of the loaded plugins (if any) threw an error or exited early
Now, we run the "default" serial number validation routine,
which checks that the serial number is not duplicated
"""
if not check_duplicates:
return
from part.models import Part
from stock.models import StockItem
if common.models.InvenTreeSetting.get_setting('SERIAL_NUMBER_GLOBALLY_UNIQUE', False):
# Serial number must be unique across *all* parts
parts = Part.objects.all()
else:
# Serial number must only be unique across this part "tree"
parts = Part.objects.filter(tree_id=self.tree_id)
stock = StockItem.objects.filter(part__in=parts, serial=serial)
if stock_item:
# Exclude existing StockItem from query
stock = stock.exclude(pk=stock_item.pk)
if stock.exists():
if raise_error:
raise ValidationError(_("Stock item with this serial number already exists") + ": " + serial)
else:
return False
else:
# This serial number is perfectly valid
return True
def find_conflicting_serial_numbers(self, serials: list):
"""For a provided list of serials, return a list of those which are conflicting."""
conflicts = []
for serial in serials:
if self.checkIfSerialNumberExists(serial, exclude_self=True):
if not self.validate_serial_number(serial):
conflicts.append(serial)
return conflicts
def getLatestSerialNumber(self):
"""Return the "latest" serial number for this Part.
def get_latest_serial_number(self):
"""Find the 'latest' serial number for this Part.
If *all* the serial numbers are integers, then this will return the highest one.
Otherwise, it will simply return the serial number most recently added.
Here we attempt to find the "highest" serial number which exists for this Part.
There are a number of edge cases where this method can fail,
but this is accepted to keep database performance at a reasonable level.
Note: Serial numbers must be unique across an entire Part "tree",
so we filter by the entire tree.
"""
parts = Part.objects.filter(tree_id=self.tree_id)
stock = StockModels.StockItem.objects.filter(part__in=parts).exclude(serial=None)
# There are no matchin StockItem objects (skip further tests)
Returns:
The latest serial number specified for this part, or None
"""
stock = StockModels.StockItem.objects.all().exclude(serial=None).exclude(serial='')
# Generate a query for any stock items for this part variant tree with non-empty serial numbers
if common.models.InvenTreeSetting.get_setting('SERIAL_NUMBER_GLOBALLY_UNIQUE', False):
# Serial numbers are unique across all parts
pass
else:
# Serial numbers are unique acros part trees
stock = stock.filter(part__tree_id=self.tree_id)
# There are no matching StockItem objects (skip further tests)
if not stock.exists():
return None
# Attempt to coerce the returned serial numbers to integers
# If *any* are not integers, fail!
try:
ordered = sorted(stock.all(), reverse=True, key=lambda n: int(n.serial))
# Sort in descending order
stock = stock.order_by('-serial_int', '-serial', '-pk')
if len(ordered) > 0:
return ordered[0].serial
# One or more of the serial numbers was non-numeric
# In this case, the "best" we can do is return the most recent
except ValueError:
return stock.last().serial
# No serial numbers found
return None
def getLatestSerialNumberInt(self):
"""Return the "latest" serial number for this Part as a integer.
If it is not an integer the result is 0
"""
latest = self.getLatestSerialNumber()
# No serial number = > 0
if latest is None:
latest = 0
# Attempt to turn into an integer and return
try:
latest = int(latest)
return latest
except Exception:
# not an integer so 0
return 0
def getSerialNumberString(self, quantity=1):
"""Return a formatted string representing the next available serial numbers, given a certain quantity of items."""
latest = self.getLatestSerialNumber()
quantity = int(quantity)
# No serial numbers can be found, assume 1 as the first serial
if latest is None:
latest = 0
# Attempt to turn into an integer
try:
latest = int(latest)
except Exception:
pass
if type(latest) is int:
if quantity >= 2:
text = '{n} - {m}'.format(n=latest + 1, m=latest + 1 + quantity)
return _('Next available serial numbers are') + ' ' + text
else:
text = str(latest + 1)
return _('Next available serial number is') + ' ' + text
else:
# Non-integer values, no option but to return latest
return _('Most recent serial number is') + ' ' + str(latest)
# Return the first serial value
return stock[0].serial
@property
def full_name(self):

View File

@ -323,12 +323,13 @@
{% endif %}
</td>
</tr>
{% if part.trackable and part.getLatestSerialNumber %}
{% with part.get_latest_serial_number as sn %}
{% if part.trackable and sn %}
<tr>
<td><span class='fas fa-hashtag'></span></td>
<td>{% trans "Latest Serial Number" %}</td>
<td>
{{ part.getLatestSerialNumber }}
{{ sn }}
<div class='btn-group float-right' role='group'>
<a class='btn btn-small btn-outline-secondary text-sm' href='#' id='serial-number-search' title='{% trans "Search for serial number" %}'>
<span class='fas fa-search'></span>
@ -337,6 +338,7 @@
</td>
</tr>
{% endif %}
{% endwith %}
{% if part.default_location %}
<tr>
<td><span class='fas fa-search-location'></span></td>

View File

@ -214,6 +214,139 @@ class ScheduleMixin:
logger.warning("unregister_tasks failed, database not ready")
class ValidationMixin:
"""Mixin class that allows custom validation for various parts of InvenTree
Custom generation and validation functionality can be provided for:
- Part names
- Part IPN (internal part number) values
- Serial numbers
- Batch codes
Notes:
- Multiple ValidationMixin plugins can be used simultaneously
- The stub methods provided here generally return None (null value).
- The "first" plugin to return a non-null value for a particular method "wins"
- In the case of "validation" functions, all loaded plugins are checked until an exception is thrown
Implementing plugins may override any of the following methods which are of interest.
For 'validation' methods, there are three 'acceptable' outcomes:
- The method determines that the value is 'invalid' and raises a django.core.exceptions.ValidationError
- The method passes and returns None (the code then moves on to the next plugin)
- The method passes and returns True (and no subsequent plugins are checked)
"""
class MixinMeta:
"""Metaclass for this mixin"""
MIXIN_NAME = "Validation"
def __init__(self):
"""Register the mixin"""
super().__init__()
self.add_mixin('validation', True, __class__)
def validate_part_name(self, name: str):
"""Perform validation on a proposed Part name
Arguments:
name: The proposed part name
Returns:
None or True
Raises:
ValidationError if the proposed name is objectionable
"""
return None
def validate_part_ipn(self, ipn: str):
"""Perform validation on a proposed Part IPN (internal part number)
Arguments:
ipn: The proposed part IPN
Returns:
None or True
Raises:
ValidationError if the proposed IPN is objectionable
"""
return None
def validate_batch_code(self, batch_code: str):
"""Validate the supplied batch code
Arguments:
batch_code: The proposed batch code (string)
Returns:
None or True
Raises:
ValidationError if the proposed batch code is objectionable
"""
return None
def generate_batch_code(self):
"""Generate a new batch code
Returns:
A new batch code (string) or None
"""
return None
def validate_serial_number(self, serial: str):
"""Validate the supplied serial number
Arguments:
serial: The proposed serial number (string)
Returns:
None or True
Raises:
ValidationError if the proposed serial is objectionable
"""
return None
def convert_serial_to_int(self, serial: str):
"""Convert a serial number (string) into an integer representation.
This integer value is used for efficient sorting based on serial numbers.
A plugin which implements this method can either return:
- An integer based on the serial string, according to some algorithm
- A fixed value, such that serial number sorting reverts to the string representation
- None (null value) to let any other plugins perform the converrsion
Note that there is no requirement for the returned integer value to be unique.
Arguments:
serial: Serial value (string)
Returns:
integer representation of the serial number, or None
"""
return None
def increment_serial_number(self, serial: str):
"""Return the next sequential serial based on the provided value.
A plugin which implements this method can either return:
- A string which represents the "next" serial number in the sequence
- None (null value) if the next value could not be determined
Arguments:
serial: Current serial value (string)
"""
return None
class UrlsMixin:
"""Mixin that enables custom URLs for the plugin."""

View File

@ -8,7 +8,8 @@ from ..base.barcodes.mixins import BarcodeMixin
from ..base.event.mixins import EventMixin
from ..base.integration.mixins import (APICallMixin, AppMixin, NavigationMixin,
PanelMixin, ScheduleMixin,
SettingsMixin, UrlsMixin)
SettingsMixin, UrlsMixin,
ValidationMixin)
from ..base.label.mixins import LabelPrintingMixin
from ..base.locate.mixins import LocateMixin
@ -25,6 +26,7 @@ __all__ = [
'ActionMixin',
'BarcodeMixin',
'LocateMixin',
'ValidationMixin',
'SingleNotificationMethod',
'BulkNotificationMethod',
]

View File

@ -398,6 +398,7 @@ class PluginsRegistry:
plg_db = None
except (IntegrityError) as error: # pragma: no cover
logger.error(f"Error initializing plugin `{plg_name}`: {error}")
handle_error(error, log_name='init')
# Append reference to plugin
plg.db = plg_db

View File

@ -0,0 +1,79 @@
"""Sample plugin which demonstrates custom validation functionality"""
from datetime import datetime
from django.core.exceptions import ValidationError
from plugin import InvenTreePlugin
from plugin.mixins import SettingsMixin, ValidationMixin
class CustomValidationMixin(SettingsMixin, ValidationMixin, InvenTreePlugin):
"""A sample plugin class for demonstrating custom validation functions"""
NAME = "CustomValidator"
SLUG = "validator"
TITLE = "Custom Validator Plugin"
DESCRIPTION = "A sample plugin for demonstrating custom validation functionality"
VERSION = "0.1"
SETTINGS = {
'ILLEGAL_PART_CHARS': {
'name': 'Illegal Part Characters',
'description': 'Characters which are not allowed to appear in Part names',
'default': '!@#$%^&*()~`'
},
'IPN_MUST_CONTAIN_Q': {
'name': 'IPN Q Requirement',
'description': 'Part IPN field must contain the character Q',
'default': False,
'validator': bool,
},
'SERIAL_MUST_BE_PALINDROME': {
'name': 'Palindromic Serials',
'description': 'Serial numbers must be palindromic',
'default': False,
'validator': bool,
},
'BATCH_CODE_PREFIX': {
'name': 'Batch prefix',
'description': 'Required prefix for batch code',
'default': '',
}
}
def validate_part_name(self, name: str):
"""Validate part name"""
illegal_chars = self.get_setting('ILLEGAL_PART_CHARS')
for c in illegal_chars:
if c in name:
raise ValidationError(f"Illegal character in part name: '{c}'")
def validate_part_ipn(self, ipn: str):
"""Validate part IPN"""
if self.get_setting('IPN_MUST_CONTAIN_Q') and 'Q' not in ipn:
raise ValidationError("IPN must contain 'Q'")
def validate_serial_number(self, serial: str):
"""Validate serial number for a given StockItem"""
if self.get_setting('SERIAL_MUST_BE_PALINDROME'):
if serial != serial[::-1]:
raise ValidationError("Serial must be a palindrome")
def validate_batch_code(self, batch_code: str):
"""Ensure that a particular batch code meets specification"""
prefix = self.get_setting('BATCH_CODE_PREFIX')
if not batch_code.startswith(prefix):
raise ValidationError(f"Batch code must start with '{prefix}'")
def generate_batch_code(self):
"""Generate a new batch code."""
now = datetime.now()
return f"BATCH-{now.year}:{now.month}:{now.day}"

View File

@ -563,23 +563,35 @@ class StockList(APIDownloadMixin, ListCreateDestroyAPIView):
# If serial numbers are specified, check that they match!
try:
serials = extract_serial_numbers(serial_numbers, quantity, part.getLatestSerialNumberInt())
serials = extract_serial_numbers(
serial_numbers,
quantity,
part.get_latest_serial_number()
)
# Determine if any of the specified serial numbers already exist!
existing = []
# Determine if any of the specified serial numbers are invalid
# Note "invalid" means either they already exist, or do not pass custom rules
invalid = []
errors = []
for serial in serials:
if part.checkIfSerialNumberExists(serial):
existing.append(serial)
try:
part.validate_serial_number(serial, raise_error=True)
except DjangoValidationError as exc:
# Catch raised error to extract specific error information
invalid.append(serial)
if len(existing) > 0:
if exc.message not in errors:
errors.append(exc.message)
msg = _("The following serial numbers already exist")
if len(errors) > 0:
msg = _("The following serial numbers already exist or are invalid")
msg += " : "
msg += ",".join([str(e) for e in existing])
msg += ",".join([str(e) for e in invalid])
raise ValidationError({
'serial_numbers': [msg],
'serial_numbers': errors + [msg]
})
except DjangoValidationError as e:

View File

@ -96,6 +96,7 @@
location: 7
quantity: 1
serial: 1000
serial_int: 1000
level: 0
tree_id: 0
lft: 0
@ -121,6 +122,7 @@
location: 7
quantity: 1
serial: 1
serial_int: 1
level: 0
tree_id: 0
lft: 0
@ -133,6 +135,7 @@
location: 7
quantity: 1
serial: 2
serial_int: 2
level: 0
tree_id: 0
lft: 0
@ -145,6 +148,7 @@
location: 7
quantity: 1
serial: 3
serial_int: 3
level: 0
tree_id: 0
lft: 0
@ -157,6 +161,7 @@
location: 7
quantity: 1
serial: 4
serial_int: 4
level: 0
tree_id: 0
lft: 0
@ -169,6 +174,7 @@
location: 7
quantity: 1
serial: 5
serial_int: 5
level: 0
tree_id: 0
lft: 0
@ -181,6 +187,7 @@
location: 7
quantity: 1
serial: 10
serial_int: 10
level: 0
tree_id: 0
lft: 0
@ -193,6 +200,7 @@
location: 7
quantity: 1
serial: 11
serial_int: 11
level: 0
tree_id: 0
lft: 0
@ -205,6 +213,7 @@
location: 7
quantity: 1
serial: 12
serial_int: 12
level: 0
tree_id: 0
lft: 0
@ -217,6 +226,7 @@
location: 7
quantity: 1
serial: 20
serial_int: 20
level: 0
tree_id: 0
lft: 0
@ -231,6 +241,7 @@
location: 7
quantity: 1
serial: 21
serial_int: 21
level: 0
tree_id: 0
lft: 0
@ -245,6 +256,7 @@
location: 7
quantity: 1
serial: 22
serial_int: 22
level: 0
tree_id: 0
lft: 0

View File

@ -180,9 +180,24 @@ class StockItemManager(TreeManager):
def generate_batch_code():
"""Generate a default 'batch code' for a new StockItem.
This uses the value of the 'STOCK_BATCH_CODE_TEMPLATE' setting (if configured),
By default, this uses the value of the 'STOCK_BATCH_CODE_TEMPLATE' setting (if configured),
which can be passed through a simple template.
Also, this function is exposed to the ValidationMixin plugin class,
allowing custom plugins to be used to generate new batch code values
"""
# First, check if any plugins can generate batch codes
from plugin.registry import registry
for plugin in registry.with_mixin('validation'):
batch = plugin.generate_batch_code()
if batch is not None:
# Return the first non-null value generated by a plugin
return batch
# If we get to this point, no plugin was able to generate a new batch code
batch_template = common.models.InvenTreeSetting.get_setting('STOCK_BATCH_CODE_TEMPLATE', '')
now = datetime.now()
@ -260,15 +275,38 @@ class StockItem(InvenTreeBarcodeMixin, MetadataMixin, MPTTModel):
This is used for efficient numerical sorting
"""
serial = getattr(self, 'serial', '')
serial = str(getattr(self, 'serial', '')).strip()
from plugin.registry import registry
# First, let any plugins convert this serial number to an integer value
# If a non-null value is returned (by any plugin) we will use that
serial_int = None
for plugin in registry.with_mixin('validation'):
serial_int = plugin.convert_serial_to_int(serial)
if serial_int is not None:
# Save the first returned result
# Ensure that it is clipped within a range allowed in the database schema
clip = 0x7fffffff
serial_int = abs(serial_int)
if serial_int > clip:
serial_int = clip
self.serial_int = serial_int
return
# If we get to this point, none of the available plugins provided an integer value
# Default value if we cannot convert to an integer
serial_int = 0
if serial is not None:
serial = str(serial).strip()
if serial not in [None, '']:
serial_int = extract_int(serial)
self.serial_int = serial_int
@ -408,16 +446,32 @@ class StockItem(InvenTreeBarcodeMixin, MetadataMixin, MPTTModel):
# If the serial number is set, make sure it is not a duplicate
if self.serial:
# Query to look for duplicate serial numbers
parts = PartModels.Part.objects.filter(tree_id=self.part.tree_id)
stock = StockItem.objects.filter(part__in=parts, serial=self.serial)
# Exclude myself from the search
if self.pk is not None:
stock = stock.exclude(pk=self.pk)
self.serial = str(self.serial).strip()
if stock.exists():
raise ValidationError({"serial": _("StockItem with this serial number already exists")})
try:
self.part.validate_serial_number(self.serial, self, raise_error=True)
except ValidationError as exc:
raise ValidationError({
'serial': exc.message,
})
def validate_batch_code(self):
"""Ensure that the batch code is valid for this StockItem.
- Validation is performed by custom plugins.
- By default, no validation checks are performed
"""
from plugin.registry import registry
for plugin in registry.with_mixin('validation'):
try:
plugin.validate_batch_code(self.batch)
except ValidationError as exc:
raise ValidationError({
'batch': exc.message
})
def clean(self):
"""Validate the StockItem object (separate to field validation).
@ -438,6 +492,8 @@ class StockItem(InvenTreeBarcodeMixin, MetadataMixin, MPTTModel):
if type(self.batch) is str:
self.batch = self.batch.strip()
self.validate_batch_code()
try:
# Trackable parts must have integer values for quantity field!
if self.part.trackable:

View File

@ -342,7 +342,11 @@ class SerializeStockItemSerializer(serializers.Serializer):
serial_numbers = data['serial_numbers']
try:
serials = InvenTree.helpers.extract_serial_numbers(serial_numbers, quantity, item.part.getLatestSerialNumberInt())
serials = InvenTree.helpers.extract_serial_numbers(
serial_numbers,
quantity,
item.part.get_latest_serial_number()
)
except DjangoValidationError as e:
raise ValidationError({
'serial_numbers': e.messages,
@ -371,7 +375,7 @@ class SerializeStockItemSerializer(serializers.Serializer):
serials = InvenTree.helpers.extract_serial_numbers(
data['serial_numbers'],
data['quantity'],
item.part.getLatestSerialNumberInt()
item.part.get_latest_serial_number()
)
item.serializeStock(

View File

@ -495,7 +495,7 @@ class StockItemTest(StockAPITestCase):
# Check that each serial number was created
for i in range(1, 11):
self.assertTrue(i in sn)
self.assertTrue(str(i) in sn)
# Check the unique stock item has been created

View File

@ -7,6 +7,7 @@ from django.db.models import Sum
from django.test import override_settings
from build.models import Build
from common.models import InvenTreeSetting
from InvenTree.helpers import InvenTreeTestCase
from InvenTree.status_codes import StockHistoryCode
from part.models import Part
@ -172,6 +173,39 @@ class StockTest(StockTestBase):
item.save()
item.full_clean()
def test_serial_numbers(self):
"""Test serial number uniqueness"""
# Ensure that 'global uniqueness' setting is enabled
InvenTreeSetting.set_setting('SERIAL_NUMBER_GLOBALLY_UNIQUE', True, self.user)
part_a = Part.objects.create(name='A', description='A', trackable=True)
part_b = Part.objects.create(name='B', description='B', trackable=True)
# Create a StockItem for part_a
StockItem.objects.create(
part=part_a,
quantity=1,
serial='ABCDE',
)
# Create a StockItem for part_a (but, will error due to identical serial)
with self.assertRaises(ValidationError):
StockItem.objects.create(
part=part_b,
quantity=1,
serial='ABCDE',
)
# Now, allow serial numbers to be duplicated between different parts
InvenTreeSetting.set_setting('SERIAL_NUMBER_GLOBALLY_UNIQUE', False, self.user)
StockItem.objects.create(
part=part_b,
quantity=1,
serial='ABCDE',
)
def test_expiry(self):
"""Test expiry date functionality for StockItem model."""
today = datetime.datetime.now().date()
@ -857,22 +891,21 @@ class VariantTest(StockTestBase):
def test_serial_numbers(self):
"""Test serial number functionality for variant / template parts."""
InvenTreeSetting.set_setting('SERIAL_NUMBER_GLOBALLY_UNIQUE', False, self.user)
chair = Part.objects.get(pk=10000)
# Operations on the top-level object
self.assertTrue(chair.checkIfSerialNumberExists(1))
self.assertTrue(chair.checkIfSerialNumberExists(2))
self.assertTrue(chair.checkIfSerialNumberExists(3))
self.assertTrue(chair.checkIfSerialNumberExists(4))
self.assertTrue(chair.checkIfSerialNumberExists(5))
[self.assertFalse(chair.validate_serial_number(i)) for i in [1, 2, 3, 4, 5, 20, 21, 22]]
self.assertTrue(chair.checkIfSerialNumberExists(20))
self.assertTrue(chair.checkIfSerialNumberExists(21))
self.assertTrue(chair.checkIfSerialNumberExists(22))
self.assertFalse(chair.validate_serial_number(20))
self.assertFalse(chair.validate_serial_number(21))
self.assertFalse(chair.validate_serial_number(22))
self.assertFalse(chair.checkIfSerialNumberExists(30))
self.assertTrue(chair.validate_serial_number(30))
self.assertEqual(chair.getLatestSerialNumber(), '22')
self.assertEqual(chair.get_latest_serial_number(), '22')
# Check for conflicting serial numbers
to_check = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
@ -883,10 +916,10 @@ class VariantTest(StockTestBase):
# Same operations on a sub-item
variant = Part.objects.get(pk=10003)
self.assertEqual(variant.getLatestSerialNumber(), '22')
self.assertEqual(variant.get_latest_serial_number(), '22')
# Create a new serial number
n = variant.getLatestSerialNumber()
n = variant.get_latest_serial_number()
item = StockItem(
part=variant,
@ -898,12 +931,6 @@ class VariantTest(StockTestBase):
with self.assertRaises(ValidationError):
item.save()
# Verify items with a non-numeric serial don't offer a next serial.
item.serial = "string"
item.save()
self.assertEqual(variant.getLatestSerialNumber(), "string")
# This should pass, although not strictly an int field now.
item.serial = int(n) + 1
item.save()
@ -915,7 +942,7 @@ class VariantTest(StockTestBase):
with self.assertRaises(ValidationError):
item.save()
item.serial += 1
item.serial = int(n) + 2
item.save()

View File

@ -11,6 +11,7 @@
<table class='table table-striped table-condensed'>
<tbody>
{% include "InvenTree/settings/setting.html" with key="SERIAL_NUMBER_GLOBALLY_UNIQUE" icon="fa-hashtag" %}
{% include "InvenTree/settings/setting.html" with key="STOCK_BATCH_CODE_TEMPLATE" icon="fa-layer-group" %}
{% include "InvenTree/settings/setting.html" with key="STOCK_ENABLE_EXPIRY" icon="fa-stopwatch" %}
{% include "InvenTree/settings/setting.html" with key="STOCK_STALE_DAYS" icon="fa-calendar" %}

View File

@ -1230,12 +1230,7 @@ function handleNestedErrors(errors, field_name, options={}) {
// Find the target (nested) field
var target = `${field_name}_${sub_field_name}_${nest_id}`;
for (var ii = errors.length-1; ii >= 0; ii--) {
var error_text = errors[ii];
addFieldErrorMessage(target, error_text, ii, options);
}
addFieldErrorMessage(target, errors, options);
}
}
}
@ -1312,13 +1307,7 @@ function handleFormErrors(errors, fields={}, options={}) {
first_error_field = field_name;
}
// Add an entry for each returned error message
for (var ii = field_errors.length-1; ii >= 0; ii--) {
var error_text = field_errors[ii];
addFieldErrorMessage(field_name, error_text, ii, options);
}
addFieldErrorMessage(field_name, field_errors, options);
}
}
@ -1341,6 +1330,16 @@ function handleFormErrors(errors, fields={}, options={}) {
*/
function addFieldErrorMessage(name, error_text, error_idx=0, options={}) {
// Handle a 'list' of error message recursively
if (typeof(error_text) == 'object') {
// Iterate backwards through the list
for (var ii = error_text.length - 1; ii >= 0; ii--) {
addFieldErrorMessage(name, error_text[ii], ii, options);
}
return;
}
field_name = getFieldName(name, options);
var field_dom = null;