feat: add vehicle recharge apply submit and detail approvals

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed 2026-09-26 20:55:43 +08:00
1 parent bdd67d44d9
commit ea8c2c0362
3 files changed
+362 -11

No files matched your search

@@ -1,4 +1,19 @@
"""车辆充值审批工作流:状态常量与核心流转单元测试。"""
from __future__ import annotations
from datetime import date, datetime
from types import SimpleNamespace
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
@@ -6,3 +21,169 @@ 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"