feat: backend read api for all entities (messages/groups/calls/sms/org/sync)

Adds the read-only API surface the upcoming Vue frontend needs to browse all
mirrored entities. Follows the existing customer-router pattern: Annotated
deps, Page[T] envelope, select+where+order_by, count() for total.

Endpoints (all GET):
* /api/departments          -> list[DepartmentOut] (path-ordered, plain list)
* /api/staff                -> Page[StaffOut] (department_id, is_on_job filters)
* /api/accounts             -> Page[AccountOut] (channel filter; unified
                             wechat+whatsapp projection built in memory)
* /api/messages             -> Page[MessageOut] (channel/owner/talker/room/
                             [start,end] filters; ordered timestamp asc)
* /api/groups               -> Page[GroupOut] (channel/owner filters)
* /api/groups/{id}/members  -> list[GroupMemberOut] (404 on unknown room)
* /api/calls                -> Page[PhoneCallOut] (channel/user/time window;
                             ordered start_call_time desc)
* /api/sms                  -> Page[SmsOut] (user/time window; create_time desc)
* /api/sync/status          -> {states, recent_runs} monitor payload

Schemas (Pydantic v2, from_attributes=True): MessageOut, GroupOut,
GroupMemberOut, PhoneCallOut, SmsOut, DepartmentOut, StaffOut, AccountOut
(unified across two tables), SyncStateOut, SyncRunOut, SyncStatusOut.
Only human-relevant fields per entity are exposed.

Tests: per-router TDD via httpx ASGITransport against an in-memory SQLite
override of get_session. A shared make_client factory in conftest.py builds
the seeded client, deduping the ~40-line fixture across the six new files.
41 new tests cover shape + every documented filter + pagination + ordering
(nulls-last verified on all desc/asc time columns).

Suite: 226 -> 267 passing. ruff check clean.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
2026-08-01 16:15:40 +08:00
parent 44d64dd237
commit ff7e9cb1d9
15 changed files with 1687 additions and 2 deletions
+24 -1
View File
@@ -8,6 +8,14 @@ Includes:
* ``GET /api/health`` -> ``{"status": "ok"}`` liveness probe. * ``GET /api/health`` -> ``{"status": "ok"}`` liveness probe.
* ``GET /api/customers`` (see :mod:`app.api.routers.customer`). * ``GET /api/customers`` (see :mod:`app.api.routers.customer`).
* ``GET /api/media/{media_hash}`` (see :mod:`app.api.routers.media`). * ``GET /api/media/{media_hash}`` (see :mod:`app.api.routers.media`).
* ``GET /api/departments`` / ``/api/staff`` / ``/api/accounts``
(see :mod:`app.api.routers.organization`).
* ``GET /api/messages`` (see :mod:`app.api.routers.messages`).
* ``GET /api/groups`` / ``/api/groups/{room_id}/members``
(see :mod:`app.api.routers.groups`).
* ``GET /api/calls`` (see :mod:`app.api.routers.calls`).
* ``GET /api/sms`` (see :mod:`app.api.routers.sms`).
* ``GET /api/sync/status`` (see :mod:`app.api.routers.sync`).
CORS is configured permissively (wildcard origin, no credentials) so the local CORS is configured permissively (wildcard origin, no credentials) so the local
frontend can call the API during development. frontend can call the API during development.
@@ -26,7 +34,16 @@ from fastapi import Depends, FastAPI
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from app.api.auth import require_auth from app.api.auth import require_auth
from app.api.routers import customer, media from app.api.routers import (
calls,
customer,
groups,
media,
messages,
organization,
sms,
sync,
)
app = FastAPI(title="yunke data mirror", dependencies=[Depends(require_auth)]) app = FastAPI(title="yunke data mirror", dependencies=[Depends(require_auth)])
@@ -39,6 +56,12 @@ app.add_middleware(
app.include_router(customer.router) app.include_router(customer.router)
app.include_router(media.router) app.include_router(media.router)
app.include_router(organization.router)
app.include_router(messages.router)
app.include_router(groups.router)
app.include_router(calls.router)
app.include_router(sms.router)
app.include_router(sync.router)
@app.get("/api/health") @app.get("/api/health")
+68
View File
@@ -0,0 +1,68 @@
"""GET /api/calls — phone-call / voice-call log list.
Read-only query over the mirrored ``phone_call`` table. ``channel`` here is the
**call-type** in {``phone_call``, ``voice_call``} (calls are not tied to a
messaging channel). Filters: ``channel``, ``user_id``, and a ms-epoch
``[start, end]`` window compared against ``start_call_time`` (inclusive).
Ordered by ``start_call_time desc`` (nulls last) — most recent calls first.
"""
from __future__ import annotations
from typing import Annotated
import sqlalchemy as sa
from fastapi import APIRouter, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_session
from app.api.schemas import Page, PhoneCallOut
from app.db.models import PhoneCall
router = APIRouter(prefix="/api", tags=["calls"])
SessionDep = Annotated[AsyncSession, Depends(get_session)]
PageParam = Annotated[int, Query(ge=1)]
PageSizeParam = Annotated[int, Query(ge=1, le=100)]
@router.get("/calls")
async def list_calls(
session: SessionDep,
page: PageParam = 1,
page_size: PageSizeParam = 20,
channel: Annotated[str | None, Query()] = None,
user_id: Annotated[str | None, Query()] = None,
start: Annotated[int | None, Query(ge=0)] = None,
end: Annotated[int | None, Query(ge=0)] = None,
) -> Page[PhoneCallOut]:
"""List phone/voice calls with optional filters, newest first."""
conditions = []
if channel is not None:
conditions.append(PhoneCall.channel == channel)
if user_id is not None:
conditions.append(PhoneCall.user_id == user_id)
if start is not None:
conditions.append(PhoneCall.start_call_time >= start)
if end is not None:
conditions.append(PhoneCall.start_call_time <= end)
total_stmt = sa.select(sa.func.count()).select_from(PhoneCall)
if conditions:
total_stmt = total_stmt.where(*conditions)
total = (await session.execute(total_stmt)).scalar_one()
list_stmt = sa.select(PhoneCall)
if conditions:
list_stmt = list_stmt.where(*conditions)
list_stmt = (
list_stmt.order_by(
sa.nullslast(PhoneCall.start_call_time.desc()),
PhoneCall.call_id.asc(),
)
.limit(page_size)
.offset((page - 1) * page_size)
)
rows = (await session.execute(list_stmt)).scalars().all()
items = [PhoneCallOut.model_validate(row) for row in rows]
return Page(items=items, total=total, page=page, page_size=page_size)
+91
View File
@@ -0,0 +1,91 @@
"""GET /api/groups, /api/groups/{room_id}/members — chat-group browse endpoints.
* ``GET /api/groups`` -> ``Page[GroupOut]`` filtered by ``channel`` /
``owner_account_id``, ordered by ``last_chat_time desc`` (nulls last) so
recently-active groups surface first.
* ``GET /api/groups/{room_id}/members`` -> ``list[GroupMemberOut]`` — the
members of a single group (small set, no pagination needed).
"""
from __future__ import annotations
from typing import Annotated
import sqlalchemy as sa
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_session
from app.api.schemas import GroupMemberOut, GroupOut, Page
from app.db.models import ChatGroup, GroupMember
router = APIRouter(prefix="/api", tags=["groups"])
SessionDep = Annotated[AsyncSession, Depends(get_session)]
PageParam = Annotated[int, Query(ge=1)]
PageSizeParam = Annotated[int, Query(ge=1, le=100)]
@router.get("/groups")
async def list_groups(
session: SessionDep,
page: PageParam = 1,
page_size: PageSizeParam = 20,
channel: Annotated[str | None, Query()] = None,
owner_account_id: Annotated[str | None, Query()] = None,
) -> Page[GroupOut]:
"""List chat groups with optional ``channel`` / ``owner_account_id`` filters."""
conditions = []
if channel is not None:
conditions.append(ChatGroup.channel == channel)
if owner_account_id is not None:
conditions.append(ChatGroup.owner_account_id == owner_account_id)
total_stmt = sa.select(sa.func.count()).select_from(ChatGroup)
if conditions:
total_stmt = total_stmt.where(*conditions)
total = (await session.execute(total_stmt)).scalar_one()
list_stmt = sa.select(ChatGroup)
if conditions:
list_stmt = list_stmt.where(*conditions)
list_stmt = (
list_stmt.order_by(
sa.nullslast(ChatGroup.last_chat_time.desc()),
ChatGroup.room_id.asc(),
)
.limit(page_size)
.offset((page - 1) * page_size)
)
rows = (await session.execute(list_stmt)).scalars().all()
items = [GroupOut.model_validate(row) for row in rows]
return Page(items=items, total=total, page=page, page_size=page_size)
@router.get("/groups/{room_id}/members")
async def list_group_members(
room_id: str, session: SessionDep
) -> list[GroupMemberOut]:
"""Return the members of a single group (plain list).
Ordered by ``member_id`` for stable rendering. Returns ``404`` when no such
group exists so the frontend can distinguish "missing room" from "empty
room".
"""
exists = (
await session.execute(
sa.select(sa.func.count())
.select_from(ChatGroup)
.where(ChatGroup.room_id == room_id)
)
).scalar_one()
if not exists:
raise HTTPException(status_code=404, detail="group not found")
stmt = (
sa.select(GroupMember)
.where(GroupMember.room_id == room_id)
.order_by(GroupMember.member_id.asc())
)
rows = (await session.execute(stmt)).scalars().all()
return [GroupMemberOut.model_validate(row) for row in rows]
+81
View File
@@ -0,0 +1,81 @@
"""GET /api/messages — chat timeline list (WeChat + WhatsApp).
Read-only query over the mirrored ``message`` table. Filters:
* ``channel`` — ``wechat`` / ``whatsapp``.
* ``owner_account_id`` — owning staff account (wxId / uid).
* ``talker_id`` (1:1) OR ``room_id`` (group) — the conversation anchor. Both
may be supplied (AND), but typically the frontend sends exactly one.
* ``start`` / ``end`` — ms-epoch window compared against ``timestamp``
(inclusive on both ends).
Ordered by ``timestamp asc`` so the timeline renders oldest-first. The index
``ix_message_channel_owner_talker_time`` covers the hot path
``(channel, owner_account_id, talker_id, timestamp)``.
"""
from __future__ import annotations
from typing import Annotated
import sqlalchemy as sa
from fastapi import APIRouter, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_session
from app.api.schemas import MessageOut, Page
from app.db.models import Message
router = APIRouter(prefix="/api", tags=["messages"])
SessionDep = Annotated[AsyncSession, Depends(get_session)]
PageParam = Annotated[int, Query(ge=1)]
PageSizeParam = Annotated[int, Query(ge=1, le=100)]
@router.get("/messages")
async def list_messages(
session: SessionDep,
page: PageParam = 1,
page_size: PageSizeParam = 50,
channel: Annotated[str | None, Query()] = None,
owner_account_id: Annotated[str | None, Query()] = None,
talker_id: Annotated[str | None, Query()] = None,
room_id: Annotated[str | None, Query()] = None,
start: Annotated[int | None, Query(ge=0)] = None,
end: Annotated[int | None, Query(ge=0)] = None,
) -> Page[MessageOut]:
"""List messages for a timeline (ordered by ``timestamp asc``)."""
conditions = []
if channel is not None:
conditions.append(Message.channel == channel)
if owner_account_id is not None:
conditions.append(Message.owner_account_id == owner_account_id)
if talker_id is not None:
conditions.append(Message.talker_id == talker_id)
if room_id is not None:
conditions.append(Message.room_id == room_id)
if start is not None:
conditions.append(Message.timestamp >= start)
if end is not None:
conditions.append(Message.timestamp <= end)
total_stmt = sa.select(sa.func.count()).select_from(Message)
if conditions:
total_stmt = total_stmt.where(*conditions)
total = (await session.execute(total_stmt)).scalar_one()
list_stmt = sa.select(Message)
if conditions:
list_stmt = list_stmt.where(*conditions)
list_stmt = (
list_stmt.order_by(
sa.nullslast(Message.timestamp.asc()),
Message.msg_id.asc(),
)
.limit(page_size)
.offset((page - 1) * page_size)
)
rows = (await session.execute(list_stmt)).scalars().all()
items = [MessageOut.model_validate(row) for row in rows]
return Page(items=items, total=total, page=page, page_size=page_size)
@@ -0,0 +1,137 @@
"""GET /api/departments, /api/staff, /api/accounts — organization browse endpoints.
Read-only queries over the mirrored ``department``, ``staff``,
``wechat_account`` and ``whatsapp_account`` tables.
* ``GET /api/departments`` -> ``list[DepartmentOut]`` ordered by ``path`` so
the frontend can assemble the tree with a stable traversal order.
* ``GET /api/staff`` -> ``Page[StaffOut]`` with optional ``department_id``
and ``is_on_job`` filters.
* ``GET /api/accounts`` -> ``Page[AccountOut]`` projecting both staff-messaging
account tables onto a unified shape (see :class:`AccountOut`). The optional
``channel`` filter restricts to one of ``wechat`` / ``whatsapp``; when
omitted, both tables are unioned in memory. Staff-account volumes are modest
(tens to low hundreds of rows), so the in-memory merge is cheap and keeps
the API contract simple.
"""
from __future__ import annotations
from typing import Annotated
import sqlalchemy as sa
from fastapi import APIRouter, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_session
from app.api.schemas import AccountOut, DepartmentOut, Page, StaffOut
from app.db.models import Department, Staff, WechatAccount, WhatsappAccount
router = APIRouter(prefix="/api", tags=["organization"])
SessionDep = Annotated[AsyncSession, Depends(get_session)]
PageParam = Annotated[int, Query(ge=1)]
PageSizeParam = Annotated[int, Query(ge=1, le=100)]
@router.get("/departments")
async def list_departments(session: SessionDep) -> list[DepartmentOut]:
"""Return every department ordered by ``path`` (nulls last).
A plain list — not paginated — because the frontend rebuilds the tree from
``parent_id``/``path`` and pagination would split the topology awkwardly.
"""
stmt = sa.select(Department).order_by(
sa.nullslast(Department.path.asc()), Department.dept_id.asc()
)
rows = (await session.execute(stmt)).scalars().all()
return [DepartmentOut.model_validate(row) for row in rows]
@router.get("/staff")
async def list_staff(
session: SessionDep,
page: PageParam = 1,
page_size: PageSizeParam = 20,
department_id: Annotated[str | None, Query()] = None,
is_on_job: Annotated[bool | None, Query()] = None,
) -> Page[StaffOut]:
"""List staff with optional ``department_id`` / ``is_on_job`` filters."""
conditions = []
if department_id is not None:
conditions.append(Staff.dept_id == department_id)
if is_on_job is not None:
conditions.append(Staff.is_on_job.is_(is_on_job))
total_stmt = sa.select(sa.func.count()).select_from(Staff)
if conditions:
total_stmt = total_stmt.where(*conditions)
total = (await session.execute(total_stmt)).scalar_one()
list_stmt = sa.select(Staff)
if conditions:
list_stmt = list_stmt.where(*conditions)
list_stmt = (
list_stmt.order_by(Staff.user_id.asc())
.limit(page_size)
.offset((page - 1) * page_size)
)
rows = (await session.execute(list_stmt)).scalars().all()
items = [StaffOut.model_validate(row) for row in rows]
return Page(items=items, total=total, page=page, page_size=page_size)
@router.get("/accounts")
async def list_accounts(
session: SessionDep,
page: PageParam = 1,
page_size: PageSizeParam = 20,
channel: Annotated[str | None, Query()] = None,
) -> Page[AccountOut]:
"""List staff messaging accounts (WeChat / WhatsApp) in a unified shape.
``channel`` optional; one of ``wechat`` / ``whatsapp``. When omitted, both
tables are merged in memory. Pagination is applied after the merge using a
stable ``(channel, account_id)`` ordering.
"""
selected_channels = [channel] if channel else ["wechat", "whatsapp"]
rows: list[AccountOut] = []
if "wechat" in selected_channels:
result = (await session.execute(sa.select(WechatAccount))).scalars().all()
for r in result:
rows.append(
AccountOut(
channel="wechat",
account_id=r.wx_id,
nickname=r.nick_name,
user_id=r.user_id,
user_name=r.user_name,
phone=r.phone,
contacts_num=r.contacts_num,
group_num=r.group_num,
is_on_job=r.is_on_job,
)
)
if "whatsapp" in selected_channels:
result = (await session.execute(sa.select(WhatsappAccount))).scalars().all()
for r in result:
rows.append(
AccountOut(
channel="whatsapp",
account_id=r.uid,
nickname=r.nick_name,
user_id=r.user_id,
user_name=r.user_name,
phone=r.phone,
contacts_num=r.contacts_num,
group_num=r.group_num,
is_on_job=r.is_on_job,
)
)
rows.sort(key=lambda a: (a.channel, a.account_id))
total = len(rows)
start = (page - 1) * page_size
items = rows[start : start + page_size]
return Page(items=items, total=total, page=page, page_size=page_size)
+63
View File
@@ -0,0 +1,63 @@
"""GET /api/sms — SMS log list.
Read-only query over the mirrored ``sms`` table. Filters: ``user_id`` and a
ms-epoch ``[start, end]`` window compared against ``create_time`` (inclusive).
Ordered by ``create_time desc`` (nulls last) — newest first.
"""
from __future__ import annotations
from typing import Annotated
import sqlalchemy as sa
from fastapi import APIRouter, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_session
from app.api.schemas import Page, SmsOut
from app.db.models import Sms
router = APIRouter(prefix="/api", tags=["sms"])
SessionDep = Annotated[AsyncSession, Depends(get_session)]
PageParam = Annotated[int, Query(ge=1)]
PageSizeParam = Annotated[int, Query(ge=1, le=100)]
@router.get("/sms")
async def list_sms(
session: SessionDep,
page: PageParam = 1,
page_size: PageSizeParam = 20,
user_id: Annotated[str | None, Query()] = None,
start: Annotated[int | None, Query(ge=0)] = None,
end: Annotated[int | None, Query(ge=0)] = None,
) -> Page[SmsOut]:
"""List SMS rows with optional filters, newest first."""
conditions = []
if user_id is not None:
conditions.append(Sms.user_id == user_id)
if start is not None:
conditions.append(Sms.create_time >= start)
if end is not None:
conditions.append(Sms.create_time <= end)
total_stmt = sa.select(sa.func.count()).select_from(Sms)
if conditions:
total_stmt = total_stmt.where(*conditions)
total = (await session.execute(total_stmt)).scalar_one()
list_stmt = sa.select(Sms)
if conditions:
list_stmt = list_stmt.where(*conditions)
list_stmt = (
list_stmt.order_by(
sa.nullslast(Sms.create_time.desc()),
Sms.sms_id.asc(),
)
.limit(page_size)
.offset((page - 1) * page_size)
)
rows = (await session.execute(list_stmt)).scalars().all()
items = [SmsOut.model_validate(row) for row in rows]
return Page(items=items, total=total, page=page, page_size=page_size)
+52
View File
@@ -0,0 +1,52 @@
"""GET /api/sync/status — sync monitor (cursors + recent runs).
Returns the current :class:`SyncState` cursors and the most recent N
:class:`SyncRun` audit rows (default 20). Read-only — purely for the operator
dashboard.
"""
from __future__ import annotations
from typing import Annotated
import sqlalchemy as sa
from fastapi import APIRouter, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_session
from app.api.schemas import SyncRunOut, SyncStateOut, SyncStatusOut
from app.db.models import SyncRun, SyncState
router = APIRouter(prefix="/api/sync", tags=["sync"])
SessionDep = Annotated[AsyncSession, Depends(get_session)]
RecentLimitParam = Annotated[int, Query(ge=1, le=200)]
@router.get("/status")
async def sync_status(
session: SessionDep,
recent: RecentLimitParam = 20,
) -> SyncStatusOut:
"""Return sync cursors plus the most recent ``recent`` runs."""
state_rows = (
await session.execute(
sa.select(SyncState).order_by(
SyncState.entity_type.asc(),
sa.nullslast(SyncState.channel.asc()),
sa.nullslast(SyncState.scope_id.asc()),
)
)
).scalars().all()
states = [SyncStateOut.model_validate(row) for row in state_rows]
run_rows = (
await session.execute(
sa.select(SyncRun)
.order_by(SyncRun.started_at.desc(), SyncRun.run_id.desc())
.limit(recent)
)
).scalars().all()
recent_runs = [SyncRunOut.model_validate(row) for row in run_rows]
return SyncStatusOut(states=states, recent_runs=recent_runs)
+180
View File
@@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import datetime as dt
from typing import Generic, TypeVar from typing import Generic, TypeVar
from pydantic import BaseModel, ConfigDict from pydantic import BaseModel, ConfigDict
@@ -41,3 +42,182 @@ class Page(BaseModel, Generic[T]):
total: int total: int
page: int page: int
page_size: int page_size: int
# ---------------------------------------------------------------------------
# Organization & people
# ---------------------------------------------------------------------------
class DepartmentOut(BaseModel):
"""云客 department node — frontend builds the tree from ``parent_id``/``path``."""
model_config = ConfigDict(from_attributes=True)
dept_id: str
parent_id: str | None = None
name: str | None = None
path: str | None = None
class StaffOut(BaseModel):
"""Staff member (employee) row."""
model_config = ConfigDict(from_attributes=True)
user_id: str
name: str | None = None
phone: str | None = None
sale_phone: str | None = None
dept_id: str | None = None
is_on_job: bool | None = None
class AccountOut(BaseModel):
"""Unified staff messaging account (WeChat or WhatsApp).
The two source tables (:class:`WechatAccount`, :class:`WhatsappAccount`)
have different PK / column names; this view projects both onto a single
shape keyed by ``channel`` + ``account_id`` so the frontend can render one
list. Built explicitly in the route handler (not via ``from_attributes``)
because no single ORM row carries this schema.
"""
channel: str
account_id: str
nickname: str | None = None
user_id: str | None = None
user_name: str | None = None
phone: str | None = None
contacts_num: int | None = None
group_num: int | None = None
is_on_job: bool | None = None
# ---------------------------------------------------------------------------
# Communication entities
# ---------------------------------------------------------------------------
class MessageOut(BaseModel):
"""Chat timeline row — a single message (text/media/voice/...)."""
model_config = ConfigDict(from_attributes=True)
msg_id: str
channel: str
owner_account_id: str | None = None
talker_id: str | None = None
room_id: str | None = None
is_send: bool | None = None
type: str | None = None
type_raw: int | None = None
text: str | None = None
file_url: str | None = None
sender_name: str | None = None
sender_head_url: str | None = None
timestamp: int | None = None
class GroupOut(BaseModel):
"""Chat group summary row."""
model_config = ConfigDict(from_attributes=True)
room_id: str
channel: str
room_name: str | None = None
owner_account_id: str | None = None
owner_account_nickname: str | None = None
member_count: int | None = None
message_count: int | None = None
last_chat_time: int | None = None
class GroupMemberOut(BaseModel):
"""Member of a chat group (composite PK: channel + room_id + member_id)."""
model_config = ConfigDict(from_attributes=True)
channel: str
room_id: str
member_id: str
nickname: str | None = None
display_name: str | None = None
avatar: str | None = None
class PhoneCallOut(BaseModel):
"""Phone / voice call record. ``channel`` is the call-type."""
model_config = ConfigDict(from_attributes=True)
call_id: str
channel: str
start_call_time: int | None = None
call_duration: int | None = None
call_status: int | None = None
incoming_call: int | None = None
user_id: str | None = None
user_id_name: str | None = None
call_number: str | None = None
sim_phone: str | None = None
class SmsOut(BaseModel):
"""SMS log row."""
model_config = ConfigDict(from_attributes=True)
sms_id: str
customer_phone: str | None = None
content_preview: str | None = None
sender: str | None = None
direction: int | None = None
user_id: str | None = None
receive_name: str | None = None
create_time: int | None = None
sim_phone: str | None = None
# ---------------------------------------------------------------------------
# Sync monitor
# ---------------------------------------------------------------------------
class SyncStateOut(BaseModel):
"""Resumable sync cursor (one per entity/channel/scope)."""
model_config = ConfigDict(from_attributes=True)
state_id: int
entity_type: str
channel: str | None = None
scope_id: str | None = None
last_version: str | None = None
last_timestamp: int | None = None
last_synced_at: dt.datetime | None = None
class SyncRunOut(BaseModel):
"""Audit row for a single sync execution."""
model_config = ConfigDict(from_attributes=True)
run_id: int
entity_type: str
channel: str | None = None
scope_id: str | None = None
status: str
started_at: dt.datetime
finished_at: dt.datetime | None = None
records_fetched: int
records_upserted: int
error: str | None = None
class SyncStatusOut(BaseModel):
"""Aggregate monitor payload — current cursors plus recent runs."""
states: list[SyncStateOut]
recent_runs: list[SyncRunOut]
+79 -1
View File
@@ -1,8 +1,22 @@
"""Shared pytest fixtures for the yunke-server test suite.""" """Shared pytest fixtures for the yunke-server test suite."""
import pytest from __future__ import annotations
from collections.abc import AsyncIterator, Awaitable, Callable
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import (
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from sqlalchemy.pool import StaticPool
from app.api import deps
from app.api.main import app
from app.config import get_settings from app.config import get_settings
from app.db.base import Base
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
@@ -34,3 +48,67 @@ def _disable_request_throttle(monkeypatch):
monkeypatch.setenv("REQUEST_JITTER", "0") monkeypatch.setenv("REQUEST_JITTER", "0")
yield yield
get_settings.cache_clear() get_settings.cache_clear()
@pytest.fixture
async def make_client() -> AsyncIterator[
Callable[[Callable[[AsyncSession], Awaitable[None]]], Awaitable[AsyncClient]]
]:
"""Factory yielding an ``AsyncClient`` over a seeded in-memory SQLite DB.
Mirrors the override pattern established in ``tests/test_api_customer.py``
(``StaticPool`` + single in-memory connection + ``get_session`` dependency
override) but lets each test supply its own seed coroutine so the read-API
test files don't each re-~40-line the boilerplate.
Usage::
async def test_x(make_client):
async def seed(session):
session.add(Department(dept_id="d1", ...))
await session.commit()
client = await make_client(seed)
r = await client.get("/api/departments")
...
The factory owns the engine + client lifetime: both are torn down
(``aclose`` / ``dispose``) and the dependency override cleared when the
test exits.
"""
created: list[tuple[AsyncClient, object]] = []
async def _factory(
seed: Callable[[AsyncSession], Awaitable[None]],
) -> AsyncClient:
engine = create_async_engine(
"sqlite+aiosqlite:///:memory:",
poolclass=StaticPool,
connect_args={"check_same_thread": False},
)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
async with SessionLocal() as session:
await seed(session)
async def _override_session() -> AsyncIterator[AsyncSession]:
async with SessionLocal() as s:
yield s
app.dependency_overrides[deps.get_session] = _override_session
client = AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
)
created.append((client, engine))
return client
try:
yield _factory
finally:
for client, engine in created:
await client.aclose()
await engine.dispose()
app.dependency_overrides.clear()
+128
View File
@@ -0,0 +1,128 @@
"""Tests for GET /api/calls — phone/voice call log.
Seeds 3 calls (2 phone_call for user u1 at 1000/3000, 1 voice_call for u2 at
2000) and asserts filters + ordering (``start_call_time desc``) + pagination.
"""
from __future__ import annotations
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import PhoneCall
async def _seed(session: AsyncSession) -> None:
session.add_all(
[
PhoneCall(
call_id="c1",
channel="phone_call",
user_id="u1",
user_id_name="Alice",
start_call_time=1000,
call_duration=30,
call_status=1,
incoming_call=0,
call_number="111",
sim_phone="111",
),
PhoneCall(
call_id="c2",
channel="voice_call",
user_id="u2",
user_id_name="Bob",
start_call_time=2000,
call_duration=60,
call_status=1,
incoming_call=1,
),
PhoneCall(
call_id="c3",
channel="phone_call",
user_id="u1",
user_id_name="Alice",
start_call_time=3000,
call_duration=45,
call_status=0,
incoming_call=0,
call_number="111",
),
]
)
await session.commit()
async def test_default_list_returns_all_ordered_desc(make_client):
"""No filters -> 3 rows, ordered by ``start_call_time desc``."""
client = await make_client(_seed)
r = await client.get("/api/calls")
assert r.status_code == 200
body = r.json()
assert body["total"] == 3
assert [it["call_id"] for it in body["items"]] == ["c3", "c2", "c1"]
for key in (
"call_id",
"channel",
"start_call_time",
"call_duration",
"call_status",
"incoming_call",
"user_id",
"user_id_name",
"call_number",
"sim_phone",
):
assert key in body["items"][0]
async def test_filter_by_user_id(make_client):
"""``?user_id=u1`` -> 2 calls for Alice."""
client = await make_client(_seed)
r = await client.get("/api/calls", params={"user_id": "u1"})
assert r.status_code == 200
body = r.json()
assert body["total"] == 2
assert {it["call_id"] for it in body["items"]} == {"c1", "c3"}
async def test_filter_by_channel_voice_call(make_client):
"""``?channel=voice_call`` -> the single voice call."""
client = await make_client(_seed)
r = await client.get("/api/calls", params={"channel": "voice_call"})
assert r.status_code == 200
body = r.json()
assert body["total"] == 1
assert body["items"][0]["call_id"] == "c2"
async def test_time_window_filter(make_client):
"""``?start=1500&end=2500`` -> only c2 (start 2000)."""
client = await make_client(_seed)
r = await client.get("/api/calls", params={"start": 1500, "end": 2500})
assert r.status_code == 200
body = r.json()
assert body["total"] == 1
assert body["items"][0]["call_id"] == "c2"
async def test_filters_compose(make_client):
"""``user_id=u1 & start=1500`` -> only c3 (u1's later call)."""
client = await make_client(_seed)
r = await client.get(
"/api/calls", params={"user_id": "u1", "start": 1500}
)
assert r.status_code == 200
body = r.json()
assert body["total"] == 1
assert body["items"][0]["call_id"] == "c3"
async def test_pagination(make_client):
"""``page_size=1`` -> 1 item, total 3."""
client = await make_client(_seed)
r = await client.get("/api/calls", params={"page": 1, "page_size": 1})
assert r.status_code == 200
body = r.json()
assert body["total"] == 3
assert len(body["items"]) == 1
assert body["items"][0]["call_id"] == "c3"
+144
View File
@@ -0,0 +1,144 @@
"""Tests for GET /api/groups and /api/groups/{room_id}/members.
Seeds 2 groups (one wechat, one whatsapp) with one member each on the wechat
group, plus a second wechat group owned by another account, then asserts list
filters + member-list shape + 404 for unknown room.
"""
from __future__ import annotations
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import ChatGroup, GroupMember
async def _seed(session: AsyncSession) -> None:
session.add_all(
[
ChatGroup(
room_id="room_a@chatroom",
channel="wechat",
room_name="Team A",
owner_account_id="wx_owner_1",
owner_account_nickname="WxOne",
member_count=2,
message_count=10,
last_chat_time=2000,
),
ChatGroup(
room_id="room_b@chatroom",
channel="wechat",
room_name="Team B",
owner_account_id="wx_owner_2",
member_count=1,
last_chat_time=1000,
),
ChatGroup(
room_id="wa_room_x@s.whatsapp.net",
channel="whatsapp",
room_name="WA Team",
owner_account_id="wa_owner_1",
member_count=3,
last_chat_time=None,
),
GroupMember(
channel="wechat",
room_id="room_a@chatroom",
member_id="mem1",
nickname="Alice",
display_name="Alice",
),
GroupMember(
channel="wechat",
room_id="room_a@chatroom",
member_id="mem2",
nickname="Bob",
display_name="Bob",
),
]
)
await session.commit()
async def test_default_list_returns_all_ordered_desc(make_client):
"""No filters -> 3 groups, ordered by ``last_chat_time desc`` (nulls last)."""
client = await make_client(_seed)
r = await client.get("/api/groups")
assert r.status_code == 200
body = r.json()
assert body["total"] == 3
# 2000, 1000, null -> [room_a, room_b, wa_room_x]
assert [it["room_id"] for it in body["items"]] == [
"room_a@chatroom",
"room_b@chatroom",
"wa_room_x@s.whatsapp.net",
]
for key in (
"room_id",
"channel",
"room_name",
"owner_account_id",
"owner_account_nickname",
"member_count",
"message_count",
"last_chat_time",
):
assert key in body["items"][0]
async def test_filter_by_channel_whatsapp(make_client):
"""``?channel=whatsapp`` -> the single WA group."""
client = await make_client(_seed)
r = await client.get("/api/groups", params={"channel": "whatsapp"})
assert r.status_code == 200
body = r.json()
assert body["total"] == 1
assert body["items"][0]["room_id"] == "wa_room_x@s.whatsapp.net"
async def test_filter_by_owner_account_id(make_client):
"""``?owner_account_id=wx_owner_1`` -> only Team A."""
client = await make_client(_seed)
r = await client.get("/api/groups", params={"owner_account_id": "wx_owner_1"})
assert r.status_code == 200
body = r.json()
assert body["total"] == 1
assert body["items"][0]["room_id"] == "room_a@chatroom"
async def test_pagination(make_client):
"""``page_size=1`` -> 1 item, total 3."""
client = await make_client(_seed)
r = await client.get("/api/groups", params={"page": 1, "page_size": 1})
assert r.status_code == 200
body = r.json()
assert body["total"] == 3
assert len(body["items"]) == 1
async def test_members_returns_list(make_client):
"""``/api/groups/room_a@chatroom/members`` -> 2 members (plain list)."""
client = await make_client(_seed)
r = await client.get("/api/groups/room_a@chatroom/members")
assert r.status_code == 200
items = r.json()
assert isinstance(items, list)
assert len(items) == 2
assert {it["member_id"] for it in items} == {"mem1", "mem2"}
for key in ("channel", "room_id", "member_id", "nickname", "display_name", "avatar"):
assert key in items[0]
async def test_members_unknown_room_404(make_client):
"""Unknown room_id -> 404 (distinguishes missing vs empty)."""
client = await make_client(_seed)
r = await client.get("/api/groups/nope@chatroom/members")
assert r.status_code == 404
async def test_members_empty_room_is_200(make_client):
"""Known room with no seeded members -> 200 + empty list (not 404)."""
client = await make_client(_seed)
r = await client.get("/api/groups/room_b@chatroom/members")
assert r.status_code == 200
assert r.json() == []
+187
View File
@@ -0,0 +1,187 @@
"""Tests for GET /api/messages — chat timeline.
Seeds 3 messages (2 for talker t_a across times 1000/3000, 1 for talker t_b at
2000) and asserts filters (channel, owner, talker, time window) + ordering
(``timestamp asc``) + pagination.
"""
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="m1",
channel="wechat",
owner_account_id="wx_owner_1",
talker_id="t_a",
room_id=None,
text="hello",
timestamp=1000,
is_send=True,
type="text",
type_raw=1,
),
Message(
msg_id="m2",
channel="wechat",
owner_account_id="wx_owner_1",
talker_id="t_b",
room_id=None,
text="hi",
timestamp=2000,
is_send=False,
type="text",
type_raw=1,
),
Message(
msg_id="m3",
channel="wechat",
owner_account_id="wx_owner_1",
talker_id="t_a",
room_id=None,
text="bye",
timestamp=3000,
is_send=True,
type="text",
type_raw=1,
),
]
)
await session.commit()
async def test_default_page_returns_all_ordered_asc(make_client):
"""No filters -> 3 rows ordered by ``timestamp asc`` (timeline order)."""
client = await make_client(_seed)
r = await client.get("/api/messages")
assert r.status_code == 200
body = r.json()
assert body["total"] == 3
assert len(body["items"]) == 3
assert [it["msg_id"] for it in body["items"]] == ["m1", "m2", "m3"]
for key in (
"msg_id",
"channel",
"owner_account_id",
"talker_id",
"room_id",
"is_send",
"type",
"type_raw",
"text",
"file_url",
"sender_name",
"sender_head_url",
"timestamp",
):
assert key in body["items"][0]
async def test_filter_by_talker_id(make_client):
"""``?talker_id=t_a`` -> 2 messages for that 1:1 conversation."""
client = await make_client(_seed)
r = await client.get("/api/messages", params={"talker_id": "t_a"})
assert r.status_code == 200
body = r.json()
assert body["total"] == 2
assert {it["msg_id"] for it in body["items"]} == {"m1", "m3"}
async def test_filter_by_owner_account_id(make_client):
"""``?owner_account_id=wx_owner_1`` -> all 3 (all owned by this account)."""
client = await make_client(_seed)
r = await client.get("/api/messages", params={"owner_account_id": "wx_owner_1"})
assert r.status_code == 200
assert r.json()["total"] == 3
async def test_filter_by_owner_account_id_no_match(make_client):
"""A non-existent owner account -> 0 rows."""
client = await make_client(_seed)
r = await client.get("/api/messages", params={"owner_account_id": "nobody"})
assert r.status_code == 200
assert r.json()["total"] == 0
async def test_time_window_filter(make_client):
"""``?start=1500&end=2500`` -> only m2 (timestamp 2000)."""
client = await make_client(_seed)
r = await client.get(
"/api/messages", params={"start": 1500, "end": 2500}
)
assert r.status_code == 200
body = r.json()
assert body["total"] == 1
assert body["items"][0]["msg_id"] == "m2"
async def test_time_window_combined_with_talker(make_client):
"""``talker_id=t_a & start=1500`` -> only m3 (t_a's later message)."""
client = await make_client(_seed)
r = await client.get(
"/api/messages", params={"talker_id": "t_a", "start": 1500}
)
assert r.status_code == 200
body = r.json()
assert body["total"] == 1
assert body["items"][0]["msg_id"] == "m3"
async def test_channel_filter(make_client):
"""``?channel=whatsapp`` -> 0 (all seeded are wechat)."""
client = await make_client(_seed)
r = await client.get("/api/messages", params={"channel": "whatsapp"})
assert r.status_code == 200
assert r.json()["total"] == 0
async def test_pagination_page_size_one(make_client):
"""``page_size=1&page=1`` -> 1 item, total still 3; page 2 -> next item."""
client = await make_client(_seed)
r1 = await client.get(
"/api/messages", params={"page": 1, "page_size": 1}
)
b1 = r1.json()
assert b1["total"] == 3
assert len(b1["items"]) == 1
assert b1["items"][0]["msg_id"] == "m1"
r2 = await client.get(
"/api/messages", params={"page": 2, "page_size": 1}
)
b2 = r2.json()
assert len(b2["items"]) == 1
assert b2["items"][0]["msg_id"] == "m2"
async def test_group_chat_filter_by_room_id(make_client):
"""``?room_id=room_x`` -> only the group message."""
async def seed(session: AsyncSession) -> None:
await _seed(session)
session.add(
Message(
msg_id="m4",
channel="wechat",
owner_account_id="wx_owner_1",
talker_id=None,
room_id="room_x",
text="group hi",
timestamp=4000,
type="text",
type_raw=1,
)
)
await session.commit()
client = await make_client(seed)
r = await client.get("/api/messages", params={"room_id": "room_x"})
assert r.status_code == 200
body = r.json()
assert body["total"] == 1
assert body["items"][0]["msg_id"] == "m4"
+206
View File
@@ -0,0 +1,206 @@
"""Tests for GET /api/departments, /api/staff, /api/accounts.
Seeds a small org (3 departments, 3 staff, 2 wechat + 1 whatsapp accounts) and
asserts shape + filters + pagination across all three endpoints.
"""
from __future__ import annotations
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import Department, Staff, WechatAccount, WhatsappAccount
async def _seed(session: AsyncSession) -> None:
session.add_all(
[
Department(dept_id="root", parent_id=None, name="Root", path="root"),
Department(
dept_id="d1", parent_id="root", name="Sales", path="root/d1"
),
Department(
dept_id="d2", parent_id="root", name="Support", path="root/d2"
),
Staff(
user_id="u1",
name="Alice",
dept_id="d1",
is_on_job=True,
phone="111",
),
Staff(
user_id="u2",
name="Bob",
dept_id="d2",
is_on_job=False,
),
Staff(user_id="u3", name="Carol", dept_id="d1", is_on_job=True),
WechatAccount(
wx_id="wx_a",
nick_name="WxOne",
user_id="u1",
user_name="Alice",
phone="111",
contacts_num=10,
group_num=2,
is_on_job=True,
),
WechatAccount(
wx_id="wx_b",
nick_name="WxTwo",
user_id="u2",
contacts_num=5,
is_on_job=False,
),
WhatsappAccount(
uid="wa_c@s.whatsapp.net",
nick_name="WaOne",
user_id="u3",
user_name="Carol",
phone="333",
contacts_num=7,
group_num=1,
is_on_job=True,
),
]
)
await session.commit()
async def test_departments_returns_all_ordered_by_path(make_client):
"""``GET /api/departments`` -> plain list, ordered by ``path``."""
client = await make_client(_seed)
r = await client.get("/api/departments")
assert r.status_code == 200
items = r.json()
assert isinstance(items, list)
assert len(items) == 3
# Path order: root, root/d1, root/d2
assert [it["dept_id"] for it in items] == ["root", "d1", "d2"]
# Shape
for key in ("dept_id", "parent_id", "name", "path"):
assert key in items[0]
async def test_staff_default_page_returns_all(make_client):
"""``GET /api/staff`` -> 3 rows, page 1 / page_size 20."""
client = await make_client(_seed)
r = await client.get("/api/staff")
assert r.status_code == 200
body = r.json()
assert body["page"] == 1
assert body["page_size"] == 20
assert body["total"] == 3
assert len(body["items"]) == 3
for key in ("user_id", "name", "phone", "sale_phone", "dept_id", "is_on_job"):
assert key in body["items"][0]
async def test_staff_filter_by_department_id(make_client):
"""``?department_id=d1`` -> 2 staff (Alice + Carol)."""
client = await make_client(_seed)
r = await client.get("/api/staff", params={"department_id": "d1"})
assert r.status_code == 200
body = r.json()
assert body["total"] == 2
assert {it["user_id"] for it in body["items"]} == {"u1", "u3"}
async def test_staff_filter_by_is_on_job(make_client):
"""``?is_on_job=true`` -> 2 on-job staff (Alice + Carol)."""
client = await make_client(_seed)
r = await client.get("/api/staff", params={"is_on_job": "true"})
assert r.status_code == 200
body = r.json()
assert body["total"] == 2
assert all(it["is_on_job"] is True for it in body["items"])
async def test_staff_filters_compose(make_client):
"""``department_id=d1 & is_on_job=true`` AND together -> still 2."""
client = await make_client(_seed)
r = await client.get(
"/api/staff", params={"department_id": "d1", "is_on_job": "true"}
)
assert r.status_code == 200
assert r.json()["total"] == 2
async def test_staff_pagination(make_client):
"""``page_size=1&page=2`` -> 1 item; total still 3."""
client = await make_client(_seed)
r = await client.get("/api/staff", params={"page": 2, "page_size": 1})
assert r.status_code == 200
body = r.json()
assert body["total"] == 3
assert len(body["items"]) == 1
async def test_accounts_default_returns_both_channels(make_client):
"""``GET /api/accounts`` with no channel -> 3 rows across both tables."""
client = await make_client(_seed)
r = await client.get("/api/accounts")
assert r.status_code == 200
body = r.json()
assert body["total"] == 3
# Sorted by (channel, account_id): wechat first, then whatsapp.
ids = [(it["channel"], it["account_id"]) for it in body["items"]]
assert ids == [
("wechat", "wx_a"),
("wechat", "wx_b"),
("whatsapp", "wa_c@s.whatsapp.net"),
]
# Unified shape exposes channel + key fields.
item = body["items"][0]
for key in (
"channel",
"account_id",
"nickname",
"user_id",
"user_name",
"phone",
"contacts_num",
"group_num",
"is_on_job",
):
assert key in item
assert item["channel"] == "wechat"
assert item["nickname"] == "WxOne"
assert item["user_name"] == "Alice"
async def test_accounts_filter_channel_wechat(make_client):
"""``?channel=wechat`` -> only the 2 wechat accounts."""
client = await make_client(_seed)
r = await client.get("/api/accounts", params={"channel": "wechat"})
assert r.status_code == 200
body = r.json()
assert body["total"] == 2
assert all(it["channel"] == "wechat" for it in body["items"])
async def test_accounts_filter_channel_whatsapp(make_client):
"""``?channel=whatsapp`` -> the single whatsapp account."""
client = await make_client(_seed)
r = await client.get("/api/accounts", params={"channel": "whatsapp"})
assert r.status_code == 200
body = r.json()
assert body["total"] == 1
assert body["items"][0]["account_id"] == "wa_c@s.whatsapp.net"
assert body["items"][0]["user_name"] == "Carol"
async def test_accounts_pagination(make_client):
"""``page_size=2&page=1`` -> 2 items, total 3; page 2 -> 1 item."""
client = await make_client(_seed)
r1 = await client.get("/api/accounts", params={"page": 1, "page_size": 2})
assert r1.status_code == 200
b1 = r1.json()
assert b1["total"] == 3
assert len(b1["items"]) == 2
r2 = await client.get("/api/accounts", params={"page": 2, "page_size": 2})
assert r2.status_code == 200
b2 = r2.json()
assert len(b2["items"]) == 1
assert b2["items"][0]["account_id"] == "wa_c@s.whatsapp.net"
+111
View File
@@ -0,0 +1,111 @@
"""Tests for GET /api/sms — SMS log.
Seeds 3 SMS rows (2 for user u1 at create_time 1000/3000, 1 for u2 at 2000)
and asserts filters + ordering (``create_time desc``) + pagination.
"""
from __future__ import annotations
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import Sms
async def _seed(session: AsyncSession) -> None:
session.add_all(
[
Sms(
sms_id="s1",
channel="sms",
user_id="u1",
receive_name="Alice",
content_preview="hello",
create_time=1000,
direction=1,
sender="111",
customer_phone="999",
),
Sms(
sms_id="s2",
channel="sms",
user_id="u2",
receive_name="Bob",
content_preview="hi",
create_time=2000,
direction=0,
),
Sms(
sms_id="s3",
channel="sms",
user_id="u1",
receive_name="Alice",
content_preview="bye",
create_time=3000,
direction=1,
),
]
)
await session.commit()
async def test_default_list_returns_all_ordered_desc(make_client):
"""No filters -> 3 rows ordered by ``create_time desc``."""
client = await make_client(_seed)
r = await client.get("/api/sms")
assert r.status_code == 200
body = r.json()
assert body["total"] == 3
assert [it["sms_id"] for it in body["items"]] == ["s3", "s2", "s1"]
for key in (
"sms_id",
"customer_phone",
"content_preview",
"sender",
"direction",
"user_id",
"receive_name",
"create_time",
"sim_phone",
):
assert key in body["items"][0]
async def test_filter_by_user_id(make_client):
"""``?user_id=u1`` -> 2 SMS for Alice."""
client = await make_client(_seed)
r = await client.get("/api/sms", params={"user_id": "u1"})
assert r.status_code == 200
body = r.json()
assert body["total"] == 2
assert {it["sms_id"] for it in body["items"]} == {"s1", "s3"}
async def test_time_window_filter(make_client):
"""``?start=1500&end=2500`` -> only s2."""
client = await make_client(_seed)
r = await client.get("/api/sms", params={"start": 1500, "end": 2500})
assert r.status_code == 200
body = r.json()
assert body["total"] == 1
assert body["items"][0]["sms_id"] == "s2"
async def test_filters_compose(make_client):
"""``user_id=u1 & end=2000`` -> only s1."""
client = await make_client(_seed)
r = await client.get("/api/sms", params={"user_id": "u1", "end": 2000})
assert r.status_code == 200
body = r.json()
assert body["total"] == 1
assert body["items"][0]["sms_id"] == "s1"
async def test_pagination(make_client):
"""``page_size=1&page=1`` -> 1 item, total 3."""
client = await make_client(_seed)
r = await client.get("/api/sms", params={"page": 1, "page_size": 1})
assert r.status_code == 200
body = r.json()
assert body["total"] == 3
assert len(body["items"]) == 1
assert body["items"][0]["sms_id"] == "s3"
+136
View File
@@ -0,0 +1,136 @@
"""Tests for GET /api/sync/status — sync monitor.
Seeds 2 SyncState cursors and 3 SyncRun audit rows and asserts the aggregate
shape ``{states, recent_runs}``, ordering of runs (newest first), and the
``recent`` cap.
"""
from __future__ import annotations
import datetime as dt
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import SyncRun, SyncState
async def _seed(session: AsyncSession) -> None:
session.add_all(
[
SyncState(
state_id=1,
entity_type="message",
channel="wechat",
scope_id="wx_owner_1",
last_version="v100",
last_timestamp=1000,
last_synced_at=dt.datetime(2026, 8, 1, 10, 0, 0, tzinfo=dt.UTC),
),
SyncState(
state_id=2,
entity_type="customer",
channel="whatsapp",
scope_id=None,
last_version="v9",
last_timestamp=2000,
),
SyncRun(
run_id=1,
entity_type="message",
channel="wechat",
scope_id="wx_owner_1",
status="success",
started_at=dt.datetime(2026, 8, 1, 9, 0, 0, tzinfo=dt.UTC),
finished_at=dt.datetime(2026, 8, 1, 9, 0, 5, tzinfo=dt.UTC),
records_fetched=100,
records_upserted=95,
),
SyncRun(
run_id=2,
entity_type="customer",
channel="whatsapp",
status="failed",
started_at=dt.datetime(2026, 8, 1, 10, 0, 0, tzinfo=dt.UTC),
records_fetched=0,
records_upserted=0,
error="upstream 500",
),
SyncRun(
run_id=3,
entity_type="message",
channel="wechat",
status="running",
started_at=dt.datetime(2026, 8, 1, 11, 0, 0, tzinfo=dt.UTC),
records_fetched=0,
records_upserted=0,
),
]
)
await session.commit()
async def test_status_returns_states_and_recent_runs(make_client):
"""``GET /api/sync/status`` -> both keys present with correct shapes."""
client = await make_client(_seed)
r = await client.get("/api/sync/status")
assert r.status_code == 200
body = r.json()
assert set(body.keys()) == {"states", "recent_runs"}
assert len(body["states"]) == 2
for key in (
"state_id",
"entity_type",
"channel",
"scope_id",
"last_version",
"last_timestamp",
"last_synced_at",
):
assert key in body["states"][0]
assert len(body["recent_runs"]) == 3
for key in (
"run_id",
"entity_type",
"channel",
"scope_id",
"status",
"started_at",
"finished_at",
"records_fetched",
"records_upserted",
"error",
):
assert key in body["recent_runs"][0]
async def test_recent_runs_ordered_newest_first(make_client):
"""Runs ordered by ``started_at desc`` -> run 3, 2, 1."""
client = await make_client(_seed)
r = await client.get("/api/sync/status")
body = r.json()
ids = [run["run_id"] for run in body["recent_runs"]]
assert ids == [3, 2, 1]
async def test_recent_limit_caps_results(make_client):
"""``?recent=2`` -> only the 2 most recent runs."""
client = await make_client(_seed)
r = await client.get("/api/sync/status", params={"recent": 2})
assert r.status_code == 200
body = r.json()
assert len(body["recent_runs"]) == 2
assert [run["run_id"] for run in body["recent_runs"]] == [3, 2]
async def test_empty_db_returns_empty_lists(make_client):
"""No seed -> 200 with empty states + recent_runs (no crash)."""
async def empty_seed(session: AsyncSession) -> None:
await session.commit()
client = await make_client(empty_seed)
r = await client.get("/api/sync/status")
assert r.status_code == 200
body = r.json()
assert body == {"states": [], "recent_runs": []}