yunke/yunke-server/tests/test_api_export.py

"""Tests for GET /api/export/chat-html — 导出 1:1/群聊天为自包含 HTML。

对齐云客微信客户/群列表「导出HTML聊天记录」：按 (channel, owner, talker_id|room_id)
拉消息（可选 start/end 时间范围），渲染为左右气泡 HTML 作为 attachment 返回下载。
"""

from __future__ import annotations

import functools
from sqlalchemy.ext.asyncio import AsyncSession

from app.api.routers.export import EXPORT_MAX_ROWS
from app.db.models import Message


async def _seed(session: AsyncSession) -> None:
    session.add_all([
        Message(msg_id="dm1", channel="wechat", owner_account_id="wx_owner",
                talker_id="wx_friend", room_id=None, text="你好",
                timestamp=1000, is_send=False, type="text", type_raw=1),
        Message(msg_id="dm2", channel="wechat", owner_account_id="wx_owner",
                talker_id="wx_friend", room_id=None, text="在的",
                timestamp=2000, is_send=True, type="text", type_raw=1),
        Message(msg_id="g1", channel="wechat", owner_account_id="wx_owner",
                talker_id="room_topic", room_id="g1@chatroom", text="群消息",
                timestamp=3000, is_send=True, type="text", type_raw=1),
        Message(msg_id="g2", channel="wechat", owner_account_id="wx_owner",
                talker_id="room_topic", room_id="g1@chatroom", text="范围外",
                timestamp=9999999, is_send=True, type="text", type_raw=1),
        # 图文消息 — OSS URL 应直接嵌入 <img>
        Message(msg_id="dm3", channel="wechat", owner_account_id="wx_owner",
                talker_id="wx_friend", room_id=None, text=None,
                timestamp=2500, is_send=False, type="image", type_raw=3,
                file_url="https://oss.example.com/x/p1.jpg"),
        # 本地 /api/media 路径 — 应显示 "[本地媒体]" 占位
        Message(msg_id="dm4", channel="wechat", owner_account_id="wx_owner",
                talker_id="wx_friend", room_id=None, text=None,
                timestamp=2600, is_send=True, type="voice", type_raw=34,
                file_url="/api/media/abc123", duration=8),
    ])
    await session.commit()


async def _seed_many(session: AsyncSession, n: int) -> None:
    """Seed ``n`` DM messages so the row cap can be exercised."""
    session.add_all([
        Message(
            msg_id=f"cap{i}",
            channel="wechat",
            owner_account_id="wx_owner",
            talker_id="wx_friend",
            room_id=None,
            text=f"msg{i}",
            timestamp=1000 + i,
            is_send=i % 2 == 0,
            type="text",
            type_raw=1,
        )
        for i in range(n)
    ])
    await session.commit()


async def test_export_dm_returns_html(make_client):
    client = await make_client(_seed)
    r = await client.get("/api/export/chat-html", params={
        "channel": "wechat", "owner_account_id": "wx_owner", "talker_id": "wx_friend"})
    assert r.status_code == 200
    assert r.headers["content-type"].startswith("text/html")
    assert "attachment" in r.headers["content-disposition"]
    body = r.text
    assert "你好" in body and "在的" in body
    assert "wx_owner" in body


async def test_export_group_with_time_range(make_client):
    client = await make_client(_seed)
    r = await client.get("/api/export/chat-html", params={
        "channel": "wechat", "owner_account_id": "wx_owner", "room_id": "g1@chatroom",
        "start": 0, "end": 5000})
    assert r.status_code == 200
    assert "群消息" in r.text
    assert "范围外" not in r.text  # end=5000 过滤掉 timestamp=9999999


async def test_export_empty(make_client):
    client = await make_client(_seed)
    r = await client.get("/api/export/chat-html", params={
        "channel": "wechat", "owner_account_id": "wx_owner", "talker_id": "none"})
    assert r.status_code == 200
    assert r.headers["content-type"].startswith("text/html")
    assert "暂无消息" in r.text


async def test_export_embeds_media(make_client):
    """图片(OSS URL)直接 <img> 嵌入;本地 /api/ 路径显示 [本地媒体] 占位。"""
    client = await make_client(_seed)
    r = await client.get("/api/export/chat-html", params={
        "channel": "wechat", "owner_account_id": "wx_owner", "talker_id": "wx_friend"})
    assert r.status_code == 200
    body = r.text
    # OSS URL → <img> with the URL escaped into src
    assert "<img" in body
    assert "https://oss.example.com/x/p1.jpg" in body
    # 本地路径 → 占位提示,不嵌 <audio>
    assert "[本地媒体]" in body
    assert "/api/media/abc123" not in body


async def test_export_caps_rows_and_emits_warning(make_client):
    """When the cap is hit, only EXPORT_MAX_ROWS rows render + a warning shows."""
    # Seed twice the cap so the truncation is unambiguous.
    seed = functools.partial(_seed_many, n=EXPORT_MAX_ROWS * 2)
    client = await make_client(seed)
    r = await client.get("/api/export/chat-html", params={
        "channel": "wechat", "owner_account_id": "wx_owner", "talker_id": "wx_friend"})
    assert r.status_code == 200
    body = r.text
    # The warning DIV is rendered (not just the CSS class in <style>).
    assert '<div class="export-warning">' in body
    assert str(EXPORT_MAX_ROWS) in body
    # The first EXPORT_MAX_ROWS seeded rows (timestamps 1000..1000+N-1) render;
    # anything past the cap must NOT appear. msg_id pattern "cap{i}" lets us
    # count rendered bubbles precisely.
    last_kept = EXPORT_MAX_ROWS - 1
    last_dropped = EXPORT_MAX_ROWS  # first row beyond the cap
    assert f"msg{last_kept}" in body
    assert f"msg{last_dropped}" not in body


async def test_export_no_warning_below_cap(make_client):
    """Sub-cap exports render no warning DIV."""
    seed = functools.partial(_seed_many, n=10)
    client = await make_client(seed)
    r = await client.get("/api/export/chat-html", params={
        "channel": "wechat", "owner_account_id": "wx_owner", "talker_id": "wx_friend"})
    assert r.status_code == 200
    # No warning DIV (the CSS class still lives in <style> — that's fine).
    assert '<div class="export-warning">' not in r.text