yunke/yunke-server/tests/test_api_calls.py

"""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"