shaguabijia-app-server/tests/test_notification_events.py

"""业务事件 → 站内通知 + 厂商推送 联动测试(services/notification_events)。

覆盖 PRD 六类真实触发:
  #3 提现成功 / #4 提现失败(含审核拒绝) / #9 官方回复 / #10 反馈奖励 /
  #11 爆料审核通过 / #12 好友下单到账。
厂商推送不真发:测试环境无凭据默认跳过;推送链路用 monkeypatch 捕获/注错验证
「有设备则推、推挂了业务不受影响」。wxpay 网络调用照旧全 monkeypatch。
"""
from __future__ import annotations

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select

from app.admin.main import admin_app
from app.admin.repositories import admin_user as admin_repo
from app.core.rewards import INVITE_COMPARE_REWARD_CENTS, PRICE_REPORT_REWARD_COINS
from app.core.security import decode_token, hash_password
from app.db.session import SessionLocal
from app.models.feedback import Feedback
from app.models.price_report import PriceReport
from app.models.wallet import CoinAccount, WithdrawOrder
from app.repositories import device as device_repo
from app.repositories import wallet as crud_wallet
from app.services import notification_events

# ===== 用户侧 helpers(同 test_withdraw / test_notifications)=====

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 _notifications(client: TestClient, token: str) -> list[dict]:
    r = client.get("/api/v1/notifications?pageSize=50", headers=_auth(token))
    assert r.status_code == 200, r.text
    return r.json()["items"]


def _seed_cash(client: TestClient, token: str, cents: int) -> None:
    """先访问 /account 触发建账户,再直接灌现金余额。"""
    client.get("/api/v1/wallet/account", headers=_auth(token))
    with SessionLocal() as db:
        acc = db.get(CoinAccount, _uid(token))
        acc.cash_balance_cents = cents
        db.commit()


def _create_withdraw(client: TestClient, token: str, monkeypatch, cents: int = 50) -> str:
    """绑微信 + 发起提现(进入 reviewing),返回 out_bill_no。"""
    monkeypatch.setattr(
        "app.integrations.wxpay.code_to_userinfo",
        lambda code: {"openid": f"openid_{_uid(token)}", "nickname": None, "avatar_url": None, "raw": {}},
    )
    _seed_cash(client, token, cents * 2)
    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
    r = client.post("/api/v1/wallet/withdraw", json={"amount_cents": cents}, headers=_auth(token))
    assert r.status_code == 200, r.text
    assert r.json()["status"] == "reviewing"
    return r.json()["out_bill_no"]


# ===== admin 侧 helpers(同 test_admin_write)=====

@pytest.fixture()
def admin_client() -> TestClient:
    return TestClient(admin_app)


@pytest.fixture()
def operator_token() -> str:
    with SessionLocal() as db:
        a = admin_repo.get_by_username(db, "ne_operator")
        if a is None:
            admin_repo.create_admin(db, username="ne_operator", password="pass1234", role="operator")
        else:
            a.password_hash = hash_password("pass1234")
            a.role = "operator"
            a.status = "active"
            db.commit()
    c = TestClient(admin_app)
    return c.post(
        "/admin/api/auth/login", json={"username": "ne_operator", "password": "pass1234"}
    ).json()["access_token"]


def _seed_feedback(uid: int) -> int:
    with SessionLocal() as db:
        fb = Feedback(user_id=uid, content="比价按钮找不到", contact="wx123", status="pending")
        db.add(fb)
        db.commit()
        return fb.id


def _seed_price_report(uid: int, store: str = "蜀大侠火锅") -> int:
    with SessionLocal() as db:
        rep = PriceReport(
            user_id=uid,
            store_name=store,
            reported_platform_id="mt",
            reported_platform_name="美团",
            reported_price_cents=990,
            images=[],
            status="pending",
        )
        db.add(rep)
        db.commit()
        return rep.id


# ===== #4 提现失败(审核拒绝路径,_refund_withdraw 收口)=====

def test_withdraw_reject_creates_failed_notification(client: TestClient, monkeypatch) -> None:
    token = _login(client, "13800005001")
    bill = _create_withdraw(client, token, monkeypatch)

    with SessionLocal() as db:
        crud_wallet.reject_withdraw(db, bill, "微信零钱未实名")

    items = _notifications(client, token)
    failed = [i for i in items if i["type"] == "withdraw_failed"]
    assert len(failed) == 1
    n = failed[0]
    assert n["cashCents"] == 50
    assert n["cashYuan"] == "0.50"
    assert n["title"] == "提现失败，款项已退回"
    assert n["extra"]["withdrawId"] == bill
    assert n["isRead"] is False
    rows = {r["label"]: r["value"] for r in n["infoRows"]}
    assert rows["失败原因"] == "微信零钱未实名"
    assert rows["退回说明"] == "款项已原路退回现金余额"


def test_withdraw_failed_event_dedup_single_unread(client: TestClient, monkeypatch) -> None:
    """同一提现单重复触发失败事件(并发查单等)→ 未读期间只落一条。"""
    token = _login(client, "13800005002")
    bill = _create_withdraw(client, token, monkeypatch)
    with SessionLocal() as db:
        crud_wallet.reject_withdraw(db, bill, "审核未通过")
        order = db.execute(
            select(WithdrawOrder).where(WithdrawOrder.out_bill_no == bill)
        ).scalar_one()
        notification_events.notify_withdraw_failed(db, order)  # 人为重复触发

    items = [i for i in _notifications(client, token) if i["type"] == "withdraw_failed"]
    assert len(items) == 1


# ===== #3 提现成功(查单归一化路径)=====

def test_withdraw_success_notification_on_status_query(client: TestClient, monkeypatch) -> None:
    token = _login(client, "13800005003")
    monkeypatch.setattr(
        "app.integrations.wxpay.create_transfer",
        lambda openid, amount_fen, out_bill_no, user_name=None: {
            "status_code": 200,
            "data": {"state": "WAIT_USER_CONFIRM", "package_info": "pkg", "transfer_bill_no": "tb"},
        },
    )
    bill = _create_withdraw(client, token, monkeypatch)
    with SessionLocal() as db:
        crud_wallet.approve_withdraw(db, bill)  # 审核通过 → 转账进 pending(等用户确认)

    assert [i for i in _notifications(client, token) if i["type"] == "withdraw_success"] == []

    # 用户确认后查单 → SUCCESS → success + 下发「提现到账」通知
    monkeypatch.setattr(
        "app.integrations.wxpay.query_transfer",
        lambda out_bill_no: {"status_code": 200, "data": {"state": "SUCCESS"}},
    )
    r = client.get("/api/v1/wallet/withdraw/status", params={"out_bill_no": bill}, headers=_auth(token))
    assert r.json()["status"] == "success"

    ok = [i for i in _notifications(client, token) if i["type"] == "withdraw_success"]
    assert len(ok) == 1
    n = ok[0]
    assert n["cashCents"] == 50
    assert n["actionText"] is None  # PRD:提现成功卡无操作行
    rows = {r["label"]: r["value"] for r in n["infoRows"]}
    assert rows["到账账户"] == "微信钱包"
    assert "到账时间" in rows

    # 再查一次(已终态,早退)→ 不重复下发
    client.get("/api/v1/wallet/withdraw/status", params={"out_bill_no": bill}, headers=_auth(token))
    assert len([i for i in _notifications(client, token) if i["type"] == "withdraw_success"]) == 1


# ===== #10 反馈奖励(admin 采纳发金币)=====

def test_feedback_approve_sends_reward_notification(
    client: TestClient, admin_client: TestClient, operator_token: str
) -> None:
    token = _login(client, "13800005004")
    fb_id = _seed_feedback(_uid(token))

    r = admin_client.post(
        f"/admin/api/feedbacks/{fb_id}/approve",
        json={"reward_coins": 300, "note": "好建议", "reply": "问题已修复上线，送您的金币请查收～"},
        headers=_auth(operator_token),
    )
    assert r.status_code == 200, r.text

    items = [i for i in _notifications(client, token) if i["type"] == "feedback_reward"]
    assert len(items) == 1
    n = items[0]
    assert n["coins"] == 300
    assert n["extra"]["feedbackId"] == str(fb_id)
    rows = {r["label"]: r["value"] for r in n["infoRows"]}
    assert rows["官方留言"] == "问题已修复上线，送您的金币请查收～"  # PRD:发奖必带官方留言
    assert "到账时间" in rows


# ===== #9 官方回复(admin 拒绝,原因/留言用户可见)=====

def test_feedback_reject_sends_reply_notification(
    client: TestClient, admin_client: TestClient, operator_token: str
) -> None:
    token = _login(client, "13800005005")
    fb_id = _seed_feedback(_uid(token))

    r = admin_client.post(
        f"/admin/api/feedbacks/{fb_id}/reject",
        json={"reason": "无法复现", "reply": "麻烦补个录屏,我们再看看~"},
        headers=_auth(operator_token),
    )
    assert r.status_code == 200, r.text

    items = [i for i in _notifications(client, token) if i["type"] == "feedback_reply"]
    assert len(items) == 1
    n = items[0]
    assert n["title"] == "傻瓜比价官方回复了您的反馈"
    assert n["extra"]["feedbackId"] == str(fb_id)
    assert n["coins"] is None


# ===== #11 爆料审核通过(admin 通过发固定金币)=====

def test_price_report_approve_sends_notification(
    client: TestClient, admin_client: TestClient, operator_token: str
) -> None:
    token = _login(client, "13800005006")
    rep_id = _seed_price_report(_uid(token), store="蜀大侠火锅")

    r = admin_client.post(
        f"/admin/api/price-reports/{rep_id}/approve", headers=_auth(operator_token)
    )
    assert r.status_code == 200, r.text

    items = [i for i in _notifications(client, token) if i["type"] == "report_approved"]
    assert len(items) == 1
    n = items[0]
    assert n["coins"] == PRICE_REPORT_REWARD_COINS
    assert n["extra"]["reportId"] == str(rep_id)
    rows = {r["label"]: r["value"] for r in n["infoRows"]}
    assert "蜀大侠火锅" in rows["奖励说明"]


# ===== #12 好友下单到账(好友首次成功比价 → 通知邀请人)=====

def test_invite_compare_reward_sends_notification(client: TestClient) -> None:
    a = _login(client, "13800005007")
    b = _login(client, "13800005008")
    code = client.get("/api/v1/invite/me", headers=_auth(a)).json()["invite_code"]
    client.post("/api/v1/invite/bind", json={"invite_code": code}, headers=_auth(b))

    r = client.post(
        "/api/v1/compare/record",
        json={"trace_id": "trace-notif-1", "status": "success"},
        headers=_auth(b),
    )
    assert r.status_code == 200, r.text

    items = [i for i in _notifications(client, a) if i["type"] == "invite_order_reward"]
    assert len(items) == 1
    n = items[0]
    assert n["cashCents"] == INVITE_COMPARE_REWARD_CENTS
    assert n["extra"]["inviteeNickname"]  # 好友昵称兜底(昵称/尾号/「好友」)非空
    # 被邀请人自己不收该通知
    assert [i for i in _notifications(client, b) if i["type"] == "invite_order_reward"] == []

    # 好友再比价 → 不再发奖也不再通知(发奖幂等 + 通知 dedup 双保险)
    client.post(
        "/api/v1/compare/record",
        json={"trace_id": "trace-notif-2", "status": "success"},
        headers=_auth(b),
    )
    assert len([i for i in _notifications(client, a) if i["type"] == "invite_order_reward"]) == 1


# ===== 厂商推送联动(有设备则推;推送失败不伤业务)=====

def _register_device(uid: int, vendor: str = "xiaomi", token: str = "regid-1") -> None:
    with SessionLocal() as db:
        device_repo.register_or_update(
            db, user_id=uid, device_id=f"dev_{uid}", push_vendor=vendor, push_token=token
        )


def test_push_sent_to_registered_device(client: TestClient, monkeypatch) -> None:
    token = _login(client, "13800005009")
    _register_device(_uid(token))

    sent: list[dict] = []
    monkeypatch.setattr(notification_events.vendor_push, "missing_settings", lambda vendor: [])

    def _capture(vendor, push_token, *, title, body, extras=None, mock=False):
        sent.append({"vendor": vendor, "token": push_token, "title": title, "body": body, "extras": extras})
        return {"ok": True}

    monkeypatch.setattr(notification_events.vendor_push, "send_notification", _capture)

    bill = _create_withdraw(client, token, monkeypatch)
    with SessionLocal() as db:
        crud_wallet.reject_withdraw(db, bill, "微信零钱未实名")

    assert len(sent) == 1
    p = sent[0]
    assert p["vendor"] == "xiaomi" and p["token"] == "regid-1"
    assert p["title"] == "提现失败，款项已退回"
    assert "0.50" in p["body"] and "微信零钱未实名" in p["body"]
    # PRD §4 push 已读联动:extras 带 type + notificationId + 跳转参数
    assert p["extras"]["type"] == "withdraw_failed"
    assert p["extras"]["withdrawId"] == bill
    nid = int(p["extras"]["notificationId"])
    assert any(i["id"] == nid for i in _notifications(client, token))


def test_push_failure_does_not_break_business(client: TestClient, monkeypatch) -> None:
    """推送炸了(哪怕不是 VendorPushError)→ 提现拒绝照常退款,站内消息照常落库。"""
    token = _login(client, "13800005010")
    _register_device(_uid(token), token="regid-2")

    monkeypatch.setattr(notification_events.vendor_push, "missing_settings", lambda vendor: [])

    def _boom(*args, **kwargs):
        raise RuntimeError("vendor api down")

    monkeypatch.setattr(notification_events.vendor_push, "send_notification", _boom)

    bill = _create_withdraw(client, token, monkeypatch)
    with SessionLocal() as db:
        crud_wallet.reject_withdraw(db, bill, "审核未通过")

    r = client.get("/api/v1/wallet/withdraw/status", params={"out_bill_no": bill}, headers=_auth(token))
    assert r.json()["status"] == "rejected"  # 业务不受影响
    r = client.get("/api/v1/wallet/account", headers=_auth(token))
    assert r.json()["cash_balance_cents"] == 100  # 已退回(seed 100 扣 50 退 50)
    assert len([i for i in _notifications(client, token) if i["type"] == "withdraw_failed"]) == 1