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

feat: upload api

Browse Source
This commit is contained in:
colaftc
2025-11-17 14:34:09 +08:00
parent dff9151651
commit d23db848ec
16 changed files with 1183 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

126
stateflow/tests/test_business_object_api.py
View File

@@ -471,6 +471,132 @@ class BusinessObjectAPITestCase(TestCase):
self.assertIsNotNone(item['completed_by'])
self.assertFalse(item['is_cancelled'])
def test_get_state_logs_api_with_parameters(self):
"""测试获取状态流转记录列表（包含参数）"""
# 未开始时应返回空列表
response = self.client.get(f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['count'], 0)
self.assertEqual(len(response.data['state_logs']), 0)
# 推进两次
self.client.post(
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
{'parameters': {'temperature': '25.5', 'humidity': '60%'}},
format='json'
)
self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/')
# 获取记录列表（默认包含参数）
response = self.client.get(f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['count'], 2)
self.assertEqual(len(response.data['state_logs']), 2)
# 验证第一条记录包含参数摘要
log1 = response.data['state_logs'][0]
self.assertIn('parameters_summary', log1)
self.assertEqual(log1['parameters_summary']['temperature'], '25.5')
self.assertEqual(log1['parameters_summary']['humidity'], '60%')
self.assertEqual(log1['state_name'], '状态1')
# 验证第二条记录（无参数）
log2 = response.data['state_logs'][1]
self.assertIn('parameters_summary', log2)
self.assertEqual(len(log2['parameters_summary']), 0)
self.assertEqual(log2['state_name'], '状态2')
def test_get_state_logs_api_without_parameters(self):
"""测试获取状态流转记录列表（不包含参数）"""
# 推进两次
self.client.post(
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
{'parameters': {'temperature': '25.5'}},
format='json'
)
self.client.post(f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/')
# 获取记录列表（不包含参数）
response = self.client.get(
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/?include_parameters=false'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['count'], 2)
# 验证不包含 parameters_summary 字段
log1 = response.data['state_logs'][0]
self.assertNotIn('parameters_summary', log1)
self.assertIn('state_name', log1)
self.assertIn('completed_at', log1)
self.assertIn('completed_by_username', log1)
def test_get_state_logs_api_exclude_cancelled(self):
"""测试获取状态流转记录列表（排除已撤销记录）"""
# 推进两次
self.client.post(
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
{'parameters': {'temperature': '25.5'}},
format='json'
)
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}/step_back/')
# 默认包含已撤销记录
response = self.client.get(f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/')
self.assertEqual(response.data['count'], 2)
# 验证第二条是已撤销的
self.assertTrue(response.data['state_logs'][1]['is_cancelled'])
# 排除已撤销记录
response = self.client.get(
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/?include_cancelled=false'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['count'], 1)
self.assertEqual(len(response.data['state_logs']), 1)
# 验证只有第一条记录（未撤销）
self.assertFalse(response.data['state_logs'][0]['is_cancelled'])
self.assertEqual(response.data['state_logs'][0]['state_name'], '状态1')
def test_get_state_logs_api_order(self):
"""测试状态流转记录按时间排序"""
# 推进三次
self.client.post(
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
{'parameters': {'temperature': '25.5'}},
format='json'
)
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.get(f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['count'], 3)
# 验证按时间顺序排列
logs = response.data['state_logs']
self.assertEqual(logs[0]['state_name'], '状态1')
self.assertEqual(logs[1]['state_name'], '状态2')
self.assertEqual(logs[2]['state_name'], '状态3')
# 验证时间递增
from datetime import datetime
time1 = datetime.fromisoformat(logs[0]['completed_at'])
time2 = datetime.fromisoformat(logs[1]['completed_at'])
time3 = datetime.fromisoformat(logs[2]['completed_at'])
self.assertLessEqual(time1, time2)
self.assertLessEqual(time2, time3)
def test_advance_when_already_completed(self):
"""测试流程已完成时尝试推进（预期错误）"""
# 完成所有状态

24
stateflow/tests/test_services.py
View File

@@ -42,9 +42,10 @@ class StateFlowServicesTestCase(TestCase):
)
def test_initial_state(self):
"""测试初始状态 - 应该是 None（未开始）"""
"""测试初始状态 - current_state 是下一个待执行节点（第一个节点）"""
current_state = self.business_object.get_current_state()
self.assertIsNone(current_state, '初始状态应该是 None（未开始）')
# current_state 返回"下一个待执行的节点"，未开始时是第一个节点
self.assertEqual(current_state.name, self.state1.name)
self.assertEqual(self.business_object.get_progress_percentage(), 0.0)
# 整体状态应该是 not_started
@@ -57,18 +58,18 @@ class StateFlowServicesTestCase(TestCase):
success, message, _ = services.advance_to_next_state(self.business_object, self.user)
self.assertTrue(success)
# current_state 应该是最后完成的状态（state1）
# current_state 返回"下一个待执行的节点"，完成state1后是state2
current_state = self.business_object.get_current_state()
self.assertEqual(current_state.name, self.state1.name)
self.assertEqual(current_state.name, self.state2.name)
self.assertAlmostEqual(self.business_object.get_progress_percentage(), 33.33, places=1)
# 推进到状态2（完成state2）
success, message, _ = services.advance_to_next_state(self.business_object, self.user)
self.assertTrue(success)
# current_state 应该是 state2
# current_state 应该是 state3（下一个待执行的）
current_state = self.business_object.get_current_state()
self.assertEqual(current_state.name, self.state2.name)
self.assertEqual(current_state.name, self.state3.name)
self.assertAlmostEqual(self.business_object.get_progress_percentage(), 66.67, places=1)
# 完成最后一个状态（state3）
@@ -76,9 +77,9 @@ class StateFlowServicesTestCase(TestCase):
self.assertTrue(success)
self.assertEqual(self.business_object.get_progress_percentage(), 100.0)
# current_state 应该是 state3（最后完成的状态）
# current_state 应该是 None（没有待执行的节点了）
current_state = self.business_object.get_current_state()
self.assertEqual(current_state.name, self.state3.name)
self.assertIsNone(current_state)
# 整体状态应该是 completed
status = services.get_overall_status(self.business_object)
@@ -116,7 +117,8 @@ class StateFlowServicesTestCase(TestCase):
timeline = self.business_object.get_timeline()
self.assertEqual(timeline[0]['status'], 'completed')
self.assertEqual(timeline[1]['status'], 'not_started')
# 第二个状态是 current_state（下一个待执行的节点）且已有完成记录，所以是 in_progress
self.assertEqual(timeline[1]['status'], 'in_progress')
self.assertEqual(timeline[2]['status'], 'not_started')
self.assertIsNotNone(timeline[0]['completed_by'])
self.assertEqual(timeline[0]['completed_by'].id, self.user.id)
@@ -141,9 +143,9 @@ class StateFlowServicesTestCase(TestCase):
cancelled_count = self.business_object.state_logs.filter(is_cancelled=True).count()
self.assertEqual(cancelled_count, 2)
# 应该回到初始状态（未开始）
# current_state 返回下一个待执行的节点（第一个节点）
current_state = self.business_object.get_current_state()
self.assertIsNone(current_state)
self.assertEqual(current_state.name, self.state1.name)
# 所有撤销的记录都应该有撤销时间
for log in self.business_object.state_logs.all():