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

refactor(backend): update expense claims service and tests

Browse Source 
- services/expense_claims.py: update expense claims service
- tests/test_expense_claim_service.py: update expense claim service tests
This commit is contained in:
caoxiaozhu
2026-05-13 15:42:25 +00:00
parent edb484e2f6
commit 98792c7584
2 changed files with 149 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
server/src/app/services/expense_claims.py
View File

@@ -535,7 +535,6 @@ class ExpenseClaimService:
if is_new_claim:
existing_draft_count = self._count_draft_claims_for_owner(
employee=employee,
employee_name=draft_owner_name,
user_id=user_id,
)
if existing_draft_count >= MAX_DRAFT_CLAIMS_PER_USER:
@@ -733,12 +732,10 @@ class ExpenseClaimService:
self,
*,
employee: Employee | None,
employee_name: str,
user_id: str | None,
) -> int:
owner_filters = self._build_draft_owner_filters(
employee=employee,
employee_name=employee_name,
user_id=user_id,
)
if not owner_filters:
@@ -752,11 +749,10 @@ class ExpenseClaimService:
)
return int(self.db.scalar(stmt) or 0)
@staticmethod
def _build_draft_owner_filters(
self,
*,
employee: Employee | None,
employee_name: str,
user_id: str | None,
) -> list[Any]:
conditions: list[Any] = []
@@ -779,10 +775,10 @@ class ExpenseClaimService:
if employee is not None:
add_condition("employee_id", employee.id)
add_condition("employee_name", employee.name)
add_condition("employee_name", employee.email)
if self._employee_name_is_unique(employee):
add_condition("employee_name", employee.name)
add_condition("employee_name", employee_name)
add_condition("employee_name", user_id)
return conditions
@@ -1607,7 +1603,25 @@ class ExpenseClaimService:
@staticmethod
def _has_privileged_claim_access(current_user: CurrentUserContext) -> bool:
return bool(set(current_user.role_codes) & PRIVILEGED_CLAIM_ROLE_CODES)
role_codes = {
str(item).strip().lower()
for item in current_user.role_codes
if str(item).strip()
}
return bool(role_codes & PRIVILEGED_CLAIM_ROLE_CODES)
def _employee_name_is_unique(self, employee: Employee) -> bool:
normalized_name = str(employee.name or "").strip()
if not normalized_name:
return False
same_name_count = int(
self.db.scalar(
select(func.count()).select_from(Employee).where(Employee.name == normalized_name)
)
or 0
)
return same_name_count == 1
def _apply_claim_scope(self, stmt: Any, current_user: CurrentUserContext) -> Any:
if self._has_privileged_claim_access(current_user):
@@ -1635,8 +1649,9 @@ class ExpenseClaimService:
if employee is not None:
add_condition("employee_id", employee.id)
add_condition("employee_name", employee.name)
add_condition("employee_name", employee.email)
if self._employee_name_is_unique(employee):
add_condition("employee_name", employee.name)
else:
add_condition("employee_id", username)
add_condition("employee_name", username)

125
server/tests/test_expense_claim_service.py
View File

@@ -9,6 +9,7 @@ from sqlalchemy.pool import StaticPool
from app.api.deps import CurrentUserContext
from app.db.base import Base
from app.models.employee import Employee
from app.models.financial_record import ExpenseClaim, ExpenseClaimItem
from app.schemas.ocr import OcrRecognizeBatchRead, OcrRecognizeDocumentRead
from app.schemas.reimbursement import ExpenseClaimItemCreate, ExpenseClaimItemUpdate
@@ -278,3 +279,127 @@ def test_delete_claim_item_removes_row_and_attachment_files(monkeypatch, tmp_pat
assert refreshed_claim.amount == Decimal("0.00")
assert refreshed_claim.invoice_count == 0
assert not attachment_root.exists()
def test_list_claims_scopes_to_current_user_id_even_when_names_duplicate() -> None:
current_user = CurrentUserContext(
username="zhangsan1@example.com",
name="张三",
role_codes=["manager"],
is_admin=False,
)
with build_session() as db:
employee_a = Employee(
employee_no="E2001",
name="张三",
email="zhangsan1@example.com",
)
employee_b = Employee(
employee_no="E2002",
name="张三",
email="zhangsan2@example.com",
)
db.add_all([employee_a, employee_b])
db.flush()
db.add_all(
[
ExpenseClaim(
claim_no="EXP-DUP-001",
employee_id=employee_a.id,
employee_name="张三",
department_name="市场部",
project_code="PRJ-A",
expense_type="travel",
reason="本人报销",
location="上海",
amount=Decimal("120.00"),
currency="CNY",
invoice_count=1,
occurred_at=datetime(2026, 5, 12, 9, 0, tzinfo=UTC),
submitted_at=datetime(2026, 5, 12, 10, 0, tzinfo=UTC),
status="submitted",
approval_stage="finance_review",
risk_flags_json=[],
),
ExpenseClaim(
claim_no="EXP-DUP-002",
employee_id=employee_b.id,
employee_name="张三",
department_name="销售部",
project_code="PRJ-B",
expense_type="meal",
reason="他人报销",
location="杭州",
amount=Decimal("300.00"),
currency="CNY",
invoice_count=1,
occurred_at=datetime(2026, 5, 12, 12, 0, tzinfo=UTC),
submitted_at=datetime(2026, 5, 12, 13, 0, tzinfo=UTC),
status="approved",
approval_stage="completed",
risk_flags_json=[],
),
]
)
db.commit()
claims = ExpenseClaimService(db).list_claims(current_user)
assert len(claims) == 1
assert claims[0].claim_no == "EXP-DUP-001"
def test_list_claims_allows_finance_to_view_all_records() -> None:
current_user = CurrentUserContext(
username="finance@example.com",
name="财务",
role_codes=["finance"],
is_admin=False,
)
with build_session() as db:
db.add_all(
[
ExpenseClaim(
claim_no="EXP-FIN-101",
employee_name="甲",
department_name="A部",
project_code="PRJ-A",
expense_type="travel",
reason="A 报销",
location="上海",
amount=Decimal("120.00"),
currency="CNY",
invoice_count=1,
occurred_at=datetime(2026, 5, 11, 9, 0, tzinfo=UTC),
submitted_at=datetime(2026, 5, 11, 10, 0, tzinfo=UTC),
status="submitted",
approval_stage="finance_review",
risk_flags_json=[],
),
ExpenseClaim(
claim_no="EXP-FIN-102",
employee_name="乙",
department_name="B部",
project_code="PRJ-B",
expense_type="meal",
reason="B 报销",
location="杭州",
amount=Decimal("300.00"),
currency="CNY",
invoice_count=1,
occurred_at=datetime(2026, 5, 11, 12, 0, tzinfo=UTC),
submitted_at=datetime(2026, 5, 11, 13, 0, tzinfo=UTC),
status="approved",
approval_stage="completed",
risk_flags_json=[],
),
]
)
db.commit()
claims = ExpenseClaimService(db).list_claims(current_user)
assert len(claims) == 2
assert {claim.claim_no for claim in claims} == {"EXP-FIN-101", "EXP-FIN-102"}