zuowei/erpnew
1
0
Fork 0
forked from erp-dev/erp
Code Pull Requests Activity

fix: process id change when plate_order and printing_order update

Browse Source
This commit is contained in:
colaftc
2026-01-15 16:35:46 +08:00
parent 9a367caab7
commit ea9f378de9
9 changed files with 384 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
api_v1/views/printing/serializers.py
View File

@@ -475,6 +475,12 @@ class PrintingJobCreateUpdateSerializer(serializers.ModelSerializer):
def update(self, instance, validated_data):
"""更新任务，使用 service 层"""
# 不允许修改 printing_order 绑定关系（避免业务对象流程不一致）
if 'printing_order' in validated_data:
new_order = validated_data.get('printing_order')
if new_order and new_order.id != getattr(instance, 'printing_order_id', None):
raise serializers.ValidationError({'printing_order': '不允许修改印染订单绑定关系'})
user = self.context['request'].user
success, message, updated_instance = PrintingJobService.update_printing_job(
instance, validated_data, user
@@ -772,6 +778,27 @@ class PlateOrderCreateUpdateSerializer(serializers.ModelSerializer):
return super().create(validated_data)
def update(self, instance, validated_data):
# 若更新涉及 process 变更，需要重建并重新绑定 BusinessObject（若有未撤销进度则拒绝）
new_process_id = validated_data.get('process', None)
if new_process_id is not None and new_process_id != getattr(instance, 'process', None):
from stateflow.models import Process
from stateflow.services import relink_business_object_for_instance
try:
new_process = Process.objects.get(id=new_process_id)
except Process.DoesNotExist:
# 理论上 validate_process 已校验；这里兜底
raise serializers.ValidationError({'process': '流程不存在'})
relinked = relink_business_object_for_instance(
instance=instance,
new_process=new_process,
default_name=f"PlateOrder-{instance.pk}",
default_description=f"开版订单 {instance.pk} 的流程实例",
)
if relinked is None:
raise serializers.ValidationError({'process': '该订单流程已存在有效进度，无法修改流程'})
plate_images = validated_data.pop('plate_image', None)
if plate_images is not None:
validated_data['plate_image'] = _build_plate_image_payload(plate_images, self._request_user)

32
api_v1/views/printing/services.py
View File

@@ -70,10 +70,40 @@ class PrintingOrderService:
Returns:
(success, message, updated_order)
"""
# 如果要修改 process，需要验证
# 如果要修改 process，需要验证并同步 relink 所有 jobs 的 BusinessObject
if 'process' in data and data['process'] != printing_order.process:
new_process = data.get('process')
if new_process is None:
return False, '流程不能为空，无法修改流程', printing_order
# 只要存在任意 job 已开始（有未撤销进度）则拒绝
if not printing_order.can_change_process():
return False, '存在已开始的印染任务，无法修改流程', printing_order
from stateflow.services import relink_business_object_for_instance
try:
with transaction.atomic():
# relink 所有 jobs（jobs 未开始，允许重建并重新绑定 BO）
for job in printing_order.printing_jobs.select_related('business_object').all():
relinked = relink_business_object_for_instance(
instance=job,
new_process=new_process,
default_name=f"PrintingJob-{job.pk}",
default_description=f"印染任务 {job.pk} 的流程实例",
)
if relinked is None:
raise RuntimeError(f'印染任务 {job.pk} 已存在有效进度，无法修改流程')
# 最后更新订单流程
printing_order.process = new_process
printing_order.save(update_fields=['process'])
except RuntimeError as e:
return False, str(e), printing_order
# process 已处理完毕，避免后续通用字段更新再次覆盖
data = {k: v for k, v in data.items() if k != 'process'}
# 更新字段
for field, value in data.items():

48
api_v1/views/printing/test_api.py
View File

@@ -514,6 +514,54 @@ class PrintingOrderAPITestCase(TestCase):
self.assertEqual(response.status_code, status.HTTP_200_OK)
order.refresh_from_db()
self.assertEqual(order.process.id, new_process.id)
def test_update_process_with_jobs_not_started_relinks_jobs(self):
"""测试：有任务但均未开始时允许改流程，并对所有 jobs 重建/绑定新的 business_object"""
order = printing_models.PrintingOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
fabric='测试布料',
width='150cm',
process=self.process,
)
category = basic_models.ProductCategory.objects.create(
name='测试类别2',
merchant=self.merchant,
)
product = basic_models.Product.objects.create(
name='测试产品2',
category=category,
merchant=self.merchant,
)
job = printing_models.PrintingJob.objects.create(
printing_order=order,
product=product,
quantity=10,
unit='件',
)
# 绑定一个旧的 business_object（未推进过，未开始）
old_bo = stateflow_models.BusinessObject.objects.create(
name=f'PrintingJob-{job.id}',
process=self.process,
)
job.business_object = old_bo
job.save()
new_process = stateflow_models.Process.objects.create(name='新流程2')
data = {'process': new_process.id}
response = self.client.patch(f'/api/v1/printing-orders/{order.id}/', data, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
order.refresh_from_db()
self.assertEqual(order.process.id, new_process.id)
job.refresh_from_db()
self.assertIsNotNone(job.business_object)
self.assertNotEqual(job.business_object.id, old_bo.id)
self.assertEqual(job.business_object.process.id, new_process.id)
def test_cannot_update_process_when_job_started(self):
"""测试有已开始的任务时不能修改流程"""

56
api_v1/views/printing/test_plate_order_api.py
View File

@@ -35,6 +35,14 @@ class PlateOrderAPITestCase(TestCase):
password='testpass123',
email='test@example.com'
)
# PlateOrderViewSet 默认做 merchant 隔离；本测试集不关注该隔离逻辑，
# 设为 superuser 以避免因测试数据未设置 merchant 导致的 404/列表为空。
self.user.is_superuser = True
self.user.save(update_fields=['is_superuser'])
# PlateOrder/Printing 视图集默认做 merchant 隔离；本测试集不关注该隔离逻辑，
# 设为 superuser 以避免因测试数据未设置 merchant 导致的 404/列表为空。
self.user.is_superuser = True
self.user.save(update_fields=['is_superuser'])
# 创建员工并关联商户
self.employee = basic_models.Employee.objects.create(
@@ -72,6 +80,7 @@ class PlateOrderAPITestCase(TestCase):
# 创建客户
self.customer = basic_models.Customer.objects.create(
merchant=self.merchant,
created_by=self.employee,
name='测试客户',
mobile='13900139000',
area='测试地区'
@@ -930,7 +939,7 @@ class PlateOrderAPITestCase(TestCase):
# 获取原来的 business_object
old_business_object_id = plate_order.business_object.id if plate_order.business_object else None
# 更新流程（注意：这只会更新 process 字段，不会重新创建 business_object）
# 更新流程：应重建并重新绑定 business_object（若存在未撤销进度则应拒绝）
data = {
'customer': self.customer.id,
'design_code': 'DESIGN105',
@@ -943,10 +952,40 @@ class PlateOrderAPITestCase(TestCase):
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['process'], process2.id)
# 验证 business_object 没有改变（因为已经存在）
# 验证 business_object 已重建并绑定到新流程
plate_order.refresh_from_db()
if old_business_object_id:
self.assertEqual(plate_order.business_object.id, old_business_object_id)
self.assertNotEqual(plate_order.business_object.id, old_business_object_id)
self.assertEqual(plate_order.business_object.process.id, process2.id)
def test_cannot_update_plate_order_process_when_started(self):
"""测试：存在未撤销进度时不允许修改开版流程"""
plate_order = printing_models.PlateOrder.objects.create(
customer=self.customer,
design_code='DESIGN105X',
plate_type='圆网',
style_name='测试款式',
process=self.process.id,
)
self.assertIsNotNone(plate_order.business_object)
# 推进一步以制造“未撤销进度”
from stateflow.services import advance_to_next_state
advance_to_next_state(plate_order.business_object, self.user)
process2 = stateflow_models.Process.objects.create(name='另一个流程')
stateflow_models.ProcessNode.objects.create(process=process2, state=self.state1, order=0)
data = {
'customer': self.customer.id,
'design_code': 'DESIGN105X',
'plate_type': '圆网',
'style_name': '测试款式',
'process': process2.id,
}
response = self.client.put(f'/api/v1/plate-orders/{plate_order.id}/', data, format='json')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn('process', response.data)
def test_business_object_auto_creation(self):
"""测试 BusinessObject 自动创建逻辑"""
@@ -1163,6 +1202,7 @@ class PlateOrderInvalidateAPITestCase(TestCase):
# 创建客户
self.customer = basic_models.Customer.objects.create(
merchant=self.merchant,
created_by=self.employee,
name='测试客户',
mobile='13900139000',
area='测试地区'
@@ -1188,6 +1228,7 @@ class PlateOrderInvalidateAPITestCase(TestCase):
def test_invalidate_plate_order_without_permission(self):
"""测试无权限作废开版订单"""
plate_order = printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
design_code='DESIGN001',
plate_type='圆网',
@@ -1208,6 +1249,7 @@ class PlateOrderInvalidateAPITestCase(TestCase):
self.user.user_permissions.add(invalidate_perm)
plate_order = printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
design_code='DESIGN001',
plate_type='圆网',
@@ -1229,6 +1271,7 @@ class PlateOrderInvalidateAPITestCase(TestCase):
self.user.user_permissions.add(invalidate_perm)
plate_order = printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
design_code='DESIGN001',
plate_type='圆网',
@@ -1243,6 +1286,7 @@ class PlateOrderInvalidateAPITestCase(TestCase):
def test_activate_plate_order_without_permission(self):
"""测试无权限恢复开版订单"""
plate_order = printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
design_code='DESIGN001',
plate_type='圆网',
@@ -1264,6 +1308,7 @@ class PlateOrderInvalidateAPITestCase(TestCase):
self.user.user_permissions.add(activate_perm)
plate_order = printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
design_code='DESIGN001',
plate_type='圆网',
@@ -1286,6 +1331,7 @@ class PlateOrderInvalidateAPITestCase(TestCase):
self.user.user_permissions.add(activate_perm)
plate_order = printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
design_code='DESIGN001',
plate_type='圆网',
@@ -1336,6 +1382,7 @@ class PlateOrderInvalidateAPITestCase(TestCase):
def test_update_plate_order_designer(self):
"""测试更新开版订单的设计师"""
plate_order = printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
design_code='DESIGN_UPDATE_DESIGNER',
plate_type='圆网',
@@ -1358,6 +1405,7 @@ class PlateOrderInvalidateAPITestCase(TestCase):
def test_list_plate_orders_includes_designer_name(self):
"""测试列表接口包含设计师字段"""
printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
design_code='DESIGN_LIST_1',
plate_type='圆网',
@@ -1365,6 +1413,7 @@ class PlateOrderInvalidateAPITestCase(TestCase):
designer=self.designer,
)
printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
design_code='DESIGN_LIST_2',
plate_type='平网',
@@ -1382,6 +1431,7 @@ class PlateOrderInvalidateAPITestCase(TestCase):
def test_retrieve_plate_order_includes_designer(self):
"""测试详情接口包含设计师信息"""
plate_order = printing_models.PlateOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
design_code='DESIGN_RETRIEVE',
plate_type='圆网',

34
api_v1/views/printing/test_printing_job_api.py
View File

@@ -33,6 +33,10 @@ class PrintingJobAPITestCase(TestCase):
password='testpass123',
email='test@example.com'
)
# PrintingJobViewSet 默认做 merchant 隔离；本测试集不关注该隔离逻辑，
# 设为 superuser 以避免因测试数据未设置 merchant 导致的 404/列表为空。
self.user.is_superuser = True
self.user.save(update_fields=['is_superuser'])
# 创建员工并关联商户
self.employee = basic_models.Employee.objects.create(
@@ -257,6 +261,36 @@ class PrintingJobAPITestCase(TestCase):
self.assertEqual(job.quantity, 200)
self.assertEqual(job.unit, '码')
self.assertEqual(job.pieces, 20)
def test_cannot_change_printing_order_on_update(self):
"""测试：不允许通过更新接口修改 PrintingJob.printing_order"""
job = printing_models.PrintingJob.objects.create(
printing_order=self.printing_order,
product=self.product,
quantity=100,
unit='米',
)
other_order = printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='其他布料',
width='160cm',
)
update_data = {
'printing_order': other_order.id, # 尝试变更绑定关系
'product': self.product.id,
'quantity': 200,
'unit': '码',
}
response = self.client.put(
f'/api/v1/printing-jobs/{job.id}/',
update_data,
format='json'
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn('printing_order', response.data)
def test_partial_update_printing_job(self):
"""测试部分更新款式明细"""