浅念
124 lines
5.8 KiB
Python
124 lines
5.8 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"] == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# is_del filter + pinned flag (2026-08-25 对齐云客员工聊天窗口)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _msg(msg_id: str, talker: str, ts: int = 1) -> Message:
|
|
return Message(
|
|
msg_id=msg_id, channel=CH, owner_account_id=OWNER,
|
|
talker_id=talker, room_id=None, timestamp=ts,
|
|
)
|
|
|
|
|
|
async def test_sessions_is_del_filter(make_client):
|
|
"""``is_del`` 过滤:delete_flag 1=已删好友;NULL/0=未删;无 customer 行保留。
|
|
|
|
云客员工聊天窗口「删除好友」开关映射 customer.delete_flag;无 customer
|
|
行的 talker(如从未入库的系统账号)视作未删,两个方向都保留。
|
|
"""
|
|
|
|
async def seed(s: AsyncSession):
|
|
s.add_all([
|
|
Customer(friend_id="f1", channel=CH, owner_account_id=OWNER, delete_flag=0),
|
|
Customer(friend_id="f2", channel=CH, owner_account_id=OWNER, delete_flag=1),
|
|
Customer(friend_id="filehelper", channel=CH, owner_account_id=OWNER, delete_flag=None),
|
|
_msg("m1", "f1", 1),
|
|
_msg("m2", "f2", 2),
|
|
_msg("m3", "filehelper", 3),
|
|
_msg("m4", "gh0st", 4), # 无 customer 行的 talker
|
|
])
|
|
await s.commit()
|
|
|
|
client = await make_client(seed)
|
|
live = await client.get("/api/sessions", params={"owner_account_id": OWNER, "channel": CH, "is_del": 0})
|
|
deleted = await client.get("/api/sessions", params={"owner_account_id": OWNER, "channel": CH, "is_del": 1})
|
|
assert live.status_code == 200 and deleted.status_code == 200
|
|
# NULL 视为未删;无 customer 行的 talker 保留
|
|
assert sorted(s["talker_id"] for s in live.json()["sessions"]) == ["f1", "filehelper", "gh0st"]
|
|
assert [s["talker_id"] for s in deleted.json()["sessions"]] == ["f2"]
|
|
|
|
|
|
async def test_sessions_pinned_filehelper(make_client):
|
|
"""filehelper 会话 pinned=True(对齐云客置顶区),普通好友/群 False。"""
|
|
|
|
async def seed(s: AsyncSession):
|
|
s.add_all([
|
|
Customer(friend_id="f1", channel=CH, owner_account_id=OWNER, delete_flag=0),
|
|
ChatGroup(room_id="room1@chatroom", channel=CH, owner_account_id=OWNER, room_name="群一号"),
|
|
_msg("p1", "f1", 1),
|
|
_msg("p2", "filehelper", 2),
|
|
Message(msg_id="p3", channel=CH, owner_account_id=OWNER, talker_id="mx", room_id="room1@chatroom", timestamp=3),
|
|
])
|
|
await s.commit()
|
|
|
|
client = await make_client(seed)
|
|
r = await client.get("/api/sessions", params={"owner_account_id": OWNER, "channel": CH})
|
|
assert r.status_code == 200
|
|
by = {s.get("talker_id") or s.get("room_id"): s for s in r.json()["sessions"]}
|
|
assert by["filehelper"]["pinned"] is True
|
|
assert by["f1"]["pinned"] is False
|
|
assert by["room1@chatroom"]["pinned"] is False
|