120 lines
3.4 KiB
Python
120 lines
3.4 KiB
Python
import pytest
|
|
from fastapi import HTTPException
|
|
from types import SimpleNamespace
|
|
|
|
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():
|
|
assert assert_at_most_one_contract([], "装货合同") == []
|
|
one = [PlanContractLink(contractId="c1", contractNo="N1")]
|
|
assert assert_at_most_one_contract(one, "装货合同") == one
|
|
|
|
|
|
def test_assert_at_most_one_contract_rejects_two():
|
|
links = [
|
|
PlanContractLink(contractId="c1", contractNo="N1"),
|
|
PlanContractLink(contractId="c2", contractNo="N2"),
|
|
]
|
|
with pytest.raises(HTTPException) as exc:
|
|
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)
|