提交

backend/tests/test_pl_arrange_auto_start_execute.py

@@ -0,0 +1,61 @@
+"""安排运输时自动开始执行已审批合同。"""
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from app.services.pl_contract_plan_qty import auto_start_execute_approved_contracts
+
+
+def test_auto_start_execute_promotes_approved_only():
+    approved = SimpleNamespace(id="c1", status="approved", update_by=None, update_time=None)
+    executing = SimpleNamespace(id="c2", status="executing", update_by=None, update_time=None)
+    by_id = {"c1": approved, "c2": executing}
+
+    db = MagicMock()
+
+    def query(_model):
+        q = MagicMock()
+        q.filter.return_value = q
+
+        def first():
+            # Retrieve id from filter call history is hard; use sequential pops
+            return None
+
+        q.first.side_effect = first
+        return q
+
+    db.query.side_effect = query
+    plan = SimpleNamespace(id="p1", contract_id=None)
+
+    call_ids = []
+
+    def fake_query(model):
+        q = MagicMock()
+        q.filter.return_value = q
+
+        def first():
+            # Peek last equality on id from filter args — use call order instead
+            cid = call_ids.pop(0) if call_ids else None
+            return by_id.get(cid)
+
+        q.first.side_effect = first
+        return q
+
+    with patch(
+        "app.services.pl_contract_plan_qty.contract_ids_linked_to_plan",
+        return_value={"c1", "c2"},
+    ):
+        # Ensure deterministic order
+        with patch(
+            "app.services.pl_contract_plan_qty.contract_ids_linked_to_plan",
+            return_value=["c1", "c2"],
+        ):
+            db.query.side_effect = fake_query
+            call_ids.extend(["c1", "c2"])
+            started = auto_start_execute_approved_contracts(
+                db, tenant_no="shanxia2.0", plan=plan, user_id="u1"
+            )
+
+    assert started == ["c1"]
+    assert approved.status == "executing"
+    assert approved.update_by == "u1"
+    assert executing.status == "executing"

backend/tests/test_pl_contract_plan_qty.py

@@ -0,0 +1,130 @@
+"""合同产品运输安排数量汇总单元测试。"""
+from types import SimpleNamespace
+
+from app.services.pl_contract_plan_qty import (
+    _bucket_for_plan,
+    accumulate_stop_products,
+    merge_product_qty_maps,
+    resolve_linked_plan_scopes,
+    sum_qtys_for_scope,
+)
+from collections import defaultdict
+
+
+def test_bucket_archived():
+    assert (
+        _bucket_for_plan(
+            archive_status="archived",
+            follow_status="arranged",
+            transport_status="arranged",
+        )
+        == "archived"
+    )
+
+
+def test_bucket_arranged_unarchived_includes_draft_archive():
+    assert (
+        _bucket_for_plan(
+            archive_status="draft",
+            follow_status="arranged",
+            transport_status="arranged",
+        )
+        == "arranged_unarchived"
+    )
+
+
+def test_bucket_skips_pending_and_cancelled():
+    assert (
+        _bucket_for_plan(
+            archive_status="unarchived",
+            follow_status="pending_arrange",
+            transport_status="pending",
+        )
+        is None
+    )
+    assert (
+        _bucket_for_plan(
+            archive_status="unarchived",
+            follow_status="cancelled",
+            transport_status="arranged",
+        )
+        is None
+    )
+
+
+def test_merge_product_qty_maps_totals():
+    out = merge_product_qty_maps(
+        archived={"p1": 120.0, "p2": 50.0},
+        arranged_unarchived={"p1": 80.0, "p3": 100.0},
+    )
+    assert out["p1"]["archivedQty"] == 120.0
+    assert out["p1"]["arrangedUnarchivedQty"] == 80.0
+    assert out["p1"]["arrangedTotalQty"] == 200.0
+    assert out["p2"]["arrangedTotalQty"] == 50.0
+    assert out["p3"]["arrangedTotalQty"] == 100.0
+
+
+def test_sum_qtys_explicit_delivery_stop_only():
+    load_products = {
+        "L1": [SimpleNamespace(product_id="p1", plan_qty=10)],
+    }
+    delivery_products = {
+        "D1": [SimpleNamespace(product_id="p1", plan_qty=40)],
+        "D2": [SimpleNamespace(product_id="p1", plan_qty=50)],
+    }
+    qtys = sum_qtys_for_scope(
+        load_products_by_load=load_products,
+        delivery_products_by_delivery=delivery_products,
+        all_load_ids={"L1"},
+        all_delivery_ids={"D1", "D2"},
+        scope={"load_ids": set(), "delivery_ids": {"D1"}, "use_all_stops": False},
+    )
+    assert qtys == {"p1": 40.0}
+
+
+def test_legacy_use_all_stops_prefers_delivery():
+    load_products = {
+        "L1": [SimpleNamespace(product_id="p1", plan_qty=99)],
+    }
+    delivery_products = {
+        "D1": [SimpleNamespace(product_id="p1", plan_qty=40)],
+    }
+    qtys = sum_qtys_for_scope(
+        load_products_by_load=load_products,
+        delivery_products_by_delivery=delivery_products,
+        all_load_ids={"L1"},
+        all_delivery_ids={"D1"},
+        scope={"load_ids": set(), "delivery_ids": set(), "use_all_stops": True},
+    )
+    assert qtys == {"p1": 40.0}
+
+
+def test_resolve_scopes_marks_legacy():
+    links = [
+        SimpleNamespace(plan_id="A", side="delivery", stop_id="D1"),
+    ]
+    scopes = resolve_linked_plan_scopes(links, ["B"])
+    assert scopes["A"]["delivery_ids"] == {"D1"}
+    assert scopes["A"]["has_explicit"] is True
+    assert scopes["B"]["use_all_stops"] is True
+
+
+def test_accumulate_stop_products():
+    acc = defaultdict(float)
+    accumulate_stop_products(
+        acc,
+        [
+            SimpleNamespace(product_id="p1", plan_qty=DecimalLike(12.5)),
+            SimpleNamespace(product_id="p1", plan_qty=7.5),
+            SimpleNamespace(product_id="", plan_qty=9),
+        ],
+    )
+    assert acc["p1"] == 20.0
+
+
+class DecimalLike:
+    def __init__(self, v):
+        self.v = v
+
+    def __float__(self):
+        return float(self.v)

backend/tests/test_pl_dibang_date_sync_on_plan_save.py

@@ -0,0 +1,336 @@
+"""计划保存时同步地磅出入库日期 — 单元测试"""
+from datetime import date, datetime
+from decimal import Decimal
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.models.db_models import DbInboundOutbound, DbPurchasePlan
+from app.models.pl_models import PlPlan, PlPlanDelivery, PlPlanLoad, PlTransportDbLink
+from app.services.pl_dibang_bridge_service import sync_linked_dibang_dates_for_plan
+
+
+def _make_query_router(mapping):
+    def query_router(model):
+        mock = MagicMock()
+        handler = mapping.get(model)
+        if handler is None:
+            return MagicMock()
+        return handler(mock)
+
+    return query_router
+
+
+@patch("app.services.pl_dibang_bridge_service._inventory_has_weighing_slip", return_value=False)
+def test_sync_same_month_updates_inventory_date_only(_mock_weighing):
+    plan = PlPlan(
+        id="plan-1",
+        plan_no="PO1",
+        plan_type="outbound",
+        plan_date=date(2026, 8, 31),
+        pickup_time=datetime(2026, 8, 31, 0, 0),
+        tenant_no="t1",
+    )
+    load = PlPlanLoad(
+        id="load-1",
+        plan_id="plan-1",
+        load_time=datetime(2026, 8, 31, 0, 0),
+        tenant_no="t1",
+        del_flag=0,
+    )
+    db_plan = DbPurchasePlan(
+        id="dbp-1",
+        tenant_no="t1",
+        plan_no="SX-OLD",
+        plan_date=date(2026, 8, 15),
+        start_date=date(2026, 8, 1),
+        end_date=date(2026, 8, 31),
+        business_type="出库",
+        product_id="prod-1",
+        product="硫酸",
+        planned_quantity=31.0,
+        customer_name="客户A",
+        del_flag=0,
+    )
+    inventory = DbInboundOutbound(
+        id="inv-1",
+        tenant_no="t1",
+        date=date(2026, 8, 15),
+        partner_company="客户A",
+        business_type="出库",
+        product_id="prod-1",
+        product="硫酸",
+        specification="98%",
+        quantity=31.0,
+        plan_no="SX-OLD",
+        status="待出库",
+        purchase_plan_id="dbp-1",
+        del_flag=0,
+    )
+    link = PlTransportDbLink(
+        id="link-1",
+        tenant_no="t1",
+        plan_id="plan-1",
+        transport_id="tr-1",
+        stop_type="load",
+        stop_id="load-1",
+        db_purchase_plan_id="dbp-1",
+        db_inventory_id="inv-1",
+        contributed_qty=Decimal("31"),
+        del_flag=0,
+    )
+
+    db = MagicMock()
+
+    def link_q(m):
+        m.filter.return_value.all.return_value = [link]
+        return m
+
+    def load_q(m):
+        m.filter.return_value.first.return_value = load
+        return m
+
+    def delivery_q(m):
+        m.filter.return_value.order_by.return_value.all.return_value = []
+        return m
+
+    def inv_q(m):
+        m.filter.return_value.first.return_value = inventory
+        return m
+
+    def plan_q(m):
+        m.filter.return_value.first.return_value = db_plan
+        return m
+
+    db.query.side_effect = _make_query_router(
+        {
+            PlTransportDbLink: link_q,
+            PlPlanLoad: load_q,
+            PlPlanDelivery: delivery_q,
+            DbInboundOutbound: inv_q,
+            DbPurchasePlan: plan_q,
+        }
+    )
+
+    warnings = sync_linked_dibang_dates_for_plan(
+        db,
+        tenant_no="t1",
+        plan=plan,
+        user_id="u1",
+        dept_id="d1",
+    )
+
+    assert warnings == []
+    assert inventory.date == date(2026, 8, 31)
+    assert inventory.update_by == "u1"
+    assert inventory.purchase_plan_id == "dbp-1"
+    assert db_plan.planned_quantity == 31.0
+
+
+@patch("app.services.pl_dibang_bridge_service._inventory_has_weighing_slip", return_value=False)
+@patch("app.services.pl_dibang_bridge_service.find_monthly_db_plan")
+@patch("app.services.pl_dibang_bridge_service._create_db_plan")
+def test_sync_cross_month_repoints_purchase_plan(mock_create, mock_find, _mock_weighing):
+    plan = PlPlan(
+        id="plan-1",
+        plan_no="PO20260830002",
+        plan_type="outbound",
+        plan_date=date(2026, 9, 1),
+        pickup_time=datetime(2026, 9, 1, 0, 0),
+        tenant_no="t1",
+    )
+    load = PlPlanLoad(
+        id="load-1",
+        plan_id="plan-1",
+        load_time=datetime(2026, 9, 1, 0, 0),
+        tenant_no="t1",
+        del_flag=0,
+    )
+    old_db_plan = DbPurchasePlan(
+        id="dbp-aug",
+        tenant_no="t1",
+        plan_no="SX-AUG",
+        plan_date=date(2026, 8, 31),
+        start_date=date(2026, 8, 1),
+        end_date=date(2026, 8, 31),
+        business_type="出库",
+        product_id="prod-1",
+        product="硫酸",
+        planned_quantity=31.0,
+        customer_name="客户A",
+        del_flag=0,
+    )
+    new_db_plan = DbPurchasePlan(
+        id="dbp-sep",
+        tenant_no="t1",
+        plan_no="SX-SEP",
+        plan_date=date(2026, 9, 1),
+        start_date=date(2026, 9, 1),
+        end_date=date(2026, 9, 30),
+        business_type="出库",
+        product_id="prod-1",
+        product="硫酸",
+        planned_quantity=10.0,
+        customer_name="客户A",
+        del_flag=0,
+    )
+    inventory = DbInboundOutbound(
+        id="inv-1",
+        tenant_no="t1",
+        date=date(2026, 8, 31),
+        partner_company="客户A",
+        business_type="出库",
+        product_id="prod-1",
+        product="硫酸",
+        specification="98%",
+        quantity=31.0,
+        plan_no="SX-AUG",
+        status="待出库",
+        purchase_plan_id="dbp-aug",
+        del_flag=0,
+    )
+    link = PlTransportDbLink(
+        id="link-1",
+        tenant_no="t1",
+        plan_id="plan-1",
+        transport_id="tr-1",
+        stop_type="load",
+        stop_id="load-1",
+        db_purchase_plan_id="dbp-aug",
+        db_inventory_id="inv-1",
+        contributed_qty=Decimal("31"),
+        del_flag=0,
+    )
+
+    mock_find.return_value = new_db_plan
+    mock_create.side_effect = AssertionError("should reuse existing September plan")
+
+    db = MagicMock()
+
+    def link_q(m):
+        m.filter.return_value.all.return_value = [link]
+        return m
+
+    def load_q(m):
+        m.filter.return_value.first.return_value = load
+        return m
+
+    def delivery_q(m):
+        m.filter.return_value.order_by.return_value.all.return_value = []
+        return m
+
+    def inv_q(m):
+        m.filter.return_value.first.return_value = inventory
+        return m
+
+    def plan_q(m):
+        m.filter.return_value.first.return_value = old_db_plan
+        return m
+
+    db.query.side_effect = _make_query_router(
+        {
+            PlTransportDbLink: link_q,
+            PlPlanLoad: load_q,
+            PlPlanDelivery: delivery_q,
+            DbInboundOutbound: inv_q,
+            DbPurchasePlan: plan_q,
+        }
+    )
+
+    warnings = sync_linked_dibang_dates_for_plan(
+        db,
+        tenant_no="t1",
+        plan=plan,
+        user_id="u1",
+        dept_id="d1",
+    )
+
+    assert warnings == []
+    assert inventory.date == date(2026, 9, 1)
+    assert inventory.purchase_plan_id == "dbp-sep"
+    assert inventory.plan_no == "SX-SEP"
+    assert link.db_purchase_plan_id == "dbp-sep"
+    assert old_db_plan.planned_quantity == 0.0
+    assert new_db_plan.planned_quantity == 41.0
+
+
+@patch("app.services.pl_dibang_bridge_service._inventory_has_weighing_slip", return_value=True)
+def test_sync_skips_when_weighing_exists(_mock_weighing):
+    plan = PlPlan(
+        id="plan-1",
+        plan_no="PO1",
+        plan_type="outbound",
+        plan_date=date(2026, 9, 1),
+        pickup_time=datetime(2026, 9, 1, 0, 0),
+        tenant_no="t1",
+    )
+    load = PlPlanLoad(
+        id="load-1",
+        plan_id="plan-1",
+        load_time=datetime(2026, 9, 1, 0, 0),
+        tenant_no="t1",
+        del_flag=0,
+    )
+    inventory = DbInboundOutbound(
+        id="inv-1",
+        tenant_no="t1",
+        date=date(2026, 8, 31),
+        partner_company="客户A",
+        business_type="出库",
+        product="硫酸",
+        quantity=31.0,
+        batch_no="CRKDH0002400",
+        status="待出库",
+        del_flag=0,
+    )
+    link = PlTransportDbLink(
+        id="link-1",
+        tenant_no="t1",
+        plan_id="plan-1",
+        transport_id="tr-1",
+        stop_type="load",
+        stop_id="load-1",
+        db_purchase_plan_id="dbp-1",
+        db_inventory_id="inv-1",
+        contributed_qty=Decimal("31"),
+        del_flag=0,
+    )
+
+    db = MagicMock()
+
+    def link_q(m):
+        m.filter.return_value.all.return_value = [link]
+        return m
+
+    def load_q(m):
+        m.filter.return_value.first.return_value = load
+        return m
+
+    def delivery_q(m):
+        m.filter.return_value.order_by.return_value.all.return_value = []
+        return m
+
+    def inv_q(m):
+        m.filter.return_value.first.return_value = inventory
+        return m
+
+    db.query.side_effect = _make_query_router(
+        {
+            PlTransportDbLink: link_q,
+            PlPlanLoad: load_q,
+            PlPlanDelivery: delivery_q,
+            DbInboundOutbound: inv_q,
+        }
+    )
+
+    warnings = sync_linked_dibang_dates_for_plan(
+        db,
+        tenant_no="t1",
+        plan=plan,
+        user_id="u1",
+        dept_id="d1",
+    )
+
+    assert len(warnings) == 1
+    assert "CRKDH0002400" in warnings[0]
+    assert inventory.date == date(2026, 8, 31)