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

from sqlalchemy.ext.asyncio import AsyncSession

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),
    ])
    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