浅念
2837500bc0
对齐云客员工聊天窗口朋友圈tab:帖子 create_time desc 分页 + wechat_account 头部(昵称/头像/别名);员工不存在不404(header全None 前端兜底)。头像提取抽共享 helper account_avatar 放 organization.py, accounts 列表与 moments 头部复用。 Co-Authored-By: Claude <noreply@anthropic.com>
80 lines
3.3 KiB
Python
80 lines
3.3 KiB
Python
"""Tests for GET /api/moments — 员工朋友圈帖子列表(create_time desc)+账号头部。"""
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.models import MomentsPost, WechatAccount
|
|
|
|
STAFF = "wxid_o"
|
|
|
|
|
|
async def test_moments_list_desc_with_header(make_client):
|
|
"""按 create_time desc 只返回该员工帖子;header 带账号昵称/头像/别名。"""
|
|
|
|
async def seed(s: AsyncSession):
|
|
s.add(WechatAccount(wx_id=STAFF, nick_name="员工昵称",
|
|
raw_json={"headImageUrl": "https://wx.qlogo.cn/a/0"}))
|
|
s.add(MomentsPost(id="p1", wechat_id=STAFF, content="新帖", create_time=200))
|
|
s.add(MomentsPost(id="p2", wechat_id=STAFF, content="旧帖", create_time=100))
|
|
s.add(MomentsPost(id="p3", wechat_id="wxid_other", content="别人", create_time=300))
|
|
await s.commit()
|
|
|
|
client = await make_client(seed)
|
|
r = await client.get("/api/moments", params={"wechat_id": STAFF, "page": 1, "page_size": 10})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["header"] == {"nickname": "员工昵称", "avatar": "https://wx.qlogo.cn/a/0", "alias": None}
|
|
assert [p["id"] for p in body["items"]] == ["p1", "p2"] # create_time desc
|
|
assert body["total"] == 2
|
|
|
|
|
|
async def test_moments_empty_staff(make_client):
|
|
"""无帖子:header 仍返回账号信息,items 空。"""
|
|
|
|
async def seed(s: AsyncSession):
|
|
s.add(WechatAccount(wx_id=STAFF, nick_name="员工昵称",
|
|
raw_json={"headImageUrl": "https://wx.qlogo.cn/a/0"}))
|
|
await s.commit()
|
|
|
|
client = await make_client(seed)
|
|
r = await client.get("/api/moments", params={"wechat_id": STAFF})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["items"] == [] and body["total"] == 0
|
|
assert body["header"]["nickname"] == "员工昵称"
|
|
|
|
|
|
async def test_moments_unknown_staff(make_client):
|
|
"""员工不存在:header 全 None,items 空——不 404(前端资料卡兜底)。"""
|
|
|
|
async def seed(s: AsyncSession):
|
|
await s.commit()
|
|
|
|
client = await make_client(seed)
|
|
r = await client.get("/api/moments", params={"wechat_id": "wxid_ghost"})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["header"] == {"nickname": None, "avatar": None, "alias": None}
|
|
assert body["items"] == [] and body["total"] == 0
|
|
|
|
|
|
async def test_moments_pagination(make_client):
|
|
"""3 帖 page_size=2:第 1 页 2 条(p1,p2),第 2 页 1 条(p3),total 恒 3。"""
|
|
|
|
async def seed(s: AsyncSession):
|
|
s.add(WechatAccount(wx_id=STAFF, nick_name="员工昵称", alias="al_o"))
|
|
s.add_all([
|
|
MomentsPost(id=f"p{i}", wechat_id=STAFF, create_time=300 - i * 100)
|
|
for i in range(1, 4)
|
|
])
|
|
await s.commit()
|
|
|
|
client = await make_client(seed)
|
|
b1 = (await client.get("/api/moments", params={"wechat_id": STAFF, "page": 1, "page_size": 2})).json()
|
|
b2 = (await client.get("/api/moments", params={"wechat_id": STAFF, "page": 2, "page_size": 2})).json()
|
|
assert [p["id"] for p in b1["items"]] == ["p1", "p2"]
|
|
assert [p["id"] for p in b2["items"]] == ["p3"]
|
|
assert b1["total"] == b2["total"] == 3
|
|
# alias 非空时映射到 header(列直读)
|
|
assert b1["header"]["alias"] == "al_o"
|