shanxia2.0/backend/tests/test_pl_vehicle_recharge_workflow.py

"""车辆充值审批工作流：状态常量与核心流转单元测试。"""
from __future__ import annotations

from datetime import date, datetime
from decimal import Decimal
from types import SimpleNamespace

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.models.bx_models import BxWorkflowDef, BxWorkflowNode
from app.models.pl_models import (
    PlVehicle,
    PlVehicleRechargeApply,
    PlVehicleRechargeApproval,
    PlVehicleRechargeRecord,
)
from app.schemas.pl.vehicle_recharge import APPLY_STATUSES, NODE_ACTION_TYPES, RECHARGE_TYPES


def test_apply_statuses_include_workflow_states():
    assert {"draft", "in_approval", "rejected", "disbursed", "cancelled"} <= APPLY_STATUSES
    assert {"approve", "disburse"} <= NODE_ACTION_TYPES
    assert {"etc", "fuel"} <= RECHARGE_TYPES


def _make_session():
    engine = create_engine("sqlite://")
    for table in (
        PlVehicle.__table__,
        PlVehicleRechargeApply.__table__,
        PlVehicleRechargeApproval.__table__,
        PlVehicleRechargeRecord.__table__,
        BxWorkflowDef.__table__,
        BxWorkflowNode.__table__,
    ):
        table.create(engine)
    return sessionmaker(bind=engine)()


def _user(uid: str = "u-applicant", name: str = "申请人") -> SimpleNamespace:
    return SimpleNamespace(id=uid, name=name, dept_id="d1", role="admin", tenant_no="t1")


def _seed_vehicle(db, plate: str = "鄂A00001") -> None:
    db.add(
        PlVehicle(
            id="veh-1",
            vehicle_type="self",
            license_plate=plate,
            tenant_no="t1",
            create_dept_id="d1",
        )
    )


def _seed_workflow(db) -> tuple[str, str, str]:
    wf_id = "wf-1"
    node_gm = "node-gm"
    node_fin = "node-fin"
    db.add(
        BxWorkflowDef(
            id=wf_id,
            biz_type="vehicle_recharge",
            wf_name="充值流程",
            status=1,
            tenant_no="t1",
            create_dept_id="d1",
        )
    )
    db.add(
        BxWorkflowNode(
            id=node_gm,
            wf_id=wf_id,
            node_name="总经理审批",
            node_order=1,
            approver_id="u-gm",
            approver_name="总经理",
            action_type="approve",
            tenant_no="t1",
            create_dept_id="d1",
        )
    )
    db.add(
        BxWorkflowNode(
            id=node_fin,
            wf_id=wf_id,
            node_name="财务发放",
            node_order=2,
            approver_id="u-fin",
            approver_name="财务",
            action_type="disburse",
            tenant_no="t1",
            create_dept_id="d1",
        )
    )
    return wf_id, node_gm, node_fin


def _patch_scope(monkeypatch, *modules) -> None:
    for mod in modules:
        monkeypatch.setattr(mod, "apply_pl_scope", lambda query, *_a, **_k: query)
        monkeypatch.setattr(mod, "assert_pl_row_in_scope", lambda *_a, **_k: None)
    from app.services import pl_vehicle_recharge_service as svc

    monkeypatch.setattr(svc, "_resolve_user_name", lambda *_a, **_k: "测试用户")
    monkeypatch.setattr(
        svc, "_resolve_applicant_id", lambda *_a, **_k: "u-applicant"
    )


def test_submit_sets_in_approval_and_first_node(monkeypatch):
    from app.services import pl_vehicle_recharge_service as svc
    from app.schemas.pl.vehicle_recharge import VehicleRechargeApplyCreate

    db = _make_session()
    _seed_vehicle(db)
    wf_id, node_gm, _node_fin = _seed_workflow(db)
    db.commit()

    _patch_scope(monkeypatch, svc)

    user = _user()
    created = svc.create_apply(
        db,
        "t1",
        user,
        VehicleRechargeApplyCreate(
            licensePlate="鄂A00001",
            rechargeType="etc",
            applyDate=date(2026, 9, 26),
        ),
    )
    assert created.status == "draft"
    assert created.approval_round == 0

    result = svc.submit_apply(db, "t1", user, created.id)
    assert result.status == "in_approval"
    assert result.approval_round == 1
    assert result.wf_id == wf_id
    assert result.current_node == node_gm


def test_resubmit_increments_round_keeps_history(monkeypatch):
    from app.services import pl_vehicle_recharge_service as svc

    db = _make_session()
    _seed_vehicle(db)
    wf_id, node_gm, _node_fin = _seed_workflow(db)
    apply = PlVehicleRechargeApply(
        id="apply-1",
        apply_no="ETC2026092601",
        license_plate="鄂A00001",
        recharge_type="etc",
        apply_date=date(2026, 9, 26),
        applicant_id="u-applicant",
        status="rejected",
        wf_id=wf_id,
        current_node=None,
        approval_round=1,
        tenant_no="t1",
        create_dept_id="d1",
    )
    db.add(apply)
    db.add(
        PlVehicleRechargeApproval(
            id="appr-1",
            apply_id="apply-1",
            approval_round=1,
            node_id=node_gm,
            node_name="总经理审批",
            approver_id="u-gm",
            approver_name="总经理",
            approval_status="rejected",
            approval_opinion="驳回",
            approval_time=datetime(2026, 9, 26, 10, 0, 0),
            tenant_no="t1",
            create_dept_id="d1",
        )
    )
    db.commit()

    _patch_scope(monkeypatch, svc)

    result = svc.submit_apply(db, "t1", _user(), "apply-1")
    assert result.approval_round == 2
    assert result.status == "in_approval"
    assert result.current_node == node_gm
    assert len(result.approvals) == 1
    assert result.approvals[0].approval_status == "rejected"


def test_approve_advances_to_next_node(monkeypatch):
    from app.services import pl_vehicle_recharge_approval_service as approval_svc
    from app.services import pl_vehicle_recharge_service as svc

    db = _make_session()
    _seed_vehicle(db)
    wf_id, node_gm, node_fin = _seed_workflow(db)
    db.add(
        PlVehicleRechargeApply(
            id="apply-1",
            apply_no="ETC2026092601",
            license_plate="鄂A00001",
            recharge_type="etc",
            apply_date=date(2026, 9, 26),
            applicant_id="u-applicant",
            status="in_approval",
            wf_id=wf_id,
            current_node=node_gm,
            approval_round=1,
            tenant_no="t1",
            create_dept_id="d1",
        )
    )
    db.commit()

    _patch_scope(monkeypatch, svc)

    result = approval_svc.approve_apply(
        db, "t1", _user("u-gm", "总经理"), "apply-1", opinion="同意"
    )
    assert result.status == "in_approval"
    assert result.current_node == node_fin
    assert any(a.approval_status == "approved" for a in result.approvals)


def test_reject_sets_rejected_and_appends_history(monkeypatch):
    from app.services import pl_vehicle_recharge_approval_service as approval_svc
    from app.services import pl_vehicle_recharge_service as svc

    db = _make_session()
    _seed_vehicle(db)
    wf_id, node_gm, _node_fin = _seed_workflow(db)
    db.add(
        PlVehicleRechargeApply(
            id="apply-1",
            apply_no="ETC2026092601",
            license_plate="鄂A00001",
            recharge_type="etc",
            apply_date=date(2026, 9, 26),
            applicant_id="u-applicant",
            status="in_approval",
            wf_id=wf_id,
            current_node=node_gm,
            approval_round=1,
            tenant_no="t1",
            create_dept_id="d1",
        )
    )
    db.commit()

    _patch_scope(monkeypatch, svc)

    result = approval_svc.reject_apply(
        db, "t1", _user("u-gm", "总经理"), "apply-1", opinion="材料不全"
    )
    assert result.status == "rejected"
    assert result.current_node is None
    assert result.approvals[-1].approval_status == "rejected"


def test_disburse_writes_record_and_disbursed(monkeypatch):
    from app.services import pl_vehicle_recharge_approval_service as approval_svc
    from app.services import pl_vehicle_recharge_service as svc
    from app.schemas.pl.vehicle_recharge import VehicleRechargeRecordCreate

    db = _make_session()
    _seed_vehicle(db)
    wf_id, _node_gm, node_fin = _seed_workflow(db)
    db.add(
        PlVehicleRechargeApply(
            id="apply-1",
            apply_no="ETC2026092601",
            license_plate="鄂A00001",
            recharge_type="etc",
            apply_date=date(2026, 9, 26),
            applicant_id="u-applicant",
            status="in_approval",
            wf_id=wf_id,
            current_node=node_fin,
            approval_round=1,
            tenant_no="t1",
            create_dept_id="d1",
        )
    )
    db.commit()

    _patch_scope(monkeypatch, svc)

    payload = VehicleRechargeRecordCreate(
        rechargeDate=date(2026, 9, 26),
        rechargeType="etc",
        amount=Decimal("100.00"),
        applyId="apply-1",
    )
    result = approval_svc.disburse_apply(
        db, "t1", _user("u-fin", "财务"), "apply-1", payload
    )
    assert result.status == "disbursed"
    assert result.record is not None
    assert result.record.amount == Decimal("100.00")
    assert any(a.approval_status == "disbursed" for a in result.approvals)


def test_approve_forbidden_on_disburse_node(monkeypatch):
    from fastapi import HTTPException

    from app.services import pl_vehicle_recharge_approval_service as approval_svc
    from app.services import pl_vehicle_recharge_service as svc

    db = _make_session()
    _seed_vehicle(db)
    wf_id, _node_gm, node_fin = _seed_workflow(db)
    db.add(
        PlVehicleRechargeApply(
            id="apply-1",
            apply_no="ETC2026092601",
            license_plate="鄂A00001",
            recharge_type="etc",
            apply_date=date(2026, 9, 26),
            applicant_id="u-applicant",
            status="in_approval",
            wf_id=wf_id,
            current_node=node_fin,
            approval_round=1,
            tenant_no="t1",
            create_dept_id="d1",
        )
    )
    db.commit()

    _patch_scope(monkeypatch, svc)

    with pytest.raises(HTTPException) as exc:
        approval_svc.approve_apply(
            db, "t1", _user("u-fin", "财务"), "apply-1", opinion="ok"
        )
    assert exc.value.status_code == 400