zuowei/erpnew
1
0
Fork 0
forked from erp-dev/erp
Code Pull Requests Activity
Files
ebd687ccdd3b7b5f7be149303a438311b6dc1590
erpnew/api_v1/views/shipment/test_api.py

2696 lines
100 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 django.test import TestCase
from django.conf import settings
from django.contrib.auth.models import Permission
from django.utils import timezone
from rest_framework.test import APIClient, APITestCase
from rest_framework import status
from django.contrib.auth import get_user_model
from basic_info import models as basic_models
from printing import models as printing_models
from shipment import models as shipment_models
from stateflow import models as stateflow_models
User = get_user_model()
class SalesItemByPrintingOrderAPITestCase(TestCase):
"""测试通过生产订单查询销售品 API"""
def setUp(self):
self.client = APIClient()
# 创建商户
self.merchant = basic_models.Merchant.objects.create(
name="测试印花厂", type=basic_models.MerchantTypeEnum.FACTORY
)
# 创建用户
self.user = User.objects.create_user(
username="testuser", password="testpass123", email="test@example.com"
)
# 创建员工并关联商户
self.employee = basic_models.Employee.objects.create(
sys_user=self.user,
merchant=self.merchant,
name="测试员工",
mobile="13800138000",
status=basic_models.EmployeeStatusEnum.ACTIVE,
)
# 创建客户
self.customer = basic_models.Customer.objects.create(
merchant=self.merchant,
name="测试客户",
mobile="13900139000",
area="测试地区",
)
# 创建流程
self.state1 = stateflow_models.State.objects.create(name="待印染")
self.state2 = stateflow_models.State.objects.create(name="印染中")
self.state3 = stateflow_models.State.objects.create(name="已完成")
self.process = stateflow_models.Process.objects.create(name="印染流程")
self.process.replace_nodes([self.state1, self.state2, self.state3])
# 创建产品分类
self.category = basic_models.ProductCategory.objects.create(
merchant=self.merchant,
name="测试分类",
)
# 创建产品
self.product = basic_models.Product.objects.create(
merchant=self.merchant,
category=self.category,
name="测试产品",
human_id="TEST001",
)
# 创建印染订单
self.printing_order = printing_models.PrintingOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
fabric="测试面料",
width="150cm",
process=self.process,
created_by=self.user,
external_order_id="EXT-PO-001",
)
# 创建印染任务
self.printing_job1 = printing_models.PrintingJob.objects.create(
merchant=self.merchant,
printing_order=self.printing_order,
product=self.product,
quantity=100,
unit="米",
created_by=self.user,
)
self.printing_job2 = printing_models.PrintingJob.objects.create(
merchant=self.merchant,
printing_order=self.printing_order,
product=self.product,
quantity=200,
unit="米",
created_by=self.user,
)
# 创建出货单
self.shipment = shipment_models.Shipment.objects.create(
merchant=self.merchant,
customer=self.customer,
shipment_date="2026-01-14",
created_by=self.user,
)
# 创建销售品 - 未关联出货单
self.sales_item1 = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="销售品1",
quantity=Decimal("50.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=self.printing_job1.id,
created_by=self.user,
)
self.sales_item2 = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="销售品2",
quantity=Decimal("30.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=self.printing_job1.id,
position="A1-01",
remark="备注信息",
created_by=self.user,
)
# 创建销售品 - 已关联出货单
self.sales_item3 = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="销售品3（已出货）",
quantity=Decimal("100.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=self.printing_job2.id,
shipment=self.shipment,
created_by=self.user,
)
# 创建与该订单无关的销售品
self.sales_item_other = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="其它销售品",
quantity=Decimal("999.00"),
unit=shipment_models.UnitChoices.PIECE,
printing_job_id=99999, # 不存在的 job
created_by=self.user,
)
# 认证用户
self.client.force_authenticate(user=self.user)
def test_get_sales_items_by_printing_order_exclude_shipped(self):
"""测试查询销售品 - 默认不包含已出货的"""
url = (
f"/api/v1/shipment/sales-items/by-printing-order/{self.printing_order.id}/"
)
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
# 应该只返回2个未出货的销售品
self.assertEqual(data["count"], 2)
# 检查返回的销售品
item_ids = [item["id"] for item in data["results"]]
self.assertIn(self.sales_item1.id, item_ids)
self.assertIn(self.sales_item2.id, item_ids)
self.assertNotIn(self.sales_item3.id, item_ids) # 已出货的不应该在列表中
self.assertNotIn(self.sales_item_other.id, item_ids) # 其它订单的也不在
def test_get_sales_items_by_printing_order_include_shipped(self):
"""测试查询销售品 - 包含已出货的"""
url = f"/api/v1/shipment/sales-items/by-printing-order/{self.printing_order.id}/?include_already_has_shipment=true"
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
# 应该返回3个销售品（包含已出货的）
self.assertEqual(data["count"], 3)
# 检查返回的销售品
item_ids = [item["id"] for item in data["results"]]
self.assertIn(self.sales_item1.id, item_ids)
self.assertIn(self.sales_item2.id, item_ids)
self.assertIn(self.sales_item3.id, item_ids) # 已出货的也应该在列表中
self.assertNotIn(self.sales_item_other.id, item_ids) # 其它订单的依然不在
def test_get_sales_items_response_format(self):
"""测试返回数据格式"""
url = (
f"/api/v1/shipment/sales-items/by-printing-order/{self.printing_order.id}/"
)
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
# 找到 sales_item2（包含 position 和 remark）
item = next(
item for item in data["results"] if item["id"] == self.sales_item2.id
)
# 检查所有字段
self.assertEqual(item["name"], "销售品2")
self.assertEqual(Decimal(item["quantity"]), Decimal("30.00"))
self.assertEqual(item["unit"], shipment_models.UnitChoices.METER)
self.assertEqual(item["unit_display"], "米")
self.assertEqual(item["position"], "A1-01")
self.assertEqual(item["remark"], "备注信息")
self.assertEqual(item["printing_job_id"], self.printing_job1.id)
self.assertEqual(item["printing_order_id"], self.printing_order.id)
self.assertEqual(item["external_order_id"], "EXT-PO-001")
self.assertIsNone(item["shipment_id"])
self.assertIsNone(item["shipment_date"])
self.assertIsNotNone(item["created_at"])
self.assertEqual(item["created_by_id"], self.user.id)
def test_get_sales_items_shipped_item_format(self):
"""测试已出货的销售品返回格式"""
url = f"/api/v1/shipment/sales-items/by-printing-order/{self.printing_order.id}/?include_already_has_shipment=true"
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
# 找到已出货的销售品
item = next(
item for item in data["results"] if item["id"] == self.sales_item3.id
)
# 检查出货单信息
self.assertEqual(item["shipment_id"], self.shipment.id)
self.assertEqual(item["shipment_date"], "2026-01-14")
def test_get_sales_items_by_external_order_id(self):
"""测试通过 external_order_id 查询销售品"""
url = "/api/v1/shipment/sales-items/by-printing-order/EXT-PO-001/"
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data["count"], 2)
item_ids = [item["id"] for item in data["results"]]
self.assertIn(self.sales_item1.id, item_ids)
self.assertIn(self.sales_item2.id, item_ids)
self.assertNotIn(self.sales_item3.id, item_ids)
def test_get_sales_items_by_external_order_id_multiple_matches(self):
"""测试 external_order_id 匹配多个生产订单时报错"""
other_order = printing_models.PrintingOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
fabric="测试面料3",
width="160cm",
process=self.process,
created_by=self.user,
external_order_id="EXT-DUPLICATED",
)
printing_models.PrintingOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
fabric="测试面料4",
width="170cm",
process=self.process,
created_by=self.user,
external_order_id="EXT-DUPLICATED",
)
printing_models.PrintingJob.objects.create(
merchant=self.merchant,
printing_order=other_order,
product=self.product,
quantity=50,
unit="米",
created_by=self.user,
)
response = self.client.get(
"/api/v1/shipment/sales-items/by-printing-order/EXT-DUPLICATED/"
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("匹配到多个生产订单", response.json()["detail"])
def test_get_sales_items_printing_order_not_found(self):
"""测试生产订单不存在"""
url = "/api/v1/shipment/sales-items/by-printing-order/99999/"
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
self.assertIn("不存在", response.json()["detail"])
def test_get_sales_items_empty_result(self):
"""测试生产订单没有关联销售品"""
# 创建一个没有销售品的订单
empty_order = printing_models.PrintingOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
fabric="测试面料2",
width="150cm",
process=self.process,
created_by=self.user,
)
url = f"/api/v1/shipment/sales-items/by-printing-order/{empty_order.id}/"
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
data = response.json()
self.assertEqual(data["count"], 0)
self.assertEqual(data["results"], [])
def test_get_sales_items_by_printing_order_excludes_soft_deleted(self):
self.sales_item2.delete_at = timezone.now()
self.sales_item2.delete_by = self.user
self.sales_item2.save(update_fields=["delete_at", "delete_by", "updated_at"])
url = (
f"/api/v1/shipment/sales-items/by-printing-order/{self.printing_order.id}/"
)
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
item_ids = [item["id"] for item in response.json()["results"]]
self.assertIn(self.sales_item1.id, item_ids)
self.assertNotIn(self.sales_item2.id, item_ids)
def test_get_sales_items_unauthenticated(self):
"""测试未认证用户"""
self.client.logout()
url = (
f"/api/v1/shipment/sales-items/by-printing-order/{self.printing_order.id}/"
)
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
class ShipmentSalesItemCustomersAPITestCase(TestCase):
"""测试由未出货销售品汇总客户 API"""
def setUp(self):
self.client = APIClient()
self.merchant = basic_models.Merchant.objects.create(
name="测试印花厂", type=basic_models.MerchantTypeEnum.FACTORY
)
self.other_merchant = basic_models.Merchant.objects.create(
name="其它商户", type=basic_models.MerchantTypeEnum.FACTORY
)
self.user = User.objects.create_user(
username="shipment_customer_user",
password="testpass123",
email="shipment_customer@example.com",
)
self.employee = basic_models.Employee.objects.create(
sys_user=self.user,
merchant=self.merchant,
name="测试员工",
mobile="13800138020",
status=basic_models.EmployeeStatusEnum.ACTIVE,
)
self.customer1 = basic_models.Customer.objects.create(
merchant=self.merchant,
name="客户A",
mobile="13900139020",
area="杭州",
)
self.customer2 = basic_models.Customer.objects.create(
merchant=self.merchant,
name="客户B",
mobile="13900139021",
area="绍兴",
)
self.other_customer = basic_models.Customer.objects.create(
merchant=self.other_merchant,
name="客户C",
mobile="13900139022",
area="苏州",
)
shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="客户A销售品1",
quantity=Decimal("10.00"),
unit=shipment_models.UnitChoices.METER,
customer_id=self.customer1.id,
created_by=self.user,
)
shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="客户A销售品2",
quantity=Decimal("20.00"),
unit=shipment_models.UnitChoices.METER,
customer_id=self.customer1.id,
created_by=self.user,
)
shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="客户B销售品1",
quantity=Decimal("30.00"),
unit=shipment_models.UnitChoices.METER,
customer_id=self.customer2.id,
created_by=self.user,
)
shipped = shipment_models.Shipment.objects.create(
merchant=self.merchant,
customer=self.customer1,
shipment_date="2026-01-14",
created_by=self.user,
)
shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="客户A已出货销售品",
quantity=Decimal("40.00"),
unit=shipment_models.UnitChoices.METER,
customer_id=self.customer1.id,
shipment=shipped,
created_by=self.user,
)
shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="无客户销售品",
quantity=Decimal("50.00"),
unit=shipment_models.UnitChoices.METER,
created_by=self.user,
)
shipment_models.SalesItem.objects.create(
merchant=self.other_merchant,
name="其它商户销售品",
quantity=Decimal("60.00"),
unit=shipment_models.UnitChoices.METER,
customer_id=self.other_customer.id,
created_by=self.user,
)
self.client.force_authenticate(user=self.user)
def test_list_customers_with_unshipped_sales_items(self):
resp = self.client.get("/api/v1/shipment/sales-items/customers/")
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["count"], 2)
self.assertIn("next", data)
self.assertIn("previous", data)
by_customer_id = {
item["customer_id"]: item
for item in data["results"]
}
self.assertEqual(set(by_customer_id.keys()), {self.customer1.id, self.customer2.id})
self.assertEqual(by_customer_id[self.customer1.id]["customer_name"], "客户A")
self.assertEqual(by_customer_id[self.customer1.id]["unshipped_sales_items_count"], 2)
self.assertEqual(by_customer_id[self.customer2.id]["unshipped_sales_items_count"], 1)
def test_list_customers_with_unshipped_sales_items_supports_limit_offset(self):
resp = self.client.get("/api/v1/shipment/sales-items/customers/?limit=1&offset=1")
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["count"], 2)
self.assertEqual(len(data["results"]), 1)
self.assertEqual(data["results"][0]["customer_id"], self.customer2.id)
def test_list_customers_with_unshipped_sales_items_unauthenticated(self):
self.client.logout()
resp = self.client.get("/api/v1/shipment/sales-items/customers/")
self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED)
class SalesItemByCustomerAPITestCase(TestCase):
"""测试按客户查询销售品 API"""
def setUp(self):
self.client = APIClient()
self.merchant = basic_models.Merchant.objects.create(
name="测试印花厂", type=basic_models.MerchantTypeEnum.FACTORY
)
self.other_merchant = basic_models.Merchant.objects.create(
name="其它印花厂", type=basic_models.MerchantTypeEnum.FACTORY
)
self.user = User.objects.create_user(
username="sales_item_by_customer_user",
password="testpass123",
email="sales_item_by_customer@example.com",
)
self.employee = basic_models.Employee.objects.create(
sys_user=self.user,
merchant=self.merchant,
name="测试员工",
mobile="13800138030",
status=basic_models.EmployeeStatusEnum.ACTIVE,
)
self.customer = basic_models.Customer.objects.create(
merchant=self.merchant,
name="测试客户",
mobile="13900139030",
area="杭州",
)
self.other_customer = basic_models.Customer.objects.create(
merchant=self.merchant,
name="其它客户",
mobile="13900139031",
area="绍兴",
)
self.foreign_customer = basic_models.Customer.objects.create(
merchant=self.other_merchant,
name="外部客户",
mobile="13900139032",
area="苏州",
)
self.state = stateflow_models.State.objects.create(name="待生产")
self.process = stateflow_models.Process.objects.create(name="生产流程")
self.process.replace_nodes([self.state])
self.category = basic_models.ProductCategory.objects.create(
merchant=self.merchant,
name="测试分类",
)
self.product = basic_models.Product.objects.create(
merchant=self.merchant,
category=self.category,
name="测试产品",
human_id="PROD001",
)
self.printing_order = printing_models.PrintingOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
fabric="测试面料",
width="150cm",
process=self.process,
created_by=self.user,
external_order_id="EXT-CUST-001",
)
self.printing_job = printing_models.PrintingJob.objects.create(
merchant=self.merchant,
printing_order=self.printing_order,
product=self.product,
quantity=100,
unit="米",
created_by=self.user,
)
self.sales_item1 = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="销售品1",
quantity=Decimal("10.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=self.printing_job.id,
customer_id=self.customer.id,
position="A1-01",
created_by=self.user,
)
self.sales_item2 = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="销售品2",
quantity=Decimal("20.00"),
unit=shipment_models.UnitChoices.PIECE,
printing_job_id=self.printing_job.id,
customer_id=self.customer.id,
created_by=self.user,
)
self.shipment = shipment_models.Shipment.objects.create(
merchant=self.merchant,
customer=self.customer,
shipment_date="2026-01-14",
created_by=self.user,
)
self.sales_item_shipped = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="已出货销售品",
quantity=Decimal("30.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=self.printing_job.id,
customer_id=self.customer.id,
shipment=self.shipment,
created_by=self.user,
)
shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="其它客户销售品",
quantity=Decimal("40.00"),
unit=shipment_models.UnitChoices.METER,
customer_id=self.other_customer.id,
created_by=self.user,
)
self.client.force_authenticate(user=self.user)
def test_get_sales_items_by_customer_exclude_shipped(self):
resp = self.client.get(
f"/api/v1/shipment/sales-items/by-customer/{self.customer.id}/"
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["count"], 2)
self.assertEqual(len(data["results"]), 2)
item = next(it for it in data["results"] if it["id"] == self.sales_item1.id)
self.assertEqual(item["customer_id"], self.customer.id)
self.assertEqual(item["customer_name"], self.customer.name)
self.assertEqual(item["printing_job_id"], self.printing_job.id)
self.assertEqual(item["printing_order_id"], self.printing_order.id)
self.assertEqual(item["external_order_id"], "EXT-CUST-001")
self.assertEqual(item["position"], "A1-01")
def test_get_sales_items_by_customer_include_shipped_and_paginate(self):
resp = self.client.get(
f"/api/v1/shipment/sales-items/by-customer/{self.customer.id}/"
"?include_already_has_shipment=true&limit=2&offset=1"
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["count"], 3)
self.assertEqual(len(data["results"]), 2)
result_ids = [item["id"] for item in data["results"]]
self.assertEqual(result_ids, [self.sales_item2.id, self.sales_item_shipped.id])
def test_get_sales_items_by_customer_supports_external_order_id_filter(self):
other_printing_order = printing_models.PrintingOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
fabric="其它面料",
width="160cm",
process=self.process,
created_by=self.user,
external_order_id="EXT-CUST-OTHER",
)
other_printing_job = printing_models.PrintingJob.objects.create(
merchant=self.merchant,
printing_order=other_printing_order,
product=self.product,
quantity=50,
unit="米",
created_by=self.user,
)
other_sales_item = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="其它外部订单销售品",
quantity=Decimal("66.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=other_printing_job.id,
customer_id=self.customer.id,
created_by=self.user,
)
resp = self.client.get(
f"/api/v1/shipment/sales-items/by-customer/{self.customer.id}/"
"?external_order_id=EXT-CUST-001"
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
result_ids = [item["id"] for item in data["results"]]
self.assertEqual(set(result_ids), {self.sales_item1.id, self.sales_item2.id})
self.assertNotIn(other_sales_item.id, result_ids)
def test_get_sales_items_by_customer_not_found(self):
resp = self.client.get("/api/v1/shipment/sales-items/by-customer/99999/")
self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
def test_get_sales_items_by_customer_excludes_soft_deleted(self):
self.sales_item2.delete_at = timezone.now()
self.sales_item2.delete_by = self.user
self.sales_item2.save(update_fields=["delete_at", "delete_by", "updated_at"])
resp = self.client.get(
f"/api/v1/shipment/sales-items/by-customer/{self.customer.id}/"
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
result_ids = [item["id"] for item in resp.json()["results"]]
self.assertIn(self.sales_item1.id, result_ids)
self.assertNotIn(self.sales_item2.id, result_ids)
def test_get_sales_items_by_customer_other_merchant_404(self):
resp = self.client.get(
f"/api/v1/shipment/sales-items/by-customer/{self.foreign_customer.id}/"
)
self.assertEqual(resp.status_code, status.HTTP_404_NOT_FOUND)
class SalesItemDetailAPITestCase(APITestCase):
"""销售品详情 API 测试"""
def setUp(self):
self.merchant = basic_models.Merchant.objects.create(
name="详情测试商户", type=basic_models.MerchantTypeEnum.FACTORY
)
self.other_merchant = basic_models.Merchant.objects.create(
name="其它详情测试商户", type=basic_models.MerchantTypeEnum.FACTORY
)
self.user = User.objects.create_user(
username="sales_item_detail_user",
password="testpass123",
email="sales_item_detail@example.com",
)
self.employee = basic_models.Employee.objects.create(
sys_user=self.user,
merchant=self.merchant,
name="详情测试员工",
mobile="13800138031",
status=basic_models.EmployeeStatusEnum.ACTIVE,
)
self.other_user = User.objects.create_user(
username="sales_item_detail_other_user",
password="testpass123",
email="sales_item_detail_other@example.com",
)
self.other_employee = basic_models.Employee.objects.create(
sys_user=self.other_user,
merchant=self.other_merchant,
name="其它详情测试员工",
mobile="13800138032",
status=basic_models.EmployeeStatusEnum.ACTIVE,
)
self.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.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)
class ShipmentCreateAPITestCase(TestCase):
"""测试创建出货单 API"""
def setUp(self):
self.client = APIClient()
# 创建商户
self.merchant = basic_models.Merchant.objects.create(
name="测试印花厂", type=basic_models.MerchantTypeEnum.FACTORY
)
# 创建用户
self.user = User.objects.create_user(
username="testuser", password="testpass123", email="test@example.com"
)
# 创建员工并关联商户
self.employee = basic_models.Employee.objects.create(
sys_user=self.user,
merchant=self.merchant,
name="测试员工",
mobile="13800138000",
status=basic_models.EmployeeStatusEnum.ACTIVE,
)
# 创建客户
self.customer = basic_models.Customer.objects.create(
merchant=self.merchant,
name="测试客户",
mobile="13900139000",
area="测试地区",
)
self.state1 = stateflow_models.State.objects.create(name="待印染")
self.state2 = stateflow_models.State.objects.create(name="印染中")
self.process = stateflow_models.Process.objects.create(name="印染流程")
self.process.replace_nodes([self.state1, self.state2])
self.category = basic_models.ProductCategory.objects.create(
merchant=self.merchant,
name="测试分类",
)
self.product = basic_models.Product.objects.create(
merchant=self.merchant,
category=self.category,
name="测试产品",
human_id="SHIPTEST001",
)
self.printing_order = printing_models.PrintingOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
fabric="测试面料",
width="150cm",
process=self.process,
created_by=self.user,
)
self.other_printing_order = printing_models.PrintingOrder.objects.create(
merchant=self.merchant,
customer=self.customer,
fabric="测试面料2",
width="160cm",
process=self.process,
created_by=self.user,
)
self.printing_job1 = printing_models.PrintingJob.objects.create(
merchant=self.merchant,
printing_order=self.printing_order,
product=self.product,
quantity=100,
unit="米",
created_by=self.user,
)
self.printing_job2 = printing_models.PrintingJob.objects.create(
merchant=self.merchant,
printing_order=self.printing_order,
product=self.product,
quantity=80,
unit="米",
created_by=self.user,
)
self.printing_job_other = printing_models.PrintingJob.objects.create(
merchant=self.merchant,
printing_order=self.other_printing_order,
product=self.product,
quantity=60,
unit="米",
created_by=self.user,
)
# 创建销售品（未关联出货单）
self.sales_item1 = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="销售品1",
quantity=Decimal("50.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=self.printing_job1.id,
created_by=self.user,
)
self.sales_item2 = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="销售品2",
quantity=Decimal("30.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=self.printing_job2.id,
created_by=self.user,
)
self.sales_item_other_order = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="销售品4（不同生产单）",
quantity=Decimal("20.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=self.printing_job_other.id,
created_by=self.user,
)
self.sales_item_missing_job = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="销售品5（缺少生产任务）",
quantity=Decimal("10.00"),
unit=shipment_models.UnitChoices.METER,
created_by=self.user,
)
# 创建已关联出货单的销售品
self.existing_shipment = shipment_models.Shipment.objects.create(
merchant=self.merchant,
customer=self.customer,
shipment_date="2026-01-13",
created_by=self.user,
)
self.sales_item_shipped = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="销售品3（已出货）",
quantity=Decimal("100.00"),
unit=shipment_models.UnitChoices.METER,
printing_job_id=self.printing_job1.id,
shipment=self.existing_shipment,
created_by=self.user,
)
# 认证用户
self.client.force_authenticate(user=self.user)
def test_create_shipment_success(self):
"""测试成功创建出货单"""
data = {
"customer": self.customer.id,
"shipment_date": "2026-01-14",
"address": "杭州市测试路 1 号",
"contact_name": "张三",
"contact_phone": "13800138000",
"area": "华东",
"remark": "测试备注",
"sales_items": [self.sales_item1.id, self.sales_item2.id],
}
response = self.client.post("/api/v1/shipment/shipments/", data, format="json")
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
result = response.json()
# 验证返回数据
self.assertIn("id", result)
self.assertEqual(result["customer"], self.customer.id)
self.assertEqual(result["customer_name"], self.customer.name)
self.assertEqual(result["shipment_date"], "2026-01-14")
self.assertEqual(result["address"], "杭州市测试路 1 号")
self.assertEqual(result["contact_name"], "张三")
self.assertEqual(result["contact_phone"], "13800138000")
self.assertEqual(result.get("area", ""), "华东")
self.assertEqual(result["remark"], "测试备注")
self.assertEqual(result["status"], shipment_models.ShipmentStatus.DRAFT)
self.assertEqual(result["status_display"], "草稿(未发布)")
self.assertEqual(result["items_count"], 2)
self.assertEqual(result["created_by_id"], self.user.id)
# 验证销售品已关联到出货单
self.sales_item1.refresh_from_db()
self.sales_item2.refresh_from_db()
self.assertEqual(self.sales_item1.shipment_id, result["id"])
self.assertEqual(self.sales_item2.shipment_id, result["id"])
def test_create_shipment_without_sales_items(self):
"""测试创建出货单但不关联销售品"""
data = {
"customer": self.customer.id,
"shipment_date": "2026-01-14",
}
response = self.client.post("/api/v1/shipment/shipments/", data, format="json")
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
result = response.json()
self.assertEqual(result["items_count"], 0)
self.assertEqual(result["status"], shipment_models.ShipmentStatus.DRAFT)
self.assertEqual(result["status_display"], "草稿(未发布)")
def test_create_shipment_customer_not_found(self):
"""测试客户不存在"""
data = {
"customer": 99999,
"shipment_date": "2026-01-14",
}
response = self.client.post("/api/v1/shipment/shipments/", data, format="json")
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("不存在", response.json()["detail"])
def test_create_shipment_sales_item_not_found(self):
"""测试销售品不存在"""
data = {
"customer": self.customer.id,
"shipment_date": "2026-01-14",
"sales_items": [99999],
}
response = self.client.post("/api/v1/shipment/shipments/", data, format="json")
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("不存在", response.json()["detail"])
def test_create_shipment_sales_item_already_shipped(self):
"""测试销售品已关联到其他出货单"""
data = {
"customer": self.customer.id,
"shipment_date": "2026-01-14",
"sales_items": [self.sales_item_shipped.id],
}
response = self.client.post("/api/v1/shipment/shipments/", data, format="json")
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("已关联", response.json()["detail"])
def test_create_shipment_rejects_sales_items_from_different_printing_orders(self):
"""测试销售品来自不同生产订单时拒绝创建"""
data = {
"customer": self.customer.id,
"shipment_date": "2026-01-14",
"sales_items": [self.sales_item1.id, self.sales_item_other_order.id],
}
response = self.client.post("/api/v1/shipment/shipments/", data, format="json")
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("同一个生产订单", response.json()["detail"])
def test_create_shipment_rejects_sales_item_without_printing_job(self):
"""测试销售品缺少生产任务时拒绝创建"""
data = {
"customer": self.customer.id,
"shipment_date": "2026-01-14",
"sales_items": [self.sales_item_missing_job.id],
}
response = self.client.post("/api/v1/shipment/shipments/", data, format="json")
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("缺少关联生产任务", response.json()["detail"])
def test_create_shipment_unauthenticated(self):
"""测试未认证用户"""
self.client.logout()
data = {
"customer": self.customer.id,
"shipment_date": "2026-01-14",
}
response = self.client.post("/api/v1/shipment/shipments/", data, format="json")
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
class ShipmentExternalCreateAPITestCase(TestCase):
"""测试创建出货单 external 版 API"""
def setUp(self):
self.client = APIClient()
# 创建商户
self.merchant = basic_models.Merchant.objects.create(
name="测试印花厂", type=basic_models.MerchantTypeEnum.FACTORY
)
# 创建用户
self.user = User.objects.create_user(
username="testuser_ext",
password="testpass123",
email="test_ext@example.com",
)
# 创建员工并关联商户
self.employee = basic_models.Employee.objects.create(
sys_user=self.user,
merchant=self.merchant,
name="测试员工Ext",
mobile="13800138002",
status=basic_models.EmployeeStatusEnum.ACTIVE,
)
# 创建客户（需要同 merchant）
self.customer = basic_models.Customer.objects.create(
merchant=self.merchant,
name="测试客户Ext",
mobile="13900139002",
area="测试地区Ext",
)
# 创建一个销售品（用于验证 external 版不会绑定任何销售品）
self.sales_item = shipment_models.SalesItem.objects.create(
merchant=self.merchant,
name="销售品-不应被绑定",
quantity=Decimal("10.00"),
unit=shipment_models.UnitChoices.METER,
created_by=self.user,
)
self.client.force_authenticate(user=self.user)
def test_create_external_shipment_success(self):
data = {
"customer": self.customer.id,
"shipment_date": "2026-01-14",
"address": "绍兴市仓库 2 号",
"contact_name": "李四",
"contact_phone": "13900139000",
"area": "华南",
"remark": "external 备注",
"external_id": "EXT-ORDER-001",
"external_finished_products": [
{"style_name": "款式A", "num_of_rolls": 2, "remark": "A备注"},
{"style_name": "款式B", "num_of_rolls": 5},
],
}
response = self.client.post(
"/api/v1/shipment/shipments/external/", data, format="json"
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
result = response.json()
self.assertEqual(result["customer"], self.customer.id)
self.assertEqual(result["external_id"], "EXT-ORDER-001")
self.assertEqual(result["address"], "绍兴市仓库 2 号")
self.assertEqual(result["contact_name"], "李四")
self.assertEqual(result["contact_phone"], "13900139000")
self.assertEqual(result.get("area", ""), "华南")
self.assertEqual(result["status"], shipment_models.ShipmentStatus.DRAFT)
self.assertEqual(result["status_display"], "草稿(未发布)")
self.assertEqual(result["items_count"], 0)
self.assertEqual(result["external_finished_products_count"], 2)
shipment_id = result["id"]
# 验证外部成品表写入并关联
self.assertEqual(
shipment_models.ExternalFinishedProduct.objects.filter(
shipment_id=shipment_id
).count(),
2,
)
efp_a = shipment_models.ExternalFinishedProduct.objects.filter(
shipment_id=shipment_id, style_name="款式A"
).first()
self.assertIsNotNone(efp_a)
self.assertEqual(getattr(efp_a, "remark", None) or "", "A备注")
# 验证不会绑定任何销售品
self.sales_item.refresh_from_db()
self.assertIsNone(self.sales_item.shipment_id)
def test_create_external_shipment_external_id_required(self):
data = {
"customer": self.customer.id,
"shipment_date": "2026-01-14",
"external_id": " ",
"external_finished_products": [
{"style_name": "款式A", "num_of_rolls": 1},
],
}
response = self.client.post(
"/api/v1/shipment/shipments/external/", data, format="json"
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("external_id", response.json())
def test_create_external_shipment_products_required(self):
data = {
"customer": self.customer.id,
"shipment_date": "2026-01-14",
"external_id": "EXT-ORDER-002",
"external_finished_products": [],
}
response = self.client.post(
"/api/v1/shipment/shipments/external/", data, format="json"
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("external_finished_products", response.json())
def test_create_external_shipment_unauthenticated(self):
self.client.logout()
data = {
"customer": self.customer.id,
"shipment_date": "2026-01-14",
"external_id": "EXT-ORDER-003",
"external_finished_products": [
{"style_name": "款式A", "num_of_rolls": 1},
],
}
response = self.client.post(
"/api/v1/shipment/shipments/external/", data, format="json"
)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
class ShipmentQueryAPITestCase(TestCase):
"""测试出货单查询 API（列表/详情）"""
def setUp(self):
self.client = APIClient()
self.merchant1 = basic_models.Merchant.objects.create(
name="商户1", type=basic_models.MerchantTypeEnum.FACTORY
)
self.merchant2 = basic_models.Merchant.objects.create(
name="商户2", type=basic_models.MerchantTypeEnum.FACTORY
)
self.user1 = User.objects.create_user(
username="u1", password="pass123", email="u1@example.com"
)
self.emp1 = basic_models.Employee.objects.create(
sys_user=self.user1,
merchant=self.merchant1,
name="员工1",
mobile="13800138010",
status=basic_models.EmployeeStatusEnum.ACTIVE,
)
self.user2 = User.objects.create_user(
username="u2", password="pass123", email="u2@example.com"
)
self.emp2 = basic_models.Employee.objects.create(
sys_user=self.user2,
merchant=self.merchant2,
name="员工2",
mobile="13800138011",
status=basic_models.EmployeeStatusEnum.ACTIVE,
)
self.customer1 = basic_models.Customer.objects.create(
merchant=self.merchant1, name="客户1", mobile="13900139010", area="A"
)
self.customer2 = basic_models.Customer.objects.create(
merchant=self.merchant2, name="客户2", mobile="13900139011", area="B"
)
self.category1 = basic_models.ProductCategory.objects.create(
merchant=self.merchant1,
name="分类1",
)
self.product1 = basic_models.Product.objects.create(
merchant=self.merchant1,
category=self.category1,
name="产品1",
human_id="SHIPMENTQ001",
mdy_image_url="https://example.com/shipment-list-image.jpg",
)
self.state1 = stateflow_models.State.objects.create(name="查询状态")
self.process1 = stateflow_models.Process.objects.create(name="查询流程")
self.process1.replace_nodes([self.state1])
self.printing_order1 = printing_models.PrintingOrder.objects.create(
merchant=self.merchant1,
customer=self.customer1,
fabric="查询面料",
width="150cm",
process=self.process1,
created_by=self.user1,
external_order_id="QUERY-PO-001",
)
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.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_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.assertEqual(result["id"], self.shipment1.id)
self.assertIn("area", result)
self.assertEqual(result["area"], "A1")
self.assertEqual(result["status"], shipment_models.ShipmentStatus.DRAFT)
self.assertEqual(result["status_display"], "草稿(未发布)")
self.assertIn("sales_items", result)
self.assertIsInstance(result["sales_items"], list)
self.assertEqual(result["sales_items"][0]["product_image_url"], "https://example.com/shipment-list-image.jpg")
self.assertIn("external_finished_products", result)
self.assertIsInstance(result["external_finished_products"], list)
def test_patch_shipment_area_success(self):
"""
新增字段 area：支持更新（PATCH）并回显。
"""
resp = self.client.patch(
f"/api/v1/shipment/shipments/{self.shipment1.id}/",
data={
"area": "更新地区",
"address": "宁波市滨海路 9 号",
"contact_name": "王五",
"contact_phone": "13700137000",
},
format="json",
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
result = resp.json()
self.assertEqual(result["id"], self.shipment1.id)
self.assertEqual(result.get("area", ""), "更新地区")
self.assertEqual(result["address"], "宁波市滨海路 9 号")
self.assertEqual(result["contact_name"], "王五")
self.assertEqual(result["contact_phone"], "13700137000")
self.shipment1.refresh_from_db()
self.assertEqual(getattr(self.shipment1, "area", ""), "更新地区")
self.assertEqual(self.shipment1.address, "宁波市滨海路 9 号")
self.assertEqual(self.shipment1.contact_name, "王五")
self.assertEqual(self.shipment1.contact_phone, "13700137000")
def test_patch_shipment_rejects_non_draft(self):
"""
非草稿状态的出货单不允许修改业务数据。
"""
from shipment.services import modify_status
modify_status(
self.shipment1,
target_status=shipment_models.ShipmentStatus.PUBLISHED,
operator=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_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_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_allows_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,
)
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)
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
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()
self.assertEqual(self.shipment.status, shipment_models.ShipmentStatus.REJECTED)
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.product = basic_models.Product.objects.create(
merchant=self.merchant,
name="测试产品",
code="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": "注意对账",
"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["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_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])
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": "已电话确认",
"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["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_modify_delivery_status_sets_timestamps(self):
delivery = shipment_models.ShipmentDelivery.objects.create(
merchant=self.merchant,
driver_name="张司机",
vehicle_trip="KD-006",
created_by=self.user,
)
resp = self.client.post(
f"/api/v1/shipment/deliveries/{delivery.id}/status/",
{"status": shipment_models.ShipmentDeliveryStatus.IN_TRANSIT},
format="json",
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
delivery.refresh_from_db()
self.assertEqual(
delivery.status, shipment_models.ShipmentDeliveryStatus.IN_TRANSIT
)
self.assertIsNotNone(delivery.started_at)
self.assertIsNone(delivery.delivered_at)
self.assertEqual(delivery.operator_id, self.employee.id)
resp = self.client.post(
f"/api/v1/shipment/deliveries/{delivery.id}/status/",
{"status": shipment_models.ShipmentDeliveryStatus.DELIVERED},
format="json",
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
delivery.refresh_from_db()
self.assertEqual(
delivery.status, shipment_models.ShipmentDeliveryStatus.DELIVERED
)
self.assertIsNotNone(delivery.delivered_at)
self.assertEqual(delivery.operator_id, self.employee.id)
def test_modify_delivery_status_rejects_invalid_transition(self):
delivery = shipment_models.ShipmentDelivery.objects.create(
merchant=self.merchant,
driver_name="张司机",
vehicle_trip="KD-007",
created_by=self.user,
)
resp = self.client.post(
f"/api/v1/shipment/deliveries/{delivery.id}/status/",
{"status": shipment_models.ShipmentDeliveryStatus.DELIVERED},
format="json",
)
self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("不允许", resp.json()["detail"])
def test_modify_delivery_status_rejects_cancelled_target(self):
delivery = shipment_models.ShipmentDelivery.objects.create(
merchant=self.merchant,
driver_name="张司机",
vehicle_trip="KD-007-B",
created_by=self.user,
)
resp = self.client.post(
f"/api/v1/shipment/deliveries/{delivery.id}/status/",
{"status": shipment_models.ShipmentDeliveryStatus.CANCELLED},
format="json",
)
self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("不是合法选项", str(resp.json()))
def test_cancel_delivery_requires_permission(self):
delivery = shipment_models.ShipmentDelivery.objects.create(
merchant=self.merchant,
driver_name="张司机",
vehicle_trip="KD-007-C",
created_by=self.user,
)
resp = self.client.post(
f"/api/v1/shipment/deliveries/{delivery.id}/cancel/",
{},
format="json",
)
self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
self.assertIn("没有权限", resp.json()["detail"])
def test_cancel_delivery_success(self):
delivery = shipment_models.ShipmentDelivery.objects.create(
merchant=self.merchant,
driver_name="张司机",
vehicle_trip="KD-007-D",
created_by=self.user,
)
permission = Permission.objects.get(codename="cancel_shipmentdelivery")
self.user.user_permissions.add(permission)
resp = self.client.post(
f"/api/v1/shipment/deliveries/{delivery.id}/cancel/",
{},
format="json",
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["status"], shipment_models.ShipmentDeliveryStatus.CANCELLED)
self.assertEqual(data["status_display"], "已取消")
self.assertEqual(data["cancelled_by_id"], self.user.id)
self.assertEqual(data["cancelled_by_name"], self.employee.name)
self.assertEqual(data["operator_id"], self.employee.id)
self.assertIsNotNone(data["cancelled_at"])
delivery.refresh_from_db()
self.assertEqual(delivery.status, shipment_models.ShipmentDeliveryStatus.CANCELLED)
self.assertEqual(delivery.cancelled_by_id, self.user.id)
self.assertIsNotNone(delivery.cancelled_at)
def test_bind_shipments_to_existing_delivery_success(self):
delivery = shipment_models.ShipmentDelivery.objects.create(
merchant=self.merchant,
driver_name="张司机",
vehicle_trip="KD-007-E",
created_by=self.user,
)
resp = self.client.post(
f"/api/v1/shipment/deliveries/{delivery.id}/bind-shipments/",
{"shipments": [self.shipment1.id, self.shipment2.id]},
format="json",
)
self.assertEqual(resp.status_code, status.HTTP_200_OK)
data = resp.json()
self.assertEqual(data["shipments_count"], 2)
shipment_ids = [item["id"] for item in data["shipments"]]
self.assertEqual(shipment_ids, [self.shipment1.id, self.shipment2.id])
self.assertEqual(data["operator_id"], self.employee.id)
self.shipment1.refresh_from_db()
self.shipment2.refresh_from_db()
self.assertEqual(self.shipment1.delivery_id, delivery.id)
self.assertEqual(self.shipment2.delivery_id, delivery.id)
def test_bind_shipments_rejects_non_approved_shipment(self):
delivery = shipment_models.ShipmentDelivery.objects.create(
merchant=self.merchant,
driver_name="张司机",
vehicle_trip="KD-007-F",
created_by=self.user,
)
unapproved_shipment = shipment_models.Shipment.objects.create(
merchant=self.merchant,
customer=self.customer,
shipment_date="2026-04-08",
created_by=self.user,
)
resp = self.client.post(
f"/api/v1/shipment/deliveries/{delivery.id}/bind-shipments/",
{"shipments": [unapproved_shipment.id]},
format="json",
)
self.assertEqual(resp.status_code, status.HTTP_400_BAD_REQUEST)
self.assertIn("未处于已审核状态", resp.json()["detail"])
def test_delete_delivery_unbinds_shipments(self):
delivery = shipment_models.ShipmentDelivery.objects.create(
merchant=self.merchant,
driver_name="张司机",
vehicle_trip="KD-008",
created_by=self.user,
)
self.shipment1.delivery = delivery
self.shipment1.save(update_fields=["delivery", "updated_at"])
resp = self.client.delete(f"/api/v1/shipment/deliveries/{delivery.id}/")
self.assertEqual(resp.status_code, status.HTTP_204_NO_CONTENT)
self.shipment1.refresh_from_db()
self.assertIsNone(self.shipment1.delivery_id)
self.assertFalse(
shipment_models.ShipmentDelivery.objects.filter(id=delivery.id).exists()
)
class ShipmentDeliveryServiceTestCase(TestCase):
def setUp(self):
self.merchant = basic_models.Merchant.objects.create(
name="送货服务测试商户", type=basic_models.MerchantTypeEnum.FACTORY
)
self.user = User.objects.create_user(
username="shipment_delivery_service_user",
password="testpass123",
)
basic_models.Employee.objects.create(
sys_user=self.user,
merchant=self.merchant,
name="送货服务测试员工",
mobile="13800138112",
status=basic_models.EmployeeStatusEnum.ACTIVE,
)
self.delivery = shipment_models.ShipmentDelivery.objects.create(
merchant=self.merchant,
driver_name="服务司机",
vehicle_trip="SERVICE-001",
created_by=self.user,
)
def test_modify_status_is_idempotent(self):
from shipment.services import modify_shipment_delivery_status
result = modify_shipment_delivery_status(
self.delivery,
target_status=shipment_models.ShipmentDeliveryStatus.PENDING,
)
self.assertEqual(result.id, self.delivery.id)
self.delivery.refresh_from_db()
self.assertEqual(
self.delivery.status, shipment_models.ShipmentDeliveryStatus.PENDING
)
self.assertIsNone(self.delivery.started_at)
def test_modify_status_cannot_rollback(self):
from shipment.services import modify_shipment_delivery_status
modify_shipment_delivery_status(
self.delivery,
target_status=shipment_models.ShipmentDeliveryStatus.IN_TRANSIT,
)
with self.assertRaisesMessage(
ValueError, "不允许将送货单状态从 送货中 修改为 待送货"
):
modify_shipment_delivery_status(
self.delivery,
target_status=shipment_models.ShipmentDeliveryStatus.PENDING,
)
def test_cancel_delivery_is_idempotent(self):
from shipment.services import cancel_shipment_delivery
result = cancel_shipment_delivery(
self.delivery,
cancelled_by=self.user,
operator=self.user,
)
first_cancelled_at = result.cancelled_at
result = cancel_shipment_delivery(
self.delivery,
cancelled_by=self.user,
operator=self.user,
)
self.assertEqual(result.status, shipment_models.ShipmentDeliveryStatus.CANCELLED)
self.assertEqual(result.cancelled_by_id, self.user.id)
self.assertEqual(result.cancelled_at, first_cancelled_at)
View Git Blame Copy Permalink