zuowei/erpnew
1
0
Fork 0
forked from erp-dev/erp
Code Pull Requests Activity

feat: shipment patch

Browse Source
This commit is contained in:
colaftc
2026-04-02 18:07:06 +08:00
parent d36d9c1d9c
commit 1be840a8c2
14 changed files with 979 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

359
api_v1/views/shipment/test_api.py
View File

@@ -817,12 +817,70 @@ class ShipmentCreateAPITestCase(TestCase):
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,
)
@@ -831,6 +889,24 @@ class ShipmentCreateAPITestCase(TestCase):
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,
)
@@ -846,6 +922,7 @@ class ShipmentCreateAPITestCase(TestCase):
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,
)
@@ -875,6 +952,8 @@ class ShipmentCreateAPITestCase(TestCase):
self.assertEqual(result["shipment_date"], "2026-01-14")
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)
@@ -896,6 +975,8 @@ class ShipmentCreateAPITestCase(TestCase):
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):
"""测试客户不存在"""
@@ -935,6 +1016,32 @@ class ShipmentCreateAPITestCase(TestCase):
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()
@@ -1017,6 +1124,8 @@ class ShipmentExternalCreateAPITestCase(TestCase):
self.assertEqual(result["customer"], self.customer.id)
self.assertEqual(result["external_id"], "EXT-ORDER-001")
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)
@@ -1160,6 +1269,8 @@ class ShipmentQueryAPITestCase(TestCase):
item = next(it for it in data["results"] if it["id"] == self.shipment1.id)
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.assertIn("external_finished_products", item)
@@ -1172,6 +1283,8 @@ class ShipmentQueryAPITestCase(TestCase):
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.assertIn("external_finished_products", result)
@@ -1221,12 +1334,258 @@ class ShipmentQueryAPITestCase(TestCase):
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 SalesItemCreateAPITestCase(APITestCase):
"""销售品创建 API 测试"""