forked from erp-dev/erp
429 lines
16 KiB
Python
429 lines
16 KiB
Python
"""
|
||
Shipment API 测试
|
||
"""
|
||
from decimal import Decimal
|
||
|
||
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 basic_info import models as basic_models
|
||
from printing import models as printing_models
|
||
from shipment import models as shipment_models
|
||
from stateflow import models as stateflow_models
|
||
|
||
User = get_user_model()
|
||
|
||
|
||
class SalesItemByPrintingOrderAPITestCase(TestCase):
|
||
"""测试通过生产订单查询销售品 API"""
|
||
|
||
def setUp(self):
|
||
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])
|
||
|
||
# 创建产品分类
|
||
self.category = basic_models.ProductCategory.objects.create(
|
||
merchant=self.merchant,
|
||
name='测试分类',
|
||
)
|
||
|
||
# 创建产品
|
||
self.product = basic_models.Product.objects.create(
|
||
merchant=self.merchant,
|
||
category=self.category,
|
||
name='测试产品',
|
||
human_id='TEST001',
|
||
)
|
||
|
||
# 创建印染订单
|
||
self.printing_order = printing_models.PrintingOrder.objects.create(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
fabric='测试面料',
|
||
width='150cm',
|
||
process=self.process,
|
||
created_by=self.user,
|
||
)
|
||
|
||
# 创建印染任务
|
||
self.printing_job1 = printing_models.PrintingJob.objects.create(
|
||
merchant=self.merchant,
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=100,
|
||
unit='米',
|
||
created_by=self.user,
|
||
)
|
||
|
||
self.printing_job2 = printing_models.PrintingJob.objects.create(
|
||
merchant=self.merchant,
|
||
printing_order=self.printing_order,
|
||
product=self.product,
|
||
quantity=200,
|
||
unit='米',
|
||
created_by=self.user,
|
||
)
|
||
|
||
# 创建出货单
|
||
self.shipment = shipment_models.Shipment.objects.create(
|
||
customer=self.customer,
|
||
shipment_date='2026-01-14',
|
||
created_by=self.user,
|
||
)
|
||
|
||
# 创建销售品 - 未关联出货单
|
||
self.sales_item1 = shipment_models.SalesItem.objects.create(
|
||
name='销售品1',
|
||
quantity=Decimal('50.00'),
|
||
unit=shipment_models.UnitChoices.METER,
|
||
printing_job_id=self.printing_job1.id,
|
||
created_by=self.user,
|
||
)
|
||
|
||
self.sales_item2 = shipment_models.SalesItem.objects.create(
|
||
name='销售品2',
|
||
quantity=Decimal('30.00'),
|
||
unit=shipment_models.UnitChoices.METER,
|
||
printing_job_id=self.printing_job1.id,
|
||
position='A1-01',
|
||
remark='备注信息',
|
||
created_by=self.user,
|
||
)
|
||
|
||
# 创建销售品 - 已关联出货单
|
||
self.sales_item3 = shipment_models.SalesItem.objects.create(
|
||
name='销售品3（已出货）',
|
||
quantity=Decimal('100.00'),
|
||
unit=shipment_models.UnitChoices.METER,
|
||
printing_job_id=self.printing_job2.id,
|
||
shipment=self.shipment,
|
||
created_by=self.user,
|
||
)
|
||
|
||
# 创建与该订单无关的销售品
|
||
self.sales_item_other = shipment_models.SalesItem.objects.create(
|
||
name='其它销售品',
|
||
quantity=Decimal('999.00'),
|
||
unit=shipment_models.UnitChoices.PIECE,
|
||
printing_job_id=99999, # 不存在的 job
|
||
created_by=self.user,
|
||
)
|
||
|
||
# 认证用户
|
||
self.client.force_authenticate(user=self.user)
|
||
|
||
def test_get_sales_items_by_printing_order_exclude_shipped(self):
|
||
"""测试查询销售品 - 默认不包含已出货的"""
|
||
url = f'/api/v1/shipment/sales-items/by-printing-order/{self.printing_order.id}/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
data = response.json()
|
||
|
||
# 应该只返回2个未出货的销售品
|
||
self.assertEqual(data['count'], 2)
|
||
|
||
# 检查返回的销售品
|
||
item_ids = [item['id'] for item in data['results']]
|
||
self.assertIn(self.sales_item1.id, item_ids)
|
||
self.assertIn(self.sales_item2.id, item_ids)
|
||
self.assertNotIn(self.sales_item3.id, item_ids) # 已出货的不应该在列表中
|
||
self.assertNotIn(self.sales_item_other.id, item_ids) # 其它订单的也不在
|
||
|
||
def test_get_sales_items_by_printing_order_include_shipped(self):
|
||
"""测试查询销售品 - 包含已出货的"""
|
||
url = f'/api/v1/shipment/sales-items/by-printing-order/{self.printing_order.id}/?include_already_has_shipment=true'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
data = response.json()
|
||
|
||
# 应该返回3个销售品（包含已出货的）
|
||
self.assertEqual(data['count'], 3)
|
||
|
||
# 检查返回的销售品
|
||
item_ids = [item['id'] for item in data['results']]
|
||
self.assertIn(self.sales_item1.id, item_ids)
|
||
self.assertIn(self.sales_item2.id, item_ids)
|
||
self.assertIn(self.sales_item3.id, item_ids) # 已出货的也应该在列表中
|
||
self.assertNotIn(self.sales_item_other.id, item_ids) # 其它订单的依然不在
|
||
|
||
def test_get_sales_items_response_format(self):
|
||
"""测试返回数据格式"""
|
||
url = f'/api/v1/shipment/sales-items/by-printing-order/{self.printing_order.id}/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
data = response.json()
|
||
|
||
# 找到 sales_item2（包含 position 和 remark）
|
||
item = next(item for item in data['results'] if item['id'] == self.sales_item2.id)
|
||
|
||
# 检查所有字段
|
||
self.assertEqual(item['name'], '销售品2')
|
||
self.assertEqual(Decimal(item['quantity']), Decimal('30.00'))
|
||
self.assertEqual(item['unit'], shipment_models.UnitChoices.METER)
|
||
self.assertEqual(item['unit_display'], '米')
|
||
self.assertEqual(item['position'], 'A1-01')
|
||
self.assertEqual(item['remark'], '备注信息')
|
||
self.assertEqual(item['printing_job_id'], self.printing_job1.id)
|
||
self.assertIsNone(item['shipment_id'])
|
||
self.assertIsNone(item['shipment_date'])
|
||
self.assertIsNotNone(item['created_at'])
|
||
self.assertEqual(item['created_by_id'], self.user.id)
|
||
|
||
def test_get_sales_items_shipped_item_format(self):
|
||
"""测试已出货的销售品返回格式"""
|
||
url = f'/api/v1/shipment/sales-items/by-printing-order/{self.printing_order.id}/?include_already_has_shipment=true'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
data = response.json()
|
||
|
||
# 找到已出货的销售品
|
||
item = next(item for item in data['results'] if item['id'] == self.sales_item3.id)
|
||
|
||
# 检查出货单信息
|
||
self.assertEqual(item['shipment_id'], self.shipment.id)
|
||
self.assertEqual(item['shipment_date'], '2026-01-14')
|
||
|
||
def test_get_sales_items_printing_order_not_found(self):
|
||
"""测试生产订单不存在"""
|
||
url = '/api/v1/shipment/sales-items/by-printing-order/99999/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
|
||
self.assertIn('不存在', response.json()['detail'])
|
||
|
||
def test_get_sales_items_empty_result(self):
|
||
"""测试生产订单没有关联销售品"""
|
||
# 创建一个没有销售品的订单
|
||
empty_order = printing_models.PrintingOrder.objects.create(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
fabric='测试面料2',
|
||
width='150cm',
|
||
process=self.process,
|
||
created_by=self.user,
|
||
)
|
||
|
||
url = f'/api/v1/shipment/sales-items/by-printing-order/{empty_order.id}/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
data = response.json()
|
||
|
||
self.assertEqual(data['count'], 0)
|
||
self.assertEqual(data['results'], [])
|
||
|
||
def test_get_sales_items_unauthenticated(self):
|
||
"""测试未认证用户"""
|
||
self.client.logout()
|
||
|
||
url = f'/api/v1/shipment/sales-items/by-printing-order/{self.printing_order.id}/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
||
|
||
|
||
class ShipmentCreateAPITestCase(TestCase):
|
||
"""测试创建出货单 API"""
|
||
|
||
def setUp(self):
|
||
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.sales_item1 = shipment_models.SalesItem.objects.create(
|
||
name='销售品1',
|
||
quantity=Decimal('50.00'),
|
||
unit=shipment_models.UnitChoices.METER,
|
||
created_by=self.user,
|
||
)
|
||
|
||
self.sales_item2 = shipment_models.SalesItem.objects.create(
|
||
name='销售品2',
|
||
quantity=Decimal('30.00'),
|
||
unit=shipment_models.UnitChoices.METER,
|
||
created_by=self.user,
|
||
)
|
||
|
||
# 创建已关联出货单的销售品
|
||
self.existing_shipment = shipment_models.Shipment.objects.create(
|
||
customer=self.customer,
|
||
shipment_date='2026-01-13',
|
||
created_by=self.user,
|
||
)
|
||
self.sales_item_shipped = shipment_models.SalesItem.objects.create(
|
||
name='销售品3（已出货）',
|
||
quantity=Decimal('100.00'),
|
||
unit=shipment_models.UnitChoices.METER,
|
||
shipment=self.existing_shipment,
|
||
created_by=self.user,
|
||
)
|
||
|
||
# 认证用户
|
||
self.client.force_authenticate(user=self.user)
|
||
|
||
def test_create_shipment_success(self):
|
||
"""测试成功创建出货单"""
|
||
data = {
|
||
'customer': self.customer.id,
|
||
'shipment_date': '2026-01-14',
|
||
'remark': '测试备注',
|
||
'sales_items': [self.sales_item1.id, self.sales_item2.id]
|
||
}
|
||
|
||
response = self.client.post('/api/v1/shipment/shipments/', data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
||
result = response.json()
|
||
|
||
# 验证返回数据
|
||
self.assertIn('id', result)
|
||
self.assertEqual(result['customer'], self.customer.id)
|
||
self.assertEqual(result['customer_name'], self.customer.name)
|
||
self.assertEqual(result['shipment_date'], '2026-01-14')
|
||
self.assertEqual(result['remark'], '测试备注')
|
||
self.assertEqual(result['items_count'], 2)
|
||
self.assertEqual(result['created_by_id'], self.user.id)
|
||
|
||
# 验证销售品已关联到出货单
|
||
self.sales_item1.refresh_from_db()
|
||
self.sales_item2.refresh_from_db()
|
||
self.assertEqual(self.sales_item1.shipment_id, result['id'])
|
||
self.assertEqual(self.sales_item2.shipment_id, result['id'])
|
||
|
||
def test_create_shipment_without_sales_items(self):
|
||
"""测试创建出货单但不关联销售品"""
|
||
data = {
|
||
'customer': self.customer.id,
|
||
'shipment_date': '2026-01-14',
|
||
}
|
||
|
||
response = self.client.post('/api/v1/shipment/shipments/', data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
||
result = response.json()
|
||
self.assertEqual(result['items_count'], 0)
|
||
|
||
def test_create_shipment_customer_not_found(self):
|
||
"""测试客户不存在"""
|
||
data = {
|
||
'customer': 99999,
|
||
'shipment_date': '2026-01-14',
|
||
}
|
||
|
||
response = self.client.post('/api/v1/shipment/shipments/', data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||
self.assertIn('不存在', response.json()['detail'])
|
||
|
||
def test_create_shipment_sales_item_not_found(self):
|
||
"""测试销售品不存在"""
|
||
data = {
|
||
'customer': self.customer.id,
|
||
'shipment_date': '2026-01-14',
|
||
'sales_items': [99999]
|
||
}
|
||
|
||
response = self.client.post('/api/v1/shipment/shipments/', data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||
self.assertIn('不存在', response.json()['detail'])
|
||
|
||
def test_create_shipment_sales_item_already_shipped(self):
|
||
"""测试销售品已关联到其他出货单"""
|
||
data = {
|
||
'customer': self.customer.id,
|
||
'shipment_date': '2026-01-14',
|
||
'sales_items': [self.sales_item_shipped.id]
|
||
}
|
||
|
||
response = self.client.post('/api/v1/shipment/shipments/', data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||
self.assertIn('已关联', response.json()['detail'])
|
||
|
||
def test_create_shipment_unauthenticated(self):
|
||
"""测试未认证用户"""
|
||
self.client.logout()
|
||
|
||
data = {
|
||
'customer': self.customer.id,
|
||
'shipment_date': '2026-01-14',
|
||
}
|
||
|
||
response = self.client.post('/api/v1/shipment/shipments/', data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|