From fbcd27b2137ce24c7b71da468b6bbcf6aadaf533 Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Sun, 30 Aug 2026 10:49:36 +0800 Subject: [PATCH] feat(pl): require load products to belong to selected contract Co-authored-by: Cursor --- backend/app/services/pl_plan_stop_service.py | 39 +++++++ .../test_pl_plan_single_contract_rules.py | 101 +++++++++++++++++- 2 files changed, 138 insertions(+), 2 deletions(-) diff --git a/backend/app/services/pl_plan_stop_service.py b/backend/app/services/pl_plan_stop_service.py index e2c9f4aae..54f50c874 100644 --- a/backend/app/services/pl_plan_stop_service.py +++ b/backend/app/services/pl_plan_stop_service.py @@ -10,6 +10,7 @@ from fastapi import HTTPException from sqlalchemy.orm import Session from app.core.utils import generate_uuid +from app.models.crm_models import CrmContractProduct from app.models.pl_models import ( PlPlan, PlPlanContract, @@ -479,6 +480,43 @@ def assert_at_most_one_contract( return links +def assert_load_products_in_selected_contract( + db: Session, + tenant_no: str, + load: PlanLoadBody, +) -> None: + if is_internal_customer_name(load.customer_name): + return + links = _normalize_contract_links(load.contracts) + if not links: + return + contract_id = (links[0].contract_id or "").strip() + if not contract_id: + return + rows = ( + db.query(CrmContractProduct) + .filter( + CrmContractProduct.contract_id == contract_id, + CrmContractProduct.tenant_no == tenant_no, + CrmContractProduct.del_flag == 0, + ) + .all() + ) + allowed = { + str(r.product_id).strip() + for r in rows + if getattr(r, "product_id", None) and str(r.product_id).strip() + } + for prod in load.products or []: + pid = str(getattr(prod, "product_id", None) or "").strip() + name = (getattr(prod, "product_name", None) or pid or "未命名").strip() + if not pid or pid not in allowed: + raise HTTPException( + status_code=422, + detail=f"装货产品「{name}」不在所选合同明细中", + ) + + def _resolve_stop_contracts( db: Session, tenant_no: str, @@ -551,6 +589,7 @@ def validate_plan_stops_write( resolved_deliveries.append(delivery) enrich_stop_addresses_from_library(db, tenant_no, load, resolved_deliveries) + assert_load_products_in_selected_contract(db, tenant_no, load) return load, resolved_deliveries diff --git a/backend/tests/test_pl_plan_single_contract_rules.py b/backend/tests/test_pl_plan_single_contract_rules.py index b9d6aa6bf..e7264590b 100644 --- a/backend/tests/test_pl_plan_single_contract_rules.py +++ b/backend/tests/test_pl_plan_single_contract_rules.py @@ -1,8 +1,12 @@ import pytest from fastapi import HTTPException +from types import SimpleNamespace -from app.schemas.pl.plan import PlanContractLink -from app.services.pl_plan_stop_service import assert_at_most_one_contract +from app.schemas.pl.plan import PlanContractLink, PlanLoadBody, PlanStopProductBody +from app.services.pl_plan_stop_service import ( + assert_at_most_one_contract, + assert_load_products_in_selected_contract, +) def test_assert_at_most_one_contract_allows_zero_or_one(): @@ -20,3 +24,96 @@ def test_assert_at_most_one_contract_rejects_two(): assert_at_most_one_contract(links, "装货合同") assert exc.value.status_code == 422 assert "只能选择 1 份" in str(exc.value.detail) + + +class _FakeQuery: + def __init__(self, rows): + self._rows = rows + + def filter(self, *args, **kwargs): + return self + + def all(self): + return self._rows + + +class _FakeDb: + def __init__(self, product_ids): + self._rows = [SimpleNamespace(product_id=pid) for pid in product_ids] + + def query(self, model): + return _FakeQuery(self._rows) + + +def _load_with_products(product_ids, contract_id="c1"): + return PlanLoadBody( + customerId="cust-1", + customerName="外部客户A", + contracts=[PlanContractLink(contractId=contract_id, contractNo="N1")], + products=[ + PlanStopProductBody( + productId=pid, + productNo=pid, + productName=f"产品{pid}", + unit="吨", + planQty=1, + ) + for pid in product_ids + ], + ) + + +def test_assert_load_products_in_contract_passes_when_all_match(): + db = _FakeDb(["p1", "p2"]) + load = _load_with_products(["p1"]) + assert_load_products_in_selected_contract(db, "shanxia2.0", load) + + +def test_assert_load_products_in_contract_rejects_outside(): + db = _FakeDb(["p1"]) + load = _load_with_products(["p2"]) + with pytest.raises(HTTPException) as exc: + assert_load_products_in_selected_contract(db, "shanxia2.0", load) + assert exc.value.status_code == 422 + assert "不在所选合同明细中" in str(exc.value.detail) + + +def test_assert_load_products_skips_when_no_contract(): + db = _FakeDb([]) + load = PlanLoadBody( + customerId="cust-1", + customerName="外部客户A", + contracts=[], + products=[ + PlanStopProductBody( + productId="p1", + productNo="p1", + productName="产品p1", + unit="吨", + planQty=1, + ) + ], + ) + assert_load_products_in_selected_contract(db, "shanxia2.0", load) + + +def test_assert_load_products_skips_internal_customer(): + from app.services.pl_internal_customer import INTERNAL_CUSTOMER_NAMES + + internal = next(iter(INTERNAL_CUSTOMER_NAMES)) + db = _FakeDb([]) + load = PlanLoadBody( + customerId="cust-1", + customerName=internal, + contracts=[], + products=[ + PlanStopProductBody( + productId="p1", + productNo="p1", + productName="产品p1", + unit="吨", + planQty=1, + ) + ], + ) + assert_load_products_in_selected_contract(db, "shanxia2.0", load)