YG-Soft/X-Financial
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(server): 适配 A/R/D 紧凑单号格式

Browse Source 
- approval_routing/service/user_agent 测试中报销单查询统一兼容 RE- 旧格式与 R+8 新格式，申请单单号断言改为短格式
- generate_claim_no 用例重命名为短前缀校验，正则改为 R[A-HJ-NP-Z2-9]{8}
- 同步更新差旅/交通/通信等财务规则表
This commit is contained in:
caoxiaozhu
2026-06-20 22:04:31 +08:00
parent 0cda750ff0
commit 8158716e23
9 changed files with 59 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
server/tests/test_expense_claim_approval_routing.py
View File

@@ -5,7 +5,7 @@ from datetime import UTC, datetime
from decimal import Decimal
import pytest
from sqlalchemy import create_engine
from sqlalchemy import create_engine, or_
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool
@@ -36,6 +36,15 @@ def build_session() -> Session:
return session_factory()
def reimbursement_claim_query(db: Session):
return db.query(ExpenseClaim).filter(
or_(
ExpenseClaim.claim_no.like("RE-%"),
ExpenseClaim.claim_no.like("R________"),
)
)
def _seed_budget_monitor_role(db: Session) -> Role:
role = Role(role_code="budget_monitor", name="预算管理员")
db.add(role)
@@ -149,7 +158,7 @@ def test_low_risk_application_skips_budget_manager_and_generates_draft() -> None
assert approved is not None
assert approved.status == "approved"
assert approved.approval_stage == APPLICATION_LINK_STATUS_STAGE
assert db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("RE-%")).count() == 1
assert reimbursement_claim_query(db).count() == 1
assert any(
isinstance(flag, dict)
and flag.get("source") == "approval_routing"

53
server/tests/test_expense_claim_service.py
View File

@@ -6,7 +6,7 @@ from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
import pytest
from sqlalchemy import create_engine
from sqlalchemy import create_engine, or_
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool
@@ -18,8 +18,8 @@ from app.models.employee import Employee
from app.models.financial_record import ExpenseClaim, ExpenseClaimItem
from app.models.organization import OrganizationUnit
from app.models.role import Role
from app.schemas.ontology import OntologyParseRequest
from app.schemas.ocr import OcrRecognizeBatchRead, OcrRecognizeDocumentRead
from app.schemas.ontology import OntologyParseRequest
from app.schemas.reimbursement import (
ExpenseClaimItemCreate,
ExpenseClaimItemUpdate,
@@ -28,19 +28,19 @@ from app.schemas.reimbursement import (
)
from app.services.agent_conversations import AgentConversationService
from app.services.budget import BudgetService
from app.services.expense_claim_budget_flow import ExpenseClaimBudgetFlowMixin
from app.services.expense_claim_attachment_storage import ExpenseClaimAttachmentStorage
from app.services.expense_claims import ExpenseClaimService
from app.services.expense_claim_budget_flow import ExpenseClaimBudgetFlowMixin
from app.services.expense_claim_workflow_constants import (
APPROVAL_DONE_STAGE,
APPLICATION_ARCHIVE_STAGE,
APPLICATION_LINK_STATUS_STAGE,
APPROVAL_DONE_STAGE,
BUDGET_MANAGER_APPROVAL_STAGE,
DIRECT_MANAGER_APPROVAL_STAGE,
FINANCE_APPROVAL_STAGE,
)
from app.services.ontology import SemanticOntologyService
from app.services.expense_claims import ExpenseClaimService
from app.services.ocr import OcrService
from app.services.ontology import SemanticOntologyService
from app.services.receipt_folder import ReceiptFolderService
@@ -121,6 +121,15 @@ def build_session() -> Session:
return session_factory()
def reimbursement_claim_query(db: Session):
return db.query(ExpenseClaim).filter(
or_(
ExpenseClaim.claim_no.like("RE-%"),
ExpenseClaim.claim_no.like("R________"),
)
)
def test_append_budget_flags_replaces_duplicate_budget_warning() -> None:
base_warning = {
"source": "budget_control",
@@ -1770,7 +1779,7 @@ def test_upsert_draft_from_ontology_updates_returned_claim_and_preserves_return_
assert manual_returns == [return_flag]
def test_generate_claim_no_uses_re_prefix_timestamp_and_random_suffix() -> None:
def test_generate_claim_no_uses_short_reimbursement_prefix_and_random_suffix() -> None:
with build_session() as db:
db.add_all(
[
@@ -1813,7 +1822,7 @@ def test_generate_claim_no_uses_re_prefix_timestamp_and_random_suffix() -> None:
service = ExpenseClaimService(db)
assert re.fullmatch(
r"RE-\d{14}-[A-HJ-NP-Z2-9]{8}",
r"R[A-HJ-NP-Z2-9]{8}",
service._generate_claim_no(datetime(2026, 5, 14, tzinfo=UTC)),
)
@@ -1831,7 +1840,7 @@ def test_upsert_draft_from_ontology_retries_claim_no_conflict() -> None:
db.flush()
db.add(
ExpenseClaim(
claim_no="RE-20260525101010-ABCDEFGH",
claim_no="RABCDEFGH",
employee_name="历史单据",
department_name="财务部",
project_code=None,
@@ -1856,9 +1865,7 @@ def test_upsert_draft_from_ontology_retries_claim_no_conflict() -> None:
)
)
service = ExpenseClaimService(db)
generated_claim_nos = iter(
["RE-20260525101010-ABCDEFGH", "RE-20260525101010-HGFEDCBA"]
)
generated_claim_nos = iter(["RABCDEFGH", "RHGFEDCBA"])
service._generate_claim_no = lambda occurred_at: next(generated_claim_nos)
result = service.upsert_draft_from_ontology(
@@ -1874,8 +1881,8 @@ def test_upsert_draft_from_ontology_retries_claim_no_conflict() -> None:
created_claim = db.get(ExpenseClaim, result["claim_id"])
assert created_claim is not None
assert created_claim.claim_no == "RE-20260525101010-HGFEDCBA"
assert result["claim_no"] == "RE-20260525101010-HGFEDCBA"
assert created_claim.claim_no == "RHGFEDCBA"
assert result["claim_no"] == "RHGFEDCBA"
def test_create_claim_item_adds_blank_draft_row_without_forcing_attachment() -> None:
@@ -5695,7 +5702,7 @@ def test_direct_manager_can_route_application_claim_to_budget_approval_then_budg
assert leader_approved is not None
assert leader_approved.status == "submitted"
assert leader_approved.approval_stage == "预算管理者审批"
assert db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("RE-%")).count() == 0
assert reimbursement_claim_query(db).count() == 0
assert any(
isinstance(flag, dict)
and flag.get("source") == "manual_approval"
@@ -5727,7 +5734,7 @@ def test_direct_manager_can_route_application_claim_to_budget_approval_then_budg
)
)
assert all(claim.claim_no != "APP-20260525-APPROVE" for claim in archived_claims)
generated_draft = db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("RE-%")).one()
generated_draft = reimbursement_claim_query(db).one()
assert generated_draft.status == "draft"
assert generated_draft.approval_stage == "待提交"
assert generated_draft.expense_type == "travel"
@@ -5949,7 +5956,7 @@ def test_direct_manager_cannot_route_application_to_missing_budget_approver() ->
db.refresh(claim)
assert claim.status == "submitted"
assert claim.approval_stage == DIRECT_MANAGER_APPROVAL_STAGE
assert db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("RE-%")).count() == 0
assert reimbursement_claim_query(db).count() == 0
def test_direct_manager_p8_executive_completes_application_without_duplicate_budget_approval() -> None:
@@ -6023,7 +6030,7 @@ def test_direct_manager_p8_executive_completes_application_without_duplicate_bud
assert approved is not None
assert approved.status == "approved"
assert approved.approval_stage == APPLICATION_LINK_STATUS_STAGE
assert db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("RE-%")).count() == 1
assert reimbursement_claim_query(db).count() == 1
assert not any(
isinstance(flag, dict)
and flag.get("next_approval_stage") == BUDGET_MANAGER_APPROVAL_STAGE
@@ -6111,7 +6118,7 @@ def test_direct_manager_budget_monitor_completes_application_claim_without_dupli
assert approved is not None
assert approved.status == "approved"
assert approved.approval_stage == "关联单据状态"
assert db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("RE-%")).count() == 1
assert reimbursement_claim_query(db).count() == 1
assert not any(
isinstance(flag, dict)
and flag.get("next_approval_stage") == "预算管理者审批"
@@ -6130,7 +6137,7 @@ def test_direct_manager_budget_monitor_completes_application_claim_without_dupli
and flag.get("budget_approval_merged_reason") == "direct_manager_is_department_budget_approver"
for flag in approved.risk_flags_json
)
generated_draft = db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("RE-%")).one()
generated_draft = reimbursement_claim_query(db).one()
assert generated_draft.status == "draft"
assert generated_draft.expense_type == "travel"
reviewer_claims = ExpenseClaimService(db).list_claims(manager_user)
@@ -6333,10 +6340,10 @@ def test_application_approval_transfers_budget_reservation_to_reimbursement_draf
assert leader_approved.approval_stage == "预算管理者审批"
assert reservation.source_type == "application"
assert reservation.source_id == claim.id
assert db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("RE-%")).count() == 0
assert reimbursement_claim_query(db).count() == 0
approved = service.approve_claim(claim.id, budget_user, opinion="预算通过")
generated_draft = db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("RE-%")).one()
generated_draft = reimbursement_claim_query(db).one()
db.refresh(reservation)
assert approved is not None
@@ -6428,7 +6435,7 @@ def test_direct_manager_approval_defaults_blank_opinion_to_agree() -> None:
and flag.get("next_approval_stage") == "预算管理者审批"
for flag in approved.risk_flags_json
)
assert db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("RE-%")).count() == 0
assert reimbursement_claim_query(db).count() == 0
def test_budget_analysis_uses_current_application_reservation_without_double_counting() -> None:

26
server/tests/test_user_agent_service.py
View File

@@ -4,14 +4,14 @@ import re
from datetime import UTC, datetime, timedelta
from decimal import Decimal
from sqlalchemy import create_engine
from sqlalchemy import create_engine, or_
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool
from app.core.agent_enums import AgentAssetType
from app.db.base import Base
from app.models.employee import Employee
from app.models.financial_record import ExpenseClaim
from app.core.agent_enums import AgentAssetType
from app.schemas.ontology import OntologyParseRequest
from app.schemas.user_agent import UserAgentCitation, UserAgentRequest, UserAgentReviewRiskBrief
from app.services.agent_assets import AgentAssetService
@@ -30,6 +30,16 @@ def build_session_factory() -> sessionmaker[Session]:
return sessionmaker(bind=engine, autoflush=False, autocommit=False)
def application_claim_query(db: Session):
return db.query(ExpenseClaim).filter(
or_(
ExpenseClaim.claim_no.like("AP-%"),
ExpenseClaim.claim_no.like("APP-%"),
ExpenseClaim.claim_no.like("A________"),
)
)
def build_application_user_agent_response(
db: Session,
message: str,
@@ -629,9 +639,9 @@ def test_user_agent_application_submit_enters_leader_review() -> None:
assert "系统已推送给 陈硕 审核，当前节点：陈硕审核中" in response.answer
assert "下方是简要单据信息" in response.answer
assert "申请信息：" not in response.answer
assert re.search(r"AP-\d{14}-[A-HJ-NP-Z2-9]{8}", response.answer)
assert re.search(r"A[A-HJ-NP-Z2-9]{8}", response.answer)
assert response.suggested_actions == []
claim = db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("AP-%")).one()
claim = application_claim_query(db).one()
assert claim.status == "submitted"
assert claim.approval_stage == "直属领导审批"
assert claim.expense_type == "travel_application"
@@ -675,7 +685,7 @@ def test_user_agent_application_submit_blocks_duplicate_business_time() -> None:
context_overrides={"manager_name": "陈硕"},
history=history,
)
first_claim = db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("AP-%")).one()
first_claim = application_claim_query(db).one()
second_response = build_application_user_agent_response(
db,
@@ -684,7 +694,7 @@ def test_user_agent_application_submit_blocks_duplicate_business_time() -> None:
history=history,
)
claims = db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("AP-%")).all()
claims = application_claim_query(db).all()
assert len(claims) == 1
assert "申请单据已生成" in first_response.answer
assert "已存在申请单" in second_response.answer
@@ -745,7 +755,7 @@ def test_user_agent_application_submit_blocks_overlapping_travel_dates() -> None
},
)
claims = db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("AP-%")).all()
claims = application_claim_query(db).all()
assert len(claims) == 1
assert "已存在申请单" in response.answer
assert "系统没有重复创建" in response.answer
@@ -841,7 +851,7 @@ def test_user_agent_application_edit_resubmits_returned_application_claim() -> N
)
db.refresh(claim)
claims = db.query(ExpenseClaim).filter(ExpenseClaim.claim_no.like("AP-%")).all()
claims = application_claim_query(db).all()
assert len(claims) == 1
assert "申请单据已修改并重新提交" in response.answer
assert response.draft_payload is not None