shaguabijia-app-server/tests/test_notifications.py

"""消息通知中心 3 接口(落库版)。

覆盖:空列表(虚拟数据已清除)、列表字段/派生/排序/分页、未读角标、标记已读(ids / all /
幂等 / 参数校验)、鉴权与用户隔离、去重键部分唯一索引。数据直接写 notification 表
(repositories/notification),不再有内存 mock。
"""
from __future__ import annotations

from datetime import datetime, timedelta, timezone

import pytest
from fastapi.testclient import TestClient
from sqlalchemy.exc import IntegrityError

from app.core.security import decode_token
from app.db.session import SessionLocal
from app.repositories import notification as notif_repo

_CST = timezone(timedelta(hours=8))


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 _uid(token: str) -> int:
    return int(decode_token(token, expected_type="access")["sub"])


def _seed_sample(token: str, type_key: str):
    """按类型插一条样例通知(复用 repo 的样例卡片内容),返回落库行。"""
    with SessionLocal() as db:
        return notif_repo.insert_sample(db, _uid(token), type_key)


def _seed(token: str, type_key: str, **kw):
    """按显式内容插一条通知(排序/分页用,可指定 sent_at)。"""
    with SessionLocal() as db:
        return notif_repo.create_notification(db, user_id=_uid(token), type_key=type_key, **kw)


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_fresh_user_has_no_notifications(client: TestClient) -> None:
    """虚拟数据已清除:新用户初始为空列表、未读 0、角标隐藏。"""
    token = _login(client, "13900010000")
    data = _fetch_all(client, token)
    assert data["items"] == []
    assert data["total"] == 0
    assert data["unreadCount"] == 0
    assert data["hasMore"] is False

    r = client.get("/api/v1/notifications/unread-count", headers=_auth(token))
    assert r.json() == {"count": 0, "badgeText": None}


def test_list_item_fields_camel_case_and_derived(client: TestClient) -> None:
    token = _login(client, "13900010001")
    _seed_sample(token, "reward_expiring")
    _seed_sample(token, "withdraw_success")
    _seed_sample(token, "perm_accessibility")
    _seed_sample(token, "feedback_reward")

    items = _fetch_all(client, token)["items"]
    assert len(items) == 4

    # 字段按 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"]  # 落库后仍恒带 +08:00

    by_type = {i["type"]: i for i in items}

    # 双金额卡:金币整数 + 现金两位小数;category/cardStyle/title/actionText 均由 catalog 派生
    expiring = by_type["reward_expiring"]
    assert expiring["category"] == "withdraw_assistant"
    assert expiring["categoryLabel"] == "提现助手"
    assert expiring["cardStyle"] == "dual_amount"
    assert expiring["title"] == "金币现金奖励即将失效"
    assert expiring["actionText"] == "立即激活您的收益"
    assert isinstance(expiring["coins"], int)
    assert expiring["cashCents"] == 1280
    assert expiring["cashYuan"] == "12.80"
    assert [row["label"] for row in expiring["infoRows"]] == ["过期说明", "过期时间"]

    # 提现成功卡:无操作行、无金币
    ws = by_type["withdraw_success"]
    assert ws["actionText"] is None
    assert ws["coins"] is None

    # 权限异常卡带 permission 参数(客户端点击时实时检测用)
    assert by_type["perm_accessibility"]["extra"] == {"permission": "accessibility"}

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

    # 新插入默认未读
    assert all(i["isRead"] is False for i in items)


def test_list_sorted_by_time_desc(client: TestClient) -> None:
    """全列表 sent_at 倒序(最新在前),不分组。"""
    token = _login(client, "13900010002")
    base = datetime(2026, 7, 1, 12, 0, tzinfo=_CST)
    _seed(token, "withdraw_success", sent_at=base - timedelta(days=2))
    newest = _seed(token, "invite_order_reward", sent_at=base)
    _seed(token, "feedback_reply", sent_at=base - timedelta(days=1))

    items = _fetch_all(client, token)["items"]
    sent_ats = [i["sentAt"] for i in items]
    assert sent_ats == sorted(sent_ats, reverse=True), "最新在前"
    assert items[0]["id"] == newest.id
    assert items[0]["type"] == "invite_order_reward"


def test_pagination(client: TestClient) -> None:
    token = _login(client, "13900010003")
    base = datetime(2026, 7, 1, 12, 0, tzinfo=_CST)
    n = 12
    for i in range(n):
        _seed(token, "withdraw_success", sent_at=base - timedelta(minutes=i))

    total = _fetch_all(client, token)["total"]
    assert total == n

    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")
    for _ in range(3):
        _seed_sample(token, "withdraw_success")

    r = client.get("/api/v1/notifications/unread-count", headers=_auth(token))
    assert r.json() == {"count": 3, "badgeText": "3"}

    # 全部读完 → 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")
    ids = [_seed_sample(token, "withdraw_success").id for _ in range(3)]
    picked = 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": 1}

    # 列表状态同步翻转
    items = {i["id"]: i for i in _fetch_all(client, token)["items"]}
    assert all(items[i]["isRead"] for i in picked)
    assert items[ids[2]]["isRead"] is False

    # 重复置读 + 不存在的 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"] == 1


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_isolated_between_users(client: TestClient) -> None:
    token_a = _login(client, "13900010007")
    token_b = _login(client, "13900010008")
    _seed_sample(token_a, "withdraw_success")
    _seed_sample(token_b, "withdraw_success")

    client.post("/api/v1/notifications/read", json={"all": True}, headers=_auth(token_a))
    assert client.get("/api/v1/notifications/unread-count", headers=_auth(token_a)).json()["count"] == 0
    assert (
        client.get("/api/v1/notifications/unread-count", headers=_auth(token_b)).json()["count"] == 1
    ), "A 清零不影响 B"


def test_dedup_key_blocks_duplicate_unread(client: TestClient) -> None:
    """同一 (user, type, dedup_key) 未读期间只允许一条(部分唯一索引拦重复未读)。"""
    token = _login(client, "13900010009")
    uid = _uid(token)
    with SessionLocal() as db:
        notif_repo.create_notification(
            db, user_id=uid, type_key="perm_accessibility", dedup_key="accessibility"
        )
    # 同键第二条(仍未读)→ 唯一索引拦截
    with pytest.raises(IntegrityError):
        with SessionLocal() as db:
            notif_repo.create_notification(
                db, user_id=uid, type_key="perm_accessibility", dedup_key="accessibility"
            )
    # 只落了一条
    assert _fetch_all(client, token)["total"] == 1