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

feat: new api for plate order (get_plate_order_by_state_status)

Browse Source
This commit is contained in:
colaftc
2025-12-25 17:18:44 +08:00
parent 2c5174a1ec
commit ff1ea2482d
21 changed files with 936 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
stateflow/tests/test_business_object_api.py
View File

@@ -6,6 +6,7 @@ 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 django.contrib.contenttypes.models import ContentType
from stateflow import models, services
User = get_user_model()
@@ -45,10 +46,14 @@ class BusinessObjectAPITestCase(TestCase):
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.process_content_type = ContentType.objects.get_for_model(models.Process)
# 创建业务对象
self.business_object = models.BusinessObject.objects.create(
name='测试业务对象',
process=self.process
process=self.process,
content_type=self.process_content_type,
object_id=self.process.id,
)
def test_reset_api(self):
@@ -178,7 +183,6 @@ class BusinessObjectAPITestCase(TestCase):
},
format='json'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertTrue(response.data['success'])
self.assertIn('parameter_record', response.data)
@@ -275,35 +279,28 @@ class BusinessObjectAPITestCase(TestCase):
# 验证记录列表
self.assertEqual(len(response.data['records']), 3)
def test_get_log_parameters_api_by_key(self):
"""测试获取指定参数的历史记录 API"""
# 推进并提供初始参数
def test_get_log_parameters_api_key_history(self):
"""测试获取指定参数 key 的历史记录"""
response = self.client.post(
f'/api/v1/stateflow/business-objects/{self.business_object.id}/advance/',
{'parameters': {'temperature': '25.5'}},
format='json'
)
state_log_id = response.data['state_log']['id']
# 补充参数
self.client.post(
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/{state_log_id}/add-parameters/',
{
'parameters': {'temperature': '26.0', 'humidity': '65%'},
'parameters': {'temperature': '26.0'},
'remark': '重测'
},
format='json'
)
# 获取 temperature 的历史
response = self.client.get(
f'/api/v1/stateflow/business-objects/{self.business_object.id}/state-logs/{state_log_id}/parameters/?key=temperature'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['key'], 'temperature')
self.assertEqual(len(response.data['history']), 2)
self.assertEqual(response.data['history'][0]['value'], '25.5')
self.assertEqual(response.data['history'][1]['value'], '26.0')
@@ -673,7 +670,9 @@ class BusinessObjectAPITestCase(TestCase):
empty_process = models.Process.objects.create(name='空流程')
empty_business_object = models.BusinessObject.objects.create(
name='空业务对象',
process=empty_process
process=empty_process,
content_type=self.process_content_type,
object_id=empty_process.id,
)
# 尝试推进（应该失败）

33
stateflow/tests/test_business_object_crud_and_filters_api.py
View File

@@ -35,6 +35,9 @@ class BusinessObjectCRUDAndFilterAPITestCase(TestCase):
self.process_b = models.Process.objects.create(name="流程B", description="用于测试过滤器B")
models.ProcessNode.objects.create(process=self.process_b, state=self.state1, order=0)
self.process_content_type = ContentType.objects.get_for_model(models.Process)
self.user_content_type = ContentType.objects.get_for_model(User)
def test_business_object_crud(self):
"""覆盖 create/list/retrieve/patch/delete 的基本 happy path"""
# create（注意：create serializer 不包含 id，需要从 DB 获取）
@@ -42,6 +45,8 @@ class BusinessObjectCRUDAndFilterAPITestCase(TestCase):
"name": "BO-CRUD-1",
"process": self.process_a.id,
"description": "测试 CRUD",
"content_type_str": "stateflow.process",
"object_id": self.process_a.id,
}
resp = self.client.post("/api/v1/stateflow/business-objects/", create_payload, format="json")
self.assertEqual(resp.status_code, status.HTTP_201_CREATED)
@@ -134,7 +139,7 @@ class BusinessObjectCRUDAndFilterAPITestCase(TestCase):
format="json",
)
self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("object_id", resp.data)
self.assertIn("detail", resp.data)
def test_business_object_filters(self):
"""覆盖 BusinessObjectFilterSet 的关键过滤条件"""
@@ -142,16 +147,29 @@ class BusinessObjectCRUDAndFilterAPITestCase(TestCase):
name="BO-InProgress-Alpha",
process=self.process_a,
description="alpha desc",
content_type=self.user_content_type,
object_id=self.user.id,
)
bo_completed = models.BusinessObject.objects.create(
name="BO-Completed-Beta",
process=self.process_a,
description="beta desc",
content_type=self.user_content_type,
object_id=self.user.id,
)
bo_other_process = models.BusinessObject.objects.create(
name="BO-OtherProcess-Gamma",
process=self.process_b,
description="gamma desc",
content_type=self.user_content_type,
object_id=self.user.id,
)
bo_unlinked = models.BusinessObject.objects.create(
name="BO-Unlinked-Legacy",
process=self.process_a,
description="legacy desc",
content_type=None,
object_id=None,
)
# 让 bo_completed 完成（该流程无必填参数，直接推进即可）
@@ -175,9 +193,9 @@ class BusinessObjectCRUDAndFilterAPITestCase(TestCase):
# process_name contains
resp = self.client.get("/api/v1/stateflow/business-objects/?process_name=流程A&limit=10&offset=0")
self.assertEqual(resp.status_code, status.HTTP_200_OK)
self.assertEqual(resp.data["count"], 2)
self.assertEqual(resp.data["count"], 3)
returned_ids = {x["id"] for x in resp.data["results"]}
self.assertSetEqual(returned_ids, {bo_in_progress.id, bo_completed.id})
self.assertSetEqual(returned_ids, {bo_in_progress.id, bo_completed.id, bo_unlinked.id})
# overall_status=completed
resp = self.client.get("/api/v1/stateflow/business-objects/?overall_status=completed&limit=10&offset=0")
@@ -189,7 +207,7 @@ class BusinessObjectCRUDAndFilterAPITestCase(TestCase):
resp = self.client.get("/api/v1/stateflow/business-objects/?overall_status=in_progress&limit=10&offset=0")
self.assertEqual(resp.status_code, status.HTTP_200_OK)
returned_ids = {x["id"] for x in resp.data["results"]}
self.assertSetEqual(returned_ids, {bo_in_progress.id, bo_other_process.id})
self.assertSetEqual(returned_ids, {bo_in_progress.id, bo_other_process.id, bo_unlinked.id})
# content_type_str + has_content_object
ct = ContentType.objects.get_for_model(models.Process)
@@ -203,12 +221,15 @@ class BusinessObjectCRUDAndFilterAPITestCase(TestCase):
resp = self.client.get("/api/v1/stateflow/business-objects/?has_content_object=true&limit=10&offset=0")
self.assertEqual(resp.status_code, status.HTTP_200_OK)
returned_ids = {x["id"] for x in resp.data["results"]}
self.assertIn(bo_linked.id, returned_ids)
self.assertSetEqual(
returned_ids,
{bo_in_progress.id, bo_completed.id, bo_other_process.id, bo_linked.id},
)
resp = self.client.get("/api/v1/stateflow/business-objects/?has_content_object=false&limit=10&offset=0")
self.assertEqual(resp.status_code, status.HTTP_200_OK)
returned_ids = {x["id"] for x in resp.data["results"]}
self.assertNotIn(bo_linked.id, returned_ids)
self.assertSetEqual(returned_ids, {bo_unlinked.id})
resp = self.client.get("/api/v1/stateflow/business-objects/?content_type_str=stateflow.process&limit=10&offset=0")
self.assertEqual(resp.status_code, status.HTTP_200_OK)

23
stateflow/tests/test_clone_business_object.py
View File

@@ -111,7 +111,8 @@ class CloneBusinessObjectServiceTestCase(TestCase):
self.business_object.refresh_from_db()
def test_clone_business_object_copies_all_records_and_parameters(self):
new_object_id = (self.business_object.object_id or 0) + 1000
new_process = models.Process.objects.create(name='流程B', description='克隆目标流程')
new_object_id = new_process.id
cloned = services.clone_business_object(
self.business_object,
new_object_id=new_object_id,
@@ -169,9 +170,10 @@ class CloneBusinessObjectServiceTestCase(TestCase):
)
def test_clone_is_independent_from_source(self):
target_process = models.Process.objects.create(name='流程B-独立', description='新的绑定对象')
cloned = services.clone_business_object(
self.business_object,
new_object_id=(self.business_object.object_id or 0) + 1000,
new_object_id=target_process.id,
expected_content_type_id=self.business_object.content_type_id,
)
@@ -186,21 +188,18 @@ class CloneBusinessObjectServiceTestCase(TestCase):
src_first_param = src_first_log.parameter_records.order_by('created_at', 'id').first()
self.assertEqual(src_first_param.parameters['temperature'], '25.5')
def test_clone_empty_business_object(self):
def test_clone_empty_business_object_requires_binding(self):
bo = models.BusinessObject.objects.create(
name='BO-empty',
process=self.process,
description='empty',
)
cloned = services.clone_business_object(
bo,
new_object_id=None,
expected_content_type_id=None,
)
self.assertNotEqual(cloned.id, bo.id)
self.assertIsNone(cloned.content_type_id)
self.assertIsNone(cloned.object_id)
self.assertEqual(cloned.state_logs.count(), 0)
with self.assertRaises(ValueError):
services.clone_business_object(
bo,
new_object_id=None,
expected_content_type_id=None,
)
def test_clone_rejects_mismatched_content_type(self):
"""content_type 不一致应拒绝克隆"""

7
stateflow/tests/test_parameters.py
View File

@@ -4,6 +4,7 @@
"""
from django.test import TestCase
from django.contrib.auth import get_user_model
from django.contrib.contenttypes.models import ContentType
from stateflow import models, services
User = get_user_model()
@@ -52,10 +53,14 @@ class ParameterManagementTestCase(TestCase):
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.process_content_type = ContentType.objects.get_for_model(models.Process)
# 创建业务对象
self.business_object = models.BusinessObject.objects.create(
name='测试业务对象',
process=self.process
process=self.process,
content_type=self.process_content_type,
object_id=self.process.id,
)
def test_advance_without_required_parameter_fails(self):

28
stateflow/tests/test_process_list_node_count_api.py
View File

@@ -37,3 +37,31 @@ class ProcessListNodeCountAPITestCase(TestCase):
self.assertEqual(item["node_count"], 2)
class ProcessNodesAPITestCase(TestCase):
def setUp(self):
self.client = APIClient()
self.user = User.objects.create_user(username="nodeuser", password="testpass")
self.client.force_authenticate(user=self.user)
self.state1 = models.State.objects.create(name="节点1", description="第一个节点")
self.state2 = models.State.objects.create(name="节点2", description="第二个节点")
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)
def test_process_nodes_endpoint_returns_ordered_states(self):
url = f"/api/v1/stateflow/processes/{self.process.id}/nodes/"
resp = self.client.get(url)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
self.assertEqual(resp.data["process_id"], self.process.id)
self.assertEqual(resp.data["count"], 2)
node_ids = [node["id"] for node in resp.data["nodes"]]
self.assertEqual(node_ids, [self.state1.id, self.state2.id])
self.assertEqual(resp.data["nodes"][0]["name"], self.state1.name)
self.assertEqual(resp.data["nodes"][1]["name"], self.state2.name)
def test_process_nodes_endpoint_404_when_missing(self):
resp = self.client.get("/api/v1/stateflow/processes/9999/nodes/")
self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)

66
stateflow/tests/test_services.py
View File

@@ -40,6 +40,23 @@ class StateFlowServicesTestCase(TestCase):
content_type=ct,
object_id=self.process.id
)
self.business_object_pending = models.BusinessObject.objects.create(
name='待开始订单',
process=self.process,
description='尚未推进的订单',
content_type=ct,
object_id=self.process.id
)
self.other_process = models.Process.objects.create(name='其他流程', description='另一个流程用于过滤测试')
models.ProcessNode.objects.create(process=self.other_process, state=self.state1, order=0)
self.other_business_object = models.BusinessObject.objects.create(
name='其他流程订单',
process=self.other_process,
description='不同流程的订单',
content_type=ct,
object_id=self.other_process.id
)
def test_initial_state(self):
"""测试初始状态 - current_state 是下一个待执行节点（第一个节点）"""
@@ -318,3 +335,52 @@ class StateFlowServicesTestCase(TestCase):
self.assertIn('state_name', node)
self.assertIn('order', node)
self.assertIsInstance(node['state'], models.State)
def test_query_business_objects_by_state_status_filters_by_completion(self):
"""验证按节点状态过滤业务对象的服务函数"""
# 初始状态：两个业务对象都未开始 state1
qs = services.query_business_objects_by_state_status(
state_ids=[self.state1.id],
status='not_started',
process_id=self.process.id,
)
ids = set(qs.values_list('id', flat=True))
self.assertIn(self.business_object.id, ids)
self.assertIn(self.business_object_pending.id, ids)
self.assertNotIn(self.other_business_object.id, ids)
# 推进第一个业务对象，变为已完成 state1
services.advance_to_next_state(self.business_object, self.user)
qs = services.query_business_objects_by_state_status(
state_ids=[self.state1.id],
status='not_started',
process_id=self.process.id,
)
ids = set(qs.values_list('id', flat=True))
self.assertNotIn(self.business_object.id, ids)
self.assertIn(self.business_object_pending.id, ids)
qs_completed = services.query_business_objects_by_state_status(
state_ids=[self.state1.id],
status='completed',
)
completed_ids = set(qs_completed.values_list('id', flat=True))
self.assertIn(self.business_object.id, completed_ids)
self.assertNotIn(self.business_object_pending.id, completed_ids)
def test_query_business_objects_by_state_status_cancelled(self):
"""撤销记录应该匹配 cancelled 状态"""
services.advance_to_next_state(self.business_object_pending, self.user)
services.reset_business_object_progress(self.business_object_pending)
qs = services.query_business_objects_by_state_status(
state_ids=[self.state1.id],
status='cancelled',
)
ids = set(qs.values_list('id', flat=True))
self.assertIn(self.business_object_pending.id, ids)
def test_query_business_objects_by_state_status_invalid_status(self):
"""非法状态值应抛出异常"""
with self.assertRaises(ValueError):
services.query_business_objects_by_state_status(status='unknown')

6
stateflow/tests/test_step_back.py
View File

@@ -3,6 +3,7 @@
"""
from django.test import TestCase
from django.contrib.auth import get_user_model
from django.contrib.contenttypes.models import ContentType
from stateflow import models, services
User = get_user_model()
@@ -29,10 +30,13 @@ class StepBackTestCase(TestCase):
models.ProcessNode.objects.create(process=self.process, state=self.state3, order=2)
# 创建业务对象
process_ct = ContentType.objects.get_for_model(models.Process)
self.business_object = models.BusinessObject.objects.create(
name='测试业务对象',
process=self.process,
description='测试回退功能'
description='测试回退功能',
content_type=process_ct,
object_id=self.process.id,
)
def test_step_back_from_not_started(self):

6
stateflow/tests/test_step_back_api.py
View File

@@ -5,6 +5,7 @@ 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 django.contrib.contenttypes.models import ContentType
from stateflow import models
User = get_user_model()
@@ -32,9 +33,12 @@ class StepBackAPITestCase(TestCase):
models.ProcessNode.objects.create(process=self.process, state=self.state3, order=2)
# 创建业务对象
process_ct = ContentType.objects.get_for_model(models.Process)
self.business_object = models.BusinessObject.objects.create(
name='测试对象',
process=self.process
process=self.process,
content_type=process_ct,
object_id=self.process.id,
)
def test_step_back_api_success(self):