inventree/InvenTree
2
0
Fork 0
mirror of https://github.com/inventree/InvenTree.git synced 2025-10-31 05:05:42 +00:00
Code Activity

[Refactor] BOM Validation (#10056)

Browse Source 
* Add "bom_validated" field to the Part model

* Check bom validity of any assemblies when a part is changed

* Improved update logic

* Fixes for circular imports

* Add additional info to BOM validation serializer

* More intelligent caching

* Refactor

* Update API filter

* Data migration to process existing BomItem entries

* Add "BOM Valid" filter to part table

* Add dashboard widget

* Display BOM validation status

* Tweak dashboard widget

* Update BomTable

* Allow locked BOM items to be validated

* Adjust get_item_hash

- preserve "some" backwards compatibility

* Bump API version

* Refactor app URL patterns

* Fix import sequence

* Tweak imports

* Fix logging message

* Fix error message

* Update src/backend/InvenTree/part/migrations/0141_auto_20250722_0303.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update exception handling

* Try info level debug

* Disable exchange rate update

* Add registry ready flag

* Add is_ready func

* Cleaner init code

* Protect against plugin access until ready

* Fix dashboard widget filter

* Adjust unit test

* Fix receiver name

* Only add plugin URLs if registry is ready

* Cleanup code

* Update playwright tests

* Update docs

* Revert changes to urls.py

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Oliver
2025-07-23 20:16:00 +10:00
committed by GitHub
parent 20477fbfcc
commit dfd9fe44a4
18 changed files with 641 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/backend/InvenTree/InvenTree/api_version.py
View File

@@ -1,12 +1,15 @@
"""InvenTree API version information."""
# InvenTree API version
INVENTREE_API_VERSION = 371
INVENTREE_API_VERSION = 372
"""Increment this API version number whenever there is a significant change to the API that any clients need to know about."""
INVENTREE_API_TEXT = """
v372 -> 2025-07-19 : https://github.com/inventree/InvenTree/pull/10056
- Adds BOM validation information to the Part API
v371 -> 2025-07-18 : https://github.com/inventree/InvenTree/pull/10042
- Adds "setup_quantity" and "attrition" fields to BomItem API endpoints
- Remove "overage" field from BomItem API endpoints

53
src/backend/InvenTree/part/api.py
View File

@@ -11,7 +11,6 @@ from django_filters.rest_framework import DjangoFilterBackend
from drf_spectacular.types import OpenApiTypes
from drf_spectacular.utils import extend_schema_field
from rest_framework import serializers
from rest_framework.exceptions import ValidationError
from rest_framework.response import Response
import InvenTree.permissions
@@ -617,32 +616,8 @@ class PartCopyBOM(CreateAPI):
class PartValidateBOM(RetrieveUpdateAPI):
"""API endpoint for 'validating' the BOM for a given Part."""
class BOMValidateSerializer(serializers.ModelSerializer):
"""Simple serializer class for validating a single BomItem instance."""
class Meta:
"""Metaclass defines serializer fields."""
model = Part
fields = ['checksum', 'valid']
checksum = serializers.CharField(read_only=True, source='bom_checksum')
valid = serializers.BooleanField(
write_only=True,
default=False,
label=_('Valid'),
help_text=_('Validate entire Bill of Materials'),
)
def validate_valid(self, valid):
"""Check that the 'valid' input was flagged."""
if not valid:
raise ValidationError(_('This option must be selected'))
queryset = Part.objects.all()
serializer_class = BOMValidateSerializer
serializer_class = part_serializers.PartBomValidateSerializer
def update(self, request, *args, **kwargs):
"""Validate the referenced BomItem instance."""
@@ -656,9 +631,14 @@ class PartValidateBOM(RetrieveUpdateAPI):
serializer = self.get_serializer(part, data=data, partial=partial)
serializer.is_valid(raise_exception=True)
part.validate_bom(request.user)
valid = str2bool(serializer.validated_data.get('valid', False))
return Response({'checksum': part.bom_checksum})
part.validate_bom(request.user, valid=valid)
# Re-serialize the response
serializer = self.get_serializer(part, many=False)
return Response(serializer.data)
class PartFilter(rest_filters.FilterSet):
@@ -883,24 +863,9 @@ class PartFilter(rest_filters.FilterSet):
)
bom_valid = rest_filters.BooleanFilter(
label=_('BOM Valid'), method='filter_bom_valid'
label=_('BOM Valid'), field_name='bom_validated'
)
def filter_bom_valid(self, queryset, name, value):
"""Filter by whether the BOM for the part is valid or not."""
# Limit queryset to active assemblies
queryset = queryset.filter(active=True, assembly=True).distinct()
# Iterate through the queryset
# TODO: We should cache BOM checksums to make this process more efficient
pks = []
for item in queryset:
if item.is_bom_valid() == value:
pks.append(item.pk)
return queryset.filter(pk__in=pks)
starred = rest_filters.BooleanFilter(label='Starred', method='filter_starred')
def filter_starred(self, queryset, name, value):

22
src/backend/InvenTree/part/migrations/0140_part_bom_validated.py Normal file
View File

@@ -0,0 +1,22 @@
# Generated by Django 4.2.23 on 2025-07-22 01:57
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("part", "0139_remove_bomitem_overage"),
]
operations = [
migrations.AddField(
model_name="part",
name="bom_validated",
field=models.BooleanField(
default=False,
help_text="Is the BOM for this part valid?",
verbose_name="BOM Validated",
),
),
]

80
src/backend/InvenTree/part/migrations/0141_auto_20250722_0303.py Normal file
View File

@@ -0,0 +1,80 @@
# Generated by Django 4.2.23 on 2025-07-22 03:03
from django.db import migrations
def cache_bom_valid(apps, schema_editor):
"""Calculate and cache the BOM validity for all parts.
Procedure:
- Find all parts which have linked BOM item(s)
- Limit to parts which have a stored BOM checksum
- For each such part, calculate and update the BOM "validity"
"""
from InvenTree.tasks import offload_task
from part.tasks import check_bom_valid
Part = apps.get_model('part', 'Part')
BomItem = apps.get_model('part', 'BomItem')
# Fetch all BomItem objects
bom_items = BomItem.objects.exclude(part=None).prefetch_related('part').distinct()
parts_to_update = set()
for item in bom_items:
# Parts associated with this BomItem
parts = []
if item.inherited:
# Find all inherited assemblies for this BomItem
parts = list(
Part.objects.filter(
tree_id=item.part.tree_id,
lft__gte=item.part.lft,
rght__lte=item.part.rght
)
)
else:
parts = [item.part]
for part in parts:
# Part has already been observed - skip
if part in parts_to_update:
continue
# Part has no BOM checksum - skip
if not part.bom_checksum:
continue
# Part has not already been validated
if not part.bom_checked_date:
continue
parts_to_update.add(part)
if len(parts_to_update) > 0:
print(f"\nScheduling {len(parts_to_update)} parts to update BOM validity.")
for part in parts_to_update:
# Offload task to recalculate the BOM checksum for this part
# The background worker will process these when the server restarts
offload_task(
check_bom_valid,
part.pk,
force_async=True,
group='part'
)
class Migration(migrations.Migration):
dependencies = [
("part", "0140_part_bom_validated"),
]
operations = [
migrations.RunPython(cache_bom_valid, migrations.RunPython.noop),
]

162
src/backend/InvenTree/part/models.py
View File

@@ -453,6 +453,12 @@ class Part(
creation_user: User who added this part to the database
responsible_owner: Owner (either user or group) which is responsible for this part (optional)
last_stocktake: Date at which last stocktake was performed for this Part
BOM (Bill of Materials) related attributes:
bom_checksum: Checksum for the BOM of this part
bom_validated: Boolean field indicating if the BOM is valid (checksum matches)
bom_checked_by: User who last checked the BOM for this part
bom_checked_date: Date when the BOM was last checked
"""
NODE_PARENT_KEY = 'variant_of'
@@ -1265,6 +1271,12 @@ class Part(
help_text=_('Is this a virtual part, such as a software product or license?'),
)
bom_validated = models.BooleanField(
default=False,
verbose_name=_('BOM Validated'),
help_text=_('Is the BOM for this part valid?'),
)
bom_checksum = models.CharField(
max_length=128,
blank=True,
@@ -1942,31 +1954,50 @@ class Part(
result_hash = hashlib.md5(str(self.id).encode())
# List *all* BOM items (including inherited ones!)
bom_items = self.get_bom_items().all().prefetch_related('sub_part')
bom_items = self.get_bom_items().all().prefetch_related('part', 'sub_part')
for item in bom_items:
result_hash.update(str(item.get_item_hash()).encode())
return str(result_hash.digest())
def is_bom_valid(self):
"""Check if the BOM is 'valid' - if the calculated checksum matches the stored value."""
return self.get_bom_hash() == self.bom_checksum or not self.has_bom
def is_bom_valid(self) -> bool:
"""Check if the BOM is 'valid'.
To be "valid", the part must:
- Have a stored "bom_checksum" value
- The stored "bom_checksum" must match the calculated checksum.
Returns:
bool: True if the BOM is valid, False otherwise
"""
if not self.bom_checksum or not self.bom_checked_date:
# If there is no BOM checksum, then the BOM is not valid
return False
return self.get_bom_hash() == self.bom_checksum
@transaction.atomic
def validate_bom(self, user):
def validate_bom(self, user, valid: bool = True):
"""Validate the BOM (mark the BOM as validated by the given User.
Arguments:
user: User who is validating the BOM
valid: If True, mark the BOM as valid (default=True)
- Calculates and stores the hash for the BOM
- Saves the current date and the checking user
"""
# Validate each line item, ignoring inherited ones
bom_items = self.get_bom_items(include_inherited=False)
bom_items = self.get_bom_items(include_inherited=False).prefetch_related(
'part', 'sub_part'
)
for item in bom_items:
item.validate_hash()
item.validate_hash(valid=valid)
self.bom_checksum = self.get_bom_hash()
self.bom_validated = valid
self.bom_checksum = self.get_bom_hash() if valid else ''
self.bom_checked_by = user
self.bom_checked_date = InvenTree.helpers.current_date()
@@ -4252,6 +4283,7 @@ class BomItem(InvenTree.models.MetadataMixin, InvenTree.models.InvenTreeModel):
rounding_multiple: Rounding quantity when calculating the required quantity for a build
note: Note field for this BOM item
checksum: Validation checksum for the particular BOM line item
validated: Boolean field indicating if this BOM item is valid (checksum matches)
inherited: This BomItem can be inherited by the BOMs of variant parts
allow_variants: Stock for part variants can be substituted for this BomItem
"""
@@ -4329,26 +4361,61 @@ class BomItem(InvenTree.models.MetadataMixin, InvenTree.models.InvenTreeModel):
def delete(self):
"""Check if this item can be deleted."""
import part.tasks as part_tasks
self.check_part_lock(self.part)
assemblies = self.get_assemblies()
super().delete()
for assembly in assemblies:
# Offload task to update the checksum for this assembly
InvenTree.tasks.offload_task(
part_tasks.check_bom_valid, assembly.pk, group='part'
)
def save(self, *args, **kwargs):
"""Enforce 'clean' operation when saving a BomItem instance."""
import part.tasks as part_tasks
self.clean()
self.check_part_lock(self.part)
check_lock = kwargs.pop('check_lock', True)
if check_lock:
self.check_part_lock(self.part)
db_instance = self.get_db_instance()
# Check if the part was changed
deltas = self.get_field_deltas()
if 'part' in deltas and (old_part := deltas['part'].get('old', None)):
self.check_part_lock(old_part)
if check_lock:
self.check_part_lock(old_part)
# Update the 'validated' field based on checksum calculation
self.validated = self.is_line_valid
super().save(*args, **kwargs)
# Do we need to recalculate the BOM hash for assemblies?
if not db_instance or any(f in deltas for f in self.hash_fields()):
# If this is a new BomItem, or if any of the fields used to calculate the hash have changed,
# then we need to recalculate the BOM checksum for all assemblies which use this BomItem
assemblies = set()
if db_instance:
# Find all assemblies which use this BomItem *after* we save
assemblies.update(db_instance.get_assemblies())
for assembly in assemblies:
# Offload task to update the checksum for this assembly
InvenTree.tasks.offload_task(
part_tasks.check_bom_valid, assembly.pk, group='part'
)
def check_part_lock(self, assembly):
"""When editing or deleting a BOM item, check if the assembly is locked.
@@ -4490,39 +4557,57 @@ class BomItem(InvenTree.models.MetadataMixin, InvenTree.models.InvenTreeModel):
help_text=_('Stock items for variant parts can be used for this BOM item'),
)
def get_item_hash(self):
"""Calculate the checksum hash of this BOM line item.
def hash_fields(self) -> list[str]:
"""Return a list of fields to be used for hashing this BOM item.
The hash is calculated from the following fields:
- part.pk
- sub_part.pk
- quantity
- reference
- optional
- inherited
- consumable
- allow_variants
These fields are used to calculate the checksum hash of this BOM item.
"""
return [
'part_id',
'sub_part_id',
'quantity',
'setup_quantity',
'attrition',
'rounding_multiple',
'reference',
'optional',
'inherited',
'consumable',
'allow_variants',
]
def get_item_hash(self) -> str:
"""Calculate the checksum hash of this BOM line item."""
# Seed the hash with the ID of this BOM item
result_hash = hashlib.md5(b'')
# The following components are used to calculate the checksum
components = [
self.part.pk,
self.sub_part.pk,
normalize(self.quantity),
self.setup_quantity,
self.attrition,
self.rounding_multiple,
self.reference,
self.optional,
self.inherited,
self.consumable,
self.allow_variants,
]
for field in self.hash_fields():
# Get the value of the field
value = getattr(self, field, None)
for component in components:
result_hash.update(str(component).encode())
# If the value is None, use an empty string
if value is None:
value = ''
# Normalize decimal values to ensure consistent representation
# These values are only included if they are non-zero
# This is to provide some backwards compatibility from before these fields were addede
if value is not None and field in [
'quantity',
'attrition',
'setup_quantity',
'rounding_multiple',
]:
try:
value = normalize(value)
if not value or value <= 0:
continue
except Exception:
pass
# Update the hash with the string representation of the value
result_hash.update(str(value).encode())
return str(result_hash.digest())
@@ -4537,7 +4622,8 @@ class BomItem(InvenTree.models.MetadataMixin, InvenTree.models.InvenTreeModel):
else:
self.checksum = ''
self.save()
# Save the BOM item (bypass lock check)
self.save(check_lock=False)
@property
def is_line_valid(self):

38
src/backend/InvenTree/part/serializers.py
View File

@@ -1210,6 +1210,44 @@ class PartSerializer(
return self.instance
class PartBomValidateSerializer(InvenTree.serializers.InvenTreeModelSerializer):
"""Serializer for Part BOM information."""
class Meta:
"""Metaclass options."""
model = Part
fields = [
'pk',
'bom_validated',
'bom_checksum',
'bom_checked_by',
'bom_checked_by_detail',
'bom_checked_date',
'valid',
]
read_only_fields = [
'bom_validated',
'bom_checksum',
'bom_checked_by',
'bom_checked_by_detail',
'bom_checked_date',
]
valid = serializers.BooleanField(
write_only=True,
default=False,
required=False,
label=_('Valid'),
help_text=_('Validate entire Bill of Materials'),
)
bom_checked_by_detail = UserSerializer(
source='bom_checked_by', many=False, read_only=True, allow_null=True
)
class PartRequirementsSerializer(InvenTree.serializers.InvenTreeModelSerializer):
"""Serializer for Part requirements."""

87
src/backend/InvenTree/part/tasks.py
View File

@@ -3,6 +3,7 @@
from datetime import datetime, timedelta
from django.core.exceptions import ValidationError
from django.db.models import Model
from django.utils.translation import gettext_lazy as _
import structlog
@@ -10,16 +11,12 @@ from opentelemetry import trace
import common.currency
import common.notifications
import company.models
import InvenTree.helpers_model
import InvenTree.tasks
import part.models as part_models
import part.stocktake
import stock.models as stock_models
from common.settings import get_global_setting
from InvenTree.tasks import (
ScheduledTask,
check_daily_holdoff,
offload_task,
record_task_success,
scheduled_task,
)
@@ -29,7 +26,7 @@ logger = structlog.get_logger('inventree')
@tracer.start_as_current_span('notify_low_stock')
def notify_low_stock(part: part_models.Part):
def notify_low_stock(part: Model):
"""Notify interested users that a part is 'low stock'.
Rules:
@@ -135,9 +132,11 @@ def notify_low_stock_if_required(part_id: int):
If true, notify the users who have subscribed to the part
"""
from part.models import Part
try:
part = part_models.Part.objects.get(pk=part_id)
except part_models.Part.DoesNotExist:
part = Part.objects.get(pk=part_id)
except Part.DoesNotExist:
logger.warning(
'notify_low_stock_if_required: Part with ID %s does not exist', part_id
)
@@ -148,7 +147,7 @@ def notify_low_stock_if_required(part_id: int):
for p in parts:
if p.is_part_low_on_stock():
InvenTree.tasks.offload_task(notify_low_stock, p, group='notification')
offload_task(notify_low_stock, p, group='notification')
@tracer.start_as_current_span('check_stale_stock')
@@ -163,6 +162,8 @@ def check_stale_stock():
to notifications for the respective parts. Each user receives one consolidated email
containing all their stale stock items.
"""
from stock.models import StockItem
# Check if stock expiry functionality is enabled
if not get_global_setting('STOCK_ENABLE_EXPIRY', False, cache=False):
logger.info('Stock expiry functionality is not enabled - exiting')
@@ -179,8 +180,8 @@ def check_stale_stock():
stale_threshold = today + timedelta(days=stale_days)
# Find stock items that are stale (expiry date within STOCK_STALE_DAYS)
stale_stock_items = stock_models.StockItem.objects.filter(
stock_models.StockItem.IN_STOCK_FILTER, # Only in-stock items
stale_stock_items = StockItem.objects.filter(
StockItem.IN_STOCK_FILTER, # Only in-stock items
expiry_date__isnull=False, # Must have an expiry date
expiry_date__lt=stale_threshold, # Expiry date is within stale threshold
).select_related('part', 'location') # Optimize queries
@@ -192,7 +193,7 @@ def check_stale_stock():
logger.info('Found %s stale stock items', stale_stock_items.count())
# Group stale stock items by user subscriptions
user_stale_items: dict[stock_models.StockItem, list[stock_models.StockItem]] = {}
user_stale_items: dict[StockItem, list[StockItem]] = {}
for stock_item in stale_stock_items:
# Get all subscribers for this part
@@ -206,9 +207,7 @@ def check_stale_stock():
# Send one consolidated notification per user
for user, items in user_stale_items.items():
try:
InvenTree.tasks.offload_task(
notify_stale_stock, user, items, group='notification'
)
offload_task(notify_stale_stock, user, items, group='notification')
except Exception as e:
logger.error(
'Error scheduling stale stock notification for user %s: %s',
@@ -222,7 +221,7 @@ def check_stale_stock():
@tracer.start_as_current_span('update_part_pricing')
def update_part_pricing(pricing: part_models.PartPricing, counter: int = 0):
def update_part_pricing(pricing: Model, counter: int = 0):
"""Update cached pricing data for the specified PartPricing instance.
Arguments:
@@ -251,6 +250,8 @@ def check_missing_pricing(limit=250):
Arguments:
limit: Maximum number of parts to process at once
"""
from part.models import Part, PartPricing
# Find any parts which have 'old' pricing information
days = int(get_global_setting('PRICING_UPDATE_DAYS', 30))
@@ -259,7 +260,7 @@ def check_missing_pricing(limit=250):
return
# Find parts for which pricing information has never been updated
results = part_models.PartPricing.objects.filter(updated=None)[:limit]
results = PartPricing.objects.filter(updated=None)[:limit]
if results.count() > 0:
logger.info('Found %s parts with empty pricing', results.count())
@@ -269,7 +270,7 @@ def check_missing_pricing(limit=250):
stale_date = datetime.now().date() - timedelta(days=days)
results = part_models.PartPricing.objects.filter(updated__lte=stale_date)[:limit]
results = PartPricing.objects.filter(updated__lte=stale_date)[:limit]
if results.count() > 0:
logger.info('Found %s stale pricing entries', results.count())
@@ -279,7 +280,7 @@ def check_missing_pricing(limit=250):
# Find any pricing data which is in the wrong currency
currency = common.currency.currency_code_default()
results = part_models.PartPricing.objects.exclude(currency=currency)
results = PartPricing.objects.exclude(currency=currency)
if results.count() > 0:
logger.info('Found %s pricing entries in the wrong currency', results.count())
@@ -288,7 +289,7 @@ def check_missing_pricing(limit=250):
pp.schedule_for_update()
# Find any parts which do not have pricing information
results = part_models.Part.objects.filter(pricing_data=None)[:limit]
results = Part.objects.filter(pricing_data=None)[:limit]
if results.count() > 0:
logger.info('Found %s parts without pricing', results.count())
@@ -309,12 +310,15 @@ def scheduled_stocktake_reports():
- Delete 'old' stocktake report files after the specified period
- Generate new reports at the specified period
"""
import part.stocktake
from part.models import PartStocktakeReport
# First let's delete any old stocktake reports
delete_n_days = int(
get_global_setting('STOCKTAKE_DELETE_REPORT_DAYS', 30, cache=False)
)
threshold = datetime.now() - timedelta(days=delete_n_days)
old_reports = part_models.PartStocktakeReport.objects.filter(date__lt=threshold)
old_reports = PartStocktakeReport.objects.filter(date__lt=threshold)
if old_reports.count() > 0:
logger.info('Deleting %s stale stocktake reports', old_reports.count())
@@ -349,12 +353,14 @@ def rebuild_parameters(template_id):
This function is called when a base template is changed,
which may cause the base unit to be adjusted.
"""
from part.models import PartParameter, PartParameterTemplate
try:
template = part_models.PartParameterTemplate.objects.get(pk=template_id)
except part_models.PartParameterTemplate.DoesNotExist:
template = PartParameterTemplate.objects.get(pk=template_id)
except PartParameterTemplate.DoesNotExist:
return
parameters = part_models.PartParameter.objects.filter(template=template)
parameters = PartParameter.objects.filter(template=template)
n = 0
@@ -373,18 +379,21 @@ def rebuild_parameters(template_id):
@tracer.start_as_current_span('rebuild_supplier_parts')
def rebuild_supplier_parts(part_id):
def rebuild_supplier_parts(part_id: int):
"""Rebuild all SupplierPart objects for a given part.
This function is called when a bart part is changed,
which may cause the native units of any supplier parts to be updated
"""
from company.models import SupplierPart
from part.models import Part
try:
prt = part_models.Part.objects.get(pk=part_id)
except part_models.Part.DoesNotExist:
prt = Part.objects.get(pk=part_id)
except Part.DoesNotExist:
return
supplier_parts = company.models.SupplierPart.objects.filter(part=prt)
supplier_parts = SupplierPart.objects.filter(part=prt)
n = supplier_parts.count()
@@ -398,3 +407,25 @@ def rebuild_supplier_parts(part_id):
if n > 0:
logger.info("Rebuilt %s supplier parts for part '%s'", n, prt.name)
@tracer.start_as_current_span('check_bom_valid')
def check_bom_valid(part_id: int):
"""Recalculate the BOM checksum for all assemblies which include the specified Part.
Arguments:
part_id: The ID of the part for which to recalculate the BOM checksum.
"""
from part.models import Part
try:
part = Part.objects.get(pk=part_id)
except Part.DoesNotExist:
logger.warning('check_bom_valid: Part with ID %s does not exist', part_id)
return
valid = part.is_bom_valid()
if valid != part.bom_validated:
part.bom_validated = valid
part.save()

87
src/backend/InvenTree/part/test_api.py
View File

@@ -911,20 +911,89 @@ class PartAPITest(PartAPITestBase):
"""Test the 'bom_valid' Part API filter."""
url = reverse('api-part-list')
n = Part.objects.filter(active=True, assembly=True).count()
# Create a new assembly
assembly = Part.objects.create(
name='Test Assembly',
description='A test assembly with a valid BOM',
category=PartCategory.objects.first(),
assembly=True,
active=True,
)
sub_part = Part.objects.create(
name='Sub Part',
description='A sub part for the assembly',
category=PartCategory.objects.first(),
component=True,
assembly=False,
active=True,
)
assembly.refresh_from_db()
sub_part.refresh_from_db()
# Link the sub part to the assembly via a BOM
bom_item = BomItem.objects.create(part=assembly, sub_part=sub_part, quantity=10)
filters = {'active': True, 'assembly': True, 'bom_valid': True}
# Initially, there are no parts with a valid BOM
response = self.get(url, {'bom_valid': False}, expected_code=200)
n1 = len(response.data)
response = self.get(url, filters)
for item in response.data:
self.assertTrue(item['assembly'])
self.assertTrue(item['active'])
self.assertEqual(len(response.data), 0)
response = self.get(url, {'bom_valid': True}, expected_code=200)
n2 = len(response.data)
# Validate the BOM assembly
assembly.validate_bom(self.user)
self.assertEqual(n1 + n2, n)
response = self.get(url, filters)
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]['pk'], assembly.pk)
# Adjust the 'quantity' of the BOM item to make it invalid
bom_item.quantity = 15
bom_item.save()
response = self.get(url, filters)
self.assertEqual(len(response.data), 0)
# Adjust it back again - should be valid again
bom_item.quantity = 10
bom_item.save()
response = self.get(url, filters)
self.assertEqual(len(response.data), 1)
# Test the BOM validation API endpoint
bom_url = reverse('api-part-bom-validate', kwargs={'pk': assembly.pk})
data = self.get(bom_url, expected_code=200).data
self.assertEqual(data['bom_validated'], True)
self.assertEqual(data['bom_checked_by'], self.user.pk)
self.assertEqual(data['bom_checked_by_detail']['username'], self.user.username)
self.assertIsNotNone(data['bom_checked_date'])
# Now, let's try to validate and invalidate the assembly BOM via the API
bom_item.quantity = 99
bom_item.save()
data = self.get(bom_url, expected_code=200).data
self.assertEqual(data['bom_validated'], False)
self.patch(bom_url, {'valid': True}, expected_code=200)
data = self.get(bom_url, expected_code=200).data
self.assertEqual(data['bom_validated'], True)
assembly.refresh_from_db()
self.assertTrue(assembly.bom_validated)
# And, we can also invalidate the BOM via the API
self.patch(bom_url, {'valid': False}, expected_code=200)
data = self.get(bom_url, expected_code=200).data
self.assertEqual(data['bom_validated'], False)
assembly.refresh_from_db()
self.assertFalse(assembly.bom_validated)
def test_filter_by_starred(self):
"""Test by 'starred' filter."""