""" 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 django.contrib.contenttypes.models import ContentType 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', 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) # 新增字段：批量推进记录（稳定输出 key，默认空数组） self.assertIn('batch_advance_records', response.data) self.assertIsInstance(response.data['batch_advance_records'], list) self.assertEqual(len(response.data['batch_advance_records']), 0) # 验证创建成功 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): """测试获取款式明细列表""" # 创建测试数据 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/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(len(response.data), 2) # 新增字段：批量推进记录（稳定输出 key） for item in response.data: self.assertIn('batch_advance_records', item) self.assertIsInstance(item['batch_advance_records'], list) self.assertEqual(len(item['batch_advance_records']), 0) 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) # 新增字段：批量推进记录（稳定输出 key） self.assertIn('batch_advance_records', response.data) self.assertIsInstance(response.data['batch_advance_records'], list) self.assertEqual(len(response.data['batch_advance_records']), 0) def test_batch_advance_records_in_list_and_detail(self): """测试 PrintingJob 序列化输出包含批量推进记录（有记录时返回明细，无记录返回空）""" job = printing_models.PrintingJob.objects.create( printing_order=self.printing_order, product=self.product, quantity=100, unit='米', size='50*60', pieces=10, description='批量记录测试' ) # 创建一条批量推进记录并关联该 job record = printing_models.PrintingJobBatchAdvanceRecord.objects.create( printing_order=self.printing_order, state=self.state1, created_by=self.user, parameters={'temperature': '25.5', 'operator': '张三'}, ) record.printing_jobs.add(job) # list: 应包含 batch_advance_records response = self.client.get('/api/v1/printing-jobs/') self.assertEqual(response.status_code, status.HTTP_200_OK) target = next(item for item in response.data if item['id'] == job.id) self.assertIn('batch_advance_records', target) self.assertEqual(len(target['batch_advance_records']), 1) rec = target['batch_advance_records'][0] self.assertEqual(rec['id'], record.id) self.assertEqual(rec['printing_order'], self.printing_order.id) self.assertEqual(rec['state'], self.state1.id) self.assertEqual(rec['state_id'], self.state1.id) self.assertEqual(rec['state_name'], self.state1.name) self.assertEqual(rec['created_by'], self.user.id) self.assertEqual(rec['created_by_username'], self.user.username) self.assertEqual(rec['created_by_name'], self.employee.name) self.assertEqual(rec['parameters']['temperature'], '25.5') self.assertEqual(rec['parameters']['operator'], '张三') # detail: 同样应包含 batch_advance_records response = self.client.get(f'/api/v1/printing-jobs/{job.id}/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn('batch_advance_records', response.data) self.assertEqual(len(response.data['batch_advance_records']), 1) 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) # 新逻辑：BusinessObject 绑定回 PrintingJob self.assertEqual(job.business_object.content_type_id, ContentType.objects.get_for_model(printing_models.PrintingJob).id) self.assertEqual(job.business_object.object_id, job.id) 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']) # 第二个状态应该是未开始（因为我们废除了 in_progress 的概念） self.assertEqual(timeline[1]['state_name'], self.state2.name) self.assertEqual(timeline[1]['status'], 'not_started') 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)