471 lines
14 KiB
Python
471 lines
14 KiB
Python
"""车辆充值审批工作流：状态常量与核心流转单元测试。"""
|
|
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"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("recharge_type", ["etc", "fuel", "shanxia"])
|
|
def test_submit_sets_in_approval_and_first_node(monkeypatch, recharge_type):
|
|
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=recharge_type,
|
|
applyDate=date(2026, 9, 26),
|
|
odometerKm=Decimal("1000"),
|
|
attachmentUrls="https://example.com/a.jpg",
|
|
),
|
|
)
|
|
assert created.recharge_type == recharge_type
|
|
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"
|
|
# 当前轮进度：全部节点均展示，尚未审批则为 pending
|
|
assert len(result.approval_flow) == 2
|
|
assert all(n.approval_status == "pending" for n in result.approval_flow)
|
|
# 历史：保留打回前的审批动作
|
|
assert len(result.approval_history) == 1
|
|
assert result.approval_history[0].approval_status == "rejected"
|
|
assert result.approval_history[0].approval_round == 1
|
|
|
|
|
|
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)
|
|
assert len(result.approval_flow) == 2
|
|
assert result.approval_flow[0].approval_status == "approved"
|
|
assert result.approval_flow[0].approver_name == "总经理"
|
|
assert result.approval_flow[0].approval_opinion == "同意"
|
|
assert result.approval_flow[1].approval_status == "pending"
|
|
assert result.approval_flow[1].node_name == "财务发放"
|
|
assert len(result.approval_history) == 1
|
|
assert result.approval_history[0].approval_status == "approved"
|
|
|
|
|
|
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"
|
|
|
|
|
|
@pytest.mark.parametrize("recharge_type", ["etc", "fuel", "shanxia"])
|
|
def test_disburse_writes_record_and_disbursed(monkeypatch, recharge_type):
|
|
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=recharge_type,
|
|
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=recharge_type,
|
|
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.recharge_type == recharge_type
|
|
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
|
|
|
|
|
|
def test_withdraw_in_approval_to_draft_keeps_round_and_approvals(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)
|
|
db.add(
|
|
PlVehicleRechargeApply(
|
|
id="a-wd",
|
|
apply_no="SXSQWD001",
|
|
license_plate="鄂A00001",
|
|
recharge_type="etc",
|
|
apply_date=date.today(),
|
|
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.add(
|
|
PlVehicleRechargeApproval(
|
|
id="ap1",
|
|
apply_id="a-wd",
|
|
node_id=node_gm,
|
|
node_name="总经理审批",
|
|
approval_round=1,
|
|
approval_status="approved",
|
|
approver_id="u-gm",
|
|
approver_name="总经理",
|
|
approval_time=datetime.now(),
|
|
tenant_no="t1",
|
|
create_dept_id="d1",
|
|
)
|
|
)
|
|
db.commit()
|
|
_patch_scope(monkeypatch, svc)
|
|
|
|
out = svc.withdraw_apply(db, "t1", _user("u-applicant"), "a-wd")
|
|
assert out.status == "draft"
|
|
assert out.current_node in (None, "")
|
|
row = db.query(PlVehicleRechargeApply).filter_by(id="a-wd").one()
|
|
assert row.status == "draft"
|
|
assert row.current_node is None
|
|
assert row.approval_round == 1
|
|
assert (
|
|
db.query(PlVehicleRechargeApproval)
|
|
.filter_by(apply_id="a-wd", del_flag=0)
|
|
.count()
|
|
== 1
|
|
)
|
|
|
|
|
|
def test_withdraw_rejects_non_applicant(monkeypatch):
|
|
from fastapi import HTTPException
|
|
from app.services import pl_vehicle_recharge_service as svc
|
|
|
|
db = _make_session()
|
|
_seed_vehicle(db)
|
|
db.add(
|
|
PlVehicleRechargeApply(
|
|
id="a-wd2",
|
|
apply_no="SXSQWD002",
|
|
license_plate="鄂A00001",
|
|
recharge_type="etc",
|
|
apply_date=date.today(),
|
|
applicant_id="u-applicant",
|
|
status="in_approval",
|
|
tenant_no="t1",
|
|
create_dept_id="d1",
|
|
)
|
|
)
|
|
db.commit()
|
|
_patch_scope(monkeypatch, svc)
|
|
|
|
with pytest.raises(HTTPException) as ei:
|
|
svc.withdraw_apply(db, "t1", _user("u-other"), "a-wd2")
|
|
assert ei.value.status_code == 403
|
|
|
|
|
|
def test_withdraw_rejects_non_in_approval(monkeypatch):
|
|
from fastapi import HTTPException
|
|
from app.services import pl_vehicle_recharge_service as svc
|
|
|
|
db = _make_session()
|
|
_seed_vehicle(db)
|
|
db.add(
|
|
PlVehicleRechargeApply(
|
|
id="a-wd3",
|
|
apply_no="SXSQWD003",
|
|
license_plate="鄂A00001",
|
|
recharge_type="etc",
|
|
apply_date=date.today(),
|
|
applicant_id="u-applicant",
|
|
status="draft",
|
|
tenant_no="t1",
|
|
create_dept_id="d1",
|
|
)
|
|
)
|
|
db.commit()
|
|
_patch_scope(monkeypatch, svc)
|
|
|
|
with pytest.raises(HTTPException) as ei:
|
|
svc.withdraw_apply(db, "t1", _user("u-applicant"), "a-wd3")
|
|
assert ei.value.status_code == 400
|