shaguabijia-app-server/tests/test_notifications.py

"""消息通知中心 3 接口(mock 数据阶段)。

覆盖:列表字段/排序/分页、未读角标、标记已读(ids / all / 幂等 / 参数校验)、
鉴权与用户间隔离。数据来自 repositories/notification_mock.py 的内存 mock。
"""
from __future__ import annotations

from fastapi.testclient import TestClient

from app.core import notification_catalog as catalog
from app.repositories import notification_mock as mock_repo


def _login(client: TestClient, phone: str) -> str:
    client.post("/api/v1/auth/sms/send", json={"phone": phone})
    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
    assert r.status_code == 200, r.text
    return r.json()["access_token"]


def _auth(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


def _fetch_all(client: TestClient, token: str) -> dict:
    r = client.get("/api/v1/notifications?pageSize=100", headers=_auth(token))
    assert r.status_code == 200, r.text
    return r.json()


def test_requires_auth(client: TestClient) -> None:
    assert client.get("/api/v1/notifications").status_code == 401
    assert client.get("/api/v1/notifications/unread-count").status_code == 401
    assert client.post("/api/v1/notifications/read", json={"all": True}).status_code == 401


def test_list_covers_all_types_with_camel_case_fields(client: TestClient) -> None:
    token = _login(client, "13900010001")
    data = _fetch_all(client, token)

    assert data["total"] >= 13
    assert data["unreadCount"] > 0
    items = data["items"]
    assert len(items) == data["total"]

    # 13 种类型全覆盖
    assert {i["type"] for i in items} == set(catalog.TYPES)

    # 字段按 PRD 契约 camelCase,且卡片要素齐全
    first = items[0]
    for key in (
        "id", "category", "categoryLabel", "type", "cardStyle", "title",
        "coins", "cashCents", "cashYuan", "infoRows", "actionText",
        "extra", "sentAt", "isRead",
    ):
        assert key in first, f"missing field {key}"
    assert "+08:00" in first["sentAt"]

    # 双金额卡:金币 + 现金都有;金币为整数、现金串两位小数
    expiring = next(i for i in items if i["type"] == "reward_expiring")
    assert isinstance(expiring["coins"], int)
    assert expiring["cashYuan"] == "12.80"
    assert expiring["cashCents"] == 1280
    assert expiring["actionText"] == "立即激活您的收益"
    assert [row["label"] for row in expiring["infoRows"]] == ["过期说明", "过期时间"]

    # 提现成功卡:无操作行、无跳转
    ws = next(i for i in items if i["type"] == "withdraw_success")
    assert ws["actionText"] is None
    assert ws["coins"] is None

    # 权限异常卡带 permission 参数(客户端点击时实时检测用)
    perm = next(i for i in items if i["type"] == "perm_accessibility")
    assert perm["extra"] == {"permission": "accessibility"}

    # 反馈奖励卡:官方留言必填(PRD §3)
    reward = next(i for i in items if i["type"] == "feedback_reward")
    assert any(row["label"] == "官方留言" and row["value"] for row in reward["infoRows"])


def test_list_sorted_by_time_desc_without_grouping(client: TestClient) -> None:
    """全列表时间倒序、不按分类分组(PRD 原文的分组已确认取消)。"""
    token = _login(client, "13900010002")
    items = _fetch_all(client, token)["items"]

    sent_ats = [i["sentAt"] for i in items]
    assert sent_ats == sorted(sent_ats, reverse=True), "全列表时间倒序,最新在前"

    # mock 样例里「好友下单(10 分钟前)」比「昨天的提现失败」新但分类靠后 →
    # 若还在按分类分组,它不可能排进前 3;以此钉死"确实没分组"。
    newest_types = {i["type"] for i in items[:3]}
    assert "invite_order_reward" in newest_types


def test_pagination(client: TestClient) -> None:
    token = _login(client, "13900010003")
    total = _fetch_all(client, token)["total"]

    page_size = 5
    seen_ids: list[int] = []
    page = 1
    while True:
        r = client.get(
            f"/api/v1/notifications?page={page}&pageSize={page_size}", headers=_auth(token)
        )
        assert r.status_code == 200
        data = r.json()
        assert data["page"] == page
        assert data["pageSize"] == page_size
        assert data["total"] == total
        seen_ids.extend(i["id"] for i in data["items"])
        if not data["hasMore"]:
            assert len(data["items"]) <= page_size
            break
        assert len(data["items"]) == page_size
        page += 1

    assert len(seen_ids) == total
    assert len(set(seen_ids)) == total, "翻页不重不漏"

    # 超出末页 → 空页而非报错
    r = client.get("/api/v1/notifications?page=99&pageSize=50", headers=_auth(token))
    assert r.status_code == 200
    assert r.json()["items"] == []
    assert r.json()["hasMore"] is False


def test_unread_count_and_badge(client: TestClient) -> None:
    token = _login(client, "13900010004")
    data = _fetch_all(client, token)
    expect_unread = sum(1 for i in data["items"] if not i["isRead"])

    r = client.get("/api/v1/notifications/unread-count", headers=_auth(token))
    assert r.status_code == 200
    body = r.json()
    assert body["count"] == expect_unread == data["unreadCount"]
    assert body["badgeText"] == str(expect_unread)  # mock 样例未读数在 1~99 区间

    # 全部读完 → count=0,badgeText=null(整个角标隐藏)
    client.post("/api/v1/notifications/read", json={"all": True}, headers=_auth(token))
    r = client.get("/api/v1/notifications/unread-count", headers=_auth(token))
    assert r.json() == {"count": 0, "badgeText": None}


def test_mark_read_by_ids_idempotent(client: TestClient) -> None:
    token = _login(client, "13900010005")
    items = _fetch_all(client, token)["items"]
    unread_ids = [i["id"] for i in items if not i["isRead"]]
    before = len(unread_ids)
    picked = unread_ids[:2]

    r = client.post("/api/v1/notifications/read", json={"ids": picked}, headers=_auth(token))
    assert r.status_code == 200
    assert r.json() == {"ok": True, "markedCount": 2, "unreadCount": before - 2}

    # 列表状态同步翻转
    items = _fetch_all(client, token)["items"]
    assert all(i["isRead"] for i in items if i["id"] in picked)

    # 重复置读 + 不存在的 id → 幂等,不报错
    r = client.post(
        "/api/v1/notifications/read", json={"ids": [*picked, 123456789]}, headers=_auth(token)
    )
    assert r.status_code == 200
    assert r.json()["markedCount"] == 0
    assert r.json()["unreadCount"] == before - 2


def test_mark_read_requires_ids_or_all(client: TestClient) -> None:
    token = _login(client, "13900010006")
    r = client.post("/api/v1/notifications/read", json={}, headers=_auth(token))
    assert r.status_code == 400
    r = client.post("/api/v1/notifications/read", json={"ids": []}, headers=_auth(token))
    assert r.status_code == 400


def test_stores_isolated_between_users(client: TestClient) -> None:
    token_a = _login(client, "13900010007")
    token_b = _login(client, "13900010008")

    client.post("/api/v1/notifications/read", json={"all": True}, headers=_auth(token_a))

    r = client.get("/api/v1/notifications/unread-count", headers=_auth(token_b))
    assert r.json()["count"] > 0, "A 清零不影响 B"


def test_reset_store_regenerates(client: TestClient) -> None:
    token = _login(client, "13900010009")
    client.post("/api/v1/notifications/read", json={"all": True}, headers=_auth(token))
    assert client.get("/api/v1/notifications/unread-count", headers=_auth(token)).json()["count"] == 0

    mock_repo.reset_store()  # 联调辅助:清空后重新生成整套样例
    assert client.get("/api/v1/notifications/unread-count", headers=_auth(token)).json()["count"] > 0