forked from erp-dev/erp
fix: added some fields to business.statements api
This commit is contained in:
231
stateflow/tests/test_business_object_crud_and_filters_api.py
Normal file
231
stateflow/tests/test_business_object_crud_and_filters_api.py
Normal file
@@ -0,0 +1,231 @@
|
||||
"""
|
||||
BusinessObject CRUD + 过滤器 API 测试
|
||||
|
||||
现有测试主要覆盖 BusinessObject 的 action（advance/reset/timeline/state-logs/...），
|
||||
这里补齐 list/retrieve/create/update/delete 以及 FilterSet 的关键分支。
|
||||
"""
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.test import TestCase
|
||||
from rest_framework import status
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from stateflow import models, services
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
|
||||
class BusinessObjectCRUDAndFilterAPITestCase(TestCase):
|
||||
def setUp(self):
|
||||
self.client = APIClient()
|
||||
self.user = User.objects.create_user(username="testuser", password="testpass")
|
||||
self.client.force_authenticate(user=self.user)
|
||||
|
||||
# 通用状态/流程（3个节点）
|
||||
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_a = models.Process.objects.create(name="流程A", description="用于测试过滤器A")
|
||||
models.ProcessNode.objects.create(process=self.process_a, state=self.state1, order=0)
|
||||
models.ProcessNode.objects.create(process=self.process_a, state=self.state2, order=1)
|
||||
models.ProcessNode.objects.create(process=self.process_a, state=self.state3, order=2)
|
||||
|
||||
self.process_b = models.Process.objects.create(name="流程B", description="用于测试过滤器B")
|
||||
models.ProcessNode.objects.create(process=self.process_b, state=self.state1, order=0)
|
||||
|
||||
def test_business_object_crud(self):
|
||||
"""覆盖 create/list/retrieve/patch/delete 的基本 happy path"""
|
||||
# create（注意：create serializer 不包含 id，需要从 DB 获取）
|
||||
create_payload = {
|
||||
"name": "BO-CRUD-1",
|
||||
"process": self.process_a.id,
|
||||
"description": "测试 CRUD",
|
||||
}
|
||||
resp = self.client.post("/api/v1/stateflow/business-objects/", create_payload, format="json")
|
||||
self.assertEqual(resp.status_code, status.HTTP_201_CREATED)
|
||||
|
||||
bo = models.BusinessObject.objects.get(name="BO-CRUD-1")
|
||||
self.assertEqual(bo.process_id, self.process_a.id)
|
||||
|
||||
# list
|
||||
resp = self.client.get("/api/v1/stateflow/business-objects/?limit=10&offset=0")
|
||||
self.assertEqual(resp.status_code, status.HTTP_200_OK)
|
||||
self.assertGreaterEqual(resp.data["count"], 1)
|
||||
|
||||
# list 返回字段校验（挑关键字段）
|
||||
results = resp.data["results"]
|
||||
item = next(x for x in results if x["id"] == bo.id)
|
||||
self.assertEqual(item["name"], "BO-CRUD-1")
|
||||
self.assertEqual(item["process"], self.process_a.id)
|
||||
self.assertEqual(item["process_name"], "流程A")
|
||||
self.assertIn("current_state_name", item)
|
||||
self.assertIn("overall_status", item)
|
||||
self.assertIn("progress_percentage", item)
|
||||
|
||||
# retrieve（详情含 timeline/state_logs 等）
|
||||
resp = self.client.get(f"/api/v1/stateflow/business-objects/{bo.id}/")
|
||||
self.assertEqual(resp.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(resp.data["id"], bo.id)
|
||||
self.assertEqual(resp.data["name"], "BO-CRUD-1")
|
||||
self.assertIn("process_detail", resp.data)
|
||||
self.assertIn("timeline", resp.data)
|
||||
self.assertIn("state_logs", resp.data)
|
||||
self.assertEqual(len(resp.data["timeline"]), 3)
|
||||
|
||||
# patch 更新 name
|
||||
resp = self.client.patch(
|
||||
f"/api/v1/stateflow/business-objects/{bo.id}/",
|
||||
{"name": "BO-CRUD-1-RENAMED"},
|
||||
format="json",
|
||||
)
|
||||
self.assertEqual(resp.status_code, status.HTTP_200_OK)
|
||||
bo.refresh_from_db()
|
||||
self.assertEqual(bo.name, "BO-CRUD-1-RENAMED")
|
||||
|
||||
# delete
|
||||
resp = self.client.delete(f"/api/v1/stateflow/business-objects/{bo.id}/")
|
||||
self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
|
||||
self.assertFalse(models.BusinessObject.objects.filter(id=bo.id).exists())
|
||||
|
||||
def test_create_business_object_with_content_type_str(self):
|
||||
"""覆盖 create serializer 的 content_type_str 解析与一致性校验"""
|
||||
# 正确：提供 content_type_str + object_id
|
||||
resp = self.client.post(
|
||||
"/api/v1/stateflow/business-objects/",
|
||||
{
|
||||
"name": "BO-Linked",
|
||||
"process": self.process_a.id,
|
||||
"content_type_str": "stateflow.process",
|
||||
"object_id": self.process_a.id,
|
||||
},
|
||||
format="json",
|
||||
)
|
||||
self.assertEqual(resp.status_code, status.HTTP_201_CREATED)
|
||||
bo = models.BusinessObject.objects.get(name="BO-Linked")
|
||||
self.assertIsNotNone(bo.content_type)
|
||||
self.assertEqual(bo.content_type.app_label, "stateflow")
|
||||
self.assertEqual(bo.content_type.model, "process")
|
||||
self.assertEqual(bo.object_id, self.process_a.id)
|
||||
|
||||
# 错误：content_type_str 格式不合法
|
||||
resp = self.client.post(
|
||||
"/api/v1/stateflow/business-objects/",
|
||||
{
|
||||
"name": "BO-Bad-CT",
|
||||
"process": self.process_a.id,
|
||||
"content_type_str": "badformat",
|
||||
"object_id": self.process_a.id,
|
||||
},
|
||||
format="json",
|
||||
)
|
||||
self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertIn("content_type_str", resp.data)
|
||||
|
||||
# 错误：提供 content_type_str 但缺 object_id
|
||||
resp = self.client.post(
|
||||
"/api/v1/stateflow/business-objects/",
|
||||
{
|
||||
"name": "BO-Missing-ObjectId",
|
||||
"process": self.process_a.id,
|
||||
"content_type_str": "stateflow.process",
|
||||
},
|
||||
format="json",
|
||||
)
|
||||
self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertIn("object_id", resp.data)
|
||||
|
||||
def test_business_object_filters(self):
|
||||
"""覆盖 BusinessObjectFilterSet 的关键过滤条件"""
|
||||
bo_in_progress = models.BusinessObject.objects.create(
|
||||
name="BO-InProgress-Alpha",
|
||||
process=self.process_a,
|
||||
description="alpha desc",
|
||||
)
|
||||
bo_completed = models.BusinessObject.objects.create(
|
||||
name="BO-Completed-Beta",
|
||||
process=self.process_a,
|
||||
description="beta desc",
|
||||
)
|
||||
bo_other_process = models.BusinessObject.objects.create(
|
||||
name="BO-OtherProcess-Gamma",
|
||||
process=self.process_b,
|
||||
description="gamma desc",
|
||||
)
|
||||
|
||||
# 让 bo_completed 完成（该流程无必填参数，直接推进即可）
|
||||
services.advance_to_next_state(bo_completed, self.user)
|
||||
services.advance_to_next_state(bo_completed, self.user)
|
||||
services.advance_to_next_state(bo_completed, self.user)
|
||||
self.assertEqual(services.get_overall_status(bo_completed), "completed")
|
||||
|
||||
# name contains
|
||||
resp = self.client.get("/api/v1/stateflow/business-objects/?name=Alpha&limit=10&offset=0")
|
||||
self.assertEqual(resp.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(resp.data["count"], 1)
|
||||
self.assertEqual(resp.data["results"][0]["id"], bo_in_progress.id)
|
||||
|
||||
# process id
|
||||
resp = self.client.get(f"/api/v1/stateflow/business-objects/?process={self.process_b.id}&limit=10&offset=0")
|
||||
self.assertEqual(resp.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(resp.data["count"], 1)
|
||||
self.assertEqual(resp.data["results"][0]["id"], bo_other_process.id)
|
||||
|
||||
# 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)
|
||||
returned_ids = {x["id"] for x in resp.data["results"]}
|
||||
self.assertSetEqual(returned_ids, {bo_in_progress.id, bo_completed.id})
|
||||
|
||||
# overall_status=completed
|
||||
resp = self.client.get("/api/v1/stateflow/business-objects/?overall_status=completed&limit=10&offset=0")
|
||||
self.assertEqual(resp.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(resp.data["count"], 1)
|
||||
self.assertEqual(resp.data["results"][0]["id"], bo_completed.id)
|
||||
|
||||
# overall_status=in_progress
|
||||
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})
|
||||
|
||||
# content_type_str + has_content_object
|
||||
ct = ContentType.objects.get_for_model(models.Process)
|
||||
bo_linked = models.BusinessObject.objects.create(
|
||||
name="BO-Linked-Process",
|
||||
process=self.process_a,
|
||||
content_type=ct,
|
||||
object_id=self.process_a.id,
|
||||
)
|
||||
|
||||
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)
|
||||
|
||||
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)
|
||||
|
||||
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)
|
||||
returned_ids = {x["id"] for x in resp.data["results"]}
|
||||
self.assertSetEqual(returned_ids, {bo_linked.id})
|
||||
|
||||
# 无效 content_type_str：返回空
|
||||
resp = self.client.get("/api/v1/stateflow/business-objects/?content_type_str=badformat&limit=10&offset=0")
|
||||
self.assertEqual(resp.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(resp.data["count"], 0)
|
||||
|
||||
|
||||
class AuthenticationGuardAPITestCase(TestCase):
|
||||
def test_stateflow_endpoints_require_authentication(self):
|
||||
"""默认权限 IsAuthenticated：未登录应 401"""
|
||||
client = APIClient()
|
||||
resp = client.get("/api/v1/stateflow/states/?limit=1&offset=0")
|
||||
self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED)
|
||||
|
||||
|
||||
39
stateflow/tests/test_process_list_node_count_api.py
Normal file
39
stateflow/tests/test_process_list_node_count_api.py
Normal file
@@ -0,0 +1,39 @@
|
||||
"""
|
||||
补充 Process 列表接口的 node_count 字段覆盖
|
||||
|
||||
ProcessViewSet.list() 会对 queryset annotate(node_count)，对应 ProcessListSerializer 的只读字段。
|
||||
"""
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.test import TestCase
|
||||
from rest_framework import status
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
from stateflow import models
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
|
||||
class ProcessListNodeCountAPITestCase(TestCase):
|
||||
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")
|
||||
|
||||
def test_process_list_contains_node_count(self):
|
||||
process = models.Process.objects.create(name="带节点计数的流程")
|
||||
models.ProcessNode.objects.create(process=process, state=self.state1, order=0)
|
||||
models.ProcessNode.objects.create(process=process, state=self.state2, order=1)
|
||||
|
||||
resp = self.client.get("/api/v1/stateflow/processes/?limit=10&offset=0")
|
||||
self.assertEqual(resp.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(resp.data["count"], 1)
|
||||
|
||||
item = resp.data["results"][0]
|
||||
self.assertIn("node_count", item)
|
||||
self.assertEqual(item["node_count"], 2)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user