forked from erp-dev/erp
82 lines
3.4 KiB
Python
82 lines
3.4 KiB
Python
"""
|
|
测试 step_back API 端点
|
|
"""
|
|
from django.test import TestCase
|
|
from rest_framework.test import APIClient
|
|
from rest_framework import status
|
|
from django.contrib.auth import get_user_model
|
|
from stateflow import models
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class StepBackAPITestCase(TestCase):
|
|
"""测试单步回退 API"""
|
|
|
|
def setUp(self):
|
|
self.client = APIClient()
|
|
|
|
# 创建用户
|
|
self.user = User.objects.create_user(username='testuser', password='testpass')
|
|
self.client.force_authenticate(user=self.user)
|
|
|
|
# 创建状态
|
|
self.state1 = models.State.objects.create(name='状态1')
|
|
self.state2 = models.State.objects.create(name='状态2')
|
|
self.state3 = models.State.objects.create(name='状态3')
|
|
|
|
# 创建流程
|
|
self.process = models.Process.objects.create(name='测试流程')
|
|
models.ProcessNode.objects.create(process=self.process, state=self.state1, order=0)
|
|
models.ProcessNode.objects.create(process=self.process, state=self.state2, order=1)
|
|
models.ProcessNode.objects.create(process=self.process, state=self.state3, order=2)
|
|
|
|
# 创建业务对象
|
|
self.business_object = models.BusinessObject.objects.create(
|
|
name='测试对象',
|
|
process=self.process
|
|
)
|
|
|
|
def test_step_back_api_success(self):
|
|
"""测试回退 API 成功"""
|
|
# 先推进两步
|
|
self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/')
|
|
self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/')
|
|
|
|
# 回退一步
|
|
response = self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/step_back/')
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertTrue(response.data['success'])
|
|
self.assertIn('状态2', response.data['message'])
|
|
self.assertIn('business_object', response.data)
|
|
|
|
def test_step_back_api_not_started(self):
|
|
"""测试未开始状态回退失败"""
|
|
response = self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/step_back/')
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
self.assertFalse(response.data['success'])
|
|
self.assertIn('没有任何状态流转记录', response.data['message'])
|
|
|
|
def test_advance_step_back_advance_cycle(self):
|
|
"""测试前进-回退-前进循环"""
|
|
# 前进
|
|
response1 = self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/')
|
|
self.assertEqual(response1.status_code, status.HTTP_200_OK)
|
|
|
|
# 回退
|
|
response2 = self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/step_back/')
|
|
self.assertEqual(response2.status_code, status.HTTP_200_OK)
|
|
|
|
# 再前进
|
|
response3 = self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/')
|
|
self.assertEqual(response3.status_code, status.HTTP_200_OK)
|
|
|
|
# 验证数据库中有已撤销的记录
|
|
cancelled_count = models.StateFlowRecord.objects.filter(
|
|
business_object=self.business_object,
|
|
is_cancelled=True
|
|
).count()
|
|
self.assertEqual(cancelled_count, 1)
|