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

3849 lines
146 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
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_by_printing_order_group_groups_by_printing_job(self):
url = (
f"/api/v1/shipment/sales-items/by-printing-order/{self.printing_order.id}/"
"group/"
)
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data["count"], 1)
self.assertEqual(len(data["results"]), 1)
group = data["results"][0]
self.assertEqual(group["printing_job_id"], self.printing_job1.id)
self.assertEqual(group["count"], 2)
self.assertEqual(
{item["id"] for item in group["items"]},
{self.sales_item1.id, self.sales_item2.id},
)
def test_get_sales_items_by_printing_order_group_supports_external_order_id(self):
response = self.client.get(
"/api/v1/shipment/sales-items/by-printing-order/EXT-PO-001/group/"
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data["count"], 1)
self.assertEqual(data["results"][0]["printing_job_id"], self.printing_job1.id)
def test_get_sales_items_by_printing_order_group_paginates_printing_job_groups(self):
url = (
f"/api/v1/shipment/sales-items/by-printing-order/{self.printing_order.id}/"
"group/?include_already_has_shipment=true&limit=1&offset=1"
)
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data["count"], 2)
self.assertEqual(len(data["results"]), 1)
self.assertEqual(data["results"][0]["printing_job_id"], self.printing_job2.id)
self.assertEqual(data["results"][0]["count"], 1)
self.assertEqual(data["results"][0]["items"][0]["id"], self.sales_item3.id)
def test_get_sales_items_by_printing_order_group_multiple_external_matches(self):
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-GROUP",
)
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-GROUP",
)
response = self.client.get(
"/api/v1/shipment/sales-items/by-printing-order/EXT-DUPLICATED-GROUP/group/"
)
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 ShipmentDeliveryByPrintingOrderAPITestCase(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="delivery_by_order_user", password="testpass123"
)
self.employee = basic_models.Employee.objects.create(
sys_user=self.user,
merchant=self.merchant,
name="送货查询员工",
status=basic_models.EmployeeStatusEnum.ACTIVE,
)
self.customer = basic_models.Customer.objects.create(
merchant=self.merchant,
name="送货查询客户",
created_by=self.employee,
)
self.state = stateflow_models.State.objects.create(name="待送货查询")
self.process = stateflow_models.Process.objects.create(name="送货查询流程")
self.process.replace_nodes([self.state])
self.product = basic_models.Product.objects.create(
merchant=self.merchant,
category=basic_models.ProductCategory.objects.create(
merchant=self.merchant,
name="送货查询分类",
),
name="送货查询产品",
)
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-DELIVERY-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=80,
unit="",
created_by=self.user,
)
self.delivery1 = shipment_models.ShipmentDelivery.objects.create(
merchant=self.merchant,
driver_name="张司机",
vehicle_trip="KD-001",
contact_phone="13800138000",
vehicle_capacity="9.6米厢车",
remark="外部备注A",
internal_remark="内部备注A",
status=shipment_models.ShipmentDeliveryStatus.IN_TRANSIT,
started_at=timezone.datetime(2026, 7, 1, 9, 0, tzinfo=timezone.get_current_timezone()),
created_by=self.user,
)
self.delivery2 = shipment_models.ShipmentDelivery.objects.create(
merchant=self.merchant,
driver_name="李司机",
vehicle_trip="KD-002",
contact_phone="13900139000",
vehicle_capacity="13米高栏",
remark="外部备注B",
internal_remark="内部备注B",
status=shipment_models.ShipmentDeliveryStatus.DELIVERED,
delivered_at=timezone.datetime(2026, 7, 3, 18, 0, tzinfo=timezone.get_current_timezone()),
created_by=self.user,
)
self.shipment1 = shipment_models.Shipment.objects.create(
merchant=self.merchant,
customer=self.customer,
shipment_date="2026-07-01",
delivery=self.delivery1,
created_by=self.user,
)
self.shipment2 = shipment_models.Shipment.objects.create(
merchant=self.merchant,
customer=self.customer,
shipment_date="2026-07-03",
delivery=self.delivery2,
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_job1.id,
shipment=self.shipment1,
customer_id=self.customer.id,
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.METER,
printing_job_id=self.printing_job1.id,
shipment=self.shipment1,
customer_id=self.customer.id,
created_by=self.user,
)
self.sales_item3 = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="送货销售品3",
quantity=Decimal("30.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=self.printing_job2.id,
shipment=self.shipment2,
customer_id=self.customer.id,
created_by=self.user,
)
other_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-OTHER-DELIVERY",
)
other_job = printing_models.PrintingJob.objects.create(
merchant=self.merchant,
printing_order=other_order,
product=self.product,
quantity=50,
unit="",
created_by=self.user,
)
other_delivery = shipment_models.ShipmentDelivery.objects.create(
merchant=self.merchant,
driver_name="王司机",
vehicle_trip="KD-999",
created_by=self.user,
)
other_shipment = shipment_models.Shipment.objects.create(
merchant=self.merchant,
customer=self.customer,
shipment_date="2026-07-04",
delivery=other_delivery,
created_by=self.user,
)
shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="其它订单销售品",
quantity=Decimal("40.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=other_job.id,
shipment=other_shipment,
customer_id=self.customer.id,
created_by=self.user,
)
self.client.force_authenticate(user=self.user)
def test_list_deliveries_by_internal_printing_order_id(self):
resp = self.client.get(
f"/api/v1/shipment/deliveries/by-printing-order/{self.printing_order.id}/"
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["count"], 2)
ids = {item["id"] for item in data["results"]}
self.assertEqual(ids, {self.delivery1.id, self.delivery2.id})
def test_list_deliveries_by_external_order_id(self):
resp = self.client.get(
"/api/v1/shipment/deliveries/by-printing-order/EXT-DELIVERY-001/"
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["count"], 2)
self.assertEqual(data["results"][0]["driver_name"], "李司机")
def test_list_deliveries_filters_driver_status_and_time_ranges(self):
resp = self.client.get(
f"/api/v1/shipment/deliveries/by-printing-order/{self.printing_order.id}/"
"?status=2&driver_name=张&vehicle_trip=KD-001&contact_phone=138"
"&vehicle_capacity=9.6&remark=外部备注A&internal_remark=内部备注A"
"&started_at_from=2026-07-01&started_at_to=2026-07-01"
"&shipment_date_from=2026-07-01&shipment_date_to=2026-07-01"
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["count"], 1)
self.assertEqual(data["results"][0]["id"], self.delivery1.id)
def test_list_deliveries_supports_pagination_and_ordering(self):
resp = self.client.get(
f"/api/v1/shipment/deliveries/by-printing-order/{self.printing_order.id}/"
"?ordering=id&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]["id"], self.delivery2.id)
def test_list_deliveries_duplicate_external_order_id_returns_500(self):
printing_models.PrintingOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
fabric="重复面料1",
width="150cm",
process=self.process,
created_by=self.user,
external_order_id="EXT-DELIVERY-DUP",
)
printing_models.PrintingOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
fabric="重复面料2",
width="150cm",
process=self.process,
created_by=self.user,
external_order_id="EXT-DELIVERY-DUP",
)
resp = self.client.get(
"/api/v1/shipment/deliveries/by-printing-order/EXT-DELIVERY-DUP/"
)
self.assertEqual(resp.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
self.assertIn("匹配到多个生产订单", resp.json()["message"])
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_filters_by_customer_name(self):
resp = self.client.get(
"/api/v1/shipment/sales-items/customers/?customer_name=客户A"
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["count"], 1)
self.assertEqual(data["results"][0]["customer_id"], self.customer1.id)
self.assertEqual(data["results"][0]["unshipped_sales_items_count"], 2)
def test_list_customers_with_unshipped_sales_items_filters_by_missing_customer_name(self):
resp = self.client.get(
"/api/v1/shipment/sales-items/customers/?customer_name=不存在"
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["count"], 0)
self.assertEqual(data["results"], [])
def test_list_customers_with_unshipped_sales_items_ignores_blank_customer_name(self):
resp = self.client.get(
"/api/v1/shipment/sales-items/customers/?customer_name=++"
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["count"], 2)
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_group_groups_by_external_order_id_and_name(self):
duplicate_sales_item = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="销售品1",
quantity=Decimal("11.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=self.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}/group/"
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["count"], 1)
self.assertEqual(len(data["results"]), 1)
external_group = data["results"][0]
self.assertEqual(external_group["external_order_id"], "EXT-CUST-001")
self.assertEqual(external_group["count"], 3)
by_name = {group["name"]: group for group in external_group["name_groups"]}
self.assertEqual(by_name["销售品1"]["count"], 2)
self.assertEqual(by_name["销售品2"]["count"], 1)
self.assertEqual(
{item["id"] for item in by_name["销售品1"]["items"]},
{self.sales_item1.id, duplicate_sales_item.id},
)
def test_get_sales_items_by_customer_group_paginates_external_order_id_groups_with_null_last(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,
)
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,
)
shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="无外部订单销售品",
quantity=Decimal("77.00"),
unit=shipment_models.UnitChoices.METER,
customer_id=self.customer.id,
created_by=self.user,
)
first_page = self.client.get(
f"/api/v1/shipment/sales-items/by-customer/{self.customer.id}/group/"
"?limit=1&offset=0"
)
second_page = self.client.get(
f"/api/v1/shipment/sales-items/by-customer/{self.customer.id}/group/"
"?limit=1&offset=1"
)
third_page = self.client.get(
f"/api/v1/shipment/sales-items/by-customer/{self.customer.id}/group/"
"?limit=1&offset=2"
)
self.assertEqual(first_page.status_code, status.HTTP_200_OK)
self.assertEqual(second_page.status_code, status.HTTP_200_OK)
self.assertEqual(third_page.status_code, status.HTTP_200_OK)
self.assertEqual(first_page.json()["count"], 3)
self.assertEqual(
first_page.json()["results"][0]["external_order_id"], "EXT-CUST-001"
)
self.assertEqual(
second_page.json()["results"][0]["external_order_id"], "EXT-CUST-OTHER"
)
self.assertIsNone(third_page.json()["results"][0]["external_order_id"])
def test_get_sales_items_by_customer_group_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}/group/"
"?external_order_id=EXT-CUST-OTHER"
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["count"], 1)
self.assertEqual(data["results"][0]["external_order_id"], "EXT-CUST-OTHER")
self.assertEqual(
data["results"][0]["name_groups"][0]["items"][0]["id"],
other_sales_item.id,
)
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": "华东" * 25,
"coordinates": "120.1551,30.2741",
"extra": {"source": "customer_address", "tags": ["snapshot"]},
"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", ""), "华东" * 25)
self.assertEqual(result["coordinates"], "120.1551,30.2741")
self.assertEqual(result["extra"], {"source": "customer_address", "tags": ["snapshot"]})
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"])
def test_create_shipment_with_address_id_keeps_address_snapshot(self):
customer_address = basic_models.CustomerAddress.objects.create(
merchant=self.merchant,
customer=self.customer,
address="客户地址表地址",
contact_name="地址联系人",
contact_phone="13900139000",
area="地址地区",
created_by=self.employee,
)
data = {
"customer": self.customer.id,
"address_id": customer_address.id,
"shipment_date": "2026-01-14",
"address": "出货单快照地址",
"contact_name": "快照联系人",
"contact_phone": "13800138000",
"area": "快照地区",
"coordinates": "121.0000,31.0000",
"extra": {"floor": "3F"},
"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.assertEqual(result["address_id"], customer_address.id)
self.assertEqual(result["address"], "出货单快照地址")
self.assertEqual(result["contact_name"], "快照联系人")
self.assertEqual(result["contact_phone"], "13800138000")
self.assertEqual(result["area"], "快照地区")
self.assertEqual(result["coordinates"], "121.0000,31.0000")
self.assertEqual(result["extra"], {"floor": "3F"})
shipment = shipment_models.Shipment.objects.get(id=result["id"])
self.assertEqual(shipment.customer_address_id, customer_address.id)
self.assertEqual(shipment.coordinates, "121.0000,31.0000")
self.assertEqual(shipment.extra, {"floor": "3F"})
def test_create_shipment_accepts_status(self):
data = {
"customer": self.customer.id,
"shipment_date": "2026-01-14",
"status": shipment_models.ShipmentStatus.APPROVED,
}
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["status"], shipment_models.ShipmentStatus.APPROVED)
self.assertEqual(result["status_display"], "已审核")
shipment = shipment_models.Shipment.objects.get(id=result["id"])
self.assertEqual(shipment.status, shipment_models.ShipmentStatus.APPROVED)
def test_create_shipment_rejects_mismatched_address_id(self):
other_customer = basic_models.Customer.objects.create(
merchant=self.merchant,
name="其它客户",
)
customer_address = basic_models.CustomerAddress.objects.create(
merchant=self.merchant,
customer=other_customer,
address="其它客户地址",
created_by=self.employee,
)
data = {
"customer": self.customer.id,
"address_id": customer_address.id,
"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_rejects_deleted_address_id(self):
customer_address = basic_models.CustomerAddress.objects.create(
merchant=self.merchant,
customer=self.customer,
address="已删除客户地址",
deleted_at=timezone.now(),
deleted_by=self.employee,
created_by=self.employee,
)
data = {
"customer": self.customer.id,
"address_id": customer_address.id,
"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"])
@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_allows_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_201_CREATED)
result = response.json()
self.assertEqual(result["items_count"], 2)
self.sales_item1.refresh_from_db()
self.sales_item_other_order.refresh_from_db()
self.assertEqual(self.sales_item1.shipment_id, result["id"])
self.assertEqual(self.sales_item_other_order.shipment_id, result["id"])
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="查询订单备注",
rolling_warn="查询滚筒预警",
)
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.assertIn("coordinates", result)
self.assertIn("extra", result)
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",
"coordinates": "120.2000,30.3000",
"extra": {"dock": "A"},
},
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.assertEqual(result["coordinates"], "120.2000,30.3000")
self.assertEqual(result["extra"], {"dock": "A"})
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")
self.assertEqual(self.shipment1.coordinates, "120.2000,30.3000")
self.assertEqual(self.shipment1.extra, {"dock": "A"})
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_allows_cancelled_shipment_without_delivery(self):
self.shipment1.status = shipment_models.ShipmentStatus.CANCELLED
self.shipment1.save(update_fields=["status", "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_200_OK)
self.assertEqual(resp.json()["area"], "取消后可修改")
self.assertEqual(resp.json()["status"], shipment_models.ShipmentStatus.CANCELLED)
self.shipment1.refresh_from_db()
self.assertEqual(self.shipment1.area, "取消后可修改")
self.assertEqual(self.shipment1.status, shipment_models.ShipmentStatus.CANCELLED)
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,
)
self.shipment1.delivery = delivery
self.shipment1.save(update_fields=["delivery", "updated_at"])
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.shipment1.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)
self.assertEqual(self.shipment1.delivery_id, delivery.id)
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)