1
0
forked from erp-dev/erp
Files
erpnew/api_v1/views/printing/test_api.py

641 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
PrintingOrder 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()
class PrintingOrderAPITestCase(TestCase):
"""测试 PrintingOrder API"""
def setUp(self):
# 清除缓存,避免 cache_page 装饰器导致的测试干扰
from django.core.cache import cache
cache.clear()
self.client = APIClient()
# 创建商户
self.merchant = basic_models.Merchant.objects.create(
name='测试印花厂',
type=basic_models.MerchantTypeEnum.FACTORY
)
# 创建用户
self.user = User.objects.create_user(
username='testuser',
password='testpass123',
email='test@example.com'
)
# 创建员工并关联商户
self.employee = basic_models.Employee.objects.create(
sys_user=self.user,
merchant=self.merchant,
name='测试员工',
mobile='13800138000',
status=basic_models.EmployeeStatusEnum.ACTIVE
)
# 创建客户
self.customer = basic_models.Customer.objects.create(
merchant=self.merchant,
name='测试客户',
mobile='13900139000',
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.client.force_authenticate(user=self.user)
# 给用户添加基础权限
view_perm = Permission.objects.get(codename='view_printingorder')
add_perm = Permission.objects.get(codename='add_printingorder')
change_perm = Permission.objects.get(codename='change_printingorder')
self.user.user_permissions.add(view_perm, add_perm, change_perm)
def test_create_printing_order(self):
"""测试创建印染订单"""
data = {
'customer': self.customer.id,
'fabric': '纯棉布料',
'width': '150cm',
'is_urgent': True,
'area': '广州',
'address': '白云区xxx',
'fabric_source': '客户提供',
'is_fabric_received': False,
'craft': '活性印花',
'description': '测试订单描述',
'outgoing_date': '2025-11-20',
'printing_warn': '注意颜色',
'rolling_warn': '注意温度',
'production_warn': '质量检查'
}
response = self.client.post('/api/v1/printing-orders/', data, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
# 检查响应数据
self.assertIn('fabric', response.data)
self.assertEqual(response.data['fabric'], '纯棉布料')
self.assertEqual(response.data['is_urgent'], True)
# 验证订单已创建 - 通过fabric查找因为响应可能不包含id
order = printing_models.PrintingOrder.objects.filter(fabric='纯棉布料').first()
self.assertIsNotNone(order)
self.assertEqual(order.customer.id, self.customer.id)
self.assertEqual(order.fabric, '纯棉布料')
def test_list_printing_orders(self):
"""测试获取订单列表"""
# 创建测试订单
printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='布料1',
width='150cm'
)
printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='布料2',
width='160cm'
)
response = self.client.get('/api/v1/printing-orders/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
# 分页响应格式
self.assertEqual(response.data['count'], 2)
def test_retrieve_printing_order(self):
"""测试获取订单详情"""
order = printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='测试布料',
width='150cm',
craft='活性印花',
description='详情测试'
)
response = self.client.get(f'/api/v1/printing-orders/{order.id}/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['fabric'], '测试布料')
self.assertEqual(response.data['craft'], '活性印花')
self.assertIn('customer_name', response.data)
self.assertEqual(response.data['customer_name'], self.customer.name)
def test_update_printing_order(self):
"""测试更新订单"""
order = printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='旧布料',
width='150cm'
)
update_data = {
'customer': self.customer.id,
'fabric': '新布料',
'width': '160cm',
'is_urgent': True,
'craft': '更新的工艺'
}
response = self.client.put(
f'/api/v1/printing-orders/{order.id}/',
update_data,
format='json'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
order.refresh_from_db()
self.assertEqual(order.fabric, '新布料')
self.assertEqual(order.width, '160cm')
self.assertEqual(order.is_urgent, True)
def test_partial_update_printing_order(self):
"""测试部分更新订单"""
order = printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='原布料',
width='150cm',
is_urgent=False
)
patch_data = {
'is_urgent': True,
'craft': '新工艺'
}
response = self.client.patch(
f'/api/v1/printing-orders/{order.id}/',
patch_data,
format='json'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
order.refresh_from_db()
self.assertEqual(order.is_urgent, True)
self.assertEqual(order.craft, '新工艺')
self.assertEqual(order.fabric, '原布料') # 未修改字段保持不变
def test_delete_printing_order_forbidden(self):
"""测试删除订单被禁用"""
# 添加删除权限以便测试destroy方法的自定义逻辑
delete_perm = Permission.objects.get(codename='delete_printingorder')
self.user.user_permissions.add(delete_perm)
order = printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='测试布料',
width='150cm'
)
response = self.client.delete(f'/api/v1/printing-orders/{order.id}/')
self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED)
self.assertIn('不支持删除', response.data['detail'])
# 验证订单仍然存在
self.assertTrue(
printing_models.PrintingOrder.objects.filter(id=order.id).exists()
)
def test_invalidate_printing_order(self):
"""测试作废订单"""
# 添加作废权限
invalidate_perm = Permission.objects.get(codename='can_invalidate_printingorder')
self.user.user_permissions.add(invalidate_perm)
order = printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='测试布料',
width='150cm',
is_invalid=False
)
response = self.client.post(f'/api/v1/printing-orders/{order.id}/invalidate/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('已作废', response.data['detail'])
order.refresh_from_db()
self.assertTrue(order.is_invalid)
def test_invalidate_without_permission(self):
"""测试没有权限时作废订单失败"""
order = printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='测试布料',
width='150cm'
)
response = self.client.post(f'/api/v1/printing-orders/{order.id}/invalidate/')
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.assertIn('没有权限', response.data['detail'])
def test_activate_printing_order(self):
"""测试恢复订单"""
# 添加恢复权限
activate_perm = Permission.objects.get(codename='can_activate_printingorder')
self.user.user_permissions.add(activate_perm)
order = printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='测试布料',
width='150cm',
is_invalid=True
)
response = self.client.post(f'/api/v1/printing-orders/{order.id}/activate/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('已恢复', response.data['detail'])
order.refresh_from_db()
self.assertFalse(order.is_invalid)
def test_activate_without_permission(self):
"""测试没有权限时恢复订单失败"""
order = printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='测试布料',
width='150cm',
is_invalid=True
)
response = self.client.post(f'/api/v1/printing-orders/{order.id}/activate/')
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.assertIn('没有权限', response.data['detail'])
def test_mark_fabric_received(self):
"""测试标记布料已收"""
order = printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='测试布料',
width='150cm',
is_fabric_received=False
)
response = self.client.post(f'/api/v1/printing-orders/{order.id}/mark_fabric_received/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('已标记布料已收', response.data['detail'])
order.refresh_from_db()
self.assertTrue(order.is_fabric_received)
def test_mark_fabric_received_already_received(self):
"""测试重复标记布料已收"""
order = printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='测试布料',
width='150cm',
is_fabric_received=True
)
response = self.client.post(f'/api/v1/printing-orders/{order.id}/mark_fabric_received/')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn('已经标记', response.data['detail'])
def test_filter_by_customer(self):
"""测试按客户过滤"""
customer2 = basic_models.Customer.objects.create(
merchant=self.merchant,
name='客户2',
mobile='13900139001'
)
printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='布料1',
width='150cm'
)
printing_models.PrintingOrder.objects.create(
customer=customer2,
fabric='布料2',
width='160cm'
)
response = self.client.get(f'/api/v1/printing-orders/?customer={self.customer.id}')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['count'], 1)
def test_filter_by_urgent(self):
"""测试按紧急状态过滤"""
printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='布料1',
width='150cm',
is_urgent=True
)
printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='布料2',
width='160cm',
is_urgent=False
)
response = self.client.get('/api/v1/printing-orders/?is_urgent=true')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['count'], 1)
self.assertTrue(response.data['results'][0]['is_urgent'])
def test_search_by_fabric(self):
"""测试按布料搜索"""
printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='纯棉布料',
width='150cm'
)
printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='涤纶布料',
width='160cm'
)
response = self.client.get('/api/v1/printing-orders/?search=纯棉')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['count'], 1)
self.assertIn('纯棉', response.data['results'][0]['fabric'])
def test_ordering(self):
"""测试排序"""
order1 = printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='布料1',
width='150cm'
)
order2 = printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='布料2',
width='160cm'
)
# 按 ID 升序
response = self.client.get('/api/v1/printing-orders/?ordering=id')
self.assertEqual(response.status_code, status.HTTP_200_OK)
results = response.data['results']
self.assertEqual(results[0]['id'], order1.id)
self.assertEqual(results[1]['id'], order2.id)
# 按 ID 降序
response = self.client.get('/api/v1/printing-orders/?ordering=-id')
results = response.data['results']
self.assertEqual(results[0]['id'], order2.id)
self.assertEqual(results[1]['id'], order1.id)
def test_human_id_generation(self):
"""测试 human_id 自动生成"""
from datetime import date
order = printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='测试布料',
width='150cm'
)
# human_id 格式: YYYYMMDD000001
self.assertIsNotNone(order.human_id)
self.assertTrue(len(order.human_id) >= 14)
today = date.today().strftime('%Y%m%d')
self.assertTrue(order.human_id.startswith(today))
def test_create_order_with_default_process(self):
"""测试创建订单使用默认流程"""
data = {
'customer': self.customer.id,
'fabric': '测试布料',
'width': '150cm',
}
response = self.client.post('/api/v1/printing-orders/', data, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
# 验证使用了默认流程
order = printing_models.PrintingOrder.objects.get(id=response.data['id'])
self.assertEqual(order.process.id, self.process.id)
def test_create_order_with_custom_process(self):
"""测试创建订单指定自定义流程"""
custom_process = stateflow_models.Process.objects.create(name='自定义流程')
data = {
'customer': self.customer.id,
'fabric': '测试布料',
'width': '150cm',
'process': custom_process.id,
}
response = self.client.post('/api/v1/printing-orders/', data, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
order = printing_models.PrintingOrder.objects.get(id=response.data['id'])
self.assertEqual(order.process.id, custom_process.id)
def test_order_progress_in_list(self):
"""测试订单列表包含进度字段"""
order = printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='测试布料',
width='150cm',
process=self.process,
)
response = self.client.get('/api/v1/printing-orders/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('progress', response.data['results'][0])
self.assertEqual(response.data['results'][0]['progress'], 0) # 没有任务时为0
def test_update_process_when_no_jobs(self):
"""测试没有任务时可以修改流程"""
order = printing_models.PrintingOrder.objects.create(
customer=self.customer,
fabric='测试布料',
width='150cm',
process=self.process,
)
new_process = stateflow_models.Process.objects.create(name='新流程')
data = {
'process': new_process.id,
}
response = self.client.patch(
f'/api/v1/printing-orders/{order.id}/',
data,
format='json'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
order.refresh_from_db()
self.assertEqual(order.process.id, new_process.id)
def test_cannot_update_process_when_job_started(self):
"""测试有已开始的任务时不能修改流程"""
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,
)
# 创建任务
job = printing_models.PrintingJob.objects.create(
printing_order=order,
product=product,
quantity=10,
unit='',
size='100x200',
pieces=5,
)
# 创建 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)
# 尝试修改流程
new_process = stateflow_models.Process.objects.create(name='新流程')
data = {
'process': new_process.id,
}
response = self.client.patch(
f'/api/v1/printing-orders/{order.id}/',
data,
format='json'
)
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, [])