zhifu-platform/tests/test_billing_statements.py

import asyncio
from uuid import uuid4

import pytest
from sqlalchemy import func, select
from test_after_sales import scenario  # noqa: F401
from test_api_documentation import validate_response
from test_money_budget import PREFIX, PRICE, approve, budget, report, reserve_sample, setup
from test_token_budget import sample

from zhifu.modules.agent_runtime.budget import change_entry
from zhifu.modules.billing.models import BillingStatement
from zhifu.modules.identity.access_models import Principal
from zhifu.operations.billing_access import change as access

pytestmark = pytest.mark.integration


async def prepared(context):
    await setup(context)
    await approve(context, PRICE)
    await budget(context, 100_000)
    identifier = uuid4()
    await reserve_sample(context, identifier.bytes)
    await change_entry(
        context["app"].state.sessions,
        context["tenant"].bytes,
        identifier.bytes,
        usage=sample()["usage"],
    )
    usage = sample()["usage"]
    # 用户授权的测试账单用于真实事务验收，费用字段来自现有验收价格的计算结果。
    return {
        "provider": PRICE["provider"],
        "reference": "integration-" + str(uuid4()),
        "evidence": "自动生成的事务测试账单，不代表供应商账单",
        "currency": "CNY",
        "lines": [
            {
                "record_id": "sample-1",
                "entry_id": str(identifier),
                "model": PRICE["model"],
                "amount": 912,
                "input_tokens": usage["input_tokens"],
                "output_tokens": usage["output_tokens"],
                "cached_tokens": usage["input_token_details"]["cache_read"],
            }
        ],
    }


async def test_import_comparison_idempotency_and_atomicity(scenario):  # noqa: F811
    body = await prepared(scenario)
    call = scenario["call"]
    path = PREFIX + "/statements/" + str(uuid4())
    before = await report(scenario)
    results = await asyncio.gather(*(call("PUT", path, "agent", body) for _ in range(4)))
    for response in results:
        validate_response(
            scenario["app"].openapi(), PREFIX + "/statements/{statement_id}", "put", response, 200
        )
        assert response.json()["items"][0]["status"] == "MATCHED"
    assert await report(scenario) == before
    assert (await call("PUT", path, "agent", {**body, "evidence": "不同证据"})).status_code == 409
    assert (await call("PUT", PREFIX + "/statements/" + str(uuid4()), "agent", body)).json()[
        "code"
    ] == "BILLING_SOURCE_EXISTS"
    invalid = {
        **body,
        "reference": "invalid",
        "lines": [
            *body["lines"],
            {**body["lines"][0], "record_id": "bad", "entry_id": str(uuid4())},
        ],
    }
    assert (
        await call("PUT", PREFIX + "/statements/" + str(uuid4()), "agent", invalid)
    ).status_code == 404
    async with scenario["app"].state.sessions() as session:
        assert (
            await session.scalar(
                select(func.count())
                .select_from(BillingStatement)
                .where(BillingStatement.tenant_id == scenario["tenant"].bytes)
            )
            == 1
        )
    response = await call("GET", PREFIX + "/statements?limit=1", "agent")
    validate_response(scenario["app"].openapi(), PREFIX + "/statements", "get", response, 200)
    assert response.json()["next_cursor"] is None


async def test_difference_approval_and_current_comparison(scenario):  # noqa: F811
    body = await prepared(scenario)
    body["lines"][0]["amount"] = 900
    identifier = uuid4()
    path = PREFIX + "/statements/" + str(identifier)
    response = await scenario["call"]("PUT", path, "agent", body)
    item = response.json()["items"][0]
    assert item["status"] == "DIFFERENT" and item["differences"] == ["amount"]
    payload = {
        "kind": "RECONCILE",
        "expected_version": item["entry_version"],
        "entry_id": body["lines"][0]["entry_id"],
        "amount": 900,
        "reason": "核实测试账单",
        "statement_id": str(identifier),
        "record_id": "sample-1",
    }
    wrong = await scenario["call"](
        "PUT", PREFIX + "/requests/" + str(uuid4()), "agent", {**payload, "amount": 901}
    )
    assert wrong.json()["code"] == "BILLING_EVIDENCE_CONFLICT"
    request_path = PREFIX + "/requests/" + str(uuid4())
    assert (await scenario["call"]("PUT", request_path, "agent", payload)).status_code == 200
    decision = {"expected_version": 1, "approve": True, "reason": "独立核对测试账单"}
    assert (
        await scenario["call"]("POST", request_path + "/decisions", "agent", decision)
    ).status_code == 403
    response = await scenario["call"]("POST", request_path + "/decisions", "reviewer", decision)
    assert response.status_code == 200, response.text
    assert (await report(scenario))["consumed"] == 900
    response = await scenario["call"]("GET", path, "agent")
    assert response.json()["items"][0]["status"] == "MATCHED"
    # 重复导入只刷新比较结果，不再次修改账本。
    assert (await scenario["call"]("PUT", path, "agent", body)).status_code == 200
    assert (await report(scenario))["consumed"] == 900
    response = await scenario["call"]("PUT", PREFIX + "/requests/" + str(uuid4()), "agent", payload)
    response = await scenario["call"](
        "POST", PREFIX + "/requests/" + response.json()["id"] + "/decisions", "reviewer", decision
    )
    assert response.json()["code"] == "BILLING_VERSION_CONFLICT"


async def test_permissions_validation_and_unmatched_records(scenario):  # noqa: F811
    body = await prepared(scenario)
    call = scenario["call"]
    path = PREFIX + "/statements/" + str(uuid4())
    assert (await call("PUT", path, "customer", body)).status_code == 403
    assert (await scenario["client"].get(PREFIX + "/statements")).status_code == 401
    for invalid in (
        {**body, "currency": "USD"},
        {**body, "lines": body["lines"] * 2},
        {**body, "lines": [{**body["lines"][0], "amount": 0.5}]},
        {**body, "lines": [{**body["lines"][0], "cached_tokens": 10**12}]},
    ):
        assert (await call("PUT", path, "agent", invalid)).status_code == 422
    body["lines"][0]["cached_tokens"] = None
    response = await call("PUT", path, "agent", body)
    assert response.json()["items"][0]["status"] == "UNKNOWN"
    unlinked = {
        **body,
        "reference": "unlinked",
        "lines": [{**body["lines"][0], "entry_id": None}],
    }
    response = await call("PUT", PREFIX + "/statements/" + str(uuid4()), "agent", unlinked)
    assert response.json()["items"][0]["status"] == "UNLINKED"
    mismatch = {**body, "reference": "route-mismatch", "provider": "https://example.org"}
    response = await call("PUT", PREFIX + "/statements/" + str(uuid4()), "agent", mismatch)
    assert response.json()["items"][0]["status"] == "ROUTE_MISMATCH"
    first = (await call("GET", PREFIX + "/statements?limit=1", "agent")).json()
    second = (
        await call("GET", PREFIX + "/statements?limit=1&after=" + first["next_cursor"], "agent")
    ).json()
    assert first["items"][0]["id"] != second["items"][0]["id"]
    async with scenario["app"].state.sessions.begin() as session:
        foreign = await session.get(Principal, scenario["accounts"]["foreign"].bytes)
        foreign.kind = "STAFF"
        tenant = foreign.tenant_id
    for permission in ("billing:read", "billing:request"):
        await access(
            scenario["app"].state.sessions,
            tenant,
            scenario["accounts"]["foreign"].bytes,
            permission,
            True,
            "integration",
            "租户隔离验证",
        )
    assert (await call("GET", path, "foreign")).status_code == 404
    assert (await call("PUT", path, "foreign", body)).status_code == 404
    assert (
        await call("PUT", PREFIX + "/statements/" + str(uuid4()), "foreign", body)
    ).status_code == 404
    assert (await call("GET", PREFIX + "/statements", "foreign")).json()["items"] == []