shanxia2.0/backend/tests/test_pl_plan_execution_lock.py

from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import pytest
from fastapi import HTTPException

from app.api.pl.plan import cancel_plan, force_void_plan
from app.services.pl_plan_execution_lock import (
    execution_lock_message,
    inventory_status_is_executed,
    pick_execution_lock_reason,
)


def test_pick_execution_lock_reason_prefers_weighed():
    assert (
        pick_execution_lock_reason(
            weighed=True,
            inventory_executed=True,
            site_feedback=True,
            manual_qty=True,
        )
        == "已过磅"
    )


def test_pick_execution_lock_reason_order():
    assert (
        pick_execution_lock_reason(
            weighed=False, inventory_executed=True, site_feedback=True, manual_qty=True
        )
        == "地磅出入库已执行"
    )
    assert (
        pick_execution_lock_reason(
            weighed=False, inventory_executed=False, site_feedback=True, manual_qty=True
        )
        == "现场已回传"
    )
    assert (
        pick_execution_lock_reason(
            weighed=False, inventory_executed=False, site_feedback=False, manual_qty=True
        )
        == "入库或出库实际数量已修改"
    )
    assert (
        pick_execution_lock_reason(
            weighed=False, inventory_executed=False, site_feedback=False, manual_qty=False
        )
        is None
    )


def test_inventory_status_is_executed_only_outside_revokable():
    assert inventory_status_is_executed("已完成") is True
    assert inventory_status_is_executed("待入库") is False
    assert inventory_status_is_executed("") is False
    assert inventory_status_is_executed(None) is False


def test_execution_lock_messages():
    assert (
        execution_lock_message("已过磅", "arrange")
        == "该计划已过磅，不能再调整，请联系管理员处理"
    )
    assert (
        execution_lock_message("现场已回传", "void")
        == "该计划现场已回传，不能作废，请联系管理员处理"
    )


def test_cancel_plan_rejects_started_plan():
    plan = SimpleNamespace(
        id="p1",
        follow_status="arranged",
        archive_status="unarchived",
        transport_status="arranged",
        transport_id="t1",
    )
    user = MagicMock(id="u1")
    with patch("app.api.pl.plan.get_tenant_no", return_value="t1"), patch(
        "app.api.pl.plan._get_plan", return_value=plan
    ), patch(
        "app.api.pl.plan.plan_execution_lock_reason", return_value="已过磅"
    ), patch("app.api.pl.plan.void_io_orders_for_plan") as void_io:
        with pytest.raises(HTTPException) as exc:
            cancel_plan("p1", body=None, db=MagicMock(), user=user)
    assert exc.value.status_code == 400
    assert exc.value.detail == "该计划已过磅，不能作废，请联系管理员处理"
    void_io.assert_not_called()


def test_force_void_rejects_non_super_admin_without_permission():
    user = MagicMock(id="u1", role="admin")
    with patch("app.api.pl.plan.get_user_menu_codes", return_value=set()):
        with pytest.raises(HTTPException) as exc:
            force_void_plan("p1", body=None, db=MagicMock(), user=user)
    assert exc.value.status_code == 403


def test_force_void_super_admin_cancels_started_plan():
    plan = SimpleNamespace(
        id="p1",
        follow_status="arranged",
        archive_status="unarchived",
        transport_status="arranged",
        transport_id="t1",
        cancel_reason=None,
        cancel_time=None,
        cancel_by=None,
        update_by=None,
    )
    user = MagicMock(id="u1", role="super_admin")
    body = SimpleNamespace(cancel_reason="过磅作废，这趟不算")
    db = MagicMock()
    with patch("app.api.pl.plan.get_tenant_no", return_value="t1"), patch(
        "app.api.pl.plan._get_plan", return_value=plan
    ), patch(
        "app.api.pl.plan.plan_execution_lock_reason", return_value="已过磅"
    ), patch("app.api.pl.plan.apply_force_void_downstream") as apply:
        result = force_void_plan("p1", body=body, db=db, user=user)
    apply.assert_called_once()
    assert result.follow_status == "cancelled"
    assert result.archive_status == "cancelled"
    assert result.transport_status == "pending"
    assert result.cancel_reason == "过磅作废，这趟不算"
    db.commit.assert_called_once()