forked from erp-dev/erp
feat: added jobs_status_summary to printing orders api
This commit is contained in:
@@ -18,6 +18,10 @@ class PrintingOrderAPITestCase(TestCase):
|
||||
"""测试 PrintingOrder API"""
|
||||
|
||||
def setUp(self):
|
||||
# 清除缓存,避免 cache_page 装饰器导致的测试干扰
|
||||
from django.core.cache import cache
|
||||
cache.clear()
|
||||
|
||||
self.client = APIClient()
|
||||
|
||||
# 创建商户
|
||||
@@ -539,3 +543,98 @@ class PrintingOrderAPITestCase(TestCase):
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertIn('已开始', str(response.data))
|
||||
|
||||
def test_jobs_status_summary_in_list(self):
|
||||
"""测试订单列表包含 jobs_status_summary 字段(PrintingJob 状态汇总)"""
|
||||
# 创建订单
|
||||
order = printing_models.PrintingOrder.objects.create(
|
||||
customer=self.customer,
|
||||
fabric='测试布料',
|
||||
width='150cm',
|
||||
process=self.process,
|
||||
)
|
||||
|
||||
# 创建产品类别和产品
|
||||
category = basic_models.ProductCategory.objects.create(
|
||||
name='测试类别',
|
||||
merchant=self.merchant,
|
||||
)
|
||||
product = basic_models.Product.objects.create(
|
||||
name='测试产品',
|
||||
category=category,
|
||||
merchant=self.merchant,
|
||||
)
|
||||
|
||||
# 创建 3 个任务
|
||||
jobs = []
|
||||
for i in range(3):
|
||||
business_object = stateflow_models.BusinessObject.objects.create(
|
||||
name=f'PrintingJob-test-{i}',
|
||||
process=self.process,
|
||||
)
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=order,
|
||||
product=product,
|
||||
quantity=10,
|
||||
unit='件',
|
||||
business_object=business_object,
|
||||
)
|
||||
jobs.append(job)
|
||||
|
||||
# 推进第一个任务的状态(完成第一个节点)
|
||||
from stateflow.services import advance_to_next_state
|
||||
advance_to_next_state(jobs[0].business_object, self.user)
|
||||
|
||||
# 推进第二个任务的状态两次(完成前两个节点)
|
||||
advance_to_next_state(jobs[1].business_object, self.user)
|
||||
advance_to_next_state(jobs[1].business_object, self.user)
|
||||
|
||||
# 第三个任务保持初始状态(待印染)
|
||||
|
||||
# 获取订单列表
|
||||
response = self.client.get('/api/v1/printing-orders/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
# 检查 jobs_status_summary 字段存在
|
||||
self.assertIn('jobs_status_summary', response.data['results'][0])
|
||||
summary = response.data['results'][0]['jobs_status_summary']
|
||||
|
||||
# 检查返回格式(列表)
|
||||
self.assertIsInstance(summary, list)
|
||||
|
||||
# 汇总结果应该包含不同状态的计数
|
||||
# - job0: 完成了第一个节点,下一个待执行是 "印染中"
|
||||
# - job1: 完成了前两个节点,下一个待执行是 "已完成"
|
||||
# - job2: 未开始,下一个待执行是 "待印染"
|
||||
state_names = {item['state_name']: item['count'] for item in summary}
|
||||
|
||||
# 验证有3个任务被正确统计
|
||||
total_count = sum(item['count'] for item in summary)
|
||||
self.assertEqual(total_count, 3)
|
||||
|
||||
# 验证返回的每个项都包含必要字段
|
||||
for item in summary:
|
||||
self.assertIn('state_name', item)
|
||||
self.assertIn('state_id', item)
|
||||
self.assertIn('count', item)
|
||||
|
||||
def test_jobs_status_summary_empty_when_no_jobs(self):
|
||||
"""测试没有任务时 jobs_status_summary 为空列表"""
|
||||
order = printing_models.PrintingOrder.objects.create(
|
||||
customer=self.customer,
|
||||
fabric='无任务订单',
|
||||
width='150cm',
|
||||
process=self.process,
|
||||
)
|
||||
|
||||
# 获取特定订单详情(避免被其他测试数据影响)
|
||||
response = self.client.get(f'/api/v1/printing-orders/{order.id}/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
# 详情接口不包含 jobs_status_summary(这是列表专用字段)
|
||||
# 改用列表接口并按特定条件过滤
|
||||
response = self.client.get('/api/v1/printing-orders/', {'fabric': '无任务订单'})
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(response.data['count'], 1)
|
||||
|
||||
summary = response.data['results'][0]['jobs_status_summary']
|
||||
self.assertEqual(summary, [])
|
||||
|
||||
Reference in New Issue
Block a user