ai-kefu-jc/tests/test_admin_push_logs.py

"""admin 推送日志查询接口(设计 §9): 鉴权/筛选/分页/不含 payload。"""
import asyncio
import os
import tempfile

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.pool import NullPool

from app.models.database import (
    Base, Conversation, CustomerProfile, Evaluation, IntentLevel, LeadPushLog,
)

# admin 路由用 `from app.main import async_session` 的模块级绑定——
# 测试必须 patch admin 模块自己的绑定(client fixture 只 patch 了 chat_rest)
import app.api.routes.admin as admin_mod


@pytest.fixture
def admin_client(monkeypatch):
    from app.main import app
    fd, path = tempfile.mkstemp(suffix=".db")
    os.close(fd)
    engine = create_async_engine(
        f"sqlite+aiosqlite:///{path}", poolclass=NullPool)

    async def create_all():
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
    asyncio.run(create_all())
    sm = async_sessionmaker(engine, expire_on_commit=False)
    monkeypatch.setattr(admin_mod, "async_session", sm)
    with TestClient(app) as c:
        yield c, sm
    asyncio.run(engine.dispose())
    try:
        os.remove(path)
    except PermissionError:
        pass


def _seed(sm):
    async def run():
        async with sm() as db:
            conv = Conversation(session_id="s-admin")
            db.add(conv)
            await db.flush()
            ev = Evaluation(conversation_id=conv.id,
                            intent_level=IntentLevel.pending, contact_info={})
            db.add(ev)
            await db.flush()
            db.add_all([
                LeadPushLog(evaluation_id=ev.id, conversation_id=conv.id,
                            push_type="lead", status="success", attempts=1,
                            outcome="accepted", response_status=200,
                            request_payload={"big": "x" * 1000}),
                LeadPushLog(evaluation_id=ev.id, conversation_id=conv.id,
                            push_type="correction", status="failed", attempts=3,
                            response_status=500, error_message="http 500",
                            request_payload={"big": "y" * 1000}),
                LeadPushLog(evaluation_id=ev.id, conversation_id=conv.id,
                            push_type="lead", status="skipped", attempts=0,
                            error_message="no valid contact",
                            request_payload=None),
            ])
            await db.commit()
    asyncio.run(run())


def test_push_logs_requires_auth(admin_client):
    c, _ = admin_client
    resp = c.get("/api/admin/push-logs")
    assert resp.json() == {"error": "unauthorized"}


def test_push_logs_list_filter_pagination(admin_client):
    c, sm = admin_client
    _seed(sm)
    assert c.post("/api/admin/login",
                  json={"password": "admin123"}).json()["ok"] is True

    body = c.get("/api/admin/push-logs").json()
    assert body["total"] == 3 and len(body["items"]) == 3
    item = body["items"][0]
    assert "request_payload" not in item          # 设计 §9: 列表不回 payload
    assert {i["status"] for i in body["items"]} == {"success", "failed", "skipped"}

    only_failed = c.get("/api/admin/push-logs", params={"status": "failed"}).json()
    assert only_failed["total"] == 1
    assert only_failed["items"][0]["push_type"] == "correction"
    assert only_failed["items"][0]["error_message"] == "http 500"
    assert only_failed["items"][0]["attempts"] == 3

    page2 = c.get("/api/admin/push-logs",
                  params={"page": 2, "size": 2}).json()
    assert page2["total"] == 3 and len(page2["items"]) == 1

    by_conv = c.get("/api/admin/push-logs",
                    params={"conversation_id": "no-such"}).json()
    assert by_conv["total"] == 0


def test_push_logs_invalid_status_ignored(admin_client):
    """非法 status 值当未筛(与 conversations 的 intent 白名单风格一致)。"""
    c, sm = admin_client
    _seed(sm)
    c.post("/api/admin/login", json={"password": "admin123"})
    body = c.get("/api/admin/push-logs", params={"status": "bogus"}).json()
    assert body["total"] == 3


def test_admin_page_has_push_logs_entry():
    """后台页静态回归:推送日志入口/视图容器/接口引用三件套在场。"""
    from pathlib import Path
    html = (Path(__file__).resolve().parent.parent / "static" / "admin.html"
            ).read_text(encoding="utf-8")
    assert 'onclick="toggleLogs()"' in html        # header 入口按钮
    assert 'id="logsView"' in html                 # 日志视图容器
    assert "/api/admin/push-logs" in html          # 数据接口引用


def test_push_logs_pending_filter_and_page_overflow(admin_client):
    """pending 筛选分支 + 超界页码(items 空、total 不变)。"""
    c, sm = admin_client
    _seed(sm)   # 3 条: success/failed/skipped

    async def add_pending():
        async with sm() as db:
            ev = (await db.execute(select(Evaluation))).scalar_one()
            conv_id = ev.conversation_id
            db.add(LeadPushLog(evaluation_id=ev.id, conversation_id=conv_id,
                               push_type="lead", status="pending", attempts=0,
                               request_payload={"external_key": "k"}))
            await db.commit()
    asyncio.run(add_pending())

    assert c.post("/api/admin/login",
                  json={"password": "admin123"}).json()["ok"] is True

    only_pending = c.get("/api/admin/push-logs",
                         params={"status": "pending"}).json()
    assert only_pending["total"] == 1
    assert only_pending["items"][0]["status"] == "pending"

    # 超界页码: items 空而 total 仍是全量(4)
    overflow = c.get("/api/admin/push-logs", params={"page": 99}).json()
    assert overflow["total"] == 4 and overflow["items"] == []