""" 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( merchant=self.merchant, customer=self.customer, shipment_date='2026-01-14', created_by=self.user, ) # 创建销售品 - 未关联出货单 self.sales_item1 = shipment_models.SalesItem.objects.create( merchant=self.merchant, 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( merchant=self.merchant, 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( merchant=self.merchant, 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( merchant=self.merchant, 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( merchant=self.merchant, name='销售品1', quantity=Decimal('50.00'), unit=shipment_models.UnitChoices.METER, created_by=self.user, ) self.sales_item2 = shipment_models.SalesItem.objects.create( merchant=self.merchant, name='销售品2', quantity=Decimal('30.00'), unit=shipment_models.UnitChoices.METER, created_by=self.user, ) # 创建已关联出货单的销售品 self.existing_shipment = shipment_models.Shipment.objects.create( merchant=self.merchant, customer=self.customer, shipment_date='2026-01-13', created_by=self.user, ) self.sales_item_shipped = shipment_models.SalesItem.objects.create( merchant=self.merchant, 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) class ShipmentExternalCreateAPITestCase(TestCase): """测试创建出货单 external 版 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_ext', password='testpass123', email='test_ext@example.com' ) # 创建员工并关联商户 self.employee = basic_models.Employee.objects.create( sys_user=self.user, merchant=self.merchant, name='测试员工Ext', mobile='13800138002', status=basic_models.EmployeeStatusEnum.ACTIVE ) # 创建客户（需要同 merchant） self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='测试客户Ext', mobile='13900139002', area='测试地区Ext' ) # 创建一个销售品（用于验证 external 版不会绑定任何销售品） self.sales_item = shipment_models.SalesItem.objects.create( merchant=self.merchant, name='销售品-不应被绑定', quantity=Decimal('10.00'), unit=shipment_models.UnitChoices.METER, created_by=self.user, ) self.client.force_authenticate(user=self.user) def test_create_external_shipment_success(self): data = { 'customer': self.customer.id, 'shipment_date': '2026-01-14', 'remark': 'external 备注', 'external_id': 'EXT-ORDER-001', 'external_finished_products': [ {'style_name': '款式A', 'num_of_rolls': 2, 'remark': 'A备注'}, {'style_name': '款式B', 'num_of_rolls': 5}, ] } response = self.client.post('/api/v1/shipment/shipments/external/', data, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) result = response.json() self.assertEqual(result['customer'], self.customer.id) self.assertEqual(result['external_id'], 'EXT-ORDER-001') self.assertEqual(result['items_count'], 0) self.assertEqual(result['external_finished_products_count'], 2) shipment_id = result['id'] # 验证外部成品表写入并关联 self.assertEqual( shipment_models.ExternalFinishedProduct.objects.filter(shipment_id=shipment_id).count(), 2 ) efp_a = shipment_models.ExternalFinishedProduct.objects.filter(shipment_id=shipment_id, style_name='款式A').first() self.assertIsNotNone(efp_a) self.assertEqual(getattr(efp_a, 'remark', None) or '', 'A备注') # 验证不会绑定任何销售品 self.sales_item.refresh_from_db() self.assertIsNone(self.sales_item.shipment_id) def test_create_external_shipment_external_id_required(self): data = { 'customer': self.customer.id, 'shipment_date': '2026-01-14', 'external_id': ' ', 'external_finished_products': [ {'style_name': '款式A', 'num_of_rolls': 1}, ] } response = self.client.post('/api/v1/shipment/shipments/external/', data, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('external_id', response.json()) def test_create_external_shipment_products_required(self): data = { 'customer': self.customer.id, 'shipment_date': '2026-01-14', 'external_id': 'EXT-ORDER-002', 'external_finished_products': [] } response = self.client.post('/api/v1/shipment/shipments/external/', data, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('external_finished_products', response.json()) def test_create_external_shipment_unauthenticated(self): self.client.logout() data = { 'customer': self.customer.id, 'shipment_date': '2026-01-14', 'external_id': 'EXT-ORDER-003', 'external_finished_products': [ {'style_name': '款式A', 'num_of_rolls': 1}, ] } response = self.client.post('/api/v1/shipment/shipments/external/', data, format='json') self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) class ShipmentQueryAPITestCase(TestCase): """测试出货单查询 API（列表/详情）""" def setUp(self): self.client = APIClient() self.merchant1 = basic_models.Merchant.objects.create( name='商户1', type=basic_models.MerchantTypeEnum.FACTORY ) self.merchant2 = basic_models.Merchant.objects.create( name='商户2', type=basic_models.MerchantTypeEnum.FACTORY ) self.user1 = User.objects.create_user( username='u1', password='pass123', email='u1@example.com' ) self.emp1 = basic_models.Employee.objects.create( sys_user=self.user1, merchant=self.merchant1, name='员工1', mobile='13800138010', status=basic_models.EmployeeStatusEnum.ACTIVE ) self.user2 = User.objects.create_user( username='u2', password='pass123', email='u2@example.com' ) self.emp2 = basic_models.Employee.objects.create( sys_user=self.user2, merchant=self.merchant2, name='员工2', mobile='13800138011', status=basic_models.EmployeeStatusEnum.ACTIVE ) self.customer1 = basic_models.Customer.objects.create( merchant=self.merchant1, name='客户1', mobile='13900139010', area='A' ) self.customer2 = basic_models.Customer.objects.create( merchant=self.merchant2, name='客户2', mobile='13900139011', area='B' ) self.shipment1 = shipment_models.Shipment.objects.create( merchant=self.merchant1, customer=self.customer1, shipment_date='2026-01-15', created_by=self.user1, remark='s1', ) self.shipment2 = shipment_models.Shipment.objects.create( merchant=self.merchant2, customer=self.customer2, shipment_date='2026-01-15', created_by=self.user2, remark='s2', ) self.client.force_authenticate(user=self.user1) def test_list_shipments_only_current_merchant(self): resp = self.client.get('/api/v1/shipment/shipments/') self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() # LimitOffsetPagination 结构：count/next/previous/results self.assertIn('count', data) self.assertIn('results', data) ids = [it['id'] for it in data['results']] self.assertIn(self.shipment1.id, ids) self.assertNotIn(self.shipment2.id, ids) # 关键：列表序列化器必须稳定输出明细数组字段（即使为空） item = next(it for it in data['results'] if it['id'] == self.shipment1.id) self.assertIn('sales_items', item) self.assertIsInstance(item['sales_items'], list) self.assertIn('external_finished_products', item) self.assertIsInstance(item['external_finished_products'], list) def test_retrieve_shipment_success(self): resp = self.client.get(f'/api/v1/shipment/shipments/{self.shipment1.id}/') self.assertEqual(resp.status_code, status.HTTP_200_OK) result = resp.json() self.assertEqual(result['id'], self.shipment1.id) self.assertIn('sales_items', result) self.assertIsInstance(result['sales_items'], list) self.assertIn('external_finished_products', result) self.assertIsInstance(result['external_finished_products'], list) def test_list_shipments_includes_external_finished_product_remark(self): """ Shipments list 需要附带 external_finished_products 明细数据（包含 remark 字段）。 """ # 给 shipment1 挂一个外部成品表 shipment_models.ExternalFinishedProduct.objects.create( shipment=self.shipment1, style_name='款式X', num_of_rolls=1, remark='X备注', created_by=self.user1, ) resp = self.client.get('/api/v1/shipment/shipments/') self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() item = next(it for it in data['results'] if it['id'] == self.shipment1.id) efps = item['external_finished_products'] self.assertIsInstance(efps, list) target = next(p for p in efps if p['style_name'] == '款式X') self.assertIn('remark', target) self.assertEqual(target['remark'], 'X备注') def test_retrieve_other_merchant_shipment_404(self): resp = self.client.get(f'/api/v1/shipment/shipments/{self.shipment2.id}/') self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) def test_list_shipments_unauthenticated(self): self.client.logout() resp = self.client.get('/api/v1/shipment/shipments/') self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED)