forked from erp-dev/erp
1088 lines
40 KiB
Python
1088 lines
40 KiB
Python
"""
|
||
PrintingJob API 测试
|
||
"""
|
||
from django.test import TestCase
|
||
from django.conf import settings
|
||
from rest_framework.test import APIClient
|
||
from rest_framework import status
|
||
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()
|
||
|
||
|
||
class PrintingJobAPITestCase(TestCase):
|
||
"""测试 PrintingJob API"""
|
||
|
||
def setUp(self):
|
||
self.client = APIClient()
|
||
|
||
# 创建商户
|
||
self.merchant = basic_models.Merchant.objects.create(
|
||
name='测试印花厂',
|
||
type=basic_models.MerchantTypeEnum.FACTORY
|
||
)
|
||
|
||
# 创建用户
|
||
self.user = User.objects.create_user(
|
||
username='testuser',
|
||
password='testpass123',
|
||
email='test@example.com'
|
||
)
|
||
|
||
# 创建员工并关联商户
|
||
self.employee = basic_models.Employee.objects.create(
|
||
sys_user=self.user,
|
||
merchant=self.merchant,
|
||
name='测试员工',
|
||
mobile='13800138000',
|
||
job_type=basic_models.EmployeeTypeEnum.PRINTER,
|
||
status=basic_models.EmployeeStatusEnum.ACTIVE
|
||
)
|
||
|
||
# 创建客户
|
||
self.customer = basic_models.Customer.objects.create(
|
||
merchant=self.merchant,
|
||
name='测试客户',
|
||
mobile='13900139000',
|
||
area='测试地区'
|
||
)
|
||
|
||
# 创建流程
|
||
self.state1 = stateflow_models.State.objects.create(name='待印染')
|
||
self.state2 = stateflow_models.State.objects.create(name='印染中')
|
||
self.state3 = stateflow_models.State.objects.create(name='已完成')
|
||
|
||
self.process = stateflow_models.Process.objects.create(name='印染流程')
|
||
self.process.replace_nodes([self.state1, self.state2, self.state3])
|
||
|
||
# 设置默认流程
|
||
settings.PRINTING_DEFAULT_PROCESS_ID = self.process.id
|
||
|
||
# 创建印染订单
|
||
self.printing_order = printing_models.PrintingOrder.objects.create(
|
||
customer=self.customer,
|
||
fabric='测试布料',
|
||
width='150cm',
|
||
process=self.process,
|
||
)
|
||
|
||
# 创建产品类别
|
||
self.category = basic_models.ProductCategory.objects.create(
|
||
merchant=self.merchant,
|
||
name='测试类别'
|
||
)
|
||
|
||
# 创建产品
|
||
self.product = basic_models.Product.objects.create(
|
||
merchant=self.merchant,
|
||
category=self.category,
|
||
name='测试产品',
|
||
human_id='TEST001',
|
||
unit=basic_models.ProductUnitEnum.METER
|
||
)
|
||
|
||
# 认证用户
|
||
self.client.force_authenticate(user=self.user)
|
||
|
||
# 给用户添加基础权限
|
||
view_perm = Permission.objects.get(codename='view_printingjob')
|
||
add_perm = Permission.objects.get(codename='add_printingjob')
|
||
change_perm = Permission.objects.get(codename='change_printingjob')
|
||
self.user.user_permissions.add(view_perm, add_perm, change_perm)
|
||
|
||
def test_create_printing_job(self):
|
||
"""测试创建印染款式明细"""
|
||
data = {
|
||
'printing_order': self.printing_order.id,
|
||
'product': self.product.id,
|
||
'quantity': 100,
|
||
'unit': '米',
|
||
'size': '50*60',
|
||
'pieces': 10,
|
||
'description': '测试备注'
|
||
}
|
||
|
||
response = self.client.post('/api/v1/printing-jobs/', data, format='json')
|
||
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
||
|
||
# 验证创建成功
|
||
job = printing_models.PrintingJob.objects.filter(
|
||
printing_order=self.printing_order,
|
||
product=self.product
|
||
).first()
|
||
self.assertIsNotNone(job)
|
||
self.assertEqual(job.quantity, 100)
|
||
self.assertEqual(job.unit, '米')
|
||
self.assertEqual(job.pieces, 10)
|
||
|
||
def test_list_printing_jobs(self):
|
||
"""测试获取款式明细列表"""
|
||
# 创建测试数据
|
||
printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=200,
|
||
unit='米',
|
||
size='60*70',
|
||
pieces=20
|
||
)
|
||
|
||
response = self.client.get('/api/v1/printing-jobs/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(len(response.data), 2)
|
||
|
||
def test_retrieve_printing_job(self):
|
||
"""测试获取款式明细详情"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10,
|
||
description='详情测试'
|
||
)
|
||
|
||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['quantity'], 100)
|
||
self.assertEqual(response.data['unit'], '米')
|
||
self.assertIn('product_name', response.data)
|
||
self.assertEqual(response.data['product_name'], self.product.name)
|
||
|
||
def test_update_printing_job(self):
|
||
"""测试更新款式明细"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
update_data = {
|
||
'printing_order': self.printing_order.id,
|
||
'product': self.product.id,
|
||
'quantity': 200,
|
||
'unit': '码',
|
||
'size': '60*70',
|
||
'pieces': 20,
|
||
'description': '更新后的备注'
|
||
}
|
||
|
||
response = self.client.put(
|
||
f'/api/v1/printing-jobs/{job.id}/',
|
||
update_data,
|
||
format='json'
|
||
)
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
job.refresh_from_db()
|
||
self.assertEqual(job.quantity, 200)
|
||
self.assertEqual(job.unit, '码')
|
||
self.assertEqual(job.pieces, 20)
|
||
|
||
def test_partial_update_printing_job(self):
|
||
"""测试部分更新款式明细"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
patch_data = {
|
||
'quantity': 150,
|
||
'pieces': 15
|
||
}
|
||
|
||
response = self.client.patch(
|
||
f'/api/v1/printing-jobs/{job.id}/',
|
||
patch_data,
|
||
format='json'
|
||
)
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
job.refresh_from_db()
|
||
self.assertEqual(job.quantity, 150)
|
||
self.assertEqual(job.pieces, 15)
|
||
self.assertEqual(job.unit, '米') # 未修改字段保持不变
|
||
|
||
def test_delete_printing_job_forbidden(self):
|
||
"""测试删除款式明细被禁用"""
|
||
# 添加删除权限以便测试destroy方法的自定义逻辑
|
||
delete_perm = Permission.objects.get(codename='delete_printingjob')
|
||
self.user.user_permissions.add(delete_perm)
|
||
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
response = self.client.delete(f'/api/v1/printing-jobs/{job.id}/')
|
||
self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)
|
||
self.assertIn('不支持删除', response.data['detail'])
|
||
|
||
# 验证明细仍然存在
|
||
self.assertTrue(
|
||
printing_models.PrintingJob.objects.filter(id=job.id).exists()
|
||
)
|
||
|
||
def test_filter_by_printing_order(self):
|
||
"""测试按印染订单过滤"""
|
||
order2 = printing_models.PrintingOrder.objects.create(
|
||
customer=self.customer,
|
||
fabric='其他布料',
|
||
width='160cm'
|
||
)
|
||
|
||
printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
printing_models.PrintingJob.objects.create(
|
||
printing_order=order2,
|
||
product=self.product,
|
||
quantity=200,
|
||
unit='米',
|
||
size='60*70',
|
||
pieces=20
|
||
)
|
||
|
||
response = self.client.get(f'/api/v1/printing-jobs/?printing_order={self.printing_order.id}')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(len(response.data), 1)
|
||
self.assertEqual(response.data[0]['printing_order'], self.printing_order.id)
|
||
|
||
def test_filter_by_product(self):
|
||
"""测试按产品过滤"""
|
||
product2 = basic_models.Product.objects.create(
|
||
merchant=self.merchant,
|
||
category=self.category,
|
||
name='产品2',
|
||
human_id='TEST002',
|
||
unit=basic_models.ProductUnitEnum.METER
|
||
)
|
||
|
||
printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=product2,
|
||
quantity=200,
|
||
unit='米',
|
||
size='60*70',
|
||
pieces=20
|
||
)
|
||
|
||
response = self.client.get(f'/api/v1/printing-jobs/?product={self.product.id}')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(len(response.data), 1)
|
||
self.assertEqual(response.data[0]['product'], self.product.id)
|
||
|
||
def test_filter_by_quantity_range(self):
|
||
"""测试按数量范围过滤"""
|
||
printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=50,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=150,
|
||
unit='米',
|
||
size='60*70',
|
||
pieces=20
|
||
)
|
||
printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=250,
|
||
unit='米',
|
||
size='70*80',
|
||
pieces=30
|
||
)
|
||
|
||
response = self.client.get('/api/v1/printing-jobs/?quantity_min=100&quantity_max=200')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(len(response.data), 1)
|
||
self.assertEqual(response.data[0]['quantity'], 150)
|
||
|
||
def test_search_by_product_name(self):
|
||
"""测试按产品名称搜索"""
|
||
product2 = basic_models.Product.objects.create(
|
||
merchant=self.merchant,
|
||
category=self.category,
|
||
name='特殊产品',
|
||
human_id='TEST003',
|
||
unit=basic_models.ProductUnitEnum.METER
|
||
)
|
||
|
||
printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=product2,
|
||
quantity=200,
|
||
unit='米',
|
||
size='60*70',
|
||
pieces=20
|
||
)
|
||
|
||
response = self.client.get('/api/v1/printing-jobs/?search=特殊')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(len(response.data), 1)
|
||
self.assertIn('特殊', response.data[0]['product_name'])
|
||
|
||
def test_ordering(self):
|
||
"""测试排序"""
|
||
job1 = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
job2 = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=200,
|
||
unit='米',
|
||
size='60*70',
|
||
pieces=20
|
||
)
|
||
|
||
# 按数量升序
|
||
response = self.client.get('/api/v1/printing-jobs/?ordering=quantity')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data[0]['quantity'], 100)
|
||
self.assertEqual(response.data[1]['quantity'], 200)
|
||
|
||
# 按数量降序
|
||
response = self.client.get('/api/v1/printing-jobs/?ordering=-quantity')
|
||
self.assertEqual(response.data[0]['quantity'], 200)
|
||
self.assertEqual(response.data[1]['quantity'], 100)
|
||
|
||
def test_validate_quantity_positive(self):
|
||
"""测试数量必须大于0"""
|
||
data = {
|
||
'printing_order': self.printing_order.id,
|
||
'product': self.product.id,
|
||
'quantity': 0, # 无效数量
|
||
'unit': '米',
|
||
'size': '50*60',
|
||
'pieces': 10
|
||
}
|
||
|
||
response = self.client.post('/api/v1/printing-jobs/', data, format='json')
|
||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||
self.assertIn('quantity', response.data)
|
||
|
||
def test_validate_pieces_can_be_zero(self):
|
||
"""测试件数可以为0或null"""
|
||
data = {
|
||
'printing_order': self.printing_order.id,
|
||
'product': self.product.id,
|
||
'quantity': 100,
|
||
'unit': '米',
|
||
'size': '50*60',
|
||
'pieces': 0 # 允许为0
|
||
}
|
||
|
||
response = self.client.post('/api/v1/printing-jobs/', data, format='json')
|
||
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
||
|
||
# 验证创建成功
|
||
job = printing_models.PrintingJob.objects.get(id=response.data['id'])
|
||
self.assertEqual(job.pieces, 0)
|
||
|
||
def test_create_job_with_business_object(self):
|
||
"""测试创建任务自动创建 BusinessObject"""
|
||
data = {
|
||
'printing_order': self.printing_order.id,
|
||
'product': self.product.id,
|
||
'quantity': 100,
|
||
'unit': '米',
|
||
'size': '50*60',
|
||
'pieces': 10
|
||
}
|
||
|
||
response = self.client.post('/api/v1/printing-jobs/', data, format='json')
|
||
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
||
|
||
# 验证 BusinessObject 已创建
|
||
job = printing_models.PrintingJob.objects.get(id=response.data['id'])
|
||
self.assertIsNotNone(job.business_object)
|
||
self.assertEqual(job.business_object.process, self.process)
|
||
|
||
def test_job_status_in_detail(self):
|
||
"""测试任务详情包含状态字段"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
# 创建 BusinessObject
|
||
business_object = stateflow_models.BusinessObject.objects.create(
|
||
name=f'PrintingJob-{job.id}',
|
||
process=self.process,
|
||
)
|
||
job.business_object = business_object
|
||
job.save()
|
||
|
||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
# 验证状态字段
|
||
self.assertIn('status', response.data)
|
||
self.assertIn('status_id', response.data)
|
||
self.assertIn('is_completed', response.data)
|
||
self.assertIn('has_started', response.data)
|
||
self.assertIn('business_object_id', response.data)
|
||
|
||
# 未推进状态，但有 BusinessObject，当前是初始状态
|
||
self.assertIn(response.data['status'], ['待印染', '未开始']) # 初始状态或未开始
|
||
self.assertFalse(response.data['is_completed'])
|
||
self.assertFalse(response.data['has_started'])
|
||
|
||
def test_job_status_in_list(self):
|
||
"""测试任务列表包含状态字段"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
response = self.client.get('/api/v1/printing-jobs/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
# 验证列表中的状态字段
|
||
self.assertIn('status', response.data[0])
|
||
self.assertIn('is_completed', response.data[0])
|
||
|
||
def test_job_completion_status(self):
|
||
"""测试任务完成状态判断"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
# 创建 BusinessObject
|
||
business_object = stateflow_models.BusinessObject.objects.create(
|
||
name=f'PrintingJob-{job.id}',
|
||
process=self.process,
|
||
)
|
||
job.business_object = business_object
|
||
job.save()
|
||
|
||
# 推进到最后一个状态（流程有3个节点，需要推进3次）
|
||
from stateflow.services import advance_to_next_state
|
||
advance_to_next_state(business_object, self.user) # 完成第1个状态
|
||
advance_to_next_state(business_object, self.user) # 完成第2个状态
|
||
advance_to_next_state(business_object, self.user) # 完成第3个状态
|
||
|
||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
# 验证完成状态
|
||
self.assertEqual(response.data['status'], '已完成')
|
||
self.assertTrue(response.data['is_completed'])
|
||
self.assertTrue(response.data['has_started'])
|
||
|
||
def test_advance_to_next_state(self):
|
||
"""测试推进到下一个状态 API"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
# 创建 BusinessObject
|
||
business_object = stateflow_models.BusinessObject.objects.create(
|
||
name=f'PrintingJob-{job.id}',
|
||
process=self.process,
|
||
)
|
||
job.business_object = business_object
|
||
job.save()
|
||
|
||
# 推进到下一个状态
|
||
response = self.client.post(f'/api/v1/printing-jobs/{job.id}/advance-to-next-state/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertIn('detail', response.data)
|
||
self.assertIn('data', response.data)
|
||
self.assertIn('已完成状态', response.data['detail'])
|
||
|
||
# 验证状态已更新
|
||
job.refresh_from_db()
|
||
self.assertTrue(job.has_started)
|
||
self.assertEqual(job.status, self.state2.name) # 应该是第二个状态（当前进行中的）
|
||
|
||
def test_advance_to_next_state_without_business_object(self):
|
||
"""测试推进到下一个状态 - 没有 BusinessObject 的情况"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
response = self.client.post(f'/api/v1/printing-jobs/{job.id}/advance-to-next-state/')
|
||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||
self.assertIn('没有关联的流程实例', response.data['detail'])
|
||
|
||
def test_advance_with_parameters(self):
|
||
"""测试推进到下一个状态 - 带参数"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
# 创建 BusinessObject
|
||
business_object = stateflow_models.BusinessObject.objects.create(
|
||
name=f'PrintingJob-{job.id}',
|
||
process=self.process,
|
||
)
|
||
job.business_object = business_object
|
||
job.save()
|
||
|
||
# 推进到下一个状态，带参数
|
||
data = {
|
||
'parameters': {
|
||
'temperature': '25.5',
|
||
'operator': '张三',
|
||
'humidity': '60%'
|
||
}
|
||
}
|
||
response = self.client.post(
|
||
f'/api/v1/printing-jobs/{job.id}/advance-to-next-state/',
|
||
data,
|
||
format='json'
|
||
)
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertIn('detail', response.data)
|
||
self.assertIn('data', response.data)
|
||
|
||
# 验证参数已保存
|
||
job.refresh_from_db()
|
||
state_log = business_object.state_logs.filter(state=self.state1, is_cancelled=False).first()
|
||
self.assertIsNotNone(state_log)
|
||
|
||
# 验证参数记录
|
||
param_record = state_log.parameter_records.first()
|
||
self.assertIsNotNone(param_record)
|
||
self.assertEqual(param_record.parameters['temperature'], '25.5')
|
||
self.assertEqual(param_record.parameters['operator'], '张三')
|
||
self.assertEqual(param_record.parameters['humidity'], '60%')
|
||
|
||
def test_advance_with_required_parameters_missing(self):
|
||
"""测试推进到下一个状态 - 缺少必填参数"""
|
||
# 创建带必填参数的状态
|
||
param1 = stateflow_models.StateParameter.objects.create(
|
||
key='temperature',
|
||
is_required=True,
|
||
description='温度'
|
||
)
|
||
param2 = stateflow_models.StateParameter.objects.create(
|
||
key='operator',
|
||
is_required=True,
|
||
description='操作员'
|
||
)
|
||
|
||
state_with_params = stateflow_models.State.objects.create(name='带参数状态')
|
||
state_with_params.parameters.add(param1, param2)
|
||
|
||
# 创建新流程，第一个状态是带必填参数的
|
||
process_with_params = stateflow_models.Process.objects.create(name='带参数流程')
|
||
process_with_params.replace_nodes([state_with_params, self.state2])
|
||
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
# 创建 BusinessObject
|
||
business_object = stateflow_models.BusinessObject.objects.create(
|
||
name=f'PrintingJob-{job.id}',
|
||
process=process_with_params,
|
||
)
|
||
job.business_object = business_object
|
||
job.save()
|
||
|
||
# 尝试推进但不提供必填参数
|
||
response = self.client.post(f'/api/v1/printing-jobs/{job.id}/advance-to-next-state/')
|
||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||
self.assertIn('缺失必填参数', response.data['detail'])
|
||
self.assertIn('temperature', response.data['detail'])
|
||
self.assertIn('operator', response.data['detail'])
|
||
|
||
def test_advance_with_partial_required_parameters(self):
|
||
"""测试推进到下一个状态 - 只提供部分必填参数"""
|
||
# 创建带必填参数的状态
|
||
param1 = stateflow_models.StateParameter.objects.create(
|
||
key='temperature',
|
||
is_required=True,
|
||
description='温度'
|
||
)
|
||
param2 = stateflow_models.StateParameter.objects.create(
|
||
key='operator',
|
||
is_required=True,
|
||
description='操作员'
|
||
)
|
||
|
||
state_with_params = stateflow_models.State.objects.create(name='带参数状态')
|
||
state_with_params.parameters.add(param1, param2)
|
||
|
||
# 创建新流程
|
||
process_with_params = stateflow_models.Process.objects.create(name='带参数流程')
|
||
process_with_params.replace_nodes([state_with_params, self.state2])
|
||
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
# 创建 BusinessObject
|
||
business_object = stateflow_models.BusinessObject.objects.create(
|
||
name=f'PrintingJob-{job.id}',
|
||
process=process_with_params,
|
||
)
|
||
job.business_object = business_object
|
||
job.save()
|
||
|
||
# 只提供一个必填参数
|
||
data = {
|
||
'parameters': {
|
||
'temperature': '25.5'
|
||
}
|
||
}
|
||
response = self.client.post(
|
||
f'/api/v1/printing-jobs/{job.id}/advance-to-next-state/',
|
||
data,
|
||
format='json'
|
||
)
|
||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||
self.assertIn('缺失必填参数', response.data['detail'])
|
||
self.assertIn('operator', response.data['detail'])
|
||
self.assertNotIn('temperature', response.data['detail']) # temperature 已提供
|
||
|
||
def test_advance_through_all_states(self):
|
||
"""测试推进到完成所有状态"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
# 创建 BusinessObject
|
||
business_object = stateflow_models.BusinessObject.objects.create(
|
||
name=f'PrintingJob-{job.id}',
|
||
process=self.process,
|
||
)
|
||
job.business_object = business_object
|
||
job.save()
|
||
|
||
# 推进 3 次完成所有状态
|
||
for i in range(3):
|
||
response = self.client.post(f'/api/v1/printing-jobs/{job.id}/advance-to-next-state/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
# 验证已完成
|
||
job.refresh_from_db()
|
||
self.assertTrue(job.is_completed)
|
||
self.assertEqual(job.status, '已完成')
|
||
|
||
def test_step_back_one_state(self):
|
||
"""测试回退一个状态 API"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
# 创建 BusinessObject
|
||
business_object = stateflow_models.BusinessObject.objects.create(
|
||
name=f'PrintingJob-{job.id}',
|
||
process=self.process,
|
||
)
|
||
job.business_object = business_object
|
||
job.save()
|
||
|
||
# 先推进两步
|
||
from stateflow.services import advance_to_next_state
|
||
advance_to_next_state(business_object, self.user)
|
||
advance_to_next_state(business_object, self.user)
|
||
|
||
# 回退一步
|
||
response = self.client.post(f'/api/v1/printing-jobs/{job.id}/step-back-one-state/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertIn('detail', response.data)
|
||
self.assertIn('data', response.data)
|
||
self.assertIn('已回退状态', response.data['detail'])
|
||
|
||
# 验证状态已回退
|
||
job.refresh_from_db()
|
||
self.assertEqual(job.status, self.state2.name) # 应该回退到第二个状态（当前）
|
||
|
||
def test_step_back_without_business_object(self):
|
||
"""测试回退状态 - 没有 BusinessObject 的情况"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
response = self.client.post(f'/api/v1/printing-jobs/{job.id}/step-back-one-state/')
|
||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||
self.assertIn('没有关联的流程实例', response.data['detail'])
|
||
|
||
def test_step_back_without_records(self):
|
||
"""测试回退状态 - 没有状态流转记录的情况"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
# 创建 BusinessObject 但不推进状态
|
||
business_object = stateflow_models.BusinessObject.objects.create(
|
||
name=f'PrintingJob-{job.id}',
|
||
process=self.process,
|
||
)
|
||
job.business_object = business_object
|
||
job.save()
|
||
|
||
response = self.client.post(f'/api/v1/printing-jobs/{job.id}/step-back-one-state/')
|
||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||
self.assertIn('没有任何状态流转记录', response.data['detail'])
|
||
|
||
def test_completed_states_list(self):
|
||
"""测试获取已完成状态列表 API"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
# 创建 BusinessObject
|
||
business_object = stateflow_models.BusinessObject.objects.create(
|
||
name=f'PrintingJob-{job.id}',
|
||
process=self.process,
|
||
)
|
||
job.business_object = business_object
|
||
job.save()
|
||
|
||
# 推进两个状态
|
||
from stateflow.services import advance_to_next_state
|
||
advance_to_next_state(business_object, self.user)
|
||
advance_to_next_state(business_object, self.user)
|
||
|
||
response = self.client.get(f'/api/v1/printing-jobs/{job.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)
|
||
self.assertEqual(len(response.data['results']), 2)
|
||
|
||
# 验证返回的状态数据
|
||
first_state = response.data['results'][0]
|
||
self.assertIn('state_id', first_state)
|
||
self.assertIn('state_name', first_state)
|
||
self.assertIn('completed_at', first_state)
|
||
self.assertIn('completed_by', first_state)
|
||
self.assertIn('is_cancelled', first_state)
|
||
self.assertFalse(first_state['is_cancelled'])
|
||
|
||
def test_completed_states_include_cancelled(self):
|
||
"""测试获取已完成状态列表 - 包含已撤销的记录"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
# 创建 BusinessObject
|
||
business_object = stateflow_models.BusinessObject.objects.create(
|
||
name=f'PrintingJob-{job.id}',
|
||
process=self.process,
|
||
)
|
||
job.business_object = business_object
|
||
job.save()
|
||
|
||
# 推进两个状态
|
||
from stateflow.services import advance_to_next_state, step_back_one_state
|
||
advance_to_next_state(business_object, self.user)
|
||
advance_to_next_state(business_object, self.user)
|
||
|
||
# 回退一次（会将最后一条记录标记为 cancelled）
|
||
step_back_one_state(business_object, self.user)
|
||
|
||
# 不包含已撤销的记录（默认）
|
||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/completed-states/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['count'], 1)
|
||
|
||
# 包含已撤销的记录
|
||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/completed-states/?include_cancelled=true')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['count'], 2)
|
||
|
||
# 验证有一条记录是已撤销的
|
||
cancelled_records = [r for r in response.data['results'] if r['is_cancelled']]
|
||
self.assertEqual(len(cancelled_records), 1)
|
||
|
||
def test_completed_states_empty(self):
|
||
"""测试获取已完成状态列表 - 没有完成任何状态"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
# 创建 BusinessObject 但不推进状态
|
||
business_object = stateflow_models.BusinessObject.objects.create(
|
||
name=f'PrintingJob-{job.id}',
|
||
process=self.process,
|
||
)
|
||
job.business_object = business_object
|
||
job.save()
|
||
|
||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/completed-states/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['count'], 0)
|
||
self.assertEqual(len(response.data['results']), 0)
|
||
|
||
def test_timeline(self):
|
||
"""测试获取流程时间线 API"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
# 创建 BusinessObject
|
||
business_object = stateflow_models.BusinessObject.objects.create(
|
||
name=f'PrintingJob-{job.id}',
|
||
process=self.process,
|
||
)
|
||
job.business_object = business_object
|
||
job.save()
|
||
|
||
# 推进一个状态
|
||
from stateflow.services import advance_to_next_state
|
||
advance_to_next_state(business_object, self.user)
|
||
|
||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/timeline/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertIn('count', response.data)
|
||
self.assertIn('results', response.data)
|
||
self.assertEqual(response.data['count'], 3) # 流程有 3 个状态
|
||
self.assertEqual(len(response.data['results']), 3)
|
||
|
||
# 验证时间线数据
|
||
timeline = response.data['results']
|
||
|
||
# 第一个状态应该是已完成
|
||
self.assertEqual(timeline[0]['state_name'], self.state1.name)
|
||
self.assertEqual(timeline[0]['status'], 'completed')
|
||
self.assertIsNotNone(timeline[0]['completed_at'])
|
||
self.assertIsNotNone(timeline[0]['completed_by'])
|
||
|
||
# 第二个状态应该是进行中
|
||
self.assertEqual(timeline[1]['state_name'], self.state2.name)
|
||
self.assertEqual(timeline[1]['status'], 'in_progress')
|
||
self.assertIsNone(timeline[1]['completed_at'])
|
||
|
||
# 第三个状态应该是未开始
|
||
self.assertEqual(timeline[2]['state_name'], self.state3.name)
|
||
self.assertEqual(timeline[2]['status'], 'not_started')
|
||
self.assertIsNone(timeline[2]['completed_at'])
|
||
|
||
def test_timeline_all_completed(self):
|
||
"""测试获取流程时间线 - 所有状态已完成"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
# 创建 BusinessObject
|
||
business_object = stateflow_models.BusinessObject.objects.create(
|
||
name=f'PrintingJob-{job.id}',
|
||
process=self.process,
|
||
)
|
||
job.business_object = business_object
|
||
job.save()
|
||
|
||
# 完成所有状态
|
||
from stateflow.services import advance_to_next_state
|
||
advance_to_next_state(business_object, self.user)
|
||
advance_to_next_state(business_object, self.user)
|
||
advance_to_next_state(business_object, self.user)
|
||
|
||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/timeline/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
# 验证所有状态都是已完成
|
||
timeline = response.data['results']
|
||
for item in timeline:
|
||
self.assertEqual(item['status'], 'completed')
|
||
self.assertIsNotNone(item['completed_at'])
|
||
|
||
def test_timeline_not_started(self):
|
||
"""测试获取流程时间线 - 未开始任何状态"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
# 创建 BusinessObject 但不推进状态
|
||
business_object = stateflow_models.BusinessObject.objects.create(
|
||
name=f'PrintingJob-{job.id}',
|
||
process=self.process,
|
||
)
|
||
job.business_object = business_object
|
||
job.save()
|
||
|
||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/timeline/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
# 验证第一个状态是未开始，因为没有推进任何状态
|
||
timeline = response.data['results']
|
||
self.assertEqual(timeline[0]['status'], 'not_started')
|
||
self.assertEqual(timeline[1]['status'], 'not_started')
|
||
self.assertEqual(timeline[2]['status'], 'not_started')
|
||
|
||
def test_timeline_excludes_cancelled(self):
|
||
"""测试获取流程时间线 - 自动过滤已撤销的记录"""
|
||
job = printing_models.PrintingJob.objects.create(
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
size='50*60',
|
||
pieces=10
|
||
)
|
||
|
||
# 创建 BusinessObject
|
||
business_object = stateflow_models.BusinessObject.objects.create(
|
||
name=f'PrintingJob-{job.id}',
|
||
process=self.process,
|
||
)
|
||
job.business_object = business_object
|
||
job.save()
|
||
|
||
# 推进两个状态然后回退一次
|
||
from stateflow.services import advance_to_next_state, step_back_one_state
|
||
advance_to_next_state(business_object, self.user)
|
||
advance_to_next_state(business_object, self.user)
|
||
step_back_one_state(business_object, self.user)
|
||
|
||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/timeline/')
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
# 验证时间线中只有一个已完成的状态（撤销的不显示）
|
||
timeline = response.data['results']
|
||
completed_states = [t for t in timeline if t['status'] == 'completed']
|
||
self.assertEqual(len(completed_states), 1)
|
||
self.assertEqual(completed_states[0]['state_name'], self.state1.name)
|
||
|