zhifu-platform/tests/container_model_acceptance.py

import asyncio
import json
from pathlib import Path
from uuid import UUID, uuid4

import httpx
from sqlalchemy import func, select
from test_after_sales import scenario  # noqa: F401

from zhifu.modules.after_sales.models import Approval, Ticket
from zhifu.modules.agent_runtime.models import AgentRun, ModelAttempt
from zhifu.modules.integrations.models import ExternalOperation


async def test_container_agent_uses_real_model_and_tools(scenario):  # noqa: F811
    async with httpx.AsyncClient(
        base_url="http://127.0.0.1:28000",
        headers=scenario["headers"]["customer"],
        timeout=10,
        trust_env=False,
    ) as client:
        response = await client.post(
            "/openapi/v1/agent/runs",
            headers={"Idempotency-Key": uuid4().hex},
            json={
                "store_id": str(scenario["store"]),
                "message": (
                    f"请查询订单 {scenario['order_id']}，商品 keyboard-1 的键盘按键失灵。"
                    "请创建数量为 1 的换货方案，并告知下一步操作。"
                ),
            },
        )
        assert response.status_code == 202, response.text
        run_id = UUID(response.json()["run_id"])
        async with asyncio.timeout(150):
            while True:
                response = await client.get(f"/openapi/v1/agent/runs/{run_id}")
                assert response.status_code == 200, response.text
                if response.json()["status"] not in {"QUEUED", "RUNNING"}:
                    break
                await asyncio.sleep(1)
        assert response.json()["status"] == "COMPLETED", response.text
    async with scenario["app"].state.sessions() as session:
        run = await session.get(AgentRun, run_id.bytes)
        calls = {
            call["name"]
            for message in run.messages
            for call in message["data"].get("tool_calls", [])
        }
        assert {"get_order", "create_exchange_proposal"} <= calls
        attempts = (
            await session.scalars(select(ModelAttempt).where(ModelAttempt.run_id == run_id.bytes))
        ).all()
        assert attempts and all(value.status == "RECEIVED" and value.usage for value in attempts)
        tickets = (
            await session.scalars(
                select(Ticket).where(Ticket.tenant_id == scenario["tenant"].bytes)
            )
        ).all()
        assert len(tickets) == 1 and tickets[0].status == "DRAFT"
        for model in (Approval, ExternalOperation):
            assert (
                await session.scalar(
                    select(func.count())
                    .select_from(model)
                    .where(model.tenant_id == scenario["tenant"].bytes)
                )
                == 0
            )
        report = {
            "status": response.json()["status"],
            "tools": sorted(calls),
            "model_calls": len(attempts),
            "total_tokens": sum(value.usage["total_tokens"] for value in attempts),
            "ticket_status": tickets[0].status,
        }
    await asyncio.to_thread(
        Path("docs/verification/container-model.json").write_text,
        json.dumps(report, indent=2),
        encoding="utf-8",
    )