浅念
62 lines
2.9 KiB
Python
62 lines
2.9 KiB
Python
"""Tests for GET /api/sessions — 账号会话列表(好友+群,今日消息数+最近联系时间)。"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.models import ChatGroup, Customer, Message
|
|
|
|
OWNER = "wx_owner_s"
|
|
CH = "wechat"
|
|
_NOW = int(time.time() * 1000)
|
|
_MS_DAY = 86_400_000
|
|
|
|
|
|
async def _seed(session: AsyncSession) -> None:
|
|
session.add_all([
|
|
Message(msg_id="t1a", channel=CH, owner_account_id=OWNER, talker_id="t1", room_id=None, is_send=True, timestamp=_NOW, type="text", type_raw=1, text="x"),
|
|
Message(msg_id="t1b", channel=CH, owner_account_id=OWNER, talker_id="t1", room_id=None, is_send=False, timestamp=_NOW - 2 * _MS_DAY, type="text", type_raw=1, text="y"),
|
|
Message(msg_id="t2a", channel=CH, owner_account_id=OWNER, talker_id="t2", room_id=None, is_send=True, timestamp=_NOW - 5 * _MS_DAY, type="text", type_raw=1, text="old"),
|
|
Message(msg_id="g1a", channel=CH, owner_account_id=OWNER, talker_id="member_x", room_id="room1@chatroom", is_send=False, timestamp=_NOW, type="text", type_raw=1, text="g"),
|
|
Message(msg_id="x1", channel=CH, owner_account_id="wx_other", talker_id="t1", room_id=None, is_send=True, timestamp=_NOW, type="text", type_raw=1, text="other"),
|
|
Customer(friend_id="t1", channel=CH, owner_account_id=OWNER, friend_nickname="Alice", friend_remark=None),
|
|
ChatGroup(room_id="room1@chatroom", channel=CH, owner_account_id=OWNER, room_name="群一号"),
|
|
])
|
|
await session.commit()
|
|
|
|
|
|
async def test_sessions_friend_and_group(make_client):
|
|
client = await make_client(_seed)
|
|
r = await client.get("/api/sessions", params={"owner_account_id": OWNER, "channel": CH})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["owner_account_id"] == OWNER
|
|
by = {(s["type"], s.get("talker_id") or s.get("room_id")): s for s in body["sessions"]}
|
|
assert ("friend", "t1") in by and ("friend", "t2") in by and ("group", "room1@chatroom") in by
|
|
assert len(body["sessions"]) == 3
|
|
t1 = by[("friend", "t1")]
|
|
assert t1["today_count"] == 1
|
|
assert t1["last_chat_time"] == _NOW
|
|
assert t1["name"] == "Alice"
|
|
assert by[("friend", "t2")]["today_count"] == 0
|
|
assert by[("friend", "t2")]["name"] == "t2"
|
|
g = by[("group", "room1@chatroom")]
|
|
assert g["name"] == "群一号" and g["today_count"] == 1
|
|
|
|
|
|
async def test_sessions_sorted_by_last_chat_time_desc(make_client):
|
|
client = await make_client(_seed)
|
|
body = (await client.get("/api/sessions", params={"owner_account_id": OWNER})).json()
|
|
last_times = [s["last_chat_time"] for s in body["sessions"]]
|
|
assert last_times == sorted(last_times, reverse=True)
|
|
|
|
|
|
async def test_sessions_empty(make_client):
|
|
async def _empty(s: AsyncSession):
|
|
await s.commit()
|
|
client = await make_client(_empty)
|
|
r = await client.get("/api/sessions", params={"owner_account_id": OWNER})
|
|
assert r.status_code == 200
|
|
assert r.json()["sessions"] == []
|