shanxia2.0/backend/tests/test_pl_arrange_auto_start_execute.py

"""安排运输时自动开始执行已审批合同。"""
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"