forked from erp-dev/erp
feat: printing module
This commit is contained in:
@@ -2,12 +2,14 @@
|
||||
PrintingJob API 测试
|
||||
"""
|
||||
from django.test import TestCase
|
||||
from django.conf import settings
|
||||
from rest_framework.test import APIClient
|
||||
from rest_framework import status
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.contrib.auth.models import Permission
|
||||
from basic_info import models as basic_models
|
||||
from printing import models as printing_models
|
||||
from stateflow import models as stateflow_models
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
@@ -49,11 +51,23 @@ class PrintingJobAPITestCase(TestCase):
|
||||
area='测试地区'
|
||||
)
|
||||
|
||||
# 创建流程
|
||||
self.state1 = stateflow_models.State.objects.create(name='待印染')
|
||||
self.state2 = stateflow_models.State.objects.create(name='印染中')
|
||||
self.state3 = stateflow_models.State.objects.create(name='已完成')
|
||||
|
||||
self.process = stateflow_models.Process.objects.create(name='印染流程')
|
||||
self.process.replace_nodes([self.state1, self.state2, self.state3])
|
||||
|
||||
# 设置默认流程
|
||||
settings.PRINTING_DEFAULT_PROCESS_ID = self.process.id
|
||||
|
||||
# 创建印染订单
|
||||
self.printing_order = printing_models.PrintingOrder.objects.create(
|
||||
customer=self.customer,
|
||||
fabric='测试布料',
|
||||
width='150cm'
|
||||
width='150cm',
|
||||
process=self.process,
|
||||
)
|
||||
|
||||
# 创建产品类别
|
||||
@@ -418,3 +432,506 @@ class PrintingJobAPITestCase(TestCase):
|
||||
response = self.client.post('/api/v1/printing-jobs/', data, format='json')
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertIn('pieces', response.data)
|
||||
|
||||
def test_create_job_with_business_object(self):
|
||||
"""测试创建任务自动创建 BusinessObject"""
|
||||
data = {
|
||||
'printing_order': self.printing_order.id,
|
||||
'product': self.product.id,
|
||||
'quantity': 100,
|
||||
'unit': '米',
|
||||
'size': '50*60',
|
||||
'pieces': 10
|
||||
}
|
||||
|
||||
response = self.client.post('/api/v1/printing-jobs/', data, format='json')
|
||||
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
||||
|
||||
# 验证 BusinessObject 已创建
|
||||
job = printing_models.PrintingJob.objects.get(id=response.data['id'])
|
||||
self.assertIsNotNone(job.business_object)
|
||||
self.assertEqual(job.business_object.process, self.process)
|
||||
|
||||
def test_job_status_in_detail(self):
|
||||
"""测试任务详情包含状态字段"""
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=self.printing_order,
|
||||
product=self.product,
|
||||
quantity=100,
|
||||
unit='米',
|
||||
size='50*60',
|
||||
pieces=10
|
||||
)
|
||||
|
||||
# 创建 BusinessObject
|
||||
business_object = stateflow_models.BusinessObject.objects.create(
|
||||
name=f'PrintingJob-{job.id}',
|
||||
process=self.process,
|
||||
)
|
||||
job.business_object = business_object
|
||||
job.save()
|
||||
|
||||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
# 验证状态字段
|
||||
self.assertIn('status', response.data)
|
||||
self.assertIn('status_id', response.data)
|
||||
self.assertIn('is_completed', response.data)
|
||||
self.assertIn('has_started', response.data)
|
||||
self.assertIn('business_object_id', response.data)
|
||||
|
||||
# 未推进状态，但有 BusinessObject，当前是初始状态
|
||||
self.assertIn(response.data['status'], ['待印染', '未开始']) # 初始状态或未开始
|
||||
self.assertFalse(response.data['is_completed'])
|
||||
self.assertFalse(response.data['has_started'])
|
||||
|
||||
def test_job_status_in_list(self):
|
||||
"""测试任务列表包含状态字段"""
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=self.printing_order,
|
||||
product=self.product,
|
||||
quantity=100,
|
||||
unit='米',
|
||||
size='50*60',
|
||||
pieces=10
|
||||
)
|
||||
|
||||
response = self.client.get('/api/v1/printing-jobs/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
# 验证列表中的状态字段
|
||||
self.assertIn('status', response.data[0])
|
||||
self.assertIn('is_completed', response.data[0])
|
||||
|
||||
def test_job_completion_status(self):
|
||||
"""测试任务完成状态判断"""
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=self.printing_order,
|
||||
product=self.product,
|
||||
quantity=100,
|
||||
unit='米',
|
||||
size='50*60',
|
||||
pieces=10
|
||||
)
|
||||
|
||||
# 创建 BusinessObject
|
||||
business_object = stateflow_models.BusinessObject.objects.create(
|
||||
name=f'PrintingJob-{job.id}',
|
||||
process=self.process,
|
||||
)
|
||||
job.business_object = business_object
|
||||
job.save()
|
||||
|
||||
# 推进到最后一个状态（流程有3个节点，需要推进3次）
|
||||
from stateflow.services import advance_to_next_state
|
||||
advance_to_next_state(business_object, self.user) # 完成第1个状态
|
||||
advance_to_next_state(business_object, self.user) # 完成第2个状态
|
||||
advance_to_next_state(business_object, self.user) # 完成第3个状态
|
||||
|
||||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
# 验证完成状态
|
||||
self.assertEqual(response.data['status'], '已完成')
|
||||
self.assertTrue(response.data['is_completed'])
|
||||
self.assertTrue(response.data['has_started'])
|
||||
|
||||
def test_advance_to_next_state(self):
|
||||
"""测试推进到下一个状态 API"""
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=self.printing_order,
|
||||
product=self.product,
|
||||
quantity=100,
|
||||
unit='米',
|
||||
size='50*60',
|
||||
pieces=10
|
||||
)
|
||||
|
||||
# 创建 BusinessObject
|
||||
business_object = stateflow_models.BusinessObject.objects.create(
|
||||
name=f'PrintingJob-{job.id}',
|
||||
process=self.process,
|
||||
)
|
||||
job.business_object = business_object
|
||||
job.save()
|
||||
|
||||
# 推进到下一个状态
|
||||
response = self.client.post(f'/api/v1/printing-jobs/{job.id}/advance-to-next-state/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertIn('detail', response.data)
|
||||
self.assertIn('data', response.data)
|
||||
self.assertIn('已完成状态', response.data['detail'])
|
||||
|
||||
# 验证状态已更新
|
||||
job.refresh_from_db()
|
||||
self.assertTrue(job.has_started)
|
||||
self.assertEqual(job.status, self.state2.name) # 应该是第二个状态（当前进行中的）
|
||||
|
||||
def test_advance_to_next_state_without_business_object(self):
|
||||
"""测试推进到下一个状态 - 没有 BusinessObject 的情况"""
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=self.printing_order,
|
||||
product=self.product,
|
||||
quantity=100,
|
||||
unit='米',
|
||||
size='50*60',
|
||||
pieces=10
|
||||
)
|
||||
|
||||
response = self.client.post(f'/api/v1/printing-jobs/{job.id}/advance-to-next-state/')
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertIn('没有关联的流程实例', response.data['detail'])
|
||||
|
||||
def test_advance_through_all_states(self):
|
||||
"""测试推进到完成所有状态"""
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=self.printing_order,
|
||||
product=self.product,
|
||||
quantity=100,
|
||||
unit='米',
|
||||
size='50*60',
|
||||
pieces=10
|
||||
)
|
||||
|
||||
# 创建 BusinessObject
|
||||
business_object = stateflow_models.BusinessObject.objects.create(
|
||||
name=f'PrintingJob-{job.id}',
|
||||
process=self.process,
|
||||
)
|
||||
job.business_object = business_object
|
||||
job.save()
|
||||
|
||||
# 推进 3 次完成所有状态
|
||||
for i in range(3):
|
||||
response = self.client.post(f'/api/v1/printing-jobs/{job.id}/advance-to-next-state/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
# 验证已完成
|
||||
job.refresh_from_db()
|
||||
self.assertTrue(job.is_completed)
|
||||
self.assertEqual(job.status, '已完成')
|
||||
|
||||
def test_step_back_one_state(self):
|
||||
"""测试回退一个状态 API"""
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=self.printing_order,
|
||||
product=self.product,
|
||||
quantity=100,
|
||||
unit='米',
|
||||
size='50*60',
|
||||
pieces=10
|
||||
)
|
||||
|
||||
# 创建 BusinessObject
|
||||
business_object = stateflow_models.BusinessObject.objects.create(
|
||||
name=f'PrintingJob-{job.id}',
|
||||
process=self.process,
|
||||
)
|
||||
job.business_object = business_object
|
||||
job.save()
|
||||
|
||||
# 先推进两步
|
||||
from stateflow.services import advance_to_next_state
|
||||
advance_to_next_state(business_object, self.user)
|
||||
advance_to_next_state(business_object, self.user)
|
||||
|
||||
# 回退一步
|
||||
response = self.client.post(f'/api/v1/printing-jobs/{job.id}/step-back-one-state/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertIn('detail', response.data)
|
||||
self.assertIn('data', response.data)
|
||||
self.assertIn('已回退状态', response.data['detail'])
|
||||
|
||||
# 验证状态已回退
|
||||
job.refresh_from_db()
|
||||
self.assertEqual(job.status, self.state2.name) # 应该回退到第二个状态（当前）
|
||||
|
||||
def test_step_back_without_business_object(self):
|
||||
"""测试回退状态 - 没有 BusinessObject 的情况"""
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=self.printing_order,
|
||||
product=self.product,
|
||||
quantity=100,
|
||||
unit='米',
|
||||
size='50*60',
|
||||
pieces=10
|
||||
)
|
||||
|
||||
response = self.client.post(f'/api/v1/printing-jobs/{job.id}/step-back-one-state/')
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertIn('没有关联的流程实例', response.data['detail'])
|
||||
|
||||
def test_step_back_without_records(self):
|
||||
"""测试回退状态 - 没有状态流转记录的情况"""
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=self.printing_order,
|
||||
product=self.product,
|
||||
quantity=100,
|
||||
unit='米',
|
||||
size='50*60',
|
||||
pieces=10
|
||||
)
|
||||
|
||||
# 创建 BusinessObject 但不推进状态
|
||||
business_object = stateflow_models.BusinessObject.objects.create(
|
||||
name=f'PrintingJob-{job.id}',
|
||||
process=self.process,
|
||||
)
|
||||
job.business_object = business_object
|
||||
job.save()
|
||||
|
||||
response = self.client.post(f'/api/v1/printing-jobs/{job.id}/step-back-one-state/')
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
self.assertIn('没有任何状态流转记录', response.data['detail'])
|
||||
|
||||
def test_completed_states_list(self):
|
||||
"""测试获取已完成状态列表 API"""
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=self.printing_order,
|
||||
product=self.product,
|
||||
quantity=100,
|
||||
unit='米',
|
||||
size='50*60',
|
||||
pieces=10
|
||||
)
|
||||
|
||||
# 创建 BusinessObject
|
||||
business_object = stateflow_models.BusinessObject.objects.create(
|
||||
name=f'PrintingJob-{job.id}',
|
||||
process=self.process,
|
||||
)
|
||||
job.business_object = business_object
|
||||
job.save()
|
||||
|
||||
# 推进两个状态
|
||||
from stateflow.services import advance_to_next_state
|
||||
advance_to_next_state(business_object, self.user)
|
||||
advance_to_next_state(business_object, self.user)
|
||||
|
||||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/completed-states/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertIn('count', response.data)
|
||||
self.assertIn('results', response.data)
|
||||
self.assertEqual(response.data['count'], 2)
|
||||
self.assertEqual(len(response.data['results']), 2)
|
||||
|
||||
# 验证返回的状态数据
|
||||
first_state = response.data['results'][0]
|
||||
self.assertIn('state_id', first_state)
|
||||
self.assertIn('state_name', first_state)
|
||||
self.assertIn('completed_at', first_state)
|
||||
self.assertIn('completed_by', first_state)
|
||||
self.assertIn('is_cancelled', first_state)
|
||||
self.assertFalse(first_state['is_cancelled'])
|
||||
|
||||
def test_completed_states_include_cancelled(self):
|
||||
"""测试获取已完成状态列表 - 包含已撤销的记录"""
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=self.printing_order,
|
||||
product=self.product,
|
||||
quantity=100,
|
||||
unit='米',
|
||||
size='50*60',
|
||||
pieces=10
|
||||
)
|
||||
|
||||
# 创建 BusinessObject
|
||||
business_object = stateflow_models.BusinessObject.objects.create(
|
||||
name=f'PrintingJob-{job.id}',
|
||||
process=self.process,
|
||||
)
|
||||
job.business_object = business_object
|
||||
job.save()
|
||||
|
||||
# 推进两个状态
|
||||
from stateflow.services import advance_to_next_state, step_back_one_state
|
||||
advance_to_next_state(business_object, self.user)
|
||||
advance_to_next_state(business_object, self.user)
|
||||
|
||||
# 回退一次（会将最后一条记录标记为 cancelled）
|
||||
step_back_one_state(business_object, self.user)
|
||||
|
||||
# 不包含已撤销的记录（默认）
|
||||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/completed-states/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(response.data['count'], 1)
|
||||
|
||||
# 包含已撤销的记录
|
||||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/completed-states/?include_cancelled=true')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(response.data['count'], 2)
|
||||
|
||||
# 验证有一条记录是已撤销的
|
||||
cancelled_records = [r for r in response.data['results'] if r['is_cancelled']]
|
||||
self.assertEqual(len(cancelled_records), 1)
|
||||
|
||||
def test_completed_states_empty(self):
|
||||
"""测试获取已完成状态列表 - 没有完成任何状态"""
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=self.printing_order,
|
||||
product=self.product,
|
||||
quantity=100,
|
||||
unit='米',
|
||||
size='50*60',
|
||||
pieces=10
|
||||
)
|
||||
|
||||
# 创建 BusinessObject 但不推进状态
|
||||
business_object = stateflow_models.BusinessObject.objects.create(
|
||||
name=f'PrintingJob-{job.id}',
|
||||
process=self.process,
|
||||
)
|
||||
job.business_object = business_object
|
||||
job.save()
|
||||
|
||||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/completed-states/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertEqual(response.data['count'], 0)
|
||||
self.assertEqual(len(response.data['results']), 0)
|
||||
|
||||
def test_timeline(self):
|
||||
"""测试获取流程时间线 API"""
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=self.printing_order,
|
||||
product=self.product,
|
||||
quantity=100,
|
||||
unit='米',
|
||||
size='50*60',
|
||||
pieces=10
|
||||
)
|
||||
|
||||
# 创建 BusinessObject
|
||||
business_object = stateflow_models.BusinessObject.objects.create(
|
||||
name=f'PrintingJob-{job.id}',
|
||||
process=self.process,
|
||||
)
|
||||
job.business_object = business_object
|
||||
job.save()
|
||||
|
||||
# 推进一个状态
|
||||
from stateflow.services import advance_to_next_state
|
||||
advance_to_next_state(business_object, self.user)
|
||||
|
||||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/timeline/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertIn('count', response.data)
|
||||
self.assertIn('results', response.data)
|
||||
self.assertEqual(response.data['count'], 3) # 流程有 3 个状态
|
||||
self.assertEqual(len(response.data['results']), 3)
|
||||
|
||||
# 验证时间线数据
|
||||
timeline = response.data['results']
|
||||
|
||||
# 第一个状态应该是已完成
|
||||
self.assertEqual(timeline[0]['state_name'], self.state1.name)
|
||||
self.assertEqual(timeline[0]['status'], 'completed')
|
||||
self.assertIsNotNone(timeline[0]['completed_at'])
|
||||
self.assertIsNotNone(timeline[0]['completed_by'])
|
||||
|
||||
# 第二个状态应该是进行中
|
||||
self.assertEqual(timeline[1]['state_name'], self.state2.name)
|
||||
self.assertEqual(timeline[1]['status'], 'in_progress')
|
||||
self.assertIsNone(timeline[1]['completed_at'])
|
||||
|
||||
# 第三个状态应该是未开始
|
||||
self.assertEqual(timeline[2]['state_name'], self.state3.name)
|
||||
self.assertEqual(timeline[2]['status'], 'not_started')
|
||||
self.assertIsNone(timeline[2]['completed_at'])
|
||||
|
||||
def test_timeline_all_completed(self):
|
||||
"""测试获取流程时间线 - 所有状态已完成"""
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=self.printing_order,
|
||||
product=self.product,
|
||||
quantity=100,
|
||||
unit='米',
|
||||
size='50*60',
|
||||
pieces=10
|
||||
)
|
||||
|
||||
# 创建 BusinessObject
|
||||
business_object = stateflow_models.BusinessObject.objects.create(
|
||||
name=f'PrintingJob-{job.id}',
|
||||
process=self.process,
|
||||
)
|
||||
job.business_object = business_object
|
||||
job.save()
|
||||
|
||||
# 完成所有状态
|
||||
from stateflow.services import advance_to_next_state
|
||||
advance_to_next_state(business_object, self.user)
|
||||
advance_to_next_state(business_object, self.user)
|
||||
advance_to_next_state(business_object, self.user)
|
||||
|
||||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/timeline/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
# 验证所有状态都是已完成
|
||||
timeline = response.data['results']
|
||||
for item in timeline:
|
||||
self.assertEqual(item['status'], 'completed')
|
||||
self.assertIsNotNone(item['completed_at'])
|
||||
|
||||
def test_timeline_not_started(self):
|
||||
"""测试获取流程时间线 - 未开始任何状态"""
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=self.printing_order,
|
||||
product=self.product,
|
||||
quantity=100,
|
||||
unit='米',
|
||||
size='50*60',
|
||||
pieces=10
|
||||
)
|
||||
|
||||
# 创建 BusinessObject 但不推进状态
|
||||
business_object = stateflow_models.BusinessObject.objects.create(
|
||||
name=f'PrintingJob-{job.id}',
|
||||
process=self.process,
|
||||
)
|
||||
job.business_object = business_object
|
||||
job.save()
|
||||
|
||||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/timeline/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
# 验证第一个状态是未开始，因为没有推进任何状态
|
||||
timeline = response.data['results']
|
||||
self.assertEqual(timeline[0]['status'], 'not_started')
|
||||
self.assertEqual(timeline[1]['status'], 'not_started')
|
||||
self.assertEqual(timeline[2]['status'], 'not_started')
|
||||
|
||||
def test_timeline_excludes_cancelled(self):
|
||||
"""测试获取流程时间线 - 自动过滤已撤销的记录"""
|
||||
job = printing_models.PrintingJob.objects.create(
|
||||
printing_order=self.printing_order,
|
||||
product=self.product,
|
||||
quantity=100,
|
||||
unit='米',
|
||||
size='50*60',
|
||||
pieces=10
|
||||
)
|
||||
|
||||
# 创建 BusinessObject
|
||||
business_object = stateflow_models.BusinessObject.objects.create(
|
||||
name=f'PrintingJob-{job.id}',
|
||||
process=self.process,
|
||||
)
|
||||
job.business_object = business_object
|
||||
job.save()
|
||||
|
||||
# 推进两个状态然后回退一次
|
||||
from stateflow.services import advance_to_next_state, step_back_one_state
|
||||
advance_to_next_state(business_object, self.user)
|
||||
advance_to_next_state(business_object, self.user)
|
||||
step_back_one_state(business_object, self.user)
|
||||
|
||||
response = self.client.get(f'/api/v1/printing-jobs/{job.id}/timeline/')
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
# 验证时间线中只有一个已完成的状态（撤销的不显示）
|
||||
timeline = response.data['results']
|
||||
completed_states = [t for t in timeline if t['status'] == 'completed']
|
||||
self.assertEqual(len(completed_states), 1)
|
||||
self.assertEqual(completed_states[0]['state_name'], self.state1.name)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user