""" Shipment API 测试 """ from decimal import Decimal from unittest.mock import patch from django.test import TestCase from django.test import override_settings from django.conf import settings from django.contrib.auth.models import Permission from django.utils import timezone from rest_framework.test import APIClient, APITestCase 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, external_order_id="EXT-PO-001", ) # 创建印染任务 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.assertEqual(item["printing_order_id"], self.printing_order.id) self.assertEqual(item["external_order_id"], "EXT-PO-001") 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_by_external_order_id(self): """测试通过 external_order_id 查询销售品""" url = "/api/v1/shipment/sales-items/by-printing-order/EXT-PO-001/" response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) data = response.json() 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) def test_get_sales_items_by_external_order_id_multiple_matches(self): """测试 external_order_id 匹配多个生产订单时报错""" other_order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric="测试面料3", width="160cm", process=self.process, created_by=self.user, external_order_id="EXT-DUPLICATED", ) printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric="测试面料4", width="170cm", process=self.process, created_by=self.user, external_order_id="EXT-DUPLICATED", ) printing_models.PrintingJob.objects.create( merchant=self.merchant, printing_order=other_order, product=self.product, quantity=50, unit="米", created_by=self.user, ) response = self.client.get( "/api/v1/shipment/sales-items/by-printing-order/EXT-DUPLICATED/" ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("匹配到多个生产订单", response.json()["detail"]) 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_by_printing_order_excludes_soft_deleted(self): self.sales_item2.delete_at = timezone.now() self.sales_item2.delete_by = self.user self.sales_item2.save(update_fields=["delete_at", "delete_by", "updated_at"]) 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) item_ids = [item["id"] for item in response.json()["results"]] self.assertIn(self.sales_item1.id, item_ids) self.assertNotIn(self.sales_item2.id, item_ids) 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 ShipmentSalesItemCustomersAPITestCase(TestCase): """测试由未出货销售品汇总客户 API""" def setUp(self): self.client = APIClient() self.merchant = basic_models.Merchant.objects.create( name="测试印花厂", type=basic_models.MerchantTypeEnum.FACTORY ) self.other_merchant = basic_models.Merchant.objects.create( name="其它商户", type=basic_models.MerchantTypeEnum.FACTORY ) self.user = User.objects.create_user( username="shipment_customer_user", password="testpass123", email="shipment_customer@example.com", ) self.employee = basic_models.Employee.objects.create( sys_user=self.user, merchant=self.merchant, name="测试员工", mobile="13800138020", status=basic_models.EmployeeStatusEnum.ACTIVE, ) self.customer1 = basic_models.Customer.objects.create( merchant=self.merchant, name="客户A", mobile="13900139020", area="杭州", ) self.customer2 = basic_models.Customer.objects.create( merchant=self.merchant, name="客户B", mobile="13900139021", area="绍兴", ) self.other_customer = basic_models.Customer.objects.create( merchant=self.other_merchant, name="客户C", mobile="13900139022", area="苏州", ) shipment_models.SalesItem.objects.create( merchant=self.merchant, name="客户A销售品1", quantity=Decimal("10.00"), unit=shipment_models.UnitChoices.METER, customer_id=self.customer1.id, created_by=self.user, ) shipment_models.SalesItem.objects.create( merchant=self.merchant, name="客户A销售品2", quantity=Decimal("20.00"), unit=shipment_models.UnitChoices.METER, customer_id=self.customer1.id, created_by=self.user, ) shipment_models.SalesItem.objects.create( merchant=self.merchant, name="客户B销售品1", quantity=Decimal("30.00"), unit=shipment_models.UnitChoices.METER, customer_id=self.customer2.id, created_by=self.user, ) shipped = shipment_models.Shipment.objects.create( merchant=self.merchant, customer=self.customer1, shipment_date="2026-01-14", created_by=self.user, ) shipment_models.SalesItem.objects.create( merchant=self.merchant, name="客户A已出货销售品", quantity=Decimal("40.00"), unit=shipment_models.UnitChoices.METER, customer_id=self.customer1.id, shipment=shipped, created_by=self.user, ) shipment_models.SalesItem.objects.create( merchant=self.merchant, name="无客户销售品", quantity=Decimal("50.00"), unit=shipment_models.UnitChoices.METER, created_by=self.user, ) shipment_models.SalesItem.objects.create( merchant=self.other_merchant, name="其它商户销售品", quantity=Decimal("60.00"), unit=shipment_models.UnitChoices.METER, customer_id=self.other_customer.id, created_by=self.user, ) self.client.force_authenticate(user=self.user) def test_list_customers_with_unshipped_sales_items(self): resp = self.client.get("/api/v1/shipment/sales-items/customers/") self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertEqual(data["count"], 2) self.assertIn("next", data) self.assertIn("previous", data) by_customer_id = { item["customer_id"]: item for item in data["results"] } self.assertEqual(set(by_customer_id.keys()), {self.customer1.id, self.customer2.id}) self.assertEqual(by_customer_id[self.customer1.id]["customer_name"], "客户A") self.assertEqual(by_customer_id[self.customer1.id]["unshipped_sales_items_count"], 2) self.assertEqual(by_customer_id[self.customer2.id]["unshipped_sales_items_count"], 1) def test_list_customers_with_unshipped_sales_items_supports_limit_offset(self): resp = self.client.get("/api/v1/shipment/sales-items/customers/?limit=1&offset=1") self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertEqual(data["count"], 2) self.assertEqual(len(data["results"]), 1) self.assertEqual(data["results"][0]["customer_id"], self.customer2.id) def test_list_customers_with_unshipped_sales_items_unauthenticated(self): self.client.logout() resp = self.client.get("/api/v1/shipment/sales-items/customers/") self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED) class SalesItemByCustomerAPITestCase(TestCase): """测试按客户查询销售品 API""" def setUp(self): self.client = APIClient() self.merchant = basic_models.Merchant.objects.create( name="测试印花厂", type=basic_models.MerchantTypeEnum.FACTORY ) self.other_merchant = basic_models.Merchant.objects.create( name="其它印花厂", type=basic_models.MerchantTypeEnum.FACTORY ) self.user = User.objects.create_user( username="sales_item_by_customer_user", password="testpass123", email="sales_item_by_customer@example.com", ) self.employee = basic_models.Employee.objects.create( sys_user=self.user, merchant=self.merchant, name="测试员工", mobile="13800138030", status=basic_models.EmployeeStatusEnum.ACTIVE, ) self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name="测试客户", mobile="13900139030", area="杭州", ) self.other_customer = basic_models.Customer.objects.create( merchant=self.merchant, name="其它客户", mobile="13900139031", area="绍兴", ) self.foreign_customer = basic_models.Customer.objects.create( merchant=self.other_merchant, name="外部客户", mobile="13900139032", area="苏州", ) self.state = stateflow_models.State.objects.create(name="待生产") self.process = stateflow_models.Process.objects.create(name="生产流程") self.process.replace_nodes([self.state]) 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="PROD001", ) self.printing_order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric="测试面料", width="150cm", process=self.process, created_by=self.user, external_order_id="EXT-CUST-001", ) self.printing_job = printing_models.PrintingJob.objects.create( merchant=self.merchant, printing_order=self.printing_order, product=self.product, quantity=100, unit="米", created_by=self.user, ) self.sales_item1 = shipment_models.SalesItem.objects.create( merchant=self.merchant, name="销售品1", quantity=Decimal("10.00"), unit=shipment_models.UnitChoices.METER, printing_job_id=self.printing_job.id, customer_id=self.customer.id, position="A1-01", created_by=self.user, ) self.sales_item2 = shipment_models.SalesItem.objects.create( merchant=self.merchant, name="销售品2", quantity=Decimal("20.00"), unit=shipment_models.UnitChoices.PIECE, printing_job_id=self.printing_job.id, customer_id=self.customer.id, 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_item_shipped = shipment_models.SalesItem.objects.create( merchant=self.merchant, name="已出货销售品", quantity=Decimal("30.00"), unit=shipment_models.UnitChoices.METER, printing_job_id=self.printing_job.id, customer_id=self.customer.id, shipment=self.shipment, created_by=self.user, ) shipment_models.SalesItem.objects.create( merchant=self.merchant, name="其它客户销售品", quantity=Decimal("40.00"), unit=shipment_models.UnitChoices.METER, customer_id=self.other_customer.id, created_by=self.user, ) self.client.force_authenticate(user=self.user) def test_get_sales_items_by_customer_exclude_shipped(self): resp = self.client.get( f"/api/v1/shipment/sales-items/by-customer/{self.customer.id}/" ) self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertEqual(data["count"], 2) self.assertEqual(len(data["results"]), 2) item = next(it for it in data["results"] if it["id"] == self.sales_item1.id) self.assertEqual(item["customer_id"], self.customer.id) self.assertEqual(item["customer_name"], self.customer.name) self.assertEqual(item["printing_job_id"], self.printing_job.id) self.assertEqual(item["printing_order_id"], self.printing_order.id) self.assertEqual(item["external_order_id"], "EXT-CUST-001") self.assertEqual(item["position"], "A1-01") def test_get_sales_items_by_customer_include_shipped_and_paginate(self): resp = self.client.get( f"/api/v1/shipment/sales-items/by-customer/{self.customer.id}/" "?include_already_has_shipment=true&limit=2&offset=1" ) self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertEqual(data["count"], 3) self.assertEqual(len(data["results"]), 2) result_ids = [item["id"] for item in data["results"]] self.assertEqual(result_ids, [self.sales_item2.id, self.sales_item_shipped.id]) def test_get_sales_items_by_customer_supports_external_order_id_filter(self): other_printing_order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric="其它面料", width="160cm", process=self.process, created_by=self.user, external_order_id="EXT-CUST-OTHER", ) other_printing_job = printing_models.PrintingJob.objects.create( merchant=self.merchant, printing_order=other_printing_order, product=self.product, quantity=50, unit="米", created_by=self.user, ) other_sales_item = shipment_models.SalesItem.objects.create( merchant=self.merchant, name="其它外部订单销售品", quantity=Decimal("66.00"), unit=shipment_models.UnitChoices.METER, printing_job_id=other_printing_job.id, customer_id=self.customer.id, created_by=self.user, ) resp = self.client.get( f"/api/v1/shipment/sales-items/by-customer/{self.customer.id}/" "?external_order_id=EXT-CUST-001" ) self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() result_ids = [item["id"] for item in data["results"]] self.assertEqual(set(result_ids), {self.sales_item1.id, self.sales_item2.id}) self.assertNotIn(other_sales_item.id, result_ids) def test_get_sales_items_by_customer_not_found(self): resp = self.client.get("/api/v1/shipment/sales-items/by-customer/99999/") self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) def test_get_sales_items_by_customer_excludes_soft_deleted(self): self.sales_item2.delete_at = timezone.now() self.sales_item2.delete_by = self.user self.sales_item2.save(update_fields=["delete_at", "delete_by", "updated_at"]) resp = self.client.get( f"/api/v1/shipment/sales-items/by-customer/{self.customer.id}/" ) self.assertEqual(resp.status_code, status.HTTP_200_OK) result_ids = [item["id"] for item in resp.json()["results"]] self.assertIn(self.sales_item1.id, result_ids) self.assertNotIn(self.sales_item2.id, result_ids) def test_get_sales_items_by_customer_other_merchant_404(self): resp = self.client.get( f"/api/v1/shipment/sales-items/by-customer/{self.foreign_customer.id}/" ) self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) class SalesItemDetailAPITestCase(APITestCase): """销售品详情 API 测试""" def setUp(self): self.merchant = basic_models.Merchant.objects.create( name="详情测试商户", type=basic_models.MerchantTypeEnum.FACTORY ) self.other_merchant = basic_models.Merchant.objects.create( name="其它详情测试商户", type=basic_models.MerchantTypeEnum.FACTORY ) self.user = User.objects.create_user( username="sales_item_detail_user", password="testpass123", email="sales_item_detail@example.com", ) self.employee = basic_models.Employee.objects.create( sys_user=self.user, merchant=self.merchant, name="详情测试员工", mobile="13800138031", status=basic_models.EmployeeStatusEnum.ACTIVE, ) self.other_user = User.objects.create_user( username="sales_item_detail_other_user", password="testpass123", email="sales_item_detail_other@example.com", ) self.other_employee = basic_models.Employee.objects.create( sys_user=self.other_user, merchant=self.other_merchant, name="其它详情测试员工", mobile="13800138032", status=basic_models.EmployeeStatusEnum.ACTIVE, ) self.same_merchant_other_user = User.objects.create_user( username="sales_item_detail_same_merchant_user", password="testpass123", email="sales_item_detail_same_merchant@example.com", ) self.same_merchant_other_employee = basic_models.Employee.objects.create( sys_user=self.same_merchant_other_user, merchant=self.merchant, name="同商户其它员工", mobile="13800138033", status=basic_models.EmployeeStatusEnum.ACTIVE, ) self.no_employee_user = User.objects.create_user( username="sales_item_detail_no_employee_user", password="testpass123", email="sales_item_detail_no_employee@example.com", ) self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name="详情测试客户", mobile="13900139033", area="杭州", ) 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="DETAIL001", mdy_image_url="https://example.com/product-image.jpg", ) self.state = stateflow_models.State.objects.create(name="详情状态") self.process = stateflow_models.Process.objects.create(name="详情流程") self.process.replace_nodes([self.state]) self.printing_order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric="详情面料", width="150cm", process=self.process, created_by=self.user, external_order_id="EXT-DETAIL-001", ) self.printing_job = printing_models.PrintingJob.objects.create( merchant=self.merchant, printing_order=self.printing_order, product=self.product, quantity=20, unit="米", created_by=self.user, ) self.new_printing_job = printing_models.PrintingJob.objects.create( merchant=self.merchant, printing_order=self.printing_order, product=self.product, quantity=30, unit="米", created_by=self.user, ) self.other_customer = basic_models.Customer.objects.create( merchant=self.other_merchant, name="其它详情测试客户", mobile="13900139034", area="苏州", ) self.other_category = basic_models.ProductCategory.objects.create( merchant=self.other_merchant, name="其它详情分类", ) self.other_product = basic_models.Product.objects.create( merchant=self.other_merchant, category=self.other_category, name="其它详情产品", human_id="DETAIL002", ) self.other_printing_order = printing_models.PrintingOrder.objects.create( merchant=self.other_merchant, customer=self.other_customer, fabric="其它详情面料", width="160cm", process=self.process, created_by=self.other_user, external_order_id="EXT-DETAIL-002", ) self.foreign_printing_job = printing_models.PrintingJob.objects.create( merchant=self.other_merchant, printing_order=self.other_printing_order, product=self.other_product, quantity=40, unit="米", created_by=self.other_user, ) self.sales_item = shipment_models.SalesItem.objects.create( merchant=self.merchant, name="详情销售品", quantity=Decimal("88.00"), unit=shipment_models.UnitChoices.METER, printing_job_id=self.printing_job.id, customer_id=self.customer.id, position="B2-03", created_by=self.user, ) self.client.force_authenticate(user=self.user) def test_get_sales_item_detail_success(self): resp = self.client.get(f"/api/v1/shipment/sales-items/{self.sales_item.id}/") self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertEqual(data["id"], self.sales_item.id) self.assertEqual(data["printing_order_id"], self.printing_order.id) self.assertEqual(data["external_order_id"], "EXT-DETAIL-001") self.assertEqual(data["customer_name"], self.customer.name) self.assertEqual(data["product_image_url"], "https://example.com/product-image.jpg") def test_get_sales_item_detail_other_merchant_404(self): self.client.force_authenticate(user=self.other_user) resp = self.client.get(f"/api/v1/shipment/sales-items/{self.sales_item.id}/") self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) def test_patch_sales_item_success_and_create_change_record(self): resp = self.client.patch( f"/api/v1/shipment/sales-items/{self.sales_item.id}/", { "quantity": "99.50", "remark": "改备注", "position": "C3-08", }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.sales_item.refresh_from_db() self.assertEqual(self.sales_item.quantity, Decimal("99.50")) self.assertEqual(self.sales_item.remark, "改备注") self.assertEqual(self.sales_item.position, "C3-08") change_record = self.sales_item.change_records.get() self.assertEqual(change_record.operator_id, self.user.id) self.assertEqual( change_record.before_values, { "quantity": "88.00", "remark": "", "position": "B2-03", }, ) self.assertEqual( change_record.after_values, { "quantity": "99.50", "remark": "改备注", "position": "C3-08", }, ) def test_patch_sales_item_rejects_relation_or_other_disallowed_fields(self): resp = self.client.patch( f"/api/v1/shipment/sales-items/{self.sales_item.id}/", { "name": "不允许改名", "customer_id": self.customer.id, }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("仅允许修改以下字段", resp.json()["detail"]) self.assertFalse(self.sales_item.change_records.exists()) def test_patch_sales_item_invalid_quantity(self): resp = self.client.patch( f"/api/v1/shipment/sales-items/{self.sales_item.id}/", { "quantity": "abc", }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("数量 abc 格式无效", resp.json()["detail"]) self.assertFalse(self.sales_item.change_records.exists()) def test_patch_sales_item_other_merchant_404(self): self.client.force_authenticate(user=self.other_user) resp = self.client.patch( f"/api/v1/shipment/sales-items/{self.sales_item.id}/", { "remark": "无权限修改", }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) def test_delete_sales_item_requires_permission(self): resp = self.client.delete(f"/api/v1/shipment/sales-items/{self.sales_item.id}/") self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) self.sales_item.refresh_from_db() self.assertIsNone(self.sales_item.delete_at) self.assertIsNone(self.sales_item.delete_by) def test_delete_sales_item_success_soft_deletes_and_hides_detail(self): permission = Permission.objects.get(codename="soft_delete_salesitem") self.user.user_permissions.add(permission) resp = self.client.delete(f"/api/v1/shipment/sales-items/{self.sales_item.id}/") self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(resp.json()["detail"], "销售品已标记为删除") self.sales_item.refresh_from_db() self.assertIsNotNone(self.sales_item.delete_at) self.assertEqual(self.sales_item.delete_by, self.user) detail_resp = self.client.get(f"/api/v1/shipment/sales-items/{self.sales_item.id}/") self.assertEqual(detail_resp.status_code, status.HTTP_404_NOT_FOUND) def test_delete_sales_item_other_merchant_404(self): permission = Permission.objects.get(codename="soft_delete_salesitem") self.other_user.user_permissions.add(permission) self.client.force_authenticate(user=self.other_user) resp = self.client.delete(f"/api/v1/shipment/sales-items/{self.sales_item.id}/") self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) def test_delete_sales_item_rejects_when_already_bound_to_shipment(self): permission = Permission.objects.get(codename="soft_delete_salesitem") self.user.user_permissions.add(permission) shipment = shipment_models.Shipment.objects.create( merchant=self.merchant, customer=self.customer, shipment_date="2026-04-03", created_by=self.user, ) self.sales_item.shipment = shipment self.sales_item.save(update_fields=["shipment", "updated_at"]) resp = self.client.delete(f"/api/v1/shipment/sales-items/{self.sales_item.id}/") self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(resp.json()["detail"], "已关联出货单的销售品不允许删除") self.sales_item.refresh_from_db() self.assertIsNone(self.sales_item.delete_at) def test_rebuild_sales_item_success_with_original_quantity(self): resp = self.client.post( f"/api/v1/shipment/sales-items/{self.sales_item.id}/rebuild/", { "new_printing_job_id": self.new_printing_job.id, }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_201_CREATED) data = resp.json() self.assertEqual(data["detail"], "销售品已重建") self.assertEqual(data["old_sales_item_id"], self.sales_item.id) self.assertNotEqual(data["new_sales_item_id"], self.sales_item.id) self.sales_item.refresh_from_db() self.assertIsNotNone(self.sales_item.delete_at) self.assertEqual(self.sales_item.delete_by, self.user) new_sales_item = shipment_models.SalesItem.objects.get(id=data["new_sales_item_id"]) self.assertEqual(new_sales_item.created_by, self.user) self.assertEqual(new_sales_item.printing_job_id, self.new_printing_job.id) self.assertEqual(new_sales_item.quantity, Decimal("88.00")) self.assertEqual(new_sales_item.name, self.sales_item.name) self.assertEqual(new_sales_item.unit, self.sales_item.unit) self.assertEqual(new_sales_item.customer_id, self.sales_item.customer_id) rebuild_record = shipment_models.SalesItemRebuildRecord.objects.get( id=data["rebuild_record_id"] ) self.assertEqual(rebuild_record.old_sales_item_id, self.sales_item.id) self.assertEqual(rebuild_record.new_sales_item_id, new_sales_item.id) self.assertEqual(rebuild_record.operator_id, self.user.id) self.assertEqual(rebuild_record.old_printing_job_id, self.printing_job.id) self.assertEqual(rebuild_record.new_printing_job_id, self.new_printing_job.id) self.assertEqual(rebuild_record.old_quantity, Decimal("88.00")) self.assertEqual(rebuild_record.new_quantity, Decimal("88.00")) def test_rebuild_sales_item_success_with_override_quantity(self): resp = self.client.post( f"/api/v1/shipment/sales-items/{self.sales_item.id}/rebuild/", { "new_printing_job_id": self.new_printing_job.id, "quantity": "66.50", }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_201_CREATED) new_sales_item = shipment_models.SalesItem.objects.get( id=resp.json()["new_sales_item_id"] ) self.assertEqual(new_sales_item.quantity, Decimal("66.50")) def test_rebuild_sales_item_not_found(self): resp = self.client.post( "/api/v1/shipment/sales-items/999999/rebuild/", { "new_printing_job_id": self.new_printing_job.id, }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) def test_rebuild_sales_item_creator_mismatch_returns_403(self): self.client.force_authenticate(user=self.same_merchant_other_user) resp = self.client.post( f"/api/v1/shipment/sales-items/{self.sales_item.id}/rebuild/", { "new_printing_job_id": self.new_printing_job.id, }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) self.assertEqual(resp.json()["detail"], "只有销售品创建者才能执行重建") def test_rebuild_sales_item_rejects_invalid_new_printing_job(self): resp = self.client.post( f"/api/v1/shipment/sales-items/{self.sales_item.id}/rebuild/", { "new_printing_job_id": self.foreign_printing_job.id, }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("不存在或不属于当前商户", resp.json()["detail"]) def test_rebuild_sales_item_rejects_when_already_bound_to_shipment(self): shipment = shipment_models.Shipment.objects.create( merchant=self.merchant, customer=self.customer, shipment_date="2026-04-03", created_by=self.user, ) self.sales_item.shipment = shipment self.sales_item.save(update_fields=["shipment", "updated_at"]) resp = self.client.post( f"/api/v1/shipment/sales-items/{self.sales_item.id}/rebuild/", { "new_printing_job_id": self.new_printing_job.id, }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertEqual(resp.json()["detail"], "已关联出货单的销售品不允许删除") def test_rebuild_sales_item_user_without_employee_returns_404(self): self.client.force_authenticate(user=self.no_employee_user) resp = self.client.post( f"/api/v1/shipment/sales-items/{self.sales_item.id}/rebuild/", { "new_printing_job_id": self.new_printing_job.id, }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) 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.state1 = stateflow_models.State.objects.create(name="待印染") self.state2 = stateflow_models.State.objects.create(name="印染中") self.process = stateflow_models.Process.objects.create(name="印染流程") self.process.replace_nodes([self.state1, self.state2]) 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="SHIPTEST001", ) 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.other_printing_order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric="测试面料2", width="160cm", 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=80, unit="米", created_by=self.user, ) self.printing_job_other = printing_models.PrintingJob.objects.create( merchant=self.merchant, printing_order=self.other_printing_order, product=self.product, quantity=60, unit="米", 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_job2.id, created_by=self.user, ) self.sales_item_other_order = shipment_models.SalesItem.objects.create( merchant=self.merchant, name="销售品4（不同生产单）", quantity=Decimal("20.00"), unit=shipment_models.UnitChoices.METER, printing_job_id=self.printing_job_other.id, created_by=self.user, ) self.sales_item_missing_job = shipment_models.SalesItem.objects.create( merchant=self.merchant, name="销售品5（缺少生产任务）", quantity=Decimal("10.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, printing_job_id=self.printing_job1.id, 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", "address": "杭州市测试路 1 号", "contact_name": "张三", "contact_phone": "13800138000", "area": "华东", "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["address"], "杭州市测试路 1 号") self.assertEqual(result["contact_name"], "张三") self.assertEqual(result["contact_phone"], "13800138000") self.assertEqual(result.get("area", ""), "华东") self.assertEqual(result["remark"], "测试备注") self.assertEqual(result["status"], shipment_models.ShipmentStatus.DRAFT) self.assertEqual(result["status_display"], "草稿(未发布)") 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"]) @override_settings(TESTING=False, SHIPMENT_CREATED_WECOM_NOTIFY_ENABLED=True) def test_create_shipment_enqueues_wecom_notification_task(self): """测试创建出货单后投递企业微信通知任务""" data = { "customer": self.customer.id, "shipment_date": "2026-01-14", "sales_items": [self.sales_item1.id, self.sales_item2.id], } with patch("shipment.tasks.notify_shipment_created_wecom.delay") as mock_delay: with self.captureOnCommitCallbacks(execute=True): response = self.client.post("/api/v1/shipment/shipments/", data, format="json") self.assertEqual(response.status_code, status.HTTP_201_CREATED) shipment_id = response.json()["id"] mock_delay.assert_called_once_with( shipment_id=shipment_id, created_by_id=self.user.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) self.assertEqual(result["status"], shipment_models.ShipmentStatus.DRAFT) self.assertEqual(result["status_display"], "草稿(未发布)") 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_rejects_sales_items_from_different_printing_orders(self): """测试销售品来自不同生产订单时拒绝创建""" data = { "customer": self.customer.id, "shipment_date": "2026-01-14", "sales_items": [self.sales_item1.id, self.sales_item_other_order.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_rejects_sales_item_without_printing_job(self): """测试销售品缺少生产任务时拒绝创建""" data = { "customer": self.customer.id, "shipment_date": "2026-01-14", "sales_items": [self.sales_item_missing_job.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", "address": "绍兴市仓库 2 号", "contact_name": "李四", "contact_phone": "13900139000", "area": "华南", "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["address"], "绍兴市仓库 2 号") self.assertEqual(result["contact_name"], "李四") self.assertEqual(result["contact_phone"], "13900139000") self.assertEqual(result.get("area", ""), "华南") self.assertEqual(result["status"], shipment_models.ShipmentStatus.DRAFT) self.assertEqual(result["status_display"], "草稿(未发布)") 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.category1 = basic_models.ProductCategory.objects.create( merchant=self.merchant1, name="分类1", ) self.product1 = basic_models.Product.objects.create( merchant=self.merchant1, category=self.category1, name="产品1", human_id="SHIPMENTQ001", mdy_image_url="https://example.com/shipment-list-image.jpg", ) self.state1 = stateflow_models.State.objects.create(name="查询状态") self.process1 = stateflow_models.Process.objects.create(name="查询流程") self.process1.replace_nodes([self.state1]) self.printing_order1 = printing_models.PrintingOrder.objects.create( merchant=self.merchant1, customer=self.customer1, fabric="查询面料", width="150cm", process=self.process1, created_by=self.user1, external_order_id="QUERY-PO-001", description="查询订单备注", ) self.printing_job1 = printing_models.PrintingJob.objects.create( merchant=self.merchant1, printing_order=self.printing_order1, product=self.product1, quantity=20, unit="米", created_by=self.user1, ) self.shipment1 = shipment_models.Shipment.objects.create( merchant=self.merchant1, customer=self.customer1, shipment_date="2026-01-15", created_by=self.user1, remark="s1", area="A1", ) self.shipment2 = shipment_models.Shipment.objects.create( merchant=self.merchant2, customer=self.customer2, shipment_date="2026-01-15", created_by=self.user2, remark="s2", area="B1", ) self.sales_item1 = shipment_models.SalesItem.objects.create( merchant=self.merchant1, shipment=self.shipment1, name="列表销售品", quantity=Decimal("12.00"), unit=shipment_models.UnitChoices.METER, printing_job_id=self.printing_job1.id, customer_id=self.customer1.id, created_by=self.user1, ) 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("address", item) self.assertIn("contact_name", item) self.assertIn("contact_phone", item) self.assertIn("area", item) self.assertIn("delivery_id", item) self.assertIsNone(item["delivery_id"]) self.assertEqual(item["fabric"], "查询面料") self.assertEqual(item["order_description"], "查询订单备注") self.assertEqual(item["area"], "A1") self.assertEqual(item["status"], shipment_models.ShipmentStatus.DRAFT) self.assertEqual(item["status_display"], "草稿(未发布)") self.assertIn("sales_items", item) self.assertIsInstance(item["sales_items"], list) self.assertEqual(item["sales_items"][0]["product_image_url"], "https://example.com/shipment-list-image.jpg") self.assertIn("external_finished_products", item) self.assertIsInstance(item["external_finished_products"], list) def test_list_shipments_supports_only_address_null_filter(self): self.shipment1.address = "" self.shipment1.save(update_fields=["address", "updated_at"]) self.shipment2.address = "" self.shipment2.save(update_fields=["address", "updated_at"]) shipment_with_address = shipment_models.Shipment.objects.create( merchant=self.merchant1, customer=self.customer1, shipment_date="2026-01-16", created_by=self.user1, address="绍兴市测试路 8 号", ) resp_true = self.client.get("/api/v1/shipment/shipments/?only_address_null=true") self.assertEqual(resp_true.status_code, status.HTTP_200_OK) true_ids = [it["id"] for it in resp_true.json()["results"]] self.assertIn(self.shipment1.id, true_ids) self.assertNotIn(shipment_with_address.id, true_ids) resp_false = self.client.get("/api/v1/shipment/shipments/?only_address_null=false") self.assertEqual(resp_false.status_code, status.HTTP_200_OK) false_ids = [it["id"] for it in resp_false.json()["results"]] self.assertIn(shipment_with_address.id, false_ids) self.assertNotIn(self.shipment1.id, false_ids) def test_list_shipments_supports_delivery_id_filter(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant1, driver_name="过滤司机", vehicle_trip="FILTER-001", created_by=self.user1, ) self.shipment1.delivery = delivery self.shipment1.save(update_fields=["delivery", "updated_at"]) other_same_merchant = shipment_models.Shipment.objects.create( merchant=self.merchant1, customer=self.customer1, shipment_date="2026-01-17", created_by=self.user1, remark="other", area="A2", ) resp = self.client.get(f"/api/v1/shipment/shipments/?delivery_id={delivery.id}") self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertEqual(data["count"], 1) self.assertEqual(data["results"][0]["id"], self.shipment1.id) self.assertEqual(data["results"][0]["delivery_id"], delivery.id) self.assertNotEqual(other_same_merchant.id, self.shipment1.id) def test_list_shipments_ignores_null_like_delivery_id_filter(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant1, driver_name="空值司机", vehicle_trip="FILTER-NULL", created_by=self.user1, ) self.shipment1.delivery = delivery self.shipment1.save(update_fields=["delivery", "updated_at"]) resp = self.client.get("/api/v1/shipment/shipments/?delivery_id=null") self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() ids = [it["id"] for it in data["results"]] self.assertIn(self.shipment1.id, ids) shipment_item = next(it for it in data["results"] if it["id"] == self.shipment1.id) self.assertEqual(shipment_item["delivery_id"], delivery.id) def test_list_shipments_supports_delivery_isnull_true_filter(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant1, driver_name="未绑定筛选司机", vehicle_trip="FILTER-ISNULL-TRUE", created_by=self.user1, ) self.shipment1.delivery = delivery self.shipment1.save(update_fields=["delivery", "updated_at"]) shipment_without_delivery = shipment_models.Shipment.objects.create( merchant=self.merchant1, customer=self.customer1, shipment_date="2026-01-18", created_by=self.user1, remark="no-delivery", area="A3", ) resp = self.client.get("/api/v1/shipment/shipments/?delivery_isnull=true") self.assertEqual(resp.status_code, status.HTTP_200_OK) ids = [it["id"] for it in resp.json()["results"]] self.assertIn(shipment_without_delivery.id, ids) self.assertNotIn(self.shipment1.id, ids) def test_list_shipments_supports_delivery_isnull_false_filter(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant1, driver_name="已绑定筛选司机", vehicle_trip="FILTER-ISNULL-FALSE", created_by=self.user1, ) self.shipment1.delivery = delivery self.shipment1.save(update_fields=["delivery", "updated_at"]) shipment_models.Shipment.objects.create( merchant=self.merchant1, customer=self.customer1, shipment_date="2026-01-19", created_by=self.user1, remark="still-null", area="A4", ) resp = self.client.get("/api/v1/shipment/shipments/?delivery_isnull=false") self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertEqual(data["count"], 1) self.assertEqual(data["results"][0]["id"], self.shipment1.id) self.assertEqual(data["results"][0]["delivery_id"], delivery.id) 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.assertIn("address", result) self.assertIn("contact_name", result) self.assertIn("contact_phone", result) self.assertIn("delivery_id", result) self.assertIsNone(result["delivery_id"]) self.assertEqual(result["fabric"], "查询面料") self.assertEqual(result["order_description"], "查询订单备注") self.assertEqual(result["id"], self.shipment1.id) self.assertIn("area", result) self.assertEqual(result["area"], "A1") self.assertEqual(result["status"], shipment_models.ShipmentStatus.DRAFT) self.assertEqual(result["status_display"], "草稿(未发布)") self.assertIn("sales_items", result) self.assertIsInstance(result["sales_items"], list) self.assertEqual(result["sales_items"][0]["product_image_url"], "https://example.com/shipment-list-image.jpg") self.assertIn("external_finished_products", result) self.assertIsInstance(result["external_finished_products"], list) def test_patch_shipment_area_success(self): """ 新增字段 area：支持更新（PATCH）并回显。 """ resp = self.client.patch( f"/api/v1/shipment/shipments/{self.shipment1.id}/", data={ "area": "更新地区", "address": "宁波市滨海路 9 号", "contact_name": "王五", "contact_phone": "13700137000", }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) result = resp.json() self.assertEqual(result["id"], self.shipment1.id) self.assertEqual(result.get("area", ""), "更新地区") self.assertEqual(result["address"], "宁波市滨海路 9 号") self.assertEqual(result["contact_name"], "王五") self.assertEqual(result["contact_phone"], "13700137000") self.shipment1.refresh_from_db() self.assertEqual(getattr(self.shipment1, "area", ""), "更新地区") self.assertEqual(self.shipment1.address, "宁波市滨海路 9 号") self.assertEqual(self.shipment1.contact_name, "王五") self.assertEqual(self.shipment1.contact_phone, "13700137000") def test_patch_shipment_allows_approved_shipment_without_delivery(self): """ 已审核但未关联送货单的出货单允许修改业务数据。 """ from shipment.services import modify_status modify_status( self.shipment1, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user1, ) modify_status( self.shipment1, target_status=shipment_models.ShipmentStatus.APPROVED, operator=self.user1, approved_by=self.user1, ) resp = self.client.patch( f"/api/v1/shipment/shipments/{self.shipment1.id}/", data={"area": "审核后可修改"}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(resp.json()["area"], "审核后可修改") self.shipment1.refresh_from_db() self.assertEqual(self.shipment1.area, "审核后可修改") def test_patch_shipment_rejects_when_bound_to_delivery(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant1, driver_name="绑定司机", vehicle_trip="LOCK-001", created_by=self.user1, ) self.shipment1.delivery = delivery self.shipment1.save(update_fields=["delivery", "updated_at"]) resp = self.client.patch( f"/api/v1/shipment/shipments/{self.shipment1.id}/", data={"area": "不允许修改"}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("已关联送货单", resp.json()["detail"]) self.shipment1.refresh_from_db() self.assertEqual(self.shipment1.area, "A1") 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_list_shipments_without_sales_items_returns_null_fabric(self): shipment_without_sales_items = shipment_models.Shipment.objects.create( merchant=self.merchant1, customer=self.customer1, shipment_date="2026-01-20", created_by=self.user1, remark="no-items", area="A5", ) resp = self.client.get("/api/v1/shipment/shipments/") self.assertEqual(resp.status_code, status.HTTP_200_OK) item = next(it for it in resp.json()["results"] if it["id"] == shipment_without_sales_items.id) self.assertIn("fabric", item) self.assertIsNone(item["fabric"]) self.assertIn("order_description", item) self.assertIsNone(item["order_description"]) 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_supports_status_filter(self): from shipment.services import modify_status modify_status( self.shipment1, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user1, ) resp = self.client.get( f"/api/v1/shipment/shipments/?status={shipment_models.ShipmentStatus.PUBLISHED}" ) self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertEqual(data["count"], 1) self.assertEqual(data["results"][0]["id"], self.shipment1.id) def test_retrieve_shipment_supports_status_filter(self): from shipment.services import modify_status modify_status( self.shipment1, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user1, ) resp = self.client.get( f"/api/v1/shipment/shipments/{self.shipment1.id}/?status={shipment_models.ShipmentStatus.PUBLISHED}" ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertEqual(resp.json()["id"], self.shipment1.id) def test_retrieve_shipment_status_filter_mismatch_returns_404(self): from shipment.services import modify_status modify_status( self.shipment1, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user1, ) resp = self.client.get( f"/api/v1/shipment/shipments/{self.shipment1.id}/?status={shipment_models.ShipmentStatus.APPROVED}" ) 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) class ShipmentStatusServiceTestCase(TestCase): """测试 shipment.services.modify_status 状态机""" def setUp(self): self.merchant = basic_models.Merchant.objects.create( name="状态测试商户", type=basic_models.MerchantTypeEnum.FACTORY ) self.user = User.objects.create_user( username="shipment_status_user", password="testpass123", email="shipment_status@example.com", ) self.employee = basic_models.Employee.objects.create( sys_user=self.user, merchant=self.merchant, name="状态测试员工", mobile="13800138100", status=basic_models.EmployeeStatusEnum.ACTIVE, ) self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name="状态测试客户", mobile="13900139100", area="杭州", ) self.shipment = shipment_models.Shipment.objects.create( merchant=self.merchant, customer=self.customer, shipment_date="2026-04-02", created_by=self.user, ) def test_modify_status_keeps_idempotent_when_same_status(self): from shipment.services import modify_status original_updated_at = self.shipment.updated_at original_status_modified_at = self.shipment.status_modified_at returned = modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.DRAFT, operator=self.user, ) self.assertEqual(returned.id, self.shipment.id) self.shipment.refresh_from_db() self.assertEqual(self.shipment.status, shipment_models.ShipmentStatus.DRAFT) self.assertEqual(self.shipment.status_modified_at, original_status_modified_at) self.assertEqual(self.shipment.updated_at, original_updated_at) def test_modify_status_allows_draft_to_published(self): from shipment.services import modify_status modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user, ) self.shipment.refresh_from_db() self.assertEqual(self.shipment.status, shipment_models.ShipmentStatus.PUBLISHED) self.assertIsNotNone(self.shipment.status_modified_at) self.assertIsNone(self.shipment.cancelled_by) self.assertIsNone(self.shipment.approved_by) def test_modify_status_rejects_draft_to_approved(self): from shipment.services import modify_status with self.assertRaisesMessage(ValueError, "不允许将出货单状态从 草稿(未发布) 修改为 已审核"): modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.APPROVED, operator=self.user, approved_by=self.user, ) def test_modify_status_requires_approved_by_when_approving(self): from shipment.services import modify_status modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user, ) with self.assertRaisesMessage(ValueError, "approved_by 不能为空"): modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.APPROVED, operator=self.user, approved_by=None, ) def test_modify_status_sets_approved_by_when_approved(self): from shipment.services import modify_status modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user, ) modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.APPROVED, operator=self.user, approved_by=self.user, ) self.shipment.refresh_from_db() self.assertEqual(self.shipment.status, shipment_models.ShipmentStatus.APPROVED) self.assertEqual(self.shipment.approved_by, self.user) self.assertIsNotNone(self.shipment.status_modified_at) def test_modify_status_rejected_unbinds_sales_items_and_snapshots_ids(self): from shipment.services import modify_status sales_item_1 = shipment_models.SalesItem.objects.create( merchant=self.merchant, shipment=self.shipment, name="驳回销售品1", quantity=Decimal("10.00"), unit=shipment_models.UnitChoices.METER, created_by=self.user, ) sales_item_2 = shipment_models.SalesItem.objects.create( merchant=self.merchant, shipment=self.shipment, name="驳回销售品2", quantity=Decimal("20.00"), unit=shipment_models.UnitChoices.METER, created_by=self.user, ) modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user, ) modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.REJECTED, operator=self.user, ) self.shipment.refresh_from_db() sales_item_1.refresh_from_db() sales_item_2.refresh_from_db() self.assertEqual(self.shipment.status, shipment_models.ShipmentStatus.REJECTED) self.assertEqual( self.shipment.rejected_sales_item_ids, [sales_item_1.id, sales_item_2.id], ) self.assertIsNone(sales_item_1.shipment_id) self.assertIsNone(sales_item_2.shipment_id) def test_modify_status_rejects_rejected_to_approved(self): from shipment.services import modify_status modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user, ) modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.REJECTED, operator=self.user, ) with self.assertRaisesMessage(ValueError, "不允许将出货单状态从 已驳回 修改为 已审核"): modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.APPROVED, operator=self.user, approved_by=self.user, ) def test_modify_status_rejects_rejected_to_published(self): from shipment.services import modify_status modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user, ) modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.REJECTED, operator=self.user, ) with self.assertRaisesMessage(ValueError, "不允许将出货单状态从 已驳回 修改为 已发布"): modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user, ) def test_modify_status_allows_cancel_from_any_non_cancelled_state(self): from shipment.services import modify_status modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user, ) modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.CANCELLED, operator=self.user, ) self.shipment.refresh_from_db() self.assertEqual(self.shipment.status, shipment_models.ShipmentStatus.CANCELLED) self.assertEqual(self.shipment.cancelled_by, self.user) self.assertIsNotNone(self.shipment.status_modified_at) def test_modify_status_rejects_change_after_cancelled(self): from shipment.services import modify_status modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user, ) modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.CANCELLED, operator=self.user, ) with self.assertRaisesMessage(ValueError, "不允许将出货单状态从 已取消 修改为 已审核"): modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.APPROVED, operator=self.user, approved_by=self.user, ) class ShipmentStatusAPITestCase(APITestCase): """测试出货单状态流转 API""" def setUp(self): self.client = APIClient() self.merchant = basic_models.Merchant.objects.create( name="状态 API 商户", type=basic_models.MerchantTypeEnum.FACTORY ) self.other_merchant = basic_models.Merchant.objects.create( name="状态 API 其它商户", type=basic_models.MerchantTypeEnum.FACTORY ) self.user = User.objects.create_user( username="shipment_status_api_user", password="testpass123", email="shipment_status_api@example.com", ) self.employee = basic_models.Employee.objects.create( sys_user=self.user, merchant=self.merchant, name="状态 API 员工", mobile="13800138101", status=basic_models.EmployeeStatusEnum.ACTIVE, ) self.other_user = User.objects.create_user( username="shipment_status_api_other_user", password="testpass123", email="shipment_status_api_other@example.com", ) self.other_employee = basic_models.Employee.objects.create( sys_user=self.other_user, merchant=self.other_merchant, name="状态 API 其它员工", mobile="13800138102", status=basic_models.EmployeeStatusEnum.ACTIVE, ) self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name="状态 API 客户", mobile="13900139101", area="杭州", ) self.shipment = shipment_models.Shipment.objects.create( merchant=self.merchant, customer=self.customer, shipment_date="2026-04-02", created_by=self.user, ) self.client.force_authenticate(user=self.user) def test_patch_status_draft_to_published_success(self): resp = self.client.patch( f"/api/v1/shipment/shipments/{self.shipment.id}/status/", {"status": shipment_models.ShipmentStatus.PUBLISHED}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.shipment.refresh_from_db() self.assertEqual(self.shipment.status, shipment_models.ShipmentStatus.PUBLISHED) self.assertIsNotNone(self.shipment.status_modified_at) self.assertIsNone(self.shipment.approved_by) self.assertIsNone(self.shipment.cancelled_by) def test_put_status_published_to_approved_sets_approved_by(self): from shipment.services import modify_status modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user, ) resp = self.client.put( f"/api/v1/shipment/shipments/{self.shipment.id}/status/", {"status": shipment_models.ShipmentStatus.APPROVED}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.shipment.refresh_from_db() self.assertEqual(self.shipment.status, shipment_models.ShipmentStatus.APPROVED) self.assertEqual(self.shipment.approved_by, self.user) def test_patch_status_published_to_rejected_success(self): from shipment.services import modify_status sales_item = shipment_models.SalesItem.objects.create( merchant=self.merchant, shipment=self.shipment, name="状态API驳回销售品", quantity=Decimal("15.00"), unit=shipment_models.UnitChoices.METER, created_by=self.user, ) modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user, ) resp = self.client.patch( f"/api/v1/shipment/shipments/{self.shipment.id}/status/", {"status": shipment_models.ShipmentStatus.REJECTED}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.shipment.refresh_from_db() sales_item.refresh_from_db() self.assertEqual(self.shipment.status, shipment_models.ShipmentStatus.REJECTED) self.assertEqual(self.shipment.rejected_sales_item_ids, [sales_item.id]) self.assertIsNone(sales_item.shipment_id) def test_patch_status_to_cancelled_sets_cancelled_by(self): from shipment.services import modify_status modify_status( self.shipment, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user, ) resp = self.client.patch( f"/api/v1/shipment/shipments/{self.shipment.id}/status/", {"status": shipment_models.ShipmentStatus.CANCELLED}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.shipment.refresh_from_db() self.assertEqual(self.shipment.status, shipment_models.ShipmentStatus.CANCELLED) self.assertEqual(self.shipment.cancelled_by, self.user) def test_patch_status_rejects_invalid_transition(self): resp = self.client.patch( f"/api/v1/shipment/shipments/{self.shipment.id}/status/", {"status": shipment_models.ShipmentStatus.APPROVED}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("不允许将出货单状态从 草稿(未发布) 修改为 已审核", resp.json()["detail"]) def test_patch_status_other_merchant_404(self): self.client.force_authenticate(user=self.other_user) resp = self.client.patch( f"/api/v1/shipment/shipments/{self.shipment.id}/status/", {"status": shipment_models.ShipmentStatus.PUBLISHED}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND) def test_patch_status_unauthenticated(self): self.client.logout() resp = self.client.patch( f"/api/v1/shipment/shipments/{self.shipment.id}/status/", {"status": shipment_models.ShipmentStatus.PUBLISHED}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED) class SalesItemCreateAPITestCase(APITestCase): """销售品创建 API 测试""" def setUp(self): 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 decimal import Decimal User = get_user_model() # 创建商户 self.merchant = basic_models.Merchant.objects.create( name="测试商户", type=basic_models.MerchantTypeEnum.FACTORY ) # 创建用户 self.user = User.objects.create_user( username="testuser_sales", password="testpass123", email="test_sales@example.com", ) # 创建员工并关联商户 self.employee = basic_models.Employee.objects.create( sys_user=self.user, merchant=self.merchant, name="测试员工Sales", mobile="13800138003", status=basic_models.EmployeeStatusEnum.ACTIVE, ) # 创建客户 self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name="测试客户Sales", mobile="13900139003", area="测试地区Sales", ) self.category = basic_models.ProductCategory.objects.create( merchant=self.merchant, name="测试分类Sales", ) # 创建产品 self.product = basic_models.Product.objects.create( merchant=self.merchant, category=self.category, name="测试产品", human_id="TEST001", unit=basic_models.ProductUnitEnum.METER, ) # 创建印染订单 self.printing_order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric="测试面料", created_by=self.user, ) # 创建印染任务 self.printing_job = printing_models.PrintingJob.objects.create( merchant=self.merchant, printing_order=self.printing_order, product=self.product, quantity=100, unit="米", pieces=1, description="测试任务", created_by=self.user, ) self.client.force_authenticate(user=self.user) def test_create_sales_item_success(self): """测试成功创建销售品""" data = { "printing_job_id": self.printing_job.id, "name": "测试销售品", "quantity": "150.50", "unit": 1, # 米 "customer_id": self.customer.id, "remark": "测试备注", "position": "A1-01", } resp = self.client.post("/api/v1/shipment/sales-items/", data, format="json") self.assertEqual(resp.status_code, status.HTTP_201_CREATED) result = resp.json() self.assertEqual(result["name"], "测试销售品") self.assertEqual(result["quantity"], "150.50") self.assertEqual(result["unit"], 1) self.assertEqual(result["unit_display"], "米") self.assertEqual(result["printing_job_id"], self.printing_job.id) self.assertEqual(result["customer_id"], self.customer.id) self.assertIsNone(result["shipment_id"]) # 待分配状态 self.assertEqual(result["created_by_id"], self.user.id) def test_create_sales_item_without_customer_id(self): """测试不指定客户ID时自动从生产订单获取""" data = { "printing_job_id": self.printing_job.id, "name": "测试销售品无客户", "quantity": "100", "unit": 1, } resp = self.client.post("/api/v1/shipment/sales-items/", data, format="json") self.assertEqual(resp.status_code, status.HTTP_201_CREATED) result = resp.json() # 客户ID应自动从生产订单获取 self.assertEqual(result["customer_id"], self.customer.id) def test_create_sales_item_invalid_unit(self): """测试无效的单位值""" data = { "printing_job_id": self.printing_job.id, "name": "测试销售品", "quantity": "100", "unit": 99, # 无效单位 } resp = self.client.post("/api/v1/shipment/sales-items/", data, format="json") self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("unit", resp.json()) def test_create_sales_item_invalid_printing_job(self): """测试不存在的生产任务""" data = { "printing_job_id": 99999, "name": "测试销售品", "quantity": "100", "unit": 1, } resp = self.client.post("/api/v1/shipment/sales-items/", data, format="json") self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("detail", resp.json()) def test_create_sales_item_missing_required_fields(self): """测试缺少必填字段""" data = { "printing_job_id": self.printing_job.id, # 缺少 name, quantity, unit } resp = self.client.post("/api/v1/shipment/sales-items/", data, format="json") self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) result = resp.json() self.assertIn("name", result) self.assertIn("quantity", result) self.assertIn("unit", result) def test_create_sales_item_unauthenticated(self): """测试未认证访问""" self.client.logout() data = { "printing_job_id": self.printing_job.id, "name": "测试销售品", "quantity": "100", "unit": 1, } resp = self.client.post("/api/v1/shipment/sales-items/", data, format="json") self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED) class ShipmentDeliveryAPITestCase(APITestCase): def setUp(self): self.client = APIClient() self.merchant = basic_models.Merchant.objects.create( name="送货测试商户", type=basic_models.MerchantTypeEnum.FACTORY ) self.other_merchant = basic_models.Merchant.objects.create( name="其它送货商户", type=basic_models.MerchantTypeEnum.FACTORY ) self.user = User.objects.create_user( username="shipment_delivery_user", password="testpass123", ) self.employee = basic_models.Employee.objects.create( sys_user=self.user, merchant=self.merchant, name="送货测试员工", mobile="13800138111", status=basic_models.EmployeeStatusEnum.ACTIVE, ) self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name="送货客户A", mobile="13900139111", area="杭州", ) self.other_customer = basic_models.Customer.objects.create( merchant=self.other_merchant, name="送货客户B", mobile="13900139112", area="绍兴", ) self.shipment1 = shipment_models.Shipment.objects.create( merchant=self.merchant, customer=self.customer, shipment_date="2026-04-03", created_by=self.user, ) self.shipment2 = shipment_models.Shipment.objects.create( merchant=self.merchant, customer=self.customer, shipment_date="2026-04-04", created_by=self.user, external_id="EXT-SHP-001", ) self.shipment3 = shipment_models.Shipment.objects.create( merchant=self.merchant, customer=self.customer, shipment_date="2026-04-05", created_by=self.user, ) self.foreign_shipment = shipment_models.Shipment.objects.create( merchant=self.other_merchant, customer=self.other_customer, shipment_date="2026-04-06", created_by=self.user, ) self.other_delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant, driver_name="旧司机", vehicle_trip="TRIP-OLD", created_by=self.user, ) self.shipment3.delivery = self.other_delivery self.shipment3.save(update_fields=["delivery", "updated_at"]) self.client.force_authenticate(user=self.user) from shipment.services import modify_status modify_status( self.shipment1, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user, ) modify_status( self.shipment1, target_status=shipment_models.ShipmentStatus.APPROVED, operator=self.user, approved_by=self.user, ) modify_status( self.shipment2, target_status=shipment_models.ShipmentStatus.PUBLISHED, operator=self.user, ) modify_status( self.shipment2, target_status=shipment_models.ShipmentStatus.APPROVED, operator=self.user, approved_by=self.user, ) def test_create_delivery_with_shipments(self): resp = self.client.post( "/api/v1/shipment/deliveries/", { "driver_name": "张司机", "vehicle_trip": "KD-001", "contact_phone": "13800138000", "vehicle_capacity": "9.6米厢车", "remark": "先装车", "internal_remark": "注意对账", "shipment_order_ids": [self.shipment2.id, self.shipment1.id], "shipments": [self.shipment1.id, self.shipment2.id], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_201_CREATED) data = resp.json() self.assertEqual(data["driver_name"], "张司机") self.assertEqual(data["vehicle_trip"], "KD-001") self.assertEqual(data["contact_phone"], "13800138000") self.assertEqual(data["vehicle_capacity"], "9.6米厢车") self.assertEqual(data["remark"], "先装车") self.assertEqual(data["internal_remark"], "注意对账") self.assertEqual(data["shipment_order_ids"], [self.shipment2.id, self.shipment1.id]) self.assertEqual(data["status"], shipment_models.ShipmentDeliveryStatus.PENDING) self.assertEqual(data["shipments_count"], 2) self.assertEqual(data["merchant_id"], self.merchant.id) self.assertEqual(data["created_by_id"], self.user.id) self.assertEqual(data["operator_id"], self.employee.id) self.assertEqual(data["operator_name"], self.employee.name) self.shipment1.refresh_from_db() self.shipment2.refresh_from_db() self.assertIsNotNone(self.shipment1.delivery_id) self.assertEqual(self.shipment1.delivery_id, self.shipment2.delivery_id) def test_create_delivery_defaults_shipment_order_ids_to_empty_list(self): resp = self.client.post( "/api/v1/shipment/deliveries/", { "driver_name": "张司机", "vehicle_trip": "KD-001-A", "shipments": [self.shipment1.id], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_201_CREATED) self.assertEqual(resp.json()["shipment_order_ids"], []) def test_create_delivery_rejects_non_array_shipment_order_ids(self): resp = self.client.post( "/api/v1/shipment/deliveries/", { "driver_name": "张司机", "vehicle_trip": "KD-001-B", "shipment_order_ids": {"bad": "value"}, }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("shipment_order_ids", resp.json()) def test_create_delivery_rejects_shipment_bound_to_other_delivery(self): resp = self.client.post( "/api/v1/shipment/deliveries/", { "driver_name": "张司机", "vehicle_trip": "KD-002", "shipments": [self.shipment3.id], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("已关联到其他送货单", resp.json()["detail"]) def test_create_delivery_rejects_non_approved_shipment(self): unapproved_shipment = shipment_models.Shipment.objects.create( merchant=self.merchant, customer=self.customer, shipment_date="2026-04-07", created_by=self.user, ) resp = self.client.post( "/api/v1/shipment/deliveries/", { "driver_name": "张司机", "vehicle_trip": "KD-002-A", "shipments": [unapproved_shipment.id], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("未处于已审核状态", resp.json()["detail"]) def test_create_delivery_rejects_foreign_merchant_shipment(self): resp = self.client.post( "/api/v1/shipment/deliveries/", { "driver_name": "张司机", "vehicle_trip": "KD-003", "shipments": [self.foreign_shipment.id], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("不属于当前商户", resp.json()["detail"]) def test_list_deliveries_supports_status_driver_and_vehicle_filters(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant, driver_name="李司机", vehicle_trip="TRIP-001", status=shipment_models.ShipmentDeliveryStatus.IN_TRANSIT, created_by=self.user, ) shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant, driver_name="王司机", vehicle_trip="TRIP-002", status=shipment_models.ShipmentDeliveryStatus.PENDING, created_by=self.user, ) resp = self.client.get( f"/api/v1/shipment/deliveries/?status={shipment_models.ShipmentDeliveryStatus.IN_TRANSIT}&driver_name=李&vehicle_trip=001" ) self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertEqual(data["count"], 1) self.assertEqual(data["results"][0]["id"], delivery.id) def test_detail_returns_nested_shipment_summaries(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant, driver_name="张司机", vehicle_trip="KD-004", created_by=self.user, ) self.shipment1.delivery = delivery self.shipment1.save(update_fields=["delivery", "updated_at"]) self.shipment2.delivery = delivery self.shipment2.save(update_fields=["delivery", "updated_at"]) resp = self.client.get(f"/api/v1/shipment/deliveries/{delivery.id}/") self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertEqual(data["shipments_count"], 2) shipment_ids = [item["id"] for item in data["shipments"]] self.assertEqual(shipment_ids, [self.shipment1.id, self.shipment2.id]) for item in data["shipments"]: self.assertIn("delivery_id", item) self.assertEqual(item["delivery_id"], delivery.id) self.assertIn("fabric", item) self.assertIn("order_description", item) first_shipment = next(item for item in data["shipments"] if item["id"] == self.shipment1.id) self.assertIsNone(first_shipment["fabric"]) self.assertIsNone(first_shipment["order_description"]) def test_patch_delivery_updates_fields_and_replaces_shipments(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant, driver_name="张司机", vehicle_trip="KD-005", created_by=self.user, ) self.shipment1.delivery = delivery self.shipment1.save(update_fields=["delivery", "updated_at"]) resp = self.client.patch( f"/api/v1/shipment/deliveries/{delivery.id}/", { "driver_name": "李司机", "vehicle_trip": "KD-005-B", "contact_phone": "13700137000", "vehicle_capacity": "13米高栏", "remark": "改派车辆", "internal_remark": "已电话确认", "shipment_order_ids": [self.shipment2.id], "shipments": [self.shipment2.id], }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertEqual(data["driver_name"], "李司机") self.assertEqual(data["vehicle_trip"], "KD-005-B") self.assertEqual(data["contact_phone"], "13700137000") self.assertEqual(data["vehicle_capacity"], "13米高栏") self.assertEqual(data["remark"], "改派车辆") self.assertEqual(data["internal_remark"], "已电话确认") self.assertEqual(data["shipment_order_ids"], [self.shipment2.id]) self.assertEqual(data["operator_id"], self.employee.id) self.assertEqual(data["shipments_count"], 1) self.assertEqual(data["shipments"][0]["id"], self.shipment2.id) self.shipment1.refresh_from_db() self.shipment2.refresh_from_db() self.assertIsNone(self.shipment1.delivery_id) self.assertEqual(self.shipment2.delivery_id, delivery.id) def test_patch_delivery_allows_null_shipment_order_ids(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant, driver_name="张司机", vehicle_trip="KD-005-N", shipment_order_ids=[self.shipment1.id], created_by=self.user, ) resp = self.client.patch( f"/api/v1/shipment/deliveries/{delivery.id}/", { "shipment_order_ids": None, }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) self.assertIsNone(resp.json()["shipment_order_ids"]) def test_patch_delivery_rejects_non_array_shipment_order_ids(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant, driver_name="张司机", vehicle_trip="KD-005-C", created_by=self.user, ) resp = self.client.patch( f"/api/v1/shipment/deliveries/{delivery.id}/", { "shipment_order_ids": "not-a-list", }, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("shipment_order_ids", resp.json()) def test_modify_delivery_status_sets_timestamps(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant, driver_name="张司机", vehicle_trip="KD-006", created_by=self.user, ) resp = self.client.post( f"/api/v1/shipment/deliveries/{delivery.id}/status/", {"status": shipment_models.ShipmentDeliveryStatus.IN_TRANSIT}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) delivery.refresh_from_db() self.assertEqual( delivery.status, shipment_models.ShipmentDeliveryStatus.IN_TRANSIT ) self.assertIsNotNone(delivery.started_at) self.assertIsNone(delivery.delivered_at) self.assertEqual(delivery.operator_id, self.employee.id) resp = self.client.post( f"/api/v1/shipment/deliveries/{delivery.id}/status/", {"status": shipment_models.ShipmentDeliveryStatus.DELIVERED}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) delivery.refresh_from_db() self.assertEqual( delivery.status, shipment_models.ShipmentDeliveryStatus.DELIVERED ) self.assertIsNotNone(delivery.delivered_at) self.assertEqual(delivery.operator_id, self.employee.id) def test_modify_delivery_status_rejects_invalid_transition(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant, driver_name="张司机", vehicle_trip="KD-007", created_by=self.user, ) resp = self.client.post( f"/api/v1/shipment/deliveries/{delivery.id}/status/", {"status": shipment_models.ShipmentDeliveryStatus.DELIVERED}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("不允许", resp.json()["detail"]) def test_modify_delivery_status_rejects_cancelled_target(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant, driver_name="张司机", vehicle_trip="KD-007-B", created_by=self.user, ) resp = self.client.post( f"/api/v1/shipment/deliveries/{delivery.id}/status/", {"status": shipment_models.ShipmentDeliveryStatus.CANCELLED}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("不是合法选项", str(resp.json())) def test_cancel_delivery_requires_permission(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant, driver_name="张司机", vehicle_trip="KD-007-C", created_by=self.user, ) resp = self.client.post( f"/api/v1/shipment/deliveries/{delivery.id}/cancel/", {}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN) self.assertIn("没有权限", resp.json()["detail"]) def test_cancel_delivery_success(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant, driver_name="张司机", vehicle_trip="KD-007-D", created_by=self.user, ) permission = Permission.objects.get(codename="cancel_shipmentdelivery") self.user.user_permissions.add(permission) resp = self.client.post( f"/api/v1/shipment/deliveries/{delivery.id}/cancel/", {}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertEqual(data["status"], shipment_models.ShipmentDeliveryStatus.CANCELLED) self.assertEqual(data["status_display"], "已取消") self.assertEqual(data["cancelled_by_id"], self.user.id) self.assertEqual(data["cancelled_by_name"], self.employee.name) self.assertEqual(data["operator_id"], self.employee.id) self.assertIsNotNone(data["cancelled_at"]) delivery.refresh_from_db() self.assertEqual(delivery.status, shipment_models.ShipmentDeliveryStatus.CANCELLED) self.assertEqual(delivery.cancelled_by_id, self.user.id) self.assertIsNotNone(delivery.cancelled_at) def test_bind_shipments_to_existing_delivery_success(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant, driver_name="张司机", vehicle_trip="KD-007-E", created_by=self.user, ) resp = self.client.post( f"/api/v1/shipment/deliveries/{delivery.id}/bind-shipments/", {"shipments": [self.shipment1.id, self.shipment2.id]}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_200_OK) data = resp.json() self.assertEqual(data["shipments_count"], 2) shipment_ids = [item["id"] for item in data["shipments"]] self.assertEqual(shipment_ids, [self.shipment1.id, self.shipment2.id]) self.assertEqual(data["operator_id"], self.employee.id) self.shipment1.refresh_from_db() self.shipment2.refresh_from_db() self.assertEqual(self.shipment1.delivery_id, delivery.id) self.assertEqual(self.shipment2.delivery_id, delivery.id) def test_bind_shipments_rejects_non_approved_shipment(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant, driver_name="张司机", vehicle_trip="KD-007-F", created_by=self.user, ) unapproved_shipment = shipment_models.Shipment.objects.create( merchant=self.merchant, customer=self.customer, shipment_date="2026-04-08", created_by=self.user, ) resp = self.client.post( f"/api/v1/shipment/deliveries/{delivery.id}/bind-shipments/", {"shipments": [unapproved_shipment.id]}, format="json", ) self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn("未处于已审核状态", resp.json()["detail"]) def test_delete_delivery_unbinds_shipments(self): delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant, driver_name="张司机", vehicle_trip="KD-008", created_by=self.user, ) self.shipment1.delivery = delivery self.shipment1.save(update_fields=["delivery", "updated_at"]) resp = self.client.delete(f"/api/v1/shipment/deliveries/{delivery.id}/") self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT) self.shipment1.refresh_from_db() self.assertIsNone(self.shipment1.delivery_id) self.assertFalse( shipment_models.ShipmentDelivery.objects.filter(id=delivery.id).exists() ) class ShipmentDeliveryServiceTestCase(TestCase): def setUp(self): self.merchant = basic_models.Merchant.objects.create( name="送货服务测试商户", type=basic_models.MerchantTypeEnum.FACTORY ) self.user = User.objects.create_user( username="shipment_delivery_service_user", password="testpass123", ) basic_models.Employee.objects.create( sys_user=self.user, merchant=self.merchant, name="送货服务测试员工", mobile="13800138112", status=basic_models.EmployeeStatusEnum.ACTIVE, ) self.delivery = shipment_models.ShipmentDelivery.objects.create( merchant=self.merchant, driver_name="服务司机", vehicle_trip="SERVICE-001", created_by=self.user, ) def test_modify_status_is_idempotent(self): from shipment.services import modify_shipment_delivery_status result = modify_shipment_delivery_status( self.delivery, target_status=shipment_models.ShipmentDeliveryStatus.PENDING, ) self.assertEqual(result.id, self.delivery.id) self.delivery.refresh_from_db() self.assertEqual( self.delivery.status, shipment_models.ShipmentDeliveryStatus.PENDING ) self.assertIsNone(self.delivery.started_at) def test_modify_status_cannot_rollback(self): from shipment.services import modify_shipment_delivery_status modify_shipment_delivery_status( self.delivery, target_status=shipment_models.ShipmentDeliveryStatus.IN_TRANSIT, ) with self.assertRaisesMessage( ValueError, "不允许将送货单状态从 送货中 修改为 待送货" ): modify_shipment_delivery_status( self.delivery, target_status=shipment_models.ShipmentDeliveryStatus.PENDING, ) def test_cancel_delivery_is_idempotent(self): from shipment.services import cancel_shipment_delivery result = cancel_shipment_delivery( self.delivery, cancelled_by=self.user, operator=self.user, ) first_cancelled_at = result.cancelled_at result = cancel_shipment_delivery( self.delivery, cancelled_by=self.user, operator=self.user, ) self.assertEqual(result.status, shipment_models.ShipmentDeliveryStatus.CANCELLED) self.assertEqual(result.cancelled_by_id, self.user.id) self.assertEqual(result.cancelled_at, first_cancelled_at)