X-Financial/server/tests/test_savings_endpoints.py

from __future__ import annotations

import uuid
from collections.abc import Generator
from datetime import UTC, date, datetime
from decimal import Decimal

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool

import app.models  # noqa: F401 - 注册完整 metadata
from app.api.deps import CurrentUserContext, get_current_user, get_db
from app.api.v1.endpoints.cfo_value import router as cfo_value_router
from app.api.v1.endpoints.savings import router
from app.db.base_class import Base
from app.models.financial_record import ExpenseClaim, ExpenseClaimItem
from app.services.savings_discovery import SavingsDiscoveryService


@pytest.fixture()
def http_context() -> Generator[
    tuple[TestClient, sessionmaker[Session], dict[str, CurrentUserContext]],
    None,
    None,
]:
    engine = create_engine(
        "sqlite+pysqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(engine)
    factory = sessionmaker(bind=engine, expire_on_commit=False)
    app = FastAPI()
    app.include_router(router, prefix="/api/v1")
    app.include_router(cfo_value_router, prefix="/api/v1")
    user_box = {"current": _user("finance-a", roles=["finance"])}

    def override_db() -> Generator[Session, None, None]:
        with factory() as db:
            yield db

    app.dependency_overrides[get_db] = override_db
    app.dependency_overrides[get_current_user] = lambda: user_box["current"]
    client = TestClient(app)
    try:
        yield client, factory, user_box
    finally:
        client.close()
        app.dependency_overrides.clear()
        Base.metadata.drop_all(engine)
        engine.dispose()


def test_savings_http_list_detail_record_and_tenant_boundary(
    http_context: tuple[
        TestClient,
        sessionmaker[Session],
        dict[str, CurrentUserContext],
    ],
) -> None:
    client, factory, user_box = http_context
    with factory() as db:
        opportunity_id = _seed_discovered_opportunity(db, user_box["current"])
        db.commit()

    list_response = client.get("/api/v1/savings/opportunities")
    assert list_response.status_code == 200
    assert list_response.json()["total"] == 1
    assert list_response.json()["items"][0]["id"] == opportunity_id

    detail_response = client.get(f"/api/v1/savings/opportunities/{opportunity_id}")
    assert detail_response.status_code == 200
    assert detail_response.json()["baseline"]["policy_version"] == "endpoint-policy-v1"
    assert len(detail_response.json()["evidence"]) == 1

    record_response = client.post(
        f"/api/v1/savings/opportunities/{opportunity_id}/realizations",
        json={
            "request_id": "endpoint-record-001",
            "expected_version": 1,
            "comment": "付款状态完成，登记待确认结果",
            "actual_gross": "200.00",
            "incremental_cost": "0.00",
            "currency": "CNY",
            "realized_at": datetime.now(UTC).isoformat(),
            "attribution_method": "server_policy_counterfactual",
            "attribution_ratio": "1.0",
            "evidence_level": "business_state",
            "evidence": [
                {
                    "evidence_key": "endpoint-payment-state-001",
                    "evidence_role": "payment_business_state",
                    "resource_type": "business_event",
                    "resource_id": "payment-event-endpoint-001",
                    "source_system": "x-financial",
                    "external_event_id": "payment-event-endpoint-001",
                    "content_hash": "a" * 64,
                    "occurred_at": datetime.now(UTC).isoformat(),
                    "verification_status": "unverified",
                    "metadata_json": {"source": "endpoint-test"},
                }
            ],
        },
    )
    assert record_response.status_code == 200
    assert record_response.json()["realization"]["status"] == "pending_confirmation"
    assert record_response.json()["opportunity"]["status"] == "realized"

    cfo_response = client.get("/api/v1/analytics/cfo-value")
    assert cfo_response.status_code == 200
    assert cfo_response.json()["kpis"]["verified_cash"]["status"] == "empty"
    assert cfo_response.json()["data_quality"]["pending_confirmation_count"] == 1

    user_box["current"] = _user("ordinary-user")
    assert client.get("/api/v1/analytics/cfo-value").status_code == 403

    user_box["current"] = _user("finance-other", tenant_id="tenant-b", roles=["finance"])
    cross_tenant = client.get(f"/api/v1/savings/opportunities/{opportunity_id}")
    assert cross_tenant.status_code == 404
    assert client.get("/api/v1/savings/opportunities").json()["total"] == 0


def _seed_discovered_opportunity(
    db: Session,
    current_user: CurrentUserContext,
) -> str:
    claim = ExpenseClaim(
        id=str(uuid.uuid4()),
        claim_no=f"BX-{uuid.uuid4().hex[:10]}",
        employee_name="测试员工",
        department_name="销售部",
        project_code="PROJECT-1",
        expense_type="hotel",
        reason="客户现场差旅",
        location="深圳",
        amount=Decimal("1000.00"),
        currency="CNY",
        invoice_count=1,
        occurred_at=datetime.now(UTC),
        status="draft",
        risk_flags_json=[],
    )
    item = ExpenseClaimItem(
        id=str(uuid.uuid4()),
        claim=claim,
        item_date=date(2026, 7, 15),
        item_type="hotel",
        item_reason="深圳住宿",
        item_location="深圳",
        item_note="",
        item_amount=Decimal("1000.00"),
    )
    db.add(claim)
    db.flush()
    opportunity = SavingsDiscoveryService(db).discover_standard_adjustments(
        claim=claim,
        items_by_id={item.id: item},
        adjustment_flags=[
            {
                "item_id": item.id,
                "message": "服务端政策重算",
                "original_amount": "1000.00",
                "reimbursable_amount": "800.00",
                "employee_absorbed_amount": "200.00",
                "policy_rule_version": "endpoint-policy-v1",
                "policy_grade": "P6",
                "policy_matched_city": "深圳",
                "calculation_fingerprint": "sha256:" + "c" * 64,
            }
        ],
        current_user=current_user,
        request_id="endpoint-discovery-001",
    )[0]
    return opportunity.id


def _user(
    username: str,
    *,
    tenant_id: str = "default",
    roles: list[str] | None = None,
) -> CurrentUserContext:
    return CurrentUserContext(
        username=username,
        name=username,
        role_codes=list(roles or []),
        is_admin=False,
        tenant_id=tenant_id,
        employee_id=username,
    )