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

fix: little bugs, feat: woodpecker settings

Browse Source
This commit is contained in:
colaftc
2025-11-14 15:11:26 +08:00
parent 107e65afc0
commit af523d0350
9 changed files with 852 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -205,6 +205,7 @@ class PlateOrderListSerializer(serializers.ModelSerializer):
merchandiser_name = serializers.CharField(source="merchandiser.name", read_only=True)
status = serializers.CharField(read_only=True)
progress_percentage = serializers.IntegerField(read_only=True)
process_name = serializers.SerializerMethodField()
class Meta:
model = models.PlateOrder
@@ -216,13 +217,25 @@ class PlateOrderListSerializer(serializers.ModelSerializer):
'merchandiser', 'merchandiser_name',
'style_name', 'fabric', 'width',
'required_completion_date', 'completion_date',
'is_ordered', 'status', 'progress_percentage',
'is_ordered', 'process', 'process_name',
'status', 'progress_percentage',
'created_at', 'updated_at'
]
read_only_fields = [
'id', 'status', 'progress_percentage',
'created_at', 'updated_at'
]
def get_process_name(self, obj):
"""获取流程名称"""
if obj.process:
try:
from stateflow.models import Process
process = Process.objects.get(id=obj.process)
return process.name
except Process.DoesNotExist:
return None
return None
class PlateOrderDetailSerializer(serializers.ModelSerializer):
@@ -238,6 +251,7 @@ class PlateOrderDetailSerializer(serializers.ModelSerializer):
progress_percentage = serializers.IntegerField(read_only=True)
business_object_id = serializers.IntegerField(source="business_object.id", read_only=True)
plate_image_url = serializers.SerializerMethodField()
process_name = serializers.SerializerMethodField()
class Meta:
model = models.PlateOrder
@@ -254,6 +268,7 @@ class PlateOrderDetailSerializer(serializers.ModelSerializer):
'sample_meter', 'required_sample_meters',
'required_completion_date', 'completion_date',
'approval_result', 'is_ordered', 'customer_feedback',
'process', 'process_name',
'status', 'status_id', 'is_completed', 'has_started',
'progress_percentage', 'business_object_id',
'created_at', 'updated_at'
@@ -272,6 +287,17 @@ class PlateOrderDetailSerializer(serializers.ModelSerializer):
return request.build_absolute_uri(obj.plate_image.url)
return obj.plate_image.url
return None
def get_process_name(self, obj):
"""获取流程名称"""
if obj.process:
try:
from stateflow.models import Process
process = Process.objects.get(id=obj.process)
return process.name
except Process.DoesNotExist:
return None
return None
class PlateOrderCreateUpdateSerializer(serializers.ModelSerializer):
@@ -290,7 +316,8 @@ class PlateOrderCreateUpdateSerializer(serializers.ModelSerializer):
"sample_rating", "difficulty_rating",
"sample_meter", "required_sample_meters",
"required_completion_date", "completion_date",
"approval_result", "is_ordered", "customer_feedback"
"approval_result", "is_ordered", "customer_feedback",
"process"
]
read_only_fields = ["id"]
@@ -324,6 +351,14 @@ class PlateOrderCreateUpdateSerializer(serializers.ModelSerializer):
raise serializers.ValidationError("所需样品米数不能为负数")
return value
def validate_process(self, value):
"""验证流程是否存在"""
if value:
from stateflow.models import Process
if not Process.objects.filter(id=value).exists():
raise serializers.ValidationError("流程不存在")
return value
def validate(self, attrs):
"""交叉验证"""
return attrs
@@ -332,16 +367,14 @@ class PlateOrderCreateUpdateSerializer(serializers.ModelSerializer):
"""创建开版订单"""
user = self.context["request"].user
# 创建 PlateOrder
# 创建 PlateOrder，save 方法会自动创建 BusinessObject
plate_order = models.PlateOrder.objects.create(**validated_data)
# 暂时不自动创建 BusinessObject，由 admin 或其他地方创建
return plate_order
def update(self, instance, validated_data):
"""更新开版订单"""
# 直接更新字段
# 直接更新字段，save 方法会自动处理 BusinessObject 的创建
for attr, value in validated_data.items():
setattr(instance, attr, value)
instance.save()

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

@@ -8,6 +8,7 @@ from django.contrib.auth import get_user_model
from django.contrib.auth.models import Permission
from basic_info import models as basic_models
from printing import models as printing_models
from stateflow import models as stateflow_models
User = get_user_model()
@@ -76,6 +77,19 @@ class PlateOrderAPITestCase(TestCase):
change_perm = Permission.objects.get(codename='change_plateorder')
delete_perm = Permission.objects.get(codename='delete_plateorder')
self.user.user_permissions.add(view_perm, add_perm, change_perm, delete_perm)
# 创建测试流程
self.state1 = stateflow_models.State.objects.create(name='画图', description='画图阶段')
self.state2 = stateflow_models.State.objects.create(name='调色', description='调色阶段')
self.state3 = stateflow_models.State.objects.create(name='套样', description='套样阶段')
self.process = stateflow_models.Process.objects.create(
name='开版流程',
description='标准开版流程'
)
stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state1, order=0)
stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state2, order=1)
stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state3, order=2)
def test_create_plate_order(self):
"""测试创建开版订单"""
@@ -401,6 +415,298 @@ class PlateOrderAPITestCase(TestCase):
else:
self.assertEqual(response.data[0]['id'], order1.id)
self.assertEqual(response.data[1]['id'], order2.id)
def test_create_plate_order_with_process(self):
"""测试创建开版订单时指定流程"""
data = {
'customer': self.customer.id,
'design_code': 'DESIGN100',
'plate_type': '圆网',
'style_name': '测试款式',
'fabric': '棉布',
'process': self.process.id,
}
response = self.client.post('/api/v1/plate-orders/', data, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(response.data['process'], self.process.id)
# 验证数据库中创建了 BusinessObject
plate_order = printing_models.PlateOrder.objects.get(id=response.data['id'])
self.assertIsNotNone(plate_order.business_object)
self.assertEqual(plate_order.business_object.process.id, self.process.id)
def test_create_plate_order_without_process(self):
"""测试创建开版订单时不指定流程（使用默认流程）"""
data = {
'customer': self.customer.id,
'design_code': 'DESIGN101',
'plate_type': '圆网',
'style_name': '测试款式',
'fabric': '棉布',
}
response = self.client.post('/api/v1/plate-orders/', data, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
# 应该使用默认流程
self.assertIsNotNone(response.data['process'])
# 验证使用了默认流程
plate_order = printing_models.PlateOrder.objects.get(id=response.data['id'])
from django.conf import settings
self.assertEqual(plate_order.process, settings.PLATE_ORDER_DEFAULT_PROCESS_ID)
def test_create_plate_order_with_invalid_process(self):
"""测试创建开版订单时指定无效的流程ID"""
data = {
'customer': self.customer.id,
'design_code': 'DESIGN102',
'plate_type': '圆网',
'style_name': '测试款式',
'fabric': '棉布',
'process': 99999, # 不存在的流程ID
}
response = self.client.post('/api/v1/plate-orders/', data, format='json')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn('process', response.data)
def test_retrieve_plate_order_with_process_name(self):
"""测试获取开版订单详情时包含流程名称"""
plate_order = printing_models.PlateOrder.objects.create(
customer=self.customer,
design_code='DESIGN103',
plate_type='圆网',
style_name='测试款式',
fabric='棉布',
process=self.process.id,
)
response = self.client.get(f'/api/v1/plate-orders/{plate_order.id}/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['process'], self.process.id)
self.assertEqual(response.data['process_name'], self.process.name)
# 验证自动创建了 business_object
self.assertIsNotNone(response.data['business_object_id'])
def test_list_plate_orders_includes_process_name(self):
"""测试列表接口包含流程名称"""
printing_models.PlateOrder.objects.create(
customer=self.customer,
design_code='DESIGN104',
plate_type='圆网',
style_name='款式1',
process=self.process.id,
)
response = self.client.get('/api/v1/plate-orders/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
if isinstance(response.data, dict):
result = response.data['results'][0]
else:
result = response.data[0]
self.assertEqual(result['process'], self.process.id)
self.assertEqual(result['process_name'], self.process.name)
def test_update_plate_order_process(self):
"""测试更新开版订单的流程"""
# 创建另一个流程
process2 = stateflow_models.Process.objects.create(
name='快速开版流程',
description='简化的开版流程'
)
stateflow_models.ProcessNode.objects.create(process=process2, state=self.state1, order=0)
plate_order = printing_models.PlateOrder.objects.create(
customer=self.customer,
design_code='DESIGN105',
plate_type='圆网',
style_name='测试款式',
process=self.process.id,
)
# 获取原来的 business_object
old_business_object_id = plate_order.business_object.id if plate_order.business_object else None
# 更新流程（注意：这只会更新 process 字段，不会重新创建 business_object）
data = {
'customer': self.customer.id,
'design_code': 'DESIGN105',
'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_200_OK)
self.assertEqual(response.data['process'], process2.id)
# 验证 business_object 没有改变（因为已经存在）
plate_order.refresh_from_db()
if old_business_object_id:
self.assertEqual(plate_order.business_object.id, old_business_object_id)
def test_business_object_auto_creation(self):
"""测试 BusinessObject 自动创建逻辑"""
# 直接创建 PlateOrder（不通过 API）
plate_order = printing_models.PlateOrder(
customer=self.customer,
design_code='DESIGN106',
plate_type='圆网',
style_name='测试款式',
process=self.process.id,
)
# 保存前没有 business_object
self.assertIsNone(plate_order.business_object)
# 保存后自动创建
plate_order.save()
plate_order.refresh_from_db()
self.assertIsNotNone(plate_order.business_object)
self.assertEqual(plate_order.business_object.process.id, self.process.id)
def test_advance_to_next_state(self):
"""测试推进到下一个状态"""
plate_order = printing_models.PlateOrder.objects.create(
customer=self.customer,
design_code='DESIGN107',
plate_type='圆网',
style_name='测试款式',
process=self.process.id,
)
# 确保有 business_object
self.assertIsNotNone(plate_order.business_object)
response = self.client.post(f'/api/v1/plate-orders/{plate_order.id}/advance-to-next-state/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('detail', response.data)
self.assertIn('data', response.data)
# 验证状态改变
plate_order.refresh_from_db()
self.assertIsNotNone(plate_order.status)
def test_advance_to_next_state_without_business_object(self):
"""测试没有 business_object 时推进状态"""
plate_order = printing_models.PlateOrder.objects.create(
customer=self.customer,
design_code='DESIGN108',
plate_type='圆网',
style_name='测试款式',
)
# 手动清除 business_object
plate_order.business_object = None
plate_order.save(update_fields=['business_object'])
response = self.client.post(f'/api/v1/plate-orders/{plate_order.id}/advance-to-next-state/')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn('没有关联的流程实例', response.data['detail'])
def test_step_back_one_state(self):
"""测试回退一步"""
plate_order = printing_models.PlateOrder.objects.create(
customer=self.customer,
design_code='DESIGN109',
plate_type='圆网',
style_name='测试款式',
process=self.process.id,
)
# 先推进一步
from stateflow import services as stateflow_services
stateflow_services.advance_to_next_state(plate_order.business_object, self.user)
# 再回退
response = self.client.post(f'/api/v1/plate-orders/{plate_order.id}/step-back-one-state/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('detail', response.data)
self.assertIn('data', response.data)
def test_completed_states(self):
"""测试查询已完成的流程列表"""
plate_order = printing_models.PlateOrder.objects.create(
customer=self.customer,
design_code='DESIGN110',
plate_type='圆网',
style_name='测试款式',
process=self.process.id,
)
# 推进几步
from stateflow import services as stateflow_services
stateflow_services.advance_to_next_state(plate_order.business_object, self.user)
stateflow_services.advance_to_next_state(plate_order.business_object, self.user)
response = self.client.get(f'/api/v1/plate-orders/{plate_order.id}/completed-states/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('count', response.data)
self.assertIn('results', response.data)
self.assertEqual(response.data['count'], 2)
def test_completed_states_include_cancelled(self):
"""测试查询包含已撤销的流程"""
plate_order = printing_models.PlateOrder.objects.create(
customer=self.customer,
design_code='DESIGN111',
plate_type='圆网',
style_name='测试款式',
process=self.process.id,
)
response = self.client.get(f'/api/v1/plate-orders/{plate_order.id}/completed-states/?include_cancelled=true')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('count', response.data)
self.assertIn('results', response.data)
def test_timeline(self):
"""测试获取流程时间线"""
plate_order = printing_models.PlateOrder.objects.create(
customer=self.customer,
design_code='DESIGN112',
plate_type='圆网',
style_name='测试款式',
process=self.process.id,
)
# 推进一步
from stateflow import services as stateflow_services
stateflow_services.advance_to_next_state(plate_order.business_object, self.user)
response = self.client.get(f'/api/v1/plate-orders/{plate_order.id}/timeline/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('count', response.data)
self.assertIn('results', response.data)
# 验证返回的数据结构
if response.data['count'] > 0:
first_item = response.data['results'][0]
self.assertIn('state_id', first_item)
self.assertIn('state_name', first_item)
self.assertIn('state_description', first_item)
self.assertIn('order', first_item)
self.assertIn('status', first_item)
def test_timeline_without_business_object(self):
"""测试没有 business_object 时获取时间线"""
plate_order = printing_models.PlateOrder.objects.create(
customer=self.customer,
design_code='DESIGN113',
plate_type='圆网',
style_name='测试款式',
)
# 手动清除 business_object
plate_order.business_object = None
plate_order.save(update_fields=['business_object'])
response = self.client.get(f'/api/v1/plate-orders/{plate_order.id}/timeline/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['count'], 0)
class PlateOrderInvalidateAPITestCase(TestCase):
@@ -448,6 +754,14 @@ class PlateOrderInvalidateAPITestCase(TestCase):
add_perm = Permission.objects.get(codename='add_plateorder')
change_perm = Permission.objects.get(codename='change_plateorder')
self.user.user_permissions.add(view_perm, add_perm, change_perm)
# 创建测试流程（用于测试）
self.state1 = stateflow_models.State.objects.create(name='画图', description='画图阶段')
self.process = stateflow_models.Process.objects.create(
name='开版流程',
description='标准开版流程'
)
stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state1, order=0)
def test_invalidate_plate_order_without_permission(self):
"""测试无权限作废开版订单"""

173
api_v1/views/printing/views.py
View File

@@ -482,9 +482,6 @@ class HasActivatePlateOrderPermission(BasePermission):
return request.user.has_perm('printing.can_activate_plateorder')
class PlateOrderFilterSet(django_filters.FilterSet):
"""开版订单过滤器"""
customer_name = django_filters.CharFilter(field_name='customer__name', lookup_expr='icontains')
@@ -523,6 +520,10 @@ class PlateOrderViewSet(viewsets.ModelViewSet):
partial_update: 部分更新开版订单
invalidate: 作废开版订单
activate: 恢复开版订单
advance_to_next_state: 推进到下一个状态
step_back_one_state: 回退一步
completed_states: 查询已完成的流程列表
timeline: 获取流程时间线
查询参数:
- customer: 客户ID
@@ -651,3 +652,169 @@ class PlateOrderViewSet(viewsets.ModelViewSet):
'detail': '开版订单已恢复',
'data': serializer.data
})
@action(detail=True, methods=['post'], url_path='advance-to-next-state')
def advance_to_next_state(self, request, pk=None):
"""
推进到下一个状态
将开版订单推进到下一个流程状态
"""
plate_order = self.get_object()
if not plate_order.business_object:
return Response(
{'detail': '该开版订单没有关联的流程实例'},
status=status.HTTP_400_BAD_REQUEST
)
from stateflow import services as stateflow_services
success, message = stateflow_services.advance_to_next_state(
plate_order.business_object, request.user
)
if not success:
return Response(
{'detail': message},
status=status.HTTP_400_BAD_REQUEST
)
# 重新获取 plate_order 以刷新状态
plate_order.refresh_from_db()
from .serializers import PlateOrderDetailSerializer
serializer = PlateOrderDetailSerializer(plate_order, context={'request': request})
return Response({
'detail': message,
'data': serializer.data
})
@action(detail=True, methods=['post'], url_path='step-back-one-state')
def step_back_one_state(self, request, pk=None):
"""
回退一步
将开版订单回退到上一个流程状态
"""
plate_order = self.get_object()
if not plate_order.business_object:
return Response(
{'detail': '该开版订单没有关联的流程实例'},
status=status.HTTP_400_BAD_REQUEST
)
from stateflow import services as stateflow_services
success, message = stateflow_services.step_back_one_state(
plate_order.business_object, request.user
)
if not success:
return Response(
{'detail': message},
status=status.HTTP_400_BAD_REQUEST
)
# 重新获取 plate_order 以刷新状态
plate_order.refresh_from_db()
from .serializers import PlateOrderDetailSerializer
serializer = PlateOrderDetailSerializer(plate_order, context={'request': request})
return Response({
'detail': message,
'data': serializer.data
})
@action(detail=True, methods=['get'], url_path='completed-states')
def completed_states(self, request, pk=None):
"""
查询已完成的流程列表
查询参数:
- include_cancelled: 是否包含已撤销的流程（true/false），默认 false
"""
plate_order = self.get_object()
if not plate_order.business_object:
return Response({
'count': 0,
'results': []
})
# 获取查询参数
include_cancelled = request.query_params.get('include_cancelled', 'false').lower() == 'true'
# 查询已完成的状态
state_logs = plate_order.business_object.state_logs.select_related('state', 'completed_by').order_by('completed_at')
if not include_cancelled:
state_logs = state_logs.filter(is_cancelled=False)
results = []
for log in state_logs:
results.append({
'id': log.id,
'state_id': log.state.id,
'state_name': log.state.name,
'completed_at': log.completed_at,
'completed_by': log.completed_by.username if log.completed_by else None,
'completed_by_name': getattr(log.completed_by, 'employee', None) and log.completed_by.employee.name or None,
'is_cancelled': log.is_cancelled,
'cancelled_at': log.cancelled_at,
})
return Response({
'count': len(results),
'results': results
})
@action(detail=True, methods=['get'], url_path='timeline')
def timeline(self, request, pk=None):
"""
获取流程时间线
返回该开版订单的完整流程时间线，包括未开始、进行中和已完成的状态
不包含已撤销的流程记录
"""
plate_order = self.get_object()
if not plate_order.business_object:
return Response({
'count': 0,
'results': []
})
from stateflow import services as stateflow_services
# 获取流程
from stateflow.models import Process
try:
process = Process.objects.get(id=plate_order.process)
except Process.DoesNotExist:
return Response({
'count': 0,
'results': []
})
timeline_data = stateflow_services.get_business_object_state_timeline(plate_order.business_object)
# 过滤掉已撤销的记录
timeline_data = [item for item in timeline_data if not item.get('is_cancelled')]
results = []
for item in timeline_data:
results.append({
'state_id': item['state'].id,
'state_name': item['state'].name,
'state_description': item['state'].description,
'order': item['order'],
'status': item['status'], # not_started, in_progress, completed
'completed_at': item['completed_at'],
'completed_by': item['completed_by'].username if item['completed_by'] else None,
'completed_by_name': getattr(item['completed_by'], 'employee', None) and item['completed_by'].employee.name or None if item['completed_by'] else None,
})
return Response({
'count': len(results),
'results': results
})