diff --git a/backend/app/api/deps.py b/backend/app/api/deps.py index e867ba0..c694d58 100644 --- a/backend/app/api/deps.py +++ b/backend/app/api/deps.py @@ -113,3 +113,22 @@ def require_admin(current_user: Annotated[User, Depends(get_current_user)]) -> U AdminUser = Annotated[User, Depends(require_admin)] + + +def require_roles(*role_codes: str): + """要求当前用户具备任一指定角色（依赖工厂）。 + + 与 require_admin 的差别：admin 不是天然放行——传入集合里显式含 + ROLE_ADMIN 才可见。用于审批等模块的粗粒度准入；节点级准入仍由 + service 按单据状态二次校验（双层：模块可见 ≠ 可操作）。 + + :raises ForbiddenError: 已登录但不具备任一指定角色（403） + """ + + def _checker(current_user: Annotated[User, Depends(get_current_user)]) -> User: + codes = set(user_repository.get_role_codes(current_user)) + if not codes & set(role_codes): + raise ForbiddenError("无权访问审批管理") + return current_user + + return _checker diff --git a/backend/app/api/router.py b/backend/app/api/router.py index 1ba9341..7e57bfb 100644 --- a/backend/app/api/router.py +++ b/backend/app/api/router.py @@ -15,6 +15,7 @@ from app.api.v1.business import businesses as business_businesses from app.api.v1.business import ratings as business_ratings from app.api.v1.business import receipts as business_receipts from app.api.v1.business import shipments as business_shipments +from app.api.v1.system import approvals as system_approvals from app.api.v1.system import business_roles as system_business_roles from app.api.v1.system import logs as system_logs from app.api.v1.system import params as system_params @@ -38,6 +39,7 @@ system_router.include_router(system_roles.router) system_router.include_router(system_business_roles.router) system_router.include_router(system_params.router) system_router.include_router(system_settings.router) +system_router.include_router(system_approvals.router) system_router.include_router(system_logs.operation_router) system_router.include_router(system_logs.login_router) v1_router.include_router(system_router) diff --git a/backend/app/api/v1/system/approvals.py b/backend/app/api/v1/system/approvals.py new file mode 100644 index 0000000..c1cf1f5 --- /dev/null +++ b/backend/app/api/v1/system/approvals.py @@ -0,0 +1,67 @@ +"""审批管理接口（2026-10-15 复核审批补充需求）。 + +准入：admin/财务/领导可见（营销后勤暂不可见——权限清单待澄清 #6， +答复后放开为「提交人可见本人单据」）。节点级准入由 service 按单据 +状态二次校验（status=1 仅财务、status=2 仅领导，admin 不代办）。 +提单不经本模块：由业务写路径（创建/编辑带二次分配方案）自动触发。 +""" + +from typing import Annotated + +from fastapi import APIRouter, Depends, Path, Query + +from app.api.deps import DbSession, RequestMetaDep, require_roles +from app.core.constants import ROLE_ADMIN, ROLE_FINANCE, ROLE_LEADER +from app.core.response import ApiResponse, PageResult, ok +from app.models import User +from app.schemas.approval import ( + ApprovalActionIn, + ApprovalOrderDetailOut, + ApprovalOrderOut, + ApprovalQuery, +) +from app.services import approval_service + +router = APIRouter(prefix="/approvals", tags=["系统管理 · 审批管理"]) + +# 模块准入：可见集合（节点准入在 service 层按状态二次判定） +_ApprovalUser = Annotated[User, Depends(require_roles(ROLE_ADMIN, ROLE_FINANCE, ROLE_LEADER))] + + +@router.get("", response_model=ApiResponse[PageResult[ApprovalOrderOut]], summary="审批单列表") +def list_orders( + session: DbSession, + _: _ApprovalUser, + query: Annotated[ApprovalQuery, Query()], +) -> ApiResponse[PageResult[ApprovalOrderOut]]: + """分页查询审批单。""" + return ok(approval_service.list_orders(session, query)) + + +@router.get("/{order_id}", response_model=ApiResponse[ApprovalOrderDetailOut], summary="审批详情") +def get_order( + session: DbSession, + _: _ApprovalUser, + order_id: Annotated[int, Path(ge=1)], +) -> ApiResponse[ApprovalOrderDetailOut]: + """审批单详情（含节点时间线）。""" + return ok(approval_service.get_order(session, order_id)) + + +@router.post( + "/{order_id}/action", + response_model=ApiResponse[ApprovalOrderDetailOut], + summary="节点动作（通过/驳回）", +) +def act_on_order( + payload: ApprovalActionIn, + session: DbSession, + current_user: _ApprovalUser, + meta: RequestMetaDep, + order_id: Annotated[int, Path(ge=1)], +) -> ApiResponse[ApprovalOrderDetailOut]: + """财务复核 / 领导审核；一驳全驳，驳回须填意见。""" + return ok( + approval_service.act(session, order_id, payload, operator=current_user, meta=meta), + message="已提交", + ) diff --git a/backend/app/core/constants.py b/backend/app/core/constants.py index c361404..db9c4f4 100644 --- a/backend/app/core/constants.py +++ b/backend/app/core/constants.py @@ -100,3 +100,5 @@ ACTION_DELETE = "删除" ACTION_RESET_PASSWORD = "重置密码" ACTION_ENABLE = "启用" ACTION_DISABLE = "禁用" +ACTION_APPROVE = "审批通过" +ACTION_REJECT = "审批驳回" diff --git a/backend/app/models/approval.py b/backend/app/models/approval.py index bfdace5..56b55b1 100644 --- a/backend/app/models/approval.py +++ b/backend/app/models/approval.py @@ -42,6 +42,10 @@ APPROVAL_ACTION_REJECT = 2 # 人工驳回 APPROVAL_ACTION_AUTO_APPROVE = 3 # 系统自动通过（开关关闭时逐节点留痕） APPROVAL_ACTION_WITHDRAW = 4 # 撤回（业务重新提交时旧在途单注销） +# 审批对象类型（approval_orders.order_type；2 为 ④ 月结预留） +APPROVAL_TYPE_SECONDARY_ALLOCATION = 1 +APPROVAL_TYPE_MONTHLY_SETTLEMENT = 2 + class ApprovalOrder(Base, TimestampMixin): """审批单：一张待审对象一条审批流（状态可流转，含 update_time）。""" diff --git a/backend/app/schemas/approval.py b/backend/app/schemas/approval.py new file mode 100644 index 0000000..75b17c0 --- /dev/null +++ b/backend/app/schemas/approval.py @@ -0,0 +1,66 @@ +"""审批管理的请求/响应模型（2026-10-15 复核审批补充需求）。""" + +from datetime import datetime + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from app.core.response import PageQuery + + +class ApprovalStepOut(BaseModel): + """节点记录（时间线；action=3 自动通过 operator 记「系统」）。""" + + model_config = ConfigDict(from_attributes=True) + + id: int + node: int + action: int + opinion: str | None = None + operator_id: int | None = None + operator_name: str + create_time: datetime + + +class ApprovalOrderOut(BaseModel): + """审批单列表行。""" + + model_config = ConfigDict(from_attributes=True) + + id: int + order_no: str + order_type: int + ref_id: int + refer_no: str + status: int + submitter_id: int | None = None + submitter_name: str + create_time: datetime + + +class ApprovalOrderDetailOut(ApprovalOrderOut): + """审批单详情：附节点时间线。""" + + steps: list[ApprovalStepOut] = Field(default_factory=list) + + +class ApprovalQuery(PageQuery): + """审批单列表查询条件。""" + + keyword: str | None = Field(default=None, max_length=32, description="审批单号/引用单号，模糊") + status: int | None = Field(default=None, ge=1, le=5, description="状态筛选") + order_type: int | None = Field(default=None, ge=1, le=2, description="对象类型筛选") + + +class ApprovalActionIn(BaseModel): + """节点动作：1 通过 / 2 驳回（驳回须填意见——空驳无从追溯）。""" + + model_config = ConfigDict(str_strip_whitespace=True) + + action: int = Field(ge=1, le=2, description="1通过 2驳回") + opinion: str | None = Field(default=None, max_length=255, description="审批意见") + + @model_validator(mode="after") + def _check_opinion(self) -> "ApprovalActionIn": + if self.action == 2 and not self.opinion: + raise ValueError("驳回必须填写审批意见") + return self diff --git a/backend/app/services/approval_service.py b/backend/app/services/approval_service.py new file mode 100644 index 0000000..e00fcba --- /dev/null +++ b/backend/app/services/approval_service.py @@ -0,0 +1,330 @@ +"""审批流业务编排（2026-10-15 复核审批补充需求）。 + +三级链路：营销后勤提交 → 财务复核 → 领导审核。行为由参数开关驱动 +（`approval.secondary_allocation.enabled` / `approval.monthly_settlement.enabled`）： + +- **关闭（默认）**：提交即生成审批单并「自动通过」全节点留痕（action=3， + operator 记「系统」）——留痕不断链，与「关闭=直接生效」的需求口径一致； +- **开启**：status=1 待财务复核 → 2 待领导审核 → 3 已通过；任一节点驳回 + 即整体驳回（status=4），一期不做退回重走（简单状态机）。 + +红线： +- `submit_order` **不 commit、不嵌套 audit** —— 由业务写路径（business_service + 的 create/update）既有 audit 上下文统一收口，违反嵌套禁令会让失败路径 + 回滚不完整（见 audit_service 模块 docstring）； +- 节点准入**严格按角色编码**：status=1 仅财务、status=2 仅领导，admin 不 + 代办 —— 放行 admin 会让双人复核流退化成管理员单点审批； +- 同一引用对象的在途单（status 1/2）在重新提交前先撤回（status=5 + + node=0/action=4 留痕），保证「一个对象至多一张在途单」。 +""" + +import logging +from datetime import date + +from sqlalchemy import func, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from app.core.constants import ( + ACTION_APPROVE, + ACTION_REJECT, + MODULE_APPROVAL, + NO_PREFIX_APPROVAL, + ROLE_FINANCE, + ROLE_LEADER, +) +from app.core.db_errors import is_duplicate_entry +from app.core.exceptions import BadRequestError, ForbiddenError, NotFoundError +from app.core.response import PageResult +from app.models import ApprovalOrder, ApprovalStep, User +from app.models.approval import ( + APPROVAL_ACTION_APPROVE, + APPROVAL_ACTION_AUTO_APPROVE, + APPROVAL_ACTION_WITHDRAW, + APPROVAL_NODE_FINANCE, + APPROVAL_NODE_LEADER, + APPROVAL_NODE_WITHDRAW, + APPROVAL_NODES, + APPROVAL_STATUS_APPROVED, + APPROVAL_STATUS_PENDING_FINANCE, + APPROVAL_STATUS_PENDING_LEADER, + APPROVAL_STATUS_REJECTED, + APPROVAL_STATUS_WITHDRAWN, +) +from app.repositories import user_repository +from app.schemas.approval import ( + ApprovalActionIn, + ApprovalOrderDetailOut, + ApprovalOrderOut, + ApprovalQuery, + ApprovalStepOut, +) +from app.services import audit_service, param_service +from app.services.audit_service import AuditRecord, RequestMeta +from app.services.no_generator import next_no + +logger = logging.getLogger(__name__) + +# 开关参数编码（与 param_registry 种子同源） +ENABLE_CODE_BY_TYPE: dict[int, str] = { + 1: "approval.secondary_allocation.enabled", + 2: "approval.monthly_settlement.enabled", +} + +_NO_RETRY = 1 # AP 单号唯一键冲突重试（next_no 的 MAX+FOR UPDATE 之外的兜底） + +# 在途状态集合（同一引用对象至多一张在途单） +_PENDING_STATUSES = (APPROVAL_STATUS_PENDING_FINANCE, APPROVAL_STATUS_PENDING_LEADER) + +# 在途状态 → 当前节点应操作的角色（admin 不代办：单人审批绕过双人复核） +_NODE_ROLE_BY_STATUS = { + APPROVAL_STATUS_PENDING_FINANCE: ROLE_FINANCE, + APPROVAL_STATUS_PENDING_LEADER: ROLE_LEADER, +} + +_ROLE_LABEL = {ROLE_FINANCE: "财务", ROLE_LEADER: "领导"} + + +def is_enabled(session: Session, order_type: int) -> bool: + """读当日参数快照判定审批开关（一次提交路径内只读一次，调用方勿重读）。""" + code = ENABLE_CODE_BY_TYPE.get(order_type) + if code is None: + raise BadRequestError("未知的审批对象类型") + return param_service.resolve_snapshot(session, date.today()).boolean(code) + + +def submit_order( + session: Session, + *, + order_type: int, + ref_id: int, + refer_no: str, + submitter: User, + retry_on_conflict: bool, +) -> ApprovalOrder: + """提单（业务写路径内部调用；不 commit、不嵌套 audit）。 + + 同一引用对象的在途单先撤回留痕再建新单；开关关闭时生成即自动通过 + （全节点 action=3 留痕）。 + + :param retry_on_conflict: AP 编号撞唯一键时是否允许本函数内部 + rollback 后重取。**仅 create 传 True**——它外层有编号重试循环， + rollback 后整体重来；update 路径必须传 False：内部 rollback 会把 + 同事务的业务行修改一并回滚，静默继续会「返回 200 库里没改」。 + False 时 IntegrityError 直接抛出 → 审计上下文回滚并记失败 → + 全局 handler 收敛 400，用户重试即可。 + """ + _withdraw_pending(session, order_type=order_type, ref_id=ref_id, operator=submitter) + + order = ApprovalOrder( + order_no=next_no( + session, + model=ApprovalOrder, + column_name="order_no", + prefix=NO_PREFIX_APPROVAL, + ), + order_type=order_type, + ref_id=ref_id, + refer_no=refer_no, + submitter_id=submitter.id, + submitter_name=submitter.username, + ) + session.add(order) + + if is_enabled(session, order_type): + order.status = APPROVAL_STATUS_PENDING_FINANCE + _flush_order(session, order, retry_on_conflict=retry_on_conflict) + logger.info("审批单待审 | order=%s", order.order_no) + return order + + # 开关关闭=直接生效：两级节点各记一条「自动通过」，留痕不断链 + order.status = APPROVAL_STATUS_APPROVED + _flush_order(session, order, retry_on_conflict=retry_on_conflict) + session.add_all( + [ + ApprovalStep( + order_id=order.id, + node=node, + action=APPROVAL_ACTION_AUTO_APPROVE, + operator_name="系统", + ) + for node in (APPROVAL_NODE_FINANCE, APPROVAL_NODE_LEADER) + ] + ) + session.flush() + logger.info("审批单自动通过 | order=%s（开关关闭）", order.order_no) + return order + + +def act( + session: Session, + order_id: int, + payload: ApprovalActionIn, + *, + operator: User, + meta: RequestMeta | None = None, +) -> ApprovalOrderDetailOut: + """节点动作（财务复核/领导审核）：状态机流转 + 节点留痕（审计上下文）。 + + 通过：status=1 → 2（下一步领导）；status=2 → 3（终态已通过）。 + 驳回：任一在途状态 → 4（整体驳回，一期不退回重走）。 + 节点留痕的 node 取**动作发生时的当前节点**（通过/驳回都记在操作节点上）。 + """ + codes = set(user_repository.get_role_codes(operator)) + verb = ACTION_APPROVE if payload.action == APPROVAL_ACTION_APPROVE else ACTION_REJECT + with audit_service.audit( + session, + operator=operator, + record=AuditRecord( + module=MODULE_APPROVAL, + action=verb, + summary=f"审批单 #{order_id} {verb}", + request_data=payload.model_dump(), + ), + meta=meta, + ): + order = session.get(ApprovalOrder, order_id, with_for_update=True) + if order is None: + raise NotFoundError("审批单不存在") + if order.status not in _PENDING_STATUSES: + raise BadRequestError("审批单已结束，不可重复操作") + + required_role = _NODE_ROLE_BY_STATUS[order.status] + if required_role not in codes: + raise ForbiddenError(f"当前节点须{_ROLE_LABEL[required_role]}角色操作") + + node_now = APPROVAL_NODES[order.status] + if payload.action == APPROVAL_ACTION_APPROVE: + order.status = ( + APPROVAL_STATUS_PENDING_LEADER + if order.status == APPROVAL_STATUS_PENDING_FINANCE + else APPROVAL_STATUS_APPROVED + ) + else: + order.status = APPROVAL_STATUS_REJECTED + + session.add( + ApprovalStep( + order_id=order.id, + node=node_now, + action=payload.action, + opinion=payload.opinion, + operator_id=operator.id, + operator_name=operator.username, + ) + ) + session.commit() + return _detail_of(session, order) + + +def list_orders(session: Session, query: ApprovalQuery) -> PageResult[ApprovalOrderOut]: + """分页查询审批单（准入由 api 层依赖判定）。""" + filters = _order_filters(query) + rows = ( + session.execute( + select(ApprovalOrder) + .where(*filters) + .order_by(ApprovalOrder.id.desc()) + .offset(query.offset) + .limit(query.page_size) + ) + .scalars() + .all() + ) + total = session.execute( + select(func.count()).select_from(ApprovalOrder).where(*filters) + ).scalar_one() + return PageResult[ApprovalOrderOut]( + records=[ApprovalOrderOut.model_validate(r) for r in rows], + total=int(total), + page_num=query.page_num, + page_size=query.page_size, + ) + + +def get_order(session: Session, order_id: int) -> ApprovalOrderDetailOut: + """审批单详情（附节点时间线）。""" + order = session.get(ApprovalOrder, order_id) + if order is None: + raise NotFoundError("审批单不存在") + return _detail_of(session, order) + + +def _withdraw_pending(session: Session, *, order_type: int, ref_id: int, operator: User) -> None: + """撤回引用对象的在途单（重新提交前调用；不 commit）。""" + pending = ( + session.execute( + select(ApprovalOrder).where( + ApprovalOrder.order_type == order_type, + ApprovalOrder.ref_id == ref_id, + ApprovalOrder.status.in_(_PENDING_STATUSES), + ) + ) + .scalars() + .all() + ) + for order in pending: + order.status = APPROVAL_STATUS_WITHDRAWN + session.add( + ApprovalStep( + order_id=order.id, + node=APPROVAL_NODE_WITHDRAW, + action=APPROVAL_ACTION_WITHDRAW, + operator_id=operator.id, + operator_name=operator.username, + ) + ) + if pending: + session.flush() + logger.info("撤回在途审批单 %s 张 | ref=%s", len(pending), ref_id) + + +def _flush_order(session: Session, order: ApprovalOrder, *, retry_on_conflict: bool) -> None: + """flush 提单；撞 AP 单号唯一键时按 retry_on_conflict 决定重取或抛出。 + + rollback 会连业务写路径的未提交变更一并回滚 —— 见 submit_order 的 + retry_on_conflict 说明（create 才允许内部重取）。 + """ + for attempt in range(_NO_RETRY + 1): + try: + session.flush() + return + except IntegrityError as exc: + session.rollback() + if is_duplicate_entry(exc) and retry_on_conflict and attempt < _NO_RETRY: + order.id = None # rollback 后实例回 transient，主键重新分配 + order.order_no = next_no( + session, + model=ApprovalOrder, + column_name="order_no", + prefix=NO_PREFIX_APPROVAL, + ) + session.add(order) + continue + raise + + +def _detail_of(session: Session, order: ApprovalOrder) -> ApprovalOrderDetailOut: + """出参组装（含 steps 时间线）。""" + steps = ( + session.execute( + select(ApprovalStep).where(ApprovalStep.order_id == order.id).order_by(ApprovalStep.id) + ) + .scalars() + .all() + ) + detail = ApprovalOrderDetailOut.model_validate(order) + detail.steps = [ApprovalStepOut.model_validate(s) for s in steps] + return detail + + +def _order_filters(query: ApprovalQuery) -> tuple: + filters = [] + if query.keyword: + like = f"%{query.keyword}%" + filters.append(ApprovalOrder.order_no.like(like) | ApprovalOrder.refer_no.like(like)) + if query.status is not None: + filters.append(ApprovalOrder.status == query.status) + if query.order_type is not None: + filters.append(ApprovalOrder.order_type == query.order_type) + return tuple(filters) diff --git a/backend/app/services/business_service.py b/backend/app/services/business_service.py index a5a29c6..3568383 100644 --- a/backend/app/services/business_service.py +++ b/backend/app/services/business_service.py @@ -45,6 +45,7 @@ from app.models import ( Shipment, User, ) +from app.models.approval import APPROVAL_TYPE_SECONDARY_ALLOCATION from app.models.business import ( BUSINESS_STATUS_ACTIVE, BUSINESS_STATUS_DONE, @@ -69,7 +70,7 @@ from app.schemas.business import ( SecondaryAllocationIn, SecondaryAllocationOut, ) -from app.services import audit_service, param_service, personnel_service +from app.services import approval_service, audit_service, param_service, personnel_service from app.services.audit_service import AuditRecord, RequestMeta from app.services.formula import converted_profit, profit_rate from app.services.no_generator import next_no @@ -389,6 +390,18 @@ def create_business( business.id, _build_allocation_rows(session, business.id, payload.secondary_allocations), ) + # 方案非空才提单：开关关闭=自动通过留痕，开启=待财务复核。 + # retry_on_conflict=True：AP 编号撞键可内部重取（外层编号 + # 重试循环兜底整体重来） + if payload.secondary_allocations: + approval_service.submit_order( + session, + order_type=APPROVAL_TYPE_SECONDARY_ALLOCATION, + ref_id=business.id, + refer_no=business.business_no, + submitter=operator, + retry_on_conflict=True, + ) customer = customer_repository.get_by_id(session, business.customer_id) out = _to_out((business, customer.company_name if customer else None, None, None)) new_id, new_no = business.id, business.business_no # 固化：commit 后不碰 ORM @@ -469,6 +482,17 @@ def update_business( business_id, _build_allocation_rows(session, business_id, payload.secondary_allocations), ) + # 方案非空提单（自动撤回旧在途单）；retry_on_conflict=False： + # 本路径无外层重试，内部 rollback 会回掉业务行修改造成静默丢失 + if payload.secondary_allocations: + approval_service.submit_order( + session, + order_type=APPROVAL_TYPE_SECONDARY_ALLOCATION, + ref_id=business_id, + refer_no=business.business_no, + submitter=operator, + retry_on_conflict=False, + ) _recalc_totals(session, business) # 锁定行上已赋值，同一身份映射实例的 _to_out 即反映新值；名称字段 # 从首查 row 取（换客户时 _ensure_customer 已给出新公司名） diff --git a/backend/tests/integration/conftest.py b/backend/tests/integration/conftest.py index 52e3b6b..13c4601 100644 --- a/backend/tests/integration/conftest.py +++ b/backend/tests/integration/conftest.py @@ -29,6 +29,7 @@ from app.core.constants import ROLE_ADMIN, ROLE_USER from app.core.db import get_session_factory from app.core.security import create_access_token, hash_password from app.models import ( + ApprovalOrder, Business, BusinessItem, BusinessRole, @@ -37,6 +38,8 @@ from app.models import ( CustomerRatingGrade, OperationLog, OwnershipChangeLog, + ParamChangeBatch, + ParamValue, Personnel, PersonnelPlan, PersonnelRegion, @@ -168,6 +171,18 @@ def _purge_test_data() -> None: session.execute(delete(Product).where(Product.product_name.like("t\\_%"))) # 人员已先删（business_role_id 外键 RESTRICT），直建的测试岗位此时可安全删 session.execute(delete(BusinessRole).where(BusinessRole.role_code.like("t\\_%"))) + + # ---- M3：审批单 / 二次分配 / 参数开关基线 ---- + # 审批单无 t_ 前缀可用，按 AP 编号前缀全量清（steps 外键 CASCADE 随删； + # ref_id 固化无外键，业务已在上一步删除，无悬挂影响） + session.execute(delete(ApprovalOrder).where(ApprovalOrder.order_no.like("AP%"))) + # 开关参数恢复种子基线：测试提交的变更批次条目 id 均大于种子终值 21 + # （V3 十九条 + V6 两条），删新条目与新批次后解析回退到种子值（开关=0） + stray_values = list(session.execute(select(ParamValue).where(ParamValue.id > 21)).scalars()) + if stray_values: + stray_batches = {v.batch_id for v in stray_values} + session.execute(delete(ParamValue).where(ParamValue.id > 21)) + session.execute(delete(ParamChangeBatch).where(ParamChangeBatch.id.in_(stray_batches))) session.commit() diff --git a/backend/tests/integration/test_approval_flow.py b/backend/tests/integration/test_approval_flow.py new file mode 100644 index 0000000..30866a0 --- /dev/null +++ b/backend/tests/integration/test_approval_flow.py @@ -0,0 +1,312 @@ +"""审批流集成测试（二次分配方案为首个接入对象）。 + +守的是（计划 T4 验收矩阵）： +- 开关关闭（种子默认 0）：提单即「自动通过」——status=3 且两级节点 + action=3 留痕（留痕不断链，关闭=直接生效的需求口径） +- 开关开启：status=1 → 财务通过 → status=2 → 领导通过 → status=3 +- 驳回：整体驳回 status=4，意见留痕；驳回必填意见（空驳 400） +- 节点准入：status=1 仅财务、status=2 仅领导；**admin 不代办**（403）； + 普通用户/营销后勤进不了模块（403） +- 再编辑撤旧建新：在途单 status=5 + node=0/action=4 撤回留痕，新单新建 +- 方案为空不提单 +- 开关读一次：同一提交路径内参数变化不影响判定（种子级默认行为锁定） + +角色用户经 make_user 造（finance/leader/marketing_ops 三新角色种子在库）。 +开关联动走参数变更批次（PUT /system/params 端到端，不直改库）。 +""" + +import uuid + +import pytest +from fastapi.testclient import TestClient + +from tests.integration.factories import business_payload, customer + +pytestmark = pytest.mark.integration + +_BUSINESSES = "/api/v1/business/businesses" +_APPROVALS = "/api/v1/system/approvals" +_PARAMS = "/api/v1/system/params" +_PERSONNEL = "/api/v1/base/personnel" + + +def _person(client: TestClient, headers: dict) -> dict: + resp = client.post( + _PERSONNEL, + json={ + "personnel_name": f"t_审批_{uuid.uuid4().hex[:8]}", + "business_role_id": _any_role_id(), + "salary_standard": "3000.00", + }, + headers=headers, + ) + assert resp.status_code == 200, resp.text + return resp.json()["data"] + + +def _any_role_id() -> int: + from sqlalchemy import select + + from app.core.constants import BUSINESS_ROLE_SALESPERSON + from app.core.db import get_session_factory + from app.models import BusinessRole + + factory = get_session_factory() + with factory() as session: + return int( + session.execute( + select(BusinessRole.id).where( + BusinessRole.role_code == BUSINESS_ROLE_SALESPERSON, + BusinessRole.status == 1, + ) + ).scalar_one() + ) + + +def _create_with_plan(client: TestClient, headers: dict, personnel_id: int) -> dict: + cust = customer(client, headers) + resp = client.post( + _BUSINESSES, + json=business_payload( + cust["id"], + secondary_allocations=[{"personnel_id": personnel_id, "ratio": "50.00"}], + ), + headers=headers, + ) + assert resp.status_code == 200, resp.text + return resp.json()["data"] + + +def _latest_order(client: TestClient, headers: dict, refer_no: str) -> dict: + resp = client.get(_APPROVALS, params={"keyword": refer_no, "pageSize": 1}, headers=headers) + assert resp.status_code == 200, resp.text + rows = resp.json()["data"]["list"] + assert rows, f"未找到审批单：{refer_no}" + detail = client.get(f"{_APPROVALS}/{rows[0]['id']}", headers=headers) + return detail.json()["data"] + + +def _set_switch(client: TestClient, admin_headers: dict, value: str) -> None: + """经参数变更批次把二次分配审批开关置值（立即生效）。""" + resp = client.post( + f"{_PARAMS}/changes", + json={ + "change_type": 1, + "remark": f"测试切换审批开关={value}", + "items": [ + { + "param_code": "approval.secondary_allocation.enabled", + "dimension_value": "", + "value": value, + } + ], + }, + headers=admin_headers, + ) + assert resp.status_code == 200, resp.text + + +@pytest.fixture +def finance(make_user, auth_headers) -> dict: + """财务角色账号（审批第二节点）。""" + user = make_user(["finance"]) + user["headers"] = auth_headers(user) + return user + + +@pytest.fixture +def leader(make_user, auth_headers) -> dict: + """领导角色账号（审批第三节点）。""" + user = make_user(["leader"]) + user["headers"] = auth_headers(user) + return user + + +@pytest.fixture +def switched_on(client: TestClient, admin: dict) -> None: + """把二次分配审批开关置 1（立即生效；用例结束由 conftest 恢复种子基线）。""" + _set_switch(client, admin["headers"], "1") + + +class TestSwitchOff: + """开关关闭（种子默认）：提单即自动通过。""" + + def test_创建带方案_自动通过全节点留痕( + self, client: TestClient, admin: dict, finance: dict + ) -> None: + p = _person(client, admin["headers"]) + data = _create_with_plan(client, admin["headers"], p["id"]) + + order = _latest_order(client, finance["headers"], data["business_no"]) + assert order["status"] == 3 + assert order["order_no"].startswith("AP") + auto_steps = [s for s in order["steps"] if s["action"] == 3] + assert len(auto_steps) == 2 + assert {s["node"] for s in auto_steps} == {1, 2} + assert all(s["operator_name"] == "系统" for s in auto_steps) + + def test_方案为空不提单(self, client: TestClient, admin: dict, finance: dict) -> None: + cust = customer(client, admin["headers"]) + resp = client.post(_BUSINESSES, json=business_payload(cust["id"]), headers=admin["headers"]) + assert resp.status_code == 200 + rows = client.get( + _APPROVALS, + params={"keyword": resp.json()["data"]["business_no"]}, + headers=finance["headers"], + ).json()["data"]["list"] + assert rows == [] + + +class TestSwitchOn: + """开关开启：两节点人工流转。""" + + @pytest.fixture + def switched_on(self, client: TestClient, admin: dict) -> None: + _set_switch(client, admin["headers"], "1") + + def test_两节点流转至通过( + self, client: TestClient, admin: dict, finance, leader, switched_on + ) -> None: + + p = _person(client, admin["headers"]) + data = _create_with_plan(client, admin["headers"], p["id"]) + + order = _latest_order(client, finance["headers"], data["business_no"]) + assert order["status"] == 1 + + # 财务通过 → 待领导 + resp = client.post( + f"{_APPROVALS}/{order['id']}/action", + json={"action": 1, "opinion": "同意"}, + headers=finance["headers"], + ) + assert resp.status_code == 200, resp.text + assert resp.json()["data"]["status"] == 2 + + # 领导通过 → 已通过 + resp = client.post( + f"{_APPROVALS}/{order['id']}/action", + json={"action": 1}, + headers=leader["headers"], + ) + assert resp.status_code == 200, resp.text + assert resp.json()["data"]["status"] == 3 + manual = [s for s in resp.json()["data"]["steps"] if s["action"] == 1] + assert {s["node"] for s in manual} == {1, 2} + + def test_财务驳回_整体驳回且意见留痕( + self, client: TestClient, admin: dict, finance, leader, switched_on + ) -> None: + _ = leader + p = _person(client, admin["headers"]) + data = _create_with_plan(client, admin["headers"], p["id"]) + order = _latest_order(client, finance["headers"], data["business_no"]) + + resp = client.post( + f"{_APPROVALS}/{order['id']}/action", + json={"action": 2, "opinion": "比例分配不合理"}, + headers=finance["headers"], + ) + assert resp.status_code == 200, resp.text + out = resp.json()["data"] + assert out["status"] == 4 + rejected = [s for s in out["steps"] if s["action"] == 2] + assert rejected and rejected[0]["opinion"] == "比例分配不合理" + + def test_驳回缺意见被拒( + self, client: TestClient, admin: dict, finance, leader, switched_on + ) -> None: + p = _person(client, admin["headers"]) + data = _create_with_plan(client, admin["headers"], p["id"]) + order = _latest_order(client, finance["headers"], data["business_no"]) + + resp = client.post( + f"{_APPROVALS}/{order['id']}/action", + json={"action": 2}, + headers=finance["headers"], + ) + assert resp.status_code == 400 + + def test_节点准入_领导不能审财务节点_admin不代办( + self, client: TestClient, admin: dict, finance, leader, switched_on + ) -> None: + p = _person(client, admin["headers"]) + data = _create_with_plan(client, admin["headers"], p["id"]) + order = _latest_order(client, leader["headers"], data["business_no"]) + + # status=1 是财务节点：领导不行，admin 也不代办 + for headers in (leader["headers"], admin["headers"]): + resp = client.post( + f"{_APPROVALS}/{order['id']}/action", + json={"action": 1}, + headers=headers, + ) + assert resp.status_code == 403 + + def test_重复操作已通过单被拒( + self, client: TestClient, admin: dict, finance, leader, switched_on + ) -> None: + + p = _person(client, admin["headers"]) + data = _create_with_plan(client, admin["headers"], p["id"]) + order = _latest_order(client, finance["headers"], data["business_no"]) + client.post( + f"{_APPROVALS}/{order['id']}/action", json={"action": 1}, headers=finance["headers"] + ) + client.post( + f"{_APPROVALS}/{order['id']}/action", json={"action": 1}, headers=leader["headers"] + ) + resp = client.post( + f"{_APPROVALS}/{order['id']}/action", json={"action": 1}, headers=finance["headers"] + ) + assert resp.status_code == 400 + + +class TestResubmit: + """再编辑撤旧建新。""" + + def test_在途单撤回并新建( + self, client: TestClient, admin: dict, finance, leader, switched_on + ) -> None: + p = _person(client, admin["headers"]) + data = _create_with_plan(client, admin["headers"], p["id"]) + first = _latest_order(client, finance["headers"], data["business_no"]) + assert first["status"] == 1 + + # 再编辑（仍带方案）→ 旧在途单撤回、新单待审 + detail = client.get(f"{_BUSINESSES}/{data['id']}", headers=admin["headers"]).json()["data"] + payload = business_payload( + detail["customer_id"], + secondary_allocations=[{"personnel_id": p["id"], "ratio": "80.00"}], + ) + resp = client.put(f"{_BUSINESSES}/{data['id']}", json=payload, headers=admin["headers"]) + assert resp.status_code == 200, resp.text + + rows = client.get( + _APPROVALS, + params={"keyword": data["business_no"], "pageSize": 10}, + headers=finance["headers"], + ).json()["data"]["list"] + statuses = {row["status"] for row in rows} + assert 5 in statuses and 1 in statuses, f"未见撤回+新建：{statuses}" + old_detail = client.get(f"{_APPROVALS}/{first['id']}", headers=finance["headers"]).json() + withdraw_steps = [s for s in old_detail["data"]["steps"] if s["action"] == 4] + assert withdraw_steps and withdraw_steps[0]["node"] == 0 + + +class TestModuleAccess: + """模块准入：普通用户与营销后勤不可见（权限清单待澄清 #6）。""" + + def test_普通用户403(self, client: TestClient, normal: dict) -> None: + assert client.get(_APPROVALS, headers=normal["headers"]).status_code == 403 + + def test_营销后勤403_暂不可见(self, client: TestClient, make_user, auth_headers) -> None: + ops = make_user(["marketing_ops"]) + ops["headers"] = auth_headers(ops) + assert client.get(_APPROVALS, headers=ops["headers"]).status_code == 403 + + def test_财务与领导可见(self, client: TestClient, make_user, auth_headers) -> None: + for code in ("finance", "leader"): + user = make_user([code]) + user["headers"] = auth_headers(user) + assert client.get(_APPROVALS, headers=user["headers"]).status_code == 200