from __future__ import annotations from collections.abc import Generator from datetime import UTC, datetime from decimal import Decimal import pytest from auth_helpers import install_legacy_header_auth_override from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from sqlalchemy.pool import StaticPool from app.api.deps import get_db from app.db.base import Base from app.main import create_app from app.models.employee import Employee from app.models.financial_record import ExpenseClaim from app.services.expense_cases import ExpenseCaseService from app.services.expense_claim_workflow_constants import ( DIRECT_MANAGER_APPROVAL_STAGE, FINANCE_APPROVAL_STAGE, ) @pytest.fixture def http_context() -> Generator[tuple[TestClient, sessionmaker[Session]], None, None]: engine = create_engine( "sqlite+pysqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) Base.metadata.create_all(bind=engine) session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False) app = create_app() install_legacy_header_auth_override(app) def override_db() -> Generator[Session, None, None]: with session_factory() as db: yield db app.dependency_overrides[get_db] = override_db client = TestClient(app) try: yield client, session_factory finally: client.close() app.dependency_overrides.clear() engine.dispose() def seed_timeline( db: Session, *, claim_no: str = "RE-TIMELINE-001", expense_type: str = "travel", relation_type: str | None = None, approval_stage: str = DIRECT_MANAGER_APPROVAL_STAGE, with_case: bool = True, tenant_id: str = "default", ) -> ExpenseClaim: manager = Employee( id="manager-1", employee_no="M001", name="李经理", email="manager@example.com", ) owner = Employee( id="owner-1", employee_no="E001", name="张三", email="owner@example.com", manager=manager, ) claim = ExpenseClaim( id="claim-1", claim_no=claim_no, employee_id=owner.id, employee_name=owner.name, department_name="市场部", expense_type=expense_type, reason="客户现场差旅", location="上海", amount=Decimal("880.00"), currency="CNY", invoice_count=1, occurred_at=datetime(2026, 7, 13, 9, 0, tzinfo=UTC), submitted_at=datetime(2026, 7, 13, 10, 0, tzinfo=UTC), status="submitted", approval_stage=approval_stage, risk_flags_json=[], ) db.add_all([manager, owner, claim]) db.flush() if with_case: ExpenseCaseService(db).record_claim_event( claim, event_type="claim_submitted", actor_id=owner.email, tenant_id=tenant_id, correlation_id="timeline-http-permission", idempotency_key="timeline-http-permission", previous_status="draft", previous_approval_stage="待提交", relation_type=relation_type, extra_payload={ "opinion": "同意，按计划执行。", "internal_route": "finance-secret-route", "archived_applications": [ { "application_claim_id": "internal-application-id", "application_claim_no": "AP-TIMELINE-001", "archive_event_id": "internal-archive-event-id", } ], }, ) db.commit() return claim @pytest.mark.parametrize( ("approval_stage", "headers"), [ ( DIRECT_MANAGER_APPROVAL_STAGE, { "x-auth-username": "owner@example.com", "x-auth-name": "Owner", "x-auth-employee-no": "E001", "x-auth-role-codes": "user", }, ), ( DIRECT_MANAGER_APPROVAL_STAGE, { "x-auth-username": "manager@example.com", "x-auth-name": "Manager", "x-auth-employee-no": "M001", "x-auth-role-codes": "approver", }, ), ( FINANCE_APPROVAL_STAGE, { "x-auth-username": "finance@example.com", "x-auth-name": "Finance", "x-auth-role-codes": "finance", }, ), ( DIRECT_MANAGER_APPROVAL_STAGE, { "x-auth-username": "admin", "x-auth-name": "Admin", "x-auth-is-admin": "true", }, ), ], ids=["owner", "current-manager", "finance", "admin"], ) def test_expense_case_timeline_allows_supported_viewers( http_context: tuple[TestClient, sessionmaker[Session]], approval_stage: str, headers: dict[str, str], ) -> None: client, session_factory = http_context with session_factory() as db: claim = seed_timeline(db, approval_stage=approval_stage) claim_id = claim.id response = client.get(f"/api/v1/expense-cases/by-claim/{claim_id}", headers=headers) assert response.status_code == 200 payload = response.json() assert set(payload) == {"id", "case_no", "current_stage", "status", "links", "events"} assert payload["case_no"] == "CASE-RE-TIMELINE-001" assert payload["links"] == [{"relation_type": "claim"}] assert len(payload["events"]) == 1 event = payload["events"][0] assert set(event) == { "id", "event_type", "actor_id", "actor_type", "payload_json", "occurred_at", } assert event["actor_id"] == "owner@example.com" assert event["payload_json"]["opinion"] == "同意，按计划执行。" assert event["payload_json"]["archived_applications"] == [ {"application_claim_no": "AP-TIMELINE-001"} ] serialized = response.text for internal_value in ( "idempotency_key", "correlation_id", "causation_id", "delivery_status", "resource_id", "internal_route", "internal-application-id", "internal-archive-event-id", ): assert internal_value not in serialized def test_expense_case_timeline_hides_claim_existence_from_unrelated_user( http_context: tuple[TestClient, sessionmaker[Session]], ) -> None: client, session_factory = http_context with session_factory() as db: claim_id = seed_timeline(db).id response = client.get( f"/api/v1/expense-cases/by-claim/{claim_id}", headers={ "x-auth-username": "unrelated@example.com", "x-auth-name": "Unrelated User", "x-auth-employee-no": "E999", "x-auth-role-codes": "user", }, ) assert response.status_code == 404 assert response.json()["detail"] == "费用单据不存在。" def test_expense_case_timeline_rejects_cross_tenant_lookup( http_context: tuple[TestClient, sessionmaker[Session]], ) -> None: client, session_factory = http_context with session_factory() as db: claim_id = seed_timeline(db, tenant_id="tenant-a").id response = client.get( f"/api/v1/expense-cases/by-claim/{claim_id}", headers={ "x-auth-username": "owner@example.com", "x-auth-name": "Owner", "x-auth-employee-no": "E001", "x-auth-role-codes": "user", "x-auth-tenant-id": "tenant-b", }, ) assert response.status_code == 404 assert response.json()["detail"] == "该单据尚未纳入统一费用事件。" def test_expense_case_timeline_returns_not_covered_for_claim_without_case( http_context: tuple[TestClient, sessionmaker[Session]], ) -> None: client, session_factory = http_context with session_factory() as db: claim_id = seed_timeline(db, with_case=False).id response = client.get( f"/api/v1/expense-cases/by-claim/{claim_id}", headers={ "x-auth-username": "owner@example.com", "x-auth-name": "Owner", "x-auth-employee-no": "E001", "x-auth-role-codes": "user", }, ) assert response.status_code == 404 assert response.json()["detail"] == "该单据尚未纳入统一费用事件。" def test_expense_case_timeline_returns_safe_summary_for_all_linked_claims( http_context: tuple[TestClient, sessionmaker[Session]], ) -> None: client, session_factory = http_context with session_factory() as db: application = seed_timeline( db, claim_no="AP-TIMELINE-001", expense_type="travel_application", relation_type="application", ) service = ExpenseCaseService(db) expense_case = service.get_timeline_for_claim(application.id) assert expense_case is not None reimbursement = ExpenseClaim( id="claim-2", claim_no="RE-TIMELINE-002", employee_id=application.employee_id, employee_name=application.employee_name, department_name=application.department_name, expense_type="travel", reason="客户现场差旅报销", location="上海", amount=Decimal("880.00"), currency="CNY", invoice_count=1, occurred_at=application.occurred_at, submitted_at=application.submitted_at, status="submitted", approval_stage=FINANCE_APPROVAL_STAGE, risk_flags_json=[], ) db.add(reimbursement) db.flush() service.record_claim_event( reimbursement, event_type="claim_submitted", actor_id="owner@example.com", expense_case=expense_case, relation_type="generated_reimbursement", correlation_id="linked-reimbursement", idempotency_key="linked-reimbursement", previous_status="draft", previous_approval_stage="待提交", ) db.commit() response = client.get( "/api/v1/expense-cases/by-claim/claim-1", headers={ "x-auth-username": "owner@example.com", "x-auth-name": "Owner", "x-auth-employee-no": "E001", "x-auth-role-codes": "user", }, ) assert response.status_code == 200 payload = response.json() assert {item["relation_type"] for item in payload["links"]} == { "application", "generated_reimbursement", } assert [event["event_type"] for event in payload["events"]] == [ "claim_submitted", "claim_submitted", ]