Files
yunke/yunke-server/tests/test_api_export.py
T
浅念 a66c3be00a chore: final Minor polish — export filename/test断言, downloadBlob 类型/清理, plan docs
address review Minors: drop redundant filename replace, assert 暂无消息,
tighten downloadBlob return type + try/finally, fix plan group time-picker wording.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-06 18:06:17 +08:00

61 lines
2.6 KiB
Python

"""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