""" 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_positive(self): """测试件数必须大于0""" data = { 'printing_order': self.printing_order.id, 'product': self.product.id, 'quantity': 100, 'unit': '米', 'size': '50*60', 'pieces': 0 # 无效件数 } response = self.client.post('/api/v1/printing-jobs/', data, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('pieces', response.data) 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_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)