57 lines
2.1 KiB
Python
57 lines
2.1 KiB
Python
"""审计测试。"""
|
|
|
|
from collections.abc import AsyncGenerator
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy import select
|
|
|
|
from app.ai_provider.types.chat import ChatChunk
|
|
from app.application.agent.runtime import AgentTurnResult
|
|
from app.application.chat.send_message import drain_pending_streams
|
|
from app.db.session import AsyncSessionLocal
|
|
from app.main import app
|
|
from app.models.safety import AuditEvent
|
|
|
|
|
|
@pytest.fixture
|
|
async def client(
|
|
monkeypatch: pytest.MonkeyPatch, auth_headers: dict[str, str]
|
|
) -> AsyncGenerator[AsyncClient, None]:
|
|
async def fake_run_agent_turn(*, on_chunk, **_kwargs: object):
|
|
await on_chunk(ChatChunk(delta_content="你好，我是测试模型回复。"))
|
|
return AgentTurnResult(content="你好，我是测试模型回复。", reasoning="")
|
|
|
|
monkeypatch.setattr("app.application.chat.send_message.run_agent_turn", fake_run_agent_turn)
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(
|
|
transport=transport,
|
|
base_url="http://test",
|
|
headers=auth_headers,
|
|
) as http_client:
|
|
yield http_client
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chat_and_memory_write_audit_events(client: AsyncClient) -> None:
|
|
session_response = await client.post("/api/v1/chat/sessions", json={"title": "审计"})
|
|
session_id = session_response.json()["session_id"]
|
|
sent = await client.post(
|
|
f"/api/v1/chat/sessions/{session_id}/messages",
|
|
json={"content": "你好", "modality": "text"},
|
|
)
|
|
assert sent.status_code == 200
|
|
await drain_pending_streams()
|
|
memory = await client.post(
|
|
"/api/v1/memory",
|
|
json={"type": "user", "title": "偏好", "content": "用户喜欢简洁回答。"},
|
|
)
|
|
assert memory.status_code == 200
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
actions = [item.action for item in (await db.execute(select(AuditEvent))).scalars().all()]
|
|
assert "run.started" in actions
|
|
assert "model.request.started" in actions
|
|
assert "model.response.completed" in actions
|
|
assert "memory.created" in actions
|