1
0
forked from erp-dev/erp

feat: mes module

This commit is contained in:
2026-05-11 21:37:41 +08:00
parent d7b388e935
commit 9359013b4c
20 changed files with 3232 additions and 133 deletions

368
mes/tests.py Normal file
View File

@@ -0,0 +1,368 @@
from django.contrib.contenttypes.models import ContentType
from django.contrib.auth import get_user_model
from django.db.models import ProtectedError
from django.test import TestCase
from basic_info.models import Customer, Employee, Merchant, MerchantTypeEnum
from mes.models import CapacityUnitEnum, Device, DeviceCategory, ProductionAssignment, ProductionAssignmentStatusEnum
from mes import services
from printing.models import PrintingOrder
class MesServiceTestCase(TestCase):
def setUp(self):
self.user = get_user_model().objects.create_user(username='mes-user', password='pass12345')
self.other_user = get_user_model().objects.create_user(username='mes-other-user', password='pass12345')
self.merchant = Merchant.objects.create(name='MES商户', type=MerchantTypeEnum.STORE)
self.other_merchant = Merchant.objects.create(name='其他商户', type=MerchantTypeEnum.STORE)
self.operator = Employee.objects.create(merchant=self.merchant, sys_user=self.user, name='操作员')
self.other_operator = Employee.objects.create(merchant=self.other_merchant, sys_user=self.other_user, name='其他操作员')
def test_create_device_category(self):
category = services.create_device_category(
merchant=self.merchant,
name='打印机',
created_by=self.user,
operator=self.operator,
)
self.assertEqual(category.merchant, self.merchant)
self.assertEqual(category.created_by, self.user)
self.assertEqual(category.operator, self.operator)
def test_create_device_category_rejects_cross_merchant_operator(self):
with self.assertRaisesMessage(ValueError, '操作人不属于当前商户'):
services.create_device_category(
merchant=self.merchant,
name='打印机',
created_by=self.user,
operator=self.other_operator,
)
def test_create_device(self):
category = services.create_device_category(
merchant=self.merchant,
name='打印机',
created_by=self.user,
operator=self.operator,
)
device = services.create_device(
merchant=self.merchant,
category=category,
name='设备A',
created_by=self.user,
operator=self.operator,
peak_capacity=120,
capacity_unit=CapacityUnitEnum.METER,
extra={'capacity': {'per_hour': 120}},
)
self.assertEqual(device.category, category)
self.assertEqual(device.peak_capacity, 120)
self.assertEqual(device.capacity_unit, CapacityUnitEnum.METER)
self.assertEqual(device.extra, {'capacity': {'per_hour': 120}})
def test_create_device_rejects_cross_merchant_category(self):
other_category = services.create_device_category(
merchant=self.other_merchant,
name='其他分类',
created_by=self.other_user,
operator=self.other_operator,
)
with self.assertRaisesMessage(ValueError, '设备分类不属于当前商户'):
services.create_device(
merchant=self.merchant,
category=other_category,
name='设备A',
created_by=self.user,
operator=self.operator,
peak_capacity=120,
)
def test_update_device(self):
category = services.create_device_category(
merchant=self.merchant,
name='打印机',
created_by=self.user,
operator=self.operator,
)
another_category = services.create_device_category(
merchant=self.merchant,
name='滚筒机',
created_by=self.user,
operator=self.operator,
)
device = services.create_device(
merchant=self.merchant,
category=category,
name='设备A',
created_by=self.user,
operator=self.operator,
peak_capacity=120,
)
updated = services.update_device(
device=device,
operator=self.operator,
name='设备B',
category=another_category,
peak_capacity=180,
capacity_unit=CapacityUnitEnum.YARD,
extra={'power': 220},
)
self.assertEqual(updated.name, '设备B')
self.assertEqual(updated.category, another_category)
self.assertEqual(updated.peak_capacity, 180)
self.assertEqual(updated.capacity_unit, CapacityUnitEnum.YARD)
self.assertEqual(updated.extra, {'power': 220})
def test_update_device_allows_clearing_extra(self):
category = services.create_device_category(
merchant=self.merchant,
name='打印机',
created_by=self.user,
operator=self.operator,
)
device = services.create_device(
merchant=self.merchant,
category=category,
name='设备A',
created_by=self.user,
operator=self.operator,
peak_capacity=120,
extra={'power': 220},
)
updated = services.update_device(device=device, operator=self.operator, extra=None)
self.assertIsNone(updated.extra)
def test_delete_device_category_is_protected_when_devices_exist(self):
category = services.create_device_category(
merchant=self.merchant,
name='打印机',
created_by=self.user,
operator=self.operator,
)
services.create_device(
merchant=self.merchant,
category=category,
name='设备A',
created_by=self.user,
operator=self.operator,
peak_capacity=120,
)
with self.assertRaises(ProtectedError):
services.delete_device_category(category=category, operator=self.operator)
def test_list_devices_for_merchant_filters_by_category(self):
category = services.create_device_category(
merchant=self.merchant,
name='打印机',
created_by=self.user,
operator=self.operator,
)
another_category = services.create_device_category(
merchant=self.merchant,
name='滚筒机',
created_by=self.user,
operator=self.operator,
)
target = services.create_device(
merchant=self.merchant,
category=category,
name='设备A',
created_by=self.user,
operator=self.operator,
peak_capacity=120,
)
services.create_device(
merchant=self.merchant,
category=another_category,
name='设备B',
created_by=self.user,
operator=self.operator,
peak_capacity=80,
)
devices = list(services.list_devices_for_merchant(merchant=self.merchant, category_id=category.id))
self.assertEqual([device.id for device in devices], [target.id])
def test_list_devices_for_merchant_filters_by_extra_json_path(self):
category = services.create_device_category(
merchant=self.merchant,
name='打印机',
created_by=self.user,
operator=self.operator,
)
target = services.create_device(
merchant=self.merchant,
category=category,
name='设备A',
created_by=self.user,
operator=self.operator,
peak_capacity=120,
extra={'capacity': {'per_hour': 120}},
)
services.create_device(
merchant=self.merchant,
category=category,
name='设备B',
created_by=self.user,
operator=self.operator,
peak_capacity=80,
extra={'capacity': {'per_hour': 80}},
)
devices = list(
services.list_devices_for_merchant(
merchant=self.merchant,
extra_json_path=['capacity', 'per_hour'],
extra_json_value=120,
)
)
self.assertEqual([device.id for device in devices], [target.id])
def test_create_production_assignment(self):
customer = Customer.objects.create(
merchant=self.merchant,
name='测试客户',
mobile='13800138000',
)
printing_order = PrintingOrder.objects.create(
merchant=self.merchant,
customer=customer,
fabric='测试面料',
width='150cm',
created_by=self.user,
)
category = services.create_device_category(
merchant=self.merchant,
name='打印机',
created_by=self.user,
operator=self.operator,
)
device = services.create_device(
merchant=self.merchant,
category=category,
name='设备A',
created_by=self.user,
operator=self.operator,
peak_capacity=120,
)
content_type = ContentType.objects.get_for_model(PrintingOrder)
assignment = services.create_production_assignment(
merchant=self.merchant,
device=device,
content_type=content_type,
object_id=printing_order.id,
assigner=self.operator,
production_quantity=300,
assignee=None,
created_by=self.user,
operator=self.operator,
extra={'batch': 'A1'},
)
self.assertEqual(assignment.device, device)
self.assertEqual(assignment.assigner, self.operator)
self.assertIsNone(assignment.assignee)
self.assertEqual(assignment.production_quantity, 300)
self.assertEqual(assignment.status, ProductionAssignmentStatusEnum.DRAFT)
self.assertEqual(assignment.extra, {'batch': 'A1'})
def test_create_production_assignment_rejects_cross_merchant_assignee(self):
customer = Customer.objects.create(
merchant=self.merchant,
name='测试客户',
mobile='13800138000',
)
printing_order = PrintingOrder.objects.create(
merchant=self.merchant,
customer=customer,
fabric='测试面料',
width='150cm',
created_by=self.user,
)
category = services.create_device_category(
merchant=self.merchant,
name='打印机',
created_by=self.user,
operator=self.operator,
)
device = services.create_device(
merchant=self.merchant,
category=category,
name='设备A',
created_by=self.user,
operator=self.operator,
peak_capacity=120,
)
content_type = ContentType.objects.get_for_model(PrintingOrder)
with self.assertRaisesMessage(ValueError, '被指派人不属于当前商户'):
services.create_production_assignment(
merchant=self.merchant,
device=device,
content_type=content_type,
object_id=printing_order.id,
assigner=self.operator,
production_quantity=300,
assignee=self.other_operator,
created_by=self.user,
operator=self.operator,
)
def test_update_production_assignment_rejects_invalid_status_transition(self):
customer = Customer.objects.create(
merchant=self.merchant,
name='测试客户',
mobile='13800138001',
)
printing_order = PrintingOrder.objects.create(
merchant=self.merchant,
customer=customer,
fabric='测试面料',
width='150cm',
created_by=self.user,
)
category = services.create_device_category(
merchant=self.merchant,
name='打印机',
created_by=self.user,
operator=self.operator,
)
device = services.create_device(
merchant=self.merchant,
category=category,
name='设备A',
created_by=self.user,
operator=self.operator,
peak_capacity=120,
)
content_type = ContentType.objects.get_for_model(PrintingOrder)
assignment = services.create_production_assignment(
merchant=self.merchant,
device=device,
content_type=content_type,
object_id=printing_order.id,
assigner=self.operator,
production_quantity=300,
created_by=self.user,
operator=self.operator,
status=ProductionAssignmentStatusEnum.ACCEPTED,
)
with self.assertRaisesMessage(ValueError, '当前状态不允许变更为目标状态'):
services.update_production_assignment(
assignment=assignment,
operator=self.operator,
status=ProductionAssignmentStatusEnum.CANCELLED,
)