shanxia2.0/backend/tests/test_pl_plan_single_contract_rules.py

import pytest
from fastapi import HTTPException
from types import SimpleNamespace

from app.schemas.pl.plan import (
    PlanContractLink,
    PlanDeliveryBody,
    PlanLoadBody,
    PlanStopProductBody,
)
from app.services.pl_internal_customer import INTERNAL_CUSTOMER_NAMES
from app.services.pl_plan_stop_service import (
    assert_at_most_one_contract,
    assert_delivery_products_in_selected_contract,
    assert_external_stop_products_have_contract_line,
    assert_load_products_in_selected_contract,
    assert_stop_contract_required,
)


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)


def test_assert_load_products_union_across_two_contracts():
    db = _FakeDb(["p1", "p2"])
    load = PlanLoadBody(
        customerId="cust-1",
        customerName="外部客户A",
        contracts=[
            PlanContractLink(contractId="c1", contractNo="N1"),
            PlanContractLink(contractId="c2", contractNo="N2"),
        ],
        products=[
            PlanStopProductBody(
                productId="p2",
                productNo="p2",
                productName="产品p2",
                unit="吨",
                planQty=1,
            )
        ],
    )
    assert_load_products_in_selected_contract(db, "shanxia2.0", load)


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=None, rows=None):
        if rows is not None:
            self._rows = rows
        else:
            self._rows = [SimpleNamespace(product_id=pid) for pid in product_ids or []]

    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_bound_contract_line_rejects_unit_different_from_contract():
    db = _FakeDb(
        rows=[
            SimpleNamespace(id="line-1", contract_id="c1", product_id="p1", unit="吨"),
        ]
    )
    load = PlanLoadBody(
        customerId="cust-1",
        customerName="外部客户A",
        contracts=[PlanContractLink(contractId="c1", contractNo="N1")],
        products=[
            PlanStopProductBody(
                productId="p1",
                productNo="p1",
                productName="工业硫酸",
                unit="kg",
                contractId="c1",
                contractProductId="line-1",
                planQty=30,
            )
        ],
    )
    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)
    assert "请按合同单位填写数量" in str(exc.value.detail)


def test_bound_contract_line_accepts_same_unit_label_and_code():
    db = _FakeDb(
        rows=[
            SimpleNamespace(id="line-1", contract_id="c1", product_id="p1", unit="dun"),
        ]
    )
    load = PlanLoadBody(
        customerId="cust-1",
        customerName="外部客户A",
        contracts=[PlanContractLink(contractId="c1", contractNo="N1")],
        products=[
            PlanStopProductBody(
                productId="p1",
                productNo="p1",
                productName="工业硫酸",
                unit="吨",
                contractId="c1",
                contractProductId="line-1",
                planQty=30,
            )
        ],
    )
    assert_load_products_in_selected_contract(db, "shanxia2.0", load)


def test_unbound_stop_product_keeps_its_own_unit():
    db = _FakeDb(
        rows=[
            SimpleNamespace(id="line-1", contract_id="c1", product_id="p1", unit="吨"),
        ]
    )
    load = PlanLoadBody(
        customerId="cust-1",
        customerName="外部客户A",
        contracts=[PlanContractLink(contractId="c1", contractNo="N1")],
        products=[
            PlanStopProductBody(
                productId="p1",
                productNo="p1",
                productName="工业硫酸",
                unit="kg",
                planQty=30,
            )
        ],
    )
    assert_load_products_in_selected_contract(db, "shanxia2.0", load)


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))
    # Empty/mismatched contract products — would 422 for external customer.
    db = _FakeDb([])
    load = PlanLoadBody(
        customerId="cust-1",
        customerName=internal,
        contracts=[PlanContractLink(contractId="c1", contractNo="N1")],
        products=[
            PlanStopProductBody(
                productId="p1",
                productNo="p1",
                productName="产品p1",
                unit="吨",
                planQty=1,
            )
        ],
    )
    assert_load_products_in_selected_contract(db, "shanxia2.0", load)


def test_assert_plan_products_outside_contract_raises():
    db = _FakeDb(["p1"])
    load = _load_with_products(["p1"])
    plan_products = [
        SimpleNamespace(product_id="p2", product_name="计划外产品"),
    ]
    with pytest.raises(HTTPException) as exc:
        assert_load_products_in_selected_contract(
            db, "shanxia2.0", load, plan_products=plan_products
        )
    assert exc.value.status_code == 422
    assert "计划外产品" in str(exc.value.detail)
    assert "不在所选合同明细中" in str(exc.value.detail)


def _delivery_with_products(product_ids, contract_id="c1"):
    return PlanDeliveryBody(
        seq=1,
        customerId="cust-2",
        customerName="外部客户B",
        contracts=[PlanContractLink(contractId=contract_id, contractNo="N2")],
        products=[
            PlanStopProductBody(
                productId=pid,
                productNo=pid,
                productName=f"产品{pid}",
                unit="吨",
                planQty=1,
            )
            for pid in product_ids
        ],
    )


def test_assert_delivery_products_in_contract_passes_when_all_match():
    db = _FakeDb(["p1", "p2"])
    delivery = _delivery_with_products(["p1"])
    assert_delivery_products_in_selected_contract(db, "shanxia2.0", delivery, 1)


def test_assert_delivery_products_in_contract_rejects_outside():
    db = _FakeDb(["p1"])
    delivery = _delivery_with_products(["p2"])
    with pytest.raises(HTTPException) as exc:
        assert_delivery_products_in_selected_contract(db, "shanxia2.0", delivery, 2)
    assert exc.value.status_code == 422
    assert "送货段#2产品" in str(exc.value.detail)
    assert "不在所选合同明细中" in str(exc.value.detail)


def test_assert_delivery_products_skips_when_no_contract():
    db = _FakeDb([])
    delivery = PlanDeliveryBody(
        seq=1,
        customerId="cust-2",
        customerName="外部客户B",
        contracts=[],
        products=[
            PlanStopProductBody(
                productId="p1",
                productNo="p1",
                productName="产品p1",
                unit="吨",
                planQty=1,
            )
        ],
    )
    assert_delivery_products_in_selected_contract(db, "shanxia2.0", delivery, 1)


def test_assert_delivery_products_skips_internal_customer():
    from app.services.pl_internal_customer import INTERNAL_CUSTOMER_NAMES

    internal = next(iter(INTERNAL_CUSTOMER_NAMES))
    db = _FakeDb([])
    delivery = PlanDeliveryBody(
        seq=1,
        customerId="cust-2",
        customerName=internal,
        contracts=[PlanContractLink(contractId="c1", contractNo="N1")],
        products=[
            PlanStopProductBody(
                productId="p1",
                productNo="p1",
                productName="产品p1",
                unit="吨",
                planQty=1,
            )
        ],
    )
    assert_delivery_products_in_selected_contract(db, "shanxia2.0", delivery, 1)


def test_formal_save_rejects_external_product_without_contract_line():
    load = _load_with_products(["p1"])
    with pytest.raises(HTTPException) as exc:
        assert_external_stop_products_have_contract_line(load, [])
    assert exc.value.status_code == 422
    assert exc.value.detail == "装货产品「产品p1」没有合同编号，请从当前合同补到原来的产品行后再保存"


def test_formal_save_accepts_external_product_bound_on_the_same_row():
    load = _load_with_products(["p1"])
    load.products[0].contract_id = "c1"
    load.products[0].contract_product_id = "line-1"
    assert assert_external_stop_products_have_contract_line(load, []) is None


def test_formal_save_skips_internal_product_without_contract_line():
    internal = next(iter(INTERNAL_CUSTOMER_NAMES))
    load = _load_with_products(["p1"])
    load.customer_name = internal
    assert assert_external_stop_products_have_contract_line(load, []) is None


def test_formal_save_names_the_delivery_segment_for_an_untagged_product():
    delivery = _delivery_with_products(["p1"])
    with pytest.raises(HTTPException) as exc:
        assert_external_stop_products_have_contract_line(
            _load_with_products([]),
            [delivery],
        )
    assert exc.value.status_code == 422
    assert "送货段#1产品" in str(exc.value.detail)


def test_assert_stop_contract_required_rejects_external_without_contract():
    with pytest.raises(HTTPException) as exc:
        assert_stop_contract_required("外部客户A", [], "装货客户")
    assert exc.value.status_code == 422
    assert exc.value.detail == "请为装货客户选择合同"


def test_assert_stop_contract_required_allows_external_with_contract():
    links = [PlanContractLink(contractId="c1", contractNo="N1")]
    assert_stop_contract_required("外部客户A", links, "装货客户") is None


def test_assert_stop_contract_required_skips_internal_customer():
    from app.services.pl_internal_customer import INTERNAL_CUSTOMER_NAMES

    internal = next(iter(INTERNAL_CUSTOMER_NAMES))
    assert_stop_contract_required(internal, [], "装货客户") is None