浅念
86 lines
3.9 KiB
Python
86 lines
3.9 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),
|
|
# 图文消息 — 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 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
|