Merge branch 'main' of https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server into feat/openobserve-api-metrics

# Conflicts:
#	app/main.py

tests/test_ad_ecpm_trace_revenue.py

@@ -0,0 +1,73 @@
+"""ad_ecpm_record.trace_id 落库 + 按 trace 聚合广告收益(元)。"""
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import delete
+
+from app.db.session import SessionLocal
+from app.models.ad_ecpm import AdEcpmRecord
+from app.repositories import ad_ecpm as crud_ecpm
+
+
+def _ecpm(trace_id: str, ecpm_raw: str, session_id: str) -> AdEcpmRecord:
+    """构造一条 Draw 展示 eCPM(不 commit;ad_session_id 全局唯一,须各不相同)。"""
+    return AdEcpmRecord(
+        user_id=1,
+        ad_type="draw",
+        feed_scene="comparison",
+        ad_session_id=session_id,
+        ecpm_raw=ecpm_raw,
+        trace_id=trace_id,
+        report_date="2020-01-02",
+        created_at=datetime(2020, 1, 2, tzinfo=UTC),
+    )
+
+
+def test_revenue_yuan_by_trace_sums_and_clamps() -> None:
+    """同一 trace 多条展示求和;收益=min(eCPM元,¥500)/1000;无展示的 trace 不出现。
+
+    ecpm 200 分→2.0 元/千次→0.002 元/次;300 分→0.003;合计 0.005。
+    """
+    db = SessionLocal()
+    try:
+        db.add_all([
+            _ecpm("t1", "200", "sess-t1-a"),
+            _ecpm("t1", "300", "sess-t1-b"),
+            _ecpm("t2", "0", "sess-t2-a"),
+            _ecpm("t_cap", "60000", "sess-cap-a"),  # 600 元 CPM > ¥500 钳顶
+        ])
+        db.flush()
+        rev = crud_ecpm.revenue_yuan_by_trace(db, ["t1", "t2", "t3", "t_cap"])
+        assert rev["t1"] == 0.005
+        assert rev["t2"] == 0.0  # 有展示但 eCPM=0 → 0 元(仍在结果里)
+        assert "t3" not in rev  # 无展示的 trace 不出现在结果里
+        assert rev["t_cap"] == 0.5  # min(¥600, ¥500)/1000,证明钳顶生效(非 0.6)
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_revenue_yuan_by_trace_empty() -> None:
+    """空 trace 列表直接返回 {}(避免 IN () 非法)。"""
+    db = SessionLocal()
+    try:
+        assert crud_ecpm.revenue_yuan_by_trace(db, []) == {}
+    finally:
+        db.close()
+
+
+def test_create_ecpm_record_persists_trace_id() -> None:
+    """create_ecpm_record 落 trace_id。"""
+    db = SessionLocal()
+    try:
+        rec = crud_ecpm.create_ecpm_record(
+            db, 1, ad_type="draw", ecpm_raw="150",
+            ad_session_id="sess-trace-persist", feed_scene="coupon",
+            trace_id="trace-xyz",
+        )
+        assert rec.trace_id == "trace-xyz"
+    finally:
+        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.ad_session_id == "sess-trace-persist"))
+        db.commit()
+        db.close()

tests/test_admin_read.py

@@ -320,3 +320,92 @@ def test_read_apis_require_auth(admin_client: TestClient) -> None:
         "/admin/api/feedbacks",
     ]:
         assert admin_client.get(path).status_code == 401, path
+
+
+def test_period_comparison_reward_coin_from_feed_scene(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    """比价奖励金币口径:按 ad_feed_reward_record.feed_scene='comparison' 的实发金币
+    (status=granted)汇总,而非查从不写入的 biz_type 桶(修复大盘该卡恒 0)。
+    too_short(未发奖)不计。"""
+    from app.models.ad_feed_reward import AdFeedRewardRecord
+
+    d = "2021-06-15"  # 独立历史日,隔离其它用例数据
+    db = SessionLocal()
+    try:
+        uid = user_repo.upsert_user_for_login(db, phone="13800008801", register_channel="sms").id
+        db.add(AdFeedRewardRecord(
+            client_event_id="cmp-fs-granted", user_id=uid, reward_date=d,
+            ecpm_raw="0", coin=123, feed_scene="comparison", status="granted",
+        ))
+        db.add(AdFeedRewardRecord(
+            client_event_id="cmp-fs-tooshort", user_id=uid, reward_date=d,
+            ecpm_raw="0", coin=99, feed_scene="comparison", status="too_short",
+        ))
+        db.commit()
+    finally:
+        db.close()
+
+    r = admin_client.get(
+        "/admin/api/stats/overview", params={"date_from": d, "date_to": d},
+        headers=_auth(admin_token),
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["period"]["coins"]["comparison_reward_coin_total"] == 123
+
+
+def test_period_coupon_reward_coin_from_feed_scene(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    """领券奖励金币口径:按 ad_feed_reward_record.feed_scene='coupon' 的实发金币汇总。"""
+    from app.models.ad_feed_reward import AdFeedRewardRecord
+
+    d = "2021-06-16"
+    db = SessionLocal()
+    try:
+        uid = user_repo.upsert_user_for_login(db, phone="13800008802", register_channel="sms").id
+        db.add(AdFeedRewardRecord(
+            client_event_id="cpn-fs-granted", user_id=uid, reward_date=d,
+            ecpm_raw="0", coin=456, feed_scene="coupon", status="granted",
+        ))
+        db.commit()
+    finally:
+        db.close()
+
+    r = admin_client.get(
+        "/admin/api/stats/overview", params={"date_from": d, "date_to": d},
+        headers=_auth(admin_token),
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["period"]["coins"]["coupon_reward_coin_total"] == 456
+
+
+def test_period_coupon_reward_excludes_reward_video(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    """领券奖励金币不再把激励视频金币算进来(reward_video/ad_reward 从领券桶拆出);
+    激励视频仍单独计入 reward_video_coin_total。"""
+    from datetime import datetime
+
+    from app.models.wallet import CoinTransaction
+
+    d = "2021-06-17"
+    db = SessionLocal()
+    try:
+        uid = user_repo.upsert_user_for_login(db, phone="13800008803", register_channel="sms").id
+        db.add(CoinTransaction(
+            user_id=uid, amount=50, balance_after=50, biz_type="reward_video",
+            ref_id="rv-split-1", created_at=datetime(2021, 6, 17, 12, 0, 0),
+        ))
+        db.commit()
+    finally:
+        db.close()
+
+    r = admin_client.get(
+        "/admin/api/stats/overview", params={"date_from": d, "date_to": d},
+        headers=_auth(admin_token),
+    )
+    assert r.status_code == 200, r.text
+    coins = r.json()["period"]["coins"]
+    assert coins["coupon_reward_coin_total"] == 0        # 激励视频不计入领券奖励
+    assert coins["reward_video_coin_total"] == 50        # 仍计入激励视频卡

tests/test_admin_roles.py

@@ -150,3 +150,80 @@ def test_create_admin_rejects_unknown_role(admin_client, super_token) -> None:
         headers=_auth(super_token),
     )
     assert r.status_code == 400
+
+
+def _login_pages(username: str, password: str = "pass1234") -> list[str]:
+    """以某账号登录,取下发的有效可见页(左侧导航过滤依据)。"""
+    c = TestClient(admin_app)
+    return c.post(
+        "/admin/api/auth/login", json={"username": username, "password": password}
+    ).json()["admin"]["pages"]
+
+
+def test_custom_role_create_pages_take_effect(admin_client, super_token) -> None:
+    # 建「自定义」账号:role=custom + 勾选页(含一个非法 key,应被过滤)
+    r = admin_client.post(
+        "/admin/api/admins",
+        json={
+            "username": "cust_user", "password": "pass1234", "role": "custom",
+            "pages_override": ["dashboard", "withdraws", "xx-bad"],
+        },
+        headers=_auth(super_token),
+    )
+    assert r.status_code == 200, r.text
+    # 登录下发的 pages == 勾选集(非法 key 过滤),真正驱动左侧导航
+    assert set(_login_pages("cust_user")) == {"dashboard", "withdraws"}
+    # 账号列表回显原始勾选集(供编辑回填),同样已过滤
+    admins = {a["username"]: a for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()}
+    assert set(admins["cust_user"]["pages_override"]) == {"dashboard", "withdraws"}
+    assert admins["cust_user"]["role"] == "custom"
+
+
+def test_custom_role_update_and_switch_back_clears_override(admin_client, super_token) -> None:
+    admin_client.post(
+        "/admin/api/admins",
+        json={
+            "username": "cust_sw", "password": "pass1234", "role": "custom",
+            "pages_override": ["dashboard"],
+        },
+        headers=_auth(super_token),
+    )
+    aid = next(
+        a["id"] for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()
+        if a["username"] == "cust_sw"
+    )
+    # 只改勾选集(角色不变)→ 生效
+    u = admin_client.patch(
+        f"/admin/api/admins/{aid}", json={"pages_override": ["dashboard", "feedbacks"]},
+        headers=_auth(super_token),
+    )
+    assert u.status_code == 200, u.text
+    assert set(_login_pages("cust_sw")) == {"dashboard", "feedbacks"}
+    # 切回普通角色 operator → override 清空,pages 跟随角色(运营页集,且不含 admins)
+    u2 = admin_client.patch(
+        f"/admin/api/admins/{aid}", json={"role": "operator"}, headers=_auth(super_token)
+    )
+    assert u2.status_code == 200, u2.text
+    admins = {a["username"]: a for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()}
+    assert admins["cust_sw"]["pages_override"] is None
+    pages = set(_login_pages("cust_sw"))
+    assert "feedbacks" in pages and "admins" not in pages  # 运营口径,自定义页已不生效
+
+
+def test_switch_operator_to_custom_sets_override(admin_client, super_token) -> None:
+    admin_client.post(
+        "/admin/api/admins",
+        json={"username": "op2cust", "password": "pass1234", "role": "operator"},
+        headers=_auth(super_token),
+    )
+    aid = next(
+        a["id"] for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()
+        if a["username"] == "op2cust"
+    )
+    u = admin_client.patch(
+        f"/admin/api/admins/{aid}",
+        json={"role": "custom", "pages_override": ["config"]},
+        headers=_auth(super_token),
+    )
+    assert u.status_code == 200, u.text
+    assert set(_login_pages("op2cust")) == {"config"}

tests/test_analytics_health.py

@@ -0,0 +1,160 @@
+"""埋点健康度聚合:纯差分函数 + admin 端点。"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from app.admin.repositories.analytics_health import _cn_day, _rates, diff_snapshots
+
+
+def _row(device, epoch, event, ts, **cum) -> dict:
+    base = {"attempted": 0, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0}
+    base.update(cum)
+    return {
+        "device_id": device, "epoch_id": epoch, "event": event,
+        "created_at": datetime(2026, 7, 1, ts, 0, tzinfo=timezone.utc),
+        "app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14",
+        **base,
+    }
+
+
+def test_diff_first_row_is_full_cumulative() -> None:
+    rows = [_row("d", "e", "video_play", 1, attempted=100, delivered=90)]
+    out = diff_snapshots(rows)
+    assert len(out) == 1
+    assert out[0]["d_attempted"] == 100
+    assert out[0]["d_delivered"] == 90
+
+
+def test_diff_consecutive_delta() -> None:
+    rows = [
+        _row("d", "e", "video_play", 1, attempted=100, delivered=90),
+        _row("d", "e", "video_play", 2, attempted=150, delivered=140),
+    ]
+    out = sorted(diff_snapshots(rows), key=lambda r: r["created_at"])
+    assert out[1]["d_attempted"] == 50
+    assert out[1]["d_delivered"] == 50
+
+
+def test_diff_epoch_reset_new_partition() -> None:
+    rows = [
+        _row("d", "e1", "video_play", 1, attempted=100),
+        _row("d", "e2", "video_play", 2, attempted=5),
+    ]
+    out = {(r["epoch_id"]): r["d_attempted"] for r in diff_snapshots(rows)}
+    assert out["e1"] == 100
+    assert out["e2"] == 5
+
+
+def test_diff_clamps_negative_on_reorder() -> None:
+    rows = [
+        _row("d", "e", "video_play", 1, attempted=100),
+        _row("d", "e", "video_play", 2, attempted=80),
+    ]
+    out = sorted(diff_snapshots(rows), key=lambda r: r["created_at"])
+    assert out[1]["d_attempted"] == 0
+
+
+def test_rates_computes_both_formulas() -> None:
+    out = _rates({"attempted": 150, "drop_capture": 0, "delivered": 140, "drop_undelivered": 10})
+    assert out["track_success_rate"] == 1.0
+    assert out["report_success_rate"] == 140 / 150
+
+
+def test_rates_zero_denominator_yields_none() -> None:
+    out = _rates({"attempted": 0, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0})
+    assert out["track_success_rate"] is None
+    assert out["report_success_rate"] is None
+
+
+def test_cn_day_beijing_boundary() -> None:
+    # UTC 15:59 → 北京 23:59 同日;UTC 16:00 → 北京 次日 00:00
+    assert _cn_day(datetime(2026, 7, 1, 15, 59, tzinfo=timezone.utc)) == "2026-07-01"
+    assert _cn_day(datetime(2026, 7, 1, 16, 0, tzinfo=timezone.utc)) == "2026-07-02"
+
+
+def test_diff_same_timestamp_ordered_by_id() -> None:
+    ts = datetime(2026, 7, 1, 1, 0, tzinfo=timezone.utc)
+    # 相同 created_at,乱序传入(id=2 在前);应按 id 排序 → id=1(cum100) 在前
+    rows = [
+        {"id": 2, "device_id": "d", "epoch_id": "e", "event": "vp", "created_at": ts,
+         "app_ver": "v", "oem": "o", "os": "s",
+         "attempted": 150, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0},
+        {"id": 1, "device_id": "d", "epoch_id": "e", "event": "vp", "created_at": ts,
+         "app_ver": "v", "oem": "o", "os": "s",
+         "attempted": 100, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0},
+    ]
+    out = diff_snapshots(rows)
+    assert out[0]["d_attempted"] == 100  # id=1 sorts first → its cumulative
+    assert out[1]["d_attempted"] == 50   # id=2 second → 150-100
+
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.admin.main import admin_app
+from app.admin.repositories import admin_user as admin_repo
+from app.db.session import SessionLocal
+
+
+@pytest.fixture()
+def admin_client() -> TestClient:
+    return TestClient(admin_app)
+
+
+@pytest.fixture()
+def admin_token() -> str:
+    db = SessionLocal()
+    try:
+        if admin_repo.get_by_username(db, "health_admin") is None:
+            admin_repo.create_admin(db, username="health_admin", password="pw", role="super_admin")
+    finally:
+        db.close()
+    c = TestClient(admin_app)
+    r = c.post("/admin/api/auth/login", json={"username": "health_admin", "password": "pw"})
+    return r.json()["access_token"]
+
+
+def _auth(t: str) -> dict:
+    return {"Authorization": f"Bearer {t}"}
+
+
+def _seed_two_snapshots(client: TestClient) -> None:
+    for attempted, delivered in ((100, 90), (150, 140)):
+        client.post("/api/v1/analytics/selfstat", json={
+            "device_id": "d-health", "epoch_id": "e-health",
+            "app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14",
+            "events": [{"event": "video_play", "attempted": attempted,
+                        "drop_capture": 0, "delivered": delivered, "drop_undelivered": 0}],
+        })
+
+
+def test_health_overview_requires_auth(admin_client: TestClient) -> None:
+    r = admin_client.get("/admin/api/analytics-health/overview",
+                         params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z"})
+    assert r.status_code == 401
+
+
+def test_health_overview(client: TestClient, admin_client: TestClient, admin_token: str) -> None:
+    _seed_two_snapshots(client)  # 播种走公开 app(ingest 端点在 app,不在 admin_app)
+    r = admin_client.get(
+        "/admin/api/analytics-health/overview",
+        params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z"},
+        headers=_auth(admin_token),
+    )
+    assert r.status_code == 200, r.text
+    data = r.json()
+    # 两快照累计 100→150,差分后 attempted 总 150(首快照 100 + 增量 50)
+    assert data["attempted"] == 150
+    assert data["track_success_rate"] == 1.0
+
+
+def test_health_breakdown(client: TestClient, admin_client: TestClient, admin_token: str) -> None:
+    _seed_two_snapshots(client)  # 播种走公开 app
+    r = admin_client.get(
+        "/admin/api/analytics-health/breakdown",
+        params={"date_from": "2026-07-01T00:00:00Z", "date_to": "2030-01-01T00:00:00Z", "dim": "event"},
+        headers=_auth(admin_token),
+    )
+    assert r.status_code == 200, r.text
+    rows = r.json()
+    assert any(row["key"] == "video_play" for row in rows)

tests/test_analytics_selfstat.py

@@ -0,0 +1,43 @@
+"""自报计数上报端点测试。"""
+from __future__ import annotations
+
+from fastapi.testclient import TestClient
+
+
+def _payload(**over) -> dict:
+    base = {
+        "device_id": "dev-1", "epoch_id": "ep-1", "sent_at": 1700000000000,
+        "app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14",
+        "batches_attempted": 10, "batches_ok": 9, "batches_fail": 1,
+        "retries": 1, "queue_depth": 2,
+        "events": [
+            {"event": "video_play", "attempted": 100, "drop_capture": 1,
+             "delivered": 95, "drop_undelivered": 2},
+        ],
+    }
+    base.update(over)
+    return base
+
+
+def test_selfstat_ingest_ok(client: TestClient) -> None:
+    r = client.post("/api/v1/analytics/selfstat", json=_payload())
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["ok"] is True
+    assert isinstance(body["snapshot_id"], int)
+
+
+def test_selfstat_ingest_empty_events(client: TestClient) -> None:
+    r = client.post("/api/v1/analytics/selfstat", json=_payload(events=[]))
+    assert r.status_code == 200, r.text
+    assert r.json()["ok"] is True
+
+
+def test_selfstat_ingest_db_error_returns_503(client: TestClient, monkeypatch) -> None:
+    """落库异常时端点回 503(计数链路稳定性守卫),不裸奔 500。"""
+    def _boom(*_args, **_kwargs):
+        raise RuntimeError("db gone")
+
+    monkeypatch.setattr("app.repositories.analytics_selfstat.record_selfstat", _boom)
+    r = client.post("/api/v1/analytics/selfstat", json=_payload())
+    assert r.status_code == 503, r.text

tests/test_auth.py

@@ -107,6 +107,67 @@ def test_sms_send_device_ip_rate_limit(client, monkeypatch) -> None:
     assert r.status_code == 200, r.text
 
 
+def test_sms_send_cooldown_reject_not_counted(client, monkeypatch) -> None:
+    """发码额度只算「成功发码」:被单号 60s 冷却挡下的重发(429)不占设备额度。
+    做法:同号狂发只成功 1 次、其余被冷却挡下;把小时额度设 2,证明换号后仍能再成功发 1 次
+    —— 若冷却重发也计数,额度早被那几次耗尽。"""
+    from app.api.v1 import auth
+    from app.core import ratelimit
+
+    monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
+    monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_HOUR_PER_DEVICE", 2)
+    ratelimit._buckets.clear()
+
+    device = "dev-cooldown"
+    phone_a = "13710137000"
+    # 首发成功(小时闸计 1/2)
+    assert client.post(
+        "/api/v1/auth/sms/send", json={"phone": phone_a, "device_id": device}
+    ).status_code == 200
+    # 同号连发 3 次:都被单号 60s 冷却挡下 → 429,且**不占**设备额度
+    for _ in range(3):
+        r = client.post(
+            "/api/v1/auth/sms/send", json={"phone": phone_a, "device_id": device}
+        )
+        assert r.status_code == 429, r.text
+    # 换号再发:设备额度只用了 1/2(冷却那几次没算)→ 仍放行(计到 2/2)
+    assert client.post(
+        "/api/v1/auth/sms/send", json={"phone": "13710137001", "device_id": device}
+    ).status_code == 200
+    # 又换号:此时小时闸已 2/2 → 429(反证成功发码确实各计了 1)
+    r = client.post(
+        "/api/v1/auth/sms/send", json={"phone": "13710137002", "device_id": device}
+    )
+    assert r.status_code == 429, r.text
+
+
+def test_sms_send_daily_cap(client, monkeypatch) -> None:
+    """每天发码上限(设备 + IP):成功发码累计到日上限即 429(用不同手机号绕开单号冷却)。
+    抬高小时闸单独测日闸;超限文案含「今日」以便前端提示明天再来。"""
+    from app.api.v1 import auth
+    from app.core import ratelimit
+
+    monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
+    monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_HOUR_PER_DEVICE", 100)  # 抬高小时闸,不干扰
+    monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_DAY_PER_DEVICE", 3)
+    ratelimit._buckets.clear()
+
+    device = "dev-daily"
+    for i in range(3):
+        r = client.post(
+            "/api/v1/auth/sms/send",
+            json={"phone": f"13720137{i:03d}", "device_id": device},
+        )
+        assert r.status_code == 200, f"第 {i + 1} 次应放行: {r.text}"
+    # 第 4 次:同设备同 IP 当日超限 → 429
+    r = client.post(
+        "/api/v1/auth/sms/send",
+        json={"phone": "13720137999", "device_id": device},
+    )
+    assert r.status_code == 429, r.text
+    assert "今日" in r.json()["detail"]
+
+
 def test_sms_login_device_ip_rate_limit(client, monkeypatch) -> None:
     """防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_LOGIN_MAX_PER_HOUR 次登录尝试,超出 429。
     conftest 默认 RATE_LIMIT_ENABLED=false(内存计数跨用例累加),本用例临时打开并清空计数隔离。"""

tests/test_board_ad_revenue.py

@@ -0,0 +1,87 @@
+"""两个看板逐行「本次广告收益」(元):按 trace_id 聚合 ad_ecpm_record。"""
+from __future__ import annotations
+
+from datetime import UTC, date, datetime
+
+from sqlalchemy import delete
+
+from app.admin.repositories import queries
+from app.admin.repositories.coupon_data import coupon_data_report, coupon_user_records
+from app.db.session import SessionLocal
+from app.models.ad_ecpm import AdEcpmRecord
+from app.models.comparison import ComparisonRecord
+from app.models.coupon_state import CouponSession
+
+
+def _ecpm(trace_id: str, ecpm_raw: str, session_id: str, scene: str) -> AdEcpmRecord:
+    return AdEcpmRecord(
+        user_id=1, ad_type="draw", feed_scene=scene, ad_session_id=session_id,
+        ecpm_raw=ecpm_raw, trace_id=trace_id, report_date="2020-01-02",
+        created_at=datetime(2020, 1, 2, tzinfo=UTC),
+    )
+
+
+def test_coupon_data_report_includes_ad_revenue() -> None:
+    """领券看板明细行带本次广告收益;200+300 分 → 0.005 元。"""
+    db = SessionLocal()
+    try:
+        db.add(CouponSession(
+            trace_id="rev-cp-1", device_id="d1", status="completed", app_env="prod",
+            platforms=["meituan-waimai"], platform_success=["meituan-waimai"],
+            started_at=datetime(2020, 1, 2, tzinfo=UTC), started_date=date(2020, 1, 2),
+        ))
+        db.add_all([
+            _ecpm("rev-cp-1", "200", "cp-sess-a", "coupon"),
+            _ecpm("rev-cp-1", "300", "cp-sess-b", "coupon"),
+        ])
+        db.flush()
+        res = coupon_data_report(db, date_from="2020-01-02", date_to="2020-01-02", app_env="prod")
+        row = next(r for r in res["items"] if r["trace_id"] == "rev-cp-1")
+        assert row["ad_revenue_yuan"] == 0.005
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_comparison_list_includes_ad_revenue() -> None:
+    """比价记录列表项带本次广告收益;200 分 → 0.002 元。
+
+    用独有 user_id 过滤,确保本行必落在第一页(避免共享测试库里同 user 记录多、分页把它挤掉)。
+    """
+    db = SessionLocal()
+    try:
+        db.add(ComparisonRecord(
+            trace_id="rev-cmp-1", user_id=987654, status="success", business_type="food",
+            created_at=datetime(2020, 1, 2, tzinfo=UTC),
+        ))
+        db.add(_ecpm("rev-cmp-1", "200", "cmp-sess-a", "comparison"))
+        db.commit()
+        items, _next, _total = queries.list_comparison_records(db, user_id=987654)
+        row = next(it for it in items if it.trace_id == "rev-cmp-1")
+        assert row.ad_revenue_yuan == 0.002
+    finally:
+        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.trace_id.in_(["rev-cmp-1"])))
+        db.execute(delete(ComparisonRecord).where(ComparisonRecord.trace_id == "rev-cmp-1"))
+        db.commit()
+        db.close()
+
+
+def test_coupon_user_records_includes_ad_revenue() -> None:
+    """手机号抽屉的用户领券记录也带本次广告收益(400 分 → 0.004 元)。"""
+    db = SessionLocal()
+    try:
+        db.add(CouponSession(
+            trace_id="rev-drawer-1", device_id="d2", status="completed", app_env="prod",
+            user_id=987655, platforms=["meituan-waimai"], platform_success=["meituan-waimai"],
+            started_at=datetime(2020, 1, 2, tzinfo=UTC), started_date=date(2020, 1, 2),
+        ))
+        db.add(_ecpm("rev-drawer-1", "400", "drawer-sess-a", "coupon"))
+        db.commit()
+        res = coupon_user_records(db, user_id=987655)
+        row = next(r for r in res["items"] if r["trace_id"] == "rev-drawer-1")
+        assert row["ad_revenue_yuan"] == 0.004
+    finally:
+        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.trace_id == "rev-drawer-1"))
+        db.execute(delete(CouponSession).where(CouponSession.trace_id == "rev-drawer-1"))
+        db.commit()
+        db.close()

tests/test_coupon_platform_success.py

@@ -0,0 +1,299 @@
+"""领券「平台成功率」埋点:coupon_id→平台映射 / 成功平台推导 / session platform_success 并集。
+
+设计:docs/guides/领券成功率指标-设计与埋点.md
+"""
+from __future__ import annotations
+
+from datetime import UTC, date, datetime
+
+from sqlalchemy import delete, func, select
+
+from app.admin.repositories.coupon_data import coupon_data_report
+from app.db.session import SessionLocal
+from app.models.coupon_state import CouponSession
+from app.repositories.coupon_state import (
+    coupon_id_to_platform,
+    merge_session_platform_success,
+    succeeded_platforms,
+)
+
+
+def _agg_session(
+    trace: str, platforms, platform_success, *, status: str = "completed"
+) -> CouponSession:
+    """构造一条聚合测试用 session(started_date 固定 2020-01-02、app_env=prod,不 commit)。"""
+    return CouponSession(
+        trace_id=trace,
+        device_id="d-agg",
+        status=status,
+        app_env="prod",
+        platforms=platforms,
+        platform_success=platform_success,
+        started_at=datetime(2020, 1, 2, tzinfo=UTC),
+        started_date=date(2020, 1, 2),
+    )
+
+
+def _make_session(db, trace_id: str, **kw) -> CouponSession:
+    row = CouponSession(
+        trace_id=trace_id,
+        device_id="dev-merge",
+        status=kw.pop("status", "started"),
+        started_at=datetime(2020, 1, 1, tzinfo=UTC),
+        started_date=date(2020, 1, 1),
+        **kw,
+    )
+    db.add(row)
+    db.commit()
+    return row
+
+
+def test_coupon_id_to_platform_prefix_mapping() -> None:
+    """coupon_id 前缀 → 三档平台 id(与客户端 couponIdToPlatform 同词表)。"""
+    assert coupon_id_to_platform("mt_banjia_zhoumo") == "meituan-waimai"
+    assert coupon_id_to_platform("mt_cps_waimai_redpacket") == "meituan-waimai"  # 大众点评CPS 也挂 mt_ → 归美团
+    assert coupon_id_to_platform("tb_vip_shangou_voucher") == "taobao-shanguang"
+    assert coupon_id_to_platform("ele_hongbao") == "taobao-shanguang"
+    assert coupon_id_to_platform("elm_hongbao") == "taobao-shanguang"
+    assert coupon_id_to_platform("jd_redpacket") == "jd-waimai"
+
+
+def test_coupon_id_to_platform_unknown_returns_none() -> None:
+    """无法识别的前缀 / 空 → None(调用方跳过,不计入平台)。"""
+    assert coupon_id_to_platform("weird_xxx") is None
+    assert coupon_id_to_platform("") is None
+    assert coupon_id_to_platform(None) is None  # type: ignore[arg-type]
+
+
+def test_succeeded_platforms_filters_status_and_dedups() -> None:
+    """只取 status∈{success,already_claimed} 的券,映射平台后去重;失败/跳过/无法识别的不计。"""
+    results = [
+        {"coupon_id": "mt_a", "status": "success"},
+        {"coupon_id": "mt_b", "status": "already_claimed"},  # 也算成功 → 仍是美团
+        {"coupon_id": "mt_c", "status": "failed"},           # 不算
+        {"coupon_id": "tb_a", "status": "success"},
+        {"coupon_id": "jd_a", "status": "skipped"},          # 不算
+        {"coupon_id": "weird", "status": "success"},         # 无法识别平台 → 跳过
+    ]
+    assert set(succeeded_platforms(results)) == {"meituan-waimai", "taobao-shanguang"}
+
+
+def test_succeeded_platforms_empty() -> None:
+    assert succeeded_platforms([]) == []
+
+
+def test_merge_platform_success_union_idempotent() -> None:
+    """按 trace_id 把成功平台并入 platform_success:并集去重 + 按 DEFAULT_PLATFORMS 保序 + 重复并入幂等。"""
+    db = SessionLocal()
+    trace = "merge-union-1"
+    try:
+        _make_session(db, trace)
+        merge_session_platform_success(db, trace, ["meituan-waimai"])
+        merge_session_platform_success(db, trace, ["jd-waimai", "meituan-waimai"])  # 并集 + 已有幂等
+        db.expire_all()
+        row = db.execute(
+            select(CouponSession).where(CouponSession.trace_id == trace)
+        ).scalar_one()
+        # 美团→淘宝→京东固定序;只含实际成功的美团/京东
+        assert row.platform_success == ["meituan-waimai", "jd-waimai"]
+    finally:
+        db.execute(delete(CouponSession).where(CouponSession.trace_id == trace))
+        db.commit()
+        db.close()
+
+
+def test_merge_platform_success_missing_row_skips() -> None:
+    """trace_id 无对应行 → 静默跳过:不建兜底行、不抛异常(设计 §5)。"""
+    db = SessionLocal()
+    try:
+        merge_session_platform_success(db, "no-such-trace", ["meituan-waimai"])
+        n = db.execute(
+            select(func.count()).select_from(CouponSession).where(
+                CouponSession.trace_id == "no-such-trace"
+            )
+        ).scalar_one()
+        assert n == 0
+    finally:
+        db.close()
+
+
+def test_coupon_data_success_rates() -> None:
+    """admin 聚合 ②整单成功率 / ③点位成功率:平台粒度,含空 platforms→全领三档、abandoned 入基数。
+
+    A 勾美团+淘宝、两个都成 → 整单成功;点位 2/2
+    B 勾三档、只成美团       → 非整单;点位 1/3
+    C 全领(platforms空→3)、三档全成 → 整单成功;点位 3/3
+    D abandoned、勾美团、无成功平台 → 非整单;点位 0/1(仍进基数)
+    发起数=4;整单成功=2 → 0.5;点位 6/9 → 0.6667
+    """
+    db = SessionLocal()
+    try:
+        db.add_all([
+            _agg_session("agg-A", ["meituan-waimai", "taobao-shanguang"],
+                         ["meituan-waimai", "taobao-shanguang"]),
+            _agg_session("agg-B", ["meituan-waimai", "taobao-shanguang", "jd-waimai"],
+                         ["meituan-waimai"]),
+            _agg_session("agg-C", [], ["meituan-waimai", "taobao-shanguang", "jd-waimai"]),
+            _agg_session("agg-D", ["meituan-waimai"], None, status="abandoned"),
+        ])
+        db.flush()  # 同会话可见,不 commit(finally 回滚保持隔离)
+        s = coupon_data_report(
+            db, date_from="2020-01-02", date_to="2020-01-02", app_env="prod"
+        )["summary"]
+        assert s["started_count"] == 4
+        assert s["full_success_count"] == 2
+        assert s["full_success_rate"] == 0.5
+        assert s["point_success_count"] == 6
+        assert s["point_total_count"] == 9
+        assert s["point_success_rate"] == round(6 / 9, 4)
+        # ③ 分平台点位成功率(§12):美团 3/4、淘宝 2/3、京东 1/2。
+        # 分平台(成功,总)= 美团(3,4)+淘宝(2,3)+京东(1,2) = (6,9),
+        # 其和正好等于上面已断言的 point_success_count=6 / point_total_count=9(和不变量)。
+        assert s["per_platform"] == {
+            "meituan-waimai": 0.75,
+            "taobao-shanguang": round(2 / 3, 4),
+            "jd-waimai": 0.5,
+        }
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_coupon_data_success_rates_empty_range() -> None:
+    """区间无 session → 发起数 0、两个率为 None(不除零)。"""
+    db = SessionLocal()
+    try:
+        s = coupon_data_report(
+            db, date_from="2019-01-01", date_to="2019-01-01", app_env="prod"
+        )["summary"]
+        assert s["started_count"] == 0
+        assert s["full_success_rate"] is None
+        assert s["point_success_rate"] is None
+    finally:
+        db.close()
+
+
+def test_coupon_data_summary_schema_exposes_rates() -> None:
+    """schema 契约:CouponDataSummary 暴露 ②③ 字段,键名与 repo 输出一致(router 直接 **summary 构造)。"""
+    from app.admin.schemas.coupon_data import CouponDataSummary
+
+    db = SessionLocal()
+    try:
+        db.add(_agg_session("agg-schema-1", ["meituan-waimai"], ["meituan-waimai"]))
+        db.flush()
+        summary = coupon_data_report(
+            db, date_from="2020-01-02", date_to="2020-01-02", app_env="prod"
+        )["summary"]
+        dumped = CouponDataSummary(**summary).model_dump()
+        assert dumped["full_success_rate"] == summary["full_success_rate"]
+        assert dumped["point_success_rate"] == summary["point_success_rate"]
+        assert dumped["full_success_count"] == summary["full_success_count"]
+        assert dumped["point_success_count"] == summary["point_success_count"]
+        assert dumped["point_total_count"] == summary["point_total_count"]
+        assert dumped["per_platform"] == summary["per_platform"]
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_step_writes_platform_success_to_session(client) -> None:
+    """/step 集成:pricebot 返 coupon_results → 本 trace 的 coupon_session.platform_success 落库(仅成功平台)。"""
+    from unittest.mock import MagicMock, patch
+
+    import httpx
+
+    from app.models.coupon_state import CouponClaimRecord, CouponDailyCompletion
+
+    trace = "int-trace-1"
+    device = "dev-int-1"
+
+    # 预置一条 started session(模拟 /session started 已先落库)
+    db = SessionLocal()
+    try:
+        db.add(CouponSession(
+            trace_id=trace, device_id=device, status="started", app_env="dev",
+            platforms=["meituan-waimai", "taobao-shanguang"],
+            started_at=datetime(2020, 1, 5, tzinfo=UTC),
+            started_date=date(2020, 1, 5),
+        ))
+        db.commit()
+    finally:
+        db.close()
+
+    fake_resp = {
+        "success": True,
+        "action": {"command": "done", "params": {
+            "information": "已领 1 张",
+            "coupon_results": [
+                {"coupon_id": "mt_x", "name": "美团券", "vendor": "meituan_internal", "status": "success"},
+                {"coupon_id": "tb_y", "name": "淘宝券", "vendor": "taobao", "status": "failed"},
+            ],
+        }},
+        "continue": False,
+    }
+
+    async def fake_post(self, url, **kw):
+        m = MagicMock()
+        m.status_code = 200
+        m.json = lambda: fake_resp
+        return m
+
+    body = {
+        "device_id": device, "trace_id": trace, "step": 5,
+        "screen_state": {"screen": {"width": 1080, "height": 2340, "density": 3.0},
+                         "foreground": {"package": "x", "activity": ""}, "windows": []},
+    }
+
+    try:
+        with patch.object(httpx.AsyncClient, "post", fake_post):
+            r = client.post("/api/v1/coupon/step", json=body)
+        assert r.status_code == 200, r.text
+
+        db = SessionLocal()
+        try:
+            row = db.execute(
+                select(CouponSession).where(CouponSession.trace_id == trace)
+            ).scalar_one()
+            assert row.platform_success == ["meituan-waimai"]  # tb 失败不计入
+            claims = db.execute(
+                select(CouponClaimRecord).where(CouponClaimRecord.device_id == device)
+            ).scalars().all()
+            assert claims and all(c.app_env == "dev" for c in claims)  # session app_env 打标
+        finally:
+            db.close()
+    finally:
+        db = SessionLocal()
+        try:
+            db.execute(delete(CouponSession).where(CouponSession.trace_id == trace))
+            db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.device_id == device))
+            db.execute(delete(CouponDailyCompletion).where(CouponDailyCompletion.device_id == device))
+            db.commit()
+        finally:
+            db.close()
+
+
+def test_coupon_data_status_filter() -> None:
+    """状态多选过滤(方案 A):整个视图按选中状态算;None/空=全部。"""
+    db = SessionLocal()
+    try:
+        db.add_all([
+            _agg_session("st-A", ["meituan-waimai"], ["meituan-waimai"], status="completed"),
+            _agg_session("st-B", ["meituan-waimai"], ["meituan-waimai"], status="started"),
+            _agg_session("st-C", ["meituan-waimai"], None, status="failed"),
+            _agg_session("st-D", ["meituan-waimai"], ["meituan-waimai"], status="abandoned"),
+        ])
+        db.flush()
+        base = dict(date_from="2020-01-02", date_to="2020-01-02", app_env="prod")
+        # None = 全部 4 发起
+        assert coupon_data_report(db, **base)["summary"]["started_count"] == 4
+        # 排除 started → 3 发起(整个视图,发起数也随之变)
+        sub = coupon_data_report(db, **base, statuses=["completed", "failed", "abandoned"])["summary"]
+        assert sub["started_count"] == 3
+        assert sub["completed_count"] == 1
+        # 只 completed → 发起数=1、整单成功率基数=1(该 completed 是整单成功)
+        comp = coupon_data_report(db, **base, statuses=["completed"])["summary"]
+        assert comp["started_count"] == 1
+        assert comp["full_success_rate"] == 1.0
+    finally:
+        db.rollback()
+        db.close()

tests/test_coupon_slots.py

@@ -0,0 +1,137 @@
+"""每券成功率(§13):app_env 落库 + coupon_slot_report 聚合。"""
+from __future__ import annotations
+
+from datetime import UTC, date, datetime
+
+from sqlalchemy import delete, select
+
+from app.admin.repositories.coupon_data import coupon_slot_report
+from app.db.session import SessionLocal
+from app.models.coupon_state import CouponClaimRecord, CouponSession
+from app.repositories.coupon_state import record_claims, session_app_env
+
+
+def test_session_app_env_lookup() -> None:
+    db = SessionLocal()
+    trace = "slot-env-1"
+    try:
+        db.add(CouponSession(
+            trace_id=trace, device_id="d-slot", status="started", app_env="dev",
+            started_at=datetime(2020, 2, 1, tzinfo=UTC), started_date=date(2020, 2, 1),
+        ))
+        db.commit()
+        assert session_app_env(db, trace) == "dev"
+        assert session_app_env(db, "no-such-trace") is None
+        assert session_app_env(db, None) is None
+    finally:
+        db.execute(delete(CouponSession).where(CouponSession.trace_id == trace))
+        db.commit()
+        db.close()
+
+
+def test_record_claims_stamps_app_env() -> None:
+    db = SessionLocal()
+    dev = "d-slot-stamp"
+    try:
+        record_claims(db, dev, None, "t-stamp",
+                      [{"coupon_id": "mt_x", "status": "success", "name": "美团券"}],
+                      app_env="prod")
+        row = db.execute(select(CouponClaimRecord).where(
+            CouponClaimRecord.device_id == dev)).scalar_one()
+        assert row.app_env == "prod"
+        # UPDATE 路:app_env=None 不覆盖已有值
+        record_claims(db, dev, None, "t-stamp",
+                      [{"coupon_id": "mt_x", "status": "already_claimed", "name": "美团券"}],
+                      app_env=None)
+        db.expire_all()
+        row = db.execute(select(CouponClaimRecord).where(
+            CouponClaimRecord.device_id == dev)).scalar_one()
+        assert row.app_env == "prod"
+        assert row.status == "already_claimed"
+    finally:
+        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.device_id == dev))
+        db.commit()
+        db.close()
+
+
+def _claim(db, device, coupon_id, status, app_env, name=None, d=date(2020, 2, 2)):
+    db.add(CouponClaimRecord(
+        device_id=device, coupon_id=coupon_id, claim_date=d,
+        status=status, coupon_name=name, app_env=app_env,
+    ))
+
+
+def test_coupon_slot_report_aggregates() -> None:
+    """按 coupon_id 聚合:tried=成功+失败(skipped 排除)、rate、env 过滤、tried 倒序。"""
+    db = SessionLocal()
+    try:
+        # mt_a(prod):dev1 success + dev2 already_claimed + dev3 failed → 2/3;dev4 skipped 不计
+        _claim(db, "sdev1", "mt_a", "success", "prod", "美团外卖红包天天领")
+        _claim(db, "sdev2", "mt_a", "already_claimed", "prod", "美团外卖红包天天领")
+        _claim(db, "sdev3", "mt_a", "failed", "prod", "美团外卖红包天天领")
+        _claim(db, "sdev4", "mt_a", "skipped", "prod", "美团外卖红包天天领")
+        _claim(db, "sdev1", "tb_b", "success", "prod", "淘宝券")          # 1/1
+        _claim(db, "sdev9", "mt_a", "failed", "dev", "美团外卖红包天天领")  # 仅 dev/全部
+        db.commit()
+
+        prod = coupon_slot_report(
+            db, date_from="2020-02-02", date_to="2020-02-02", app_env="prod"
+        )["items"]
+        by_id = {r["coupon_id"]: r for r in prod}
+        assert by_id["mt_a"]["tried"] == 3            # skipped 排除
+        assert by_id["mt_a"]["succeeded"] == 2
+        assert by_id["mt_a"]["success_rate"] == round(2 / 3, 4)
+        assert by_id["mt_a"]["platform"] == "meituan-waimai"
+        assert by_id["mt_a"]["coupon_name"] == "美团外卖红包天天领"
+        assert by_id["tb_b"]["success_rate"] == 1.0
+        assert by_id["tb_b"]["platform"] == "taobao-shanguang"
+        assert [r["coupon_id"] for r in prod] == ["mt_a", "tb_b"]  # tried 倒序
+
+        dev = coupon_slot_report(
+            db, date_from="2020-02-02", date_to="2020-02-02", app_env="dev"
+        )["items"]
+        assert {r["coupon_id"]: r["tried"] for r in dev} == {"mt_a": 1}
+        allenv = coupon_slot_report(
+            db, date_from="2020-02-02", date_to="2020-02-02", app_env=None
+        )["items"]
+        assert {r["coupon_id"]: r["tried"] for r in allenv}["mt_a"] == 4
+    finally:
+        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.claim_date == date(2020, 2, 2)))
+        db.commit()
+        db.close()
+
+
+def test_coupon_slots_endpoint() -> None:
+    from fastapi.testclient import TestClient
+
+    from app.admin.main import admin_app
+    from app.admin.repositories import admin_user as admin_repo
+    from app.admin.security import create_admin_token
+
+    db = SessionLocal()
+    try:
+        _claim(db, "epdev1", "mt_ep", "success", "prod", "端点券", d=date(2020, 2, 3))
+        _claim(db, "epdev2", "mt_ep", "failed", "prod", "端点券", d=date(2020, 2, 3))
+        admin = admin_repo.create_admin(
+            db, username="slot_admin", password="pass1234", role="super_admin"
+        )
+        token, _exp = create_admin_token(admin_id=admin.id, role=admin.role)
+        db.commit()
+    finally:
+        db.close()
+    try:
+        c = TestClient(admin_app)
+        r = c.get(
+            "/admin/api/coupon-data/coupons",
+            params={"date_from": "2020-02-03", "date_to": "2020-02-03", "app_env": "prod"},
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        assert r.status_code == 200, r.text
+        row = next(x for x in r.json()["items"] if x["coupon_id"] == "mt_ep")
+        assert row["tried"] == 2 and row["succeeded"] == 1 and row["success_rate"] == 0.5
+        assert row["coupon_name"] == "端点券" and row["platform"] == "meituan-waimai"
+    finally:
+        db = SessionLocal()
+        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.claim_date == date(2020, 2, 3)))
+        db.commit()
+        db.close()

tests/test_inactivity_reset.py

@@ -0,0 +1,478 @@
+"""15 天不活跃清零:模型 / 活跃口径 / 清零 / 预警 / 配置 / worker。"""
+from __future__ import annotations
+
+from datetime import date, datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import delete, select, update
+
+from app.db.session import SessionLocal
+from app.models.inactivity import InactivityNotificationLog, InactivityResetLog
+from app.repositories import activity
+
+
+def test_reset_and_notification_models_persist() -> None:
+    db = SessionLocal()
+    try:
+        db.add(InactivityResetLog(
+            user_id=1, coin_balance_before=10, cash_balance_cents_before=20,
+            invite_cash_balance_cents_before=30,
+            last_active_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
+            inactive_days=15, reason="inactive_15d",
+        ))
+        db.add(InactivityNotificationLog(
+            user_id=1, stage=7, inactive_days=8, coin_balance=10,
+            cash_balance_cents=20, invite_cash_balance_cents=30,
+            channel="log", status="placeholder",
+        ))
+        db.commit()
+        r = db.execute(select(InactivityResetLog).where(InactivityResetLog.user_id == 1)).scalar_one()
+        assert r.reason == "inactive_15d" and r.reset_at is not None
+        n = db.execute(select(InactivityNotificationLog).where(InactivityNotificationLog.user_id == 1)).scalar_one()
+        assert n.stage == 7 and n.created_at is not None
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_reset_cutoff_is_cn_midnight_of_today_minus_days_minus_1() -> None:
+    # RESET_DAYS=15, today=1/20 → cutoff = 北京 00:00 of 1/6 = 1/5 16:00 UTC
+    cutoff = activity.reset_cutoff(15, today=date(2026, 1, 20))
+    assert cutoff == datetime(2026, 1, 5, 16, 0, tzinfo=timezone.utc)
+
+
+def test_active_event_constants() -> None:
+    # 首页可见 = event=show + page=home 组合,不在纯 event 名集合里
+    assert activity.HOME_VIEW_EVENT == "show" and activity.HOME_VIEW_PAGE == "home"
+    assert activity.HOME_VIEW_EVENT not in activity.ACTIVE_EVENTS
+    assert "real_compare_start" in activity.ACTIVE_EVENTS
+    assert "real_coupon_start" in activity.ACTIVE_EVENTS
+    assert activity.ACTIVE_ENGAGE_TYPE == "claim_started"
+
+
+def test_as_utc_normalizes() -> None:
+    assert activity.as_utc(datetime(2026, 1, 1)) == datetime(2026, 1, 1, tzinfo=timezone.utc)
+    cn = datetime(2026, 1, 1, tzinfo=activity.CN_TZ)  # 北京 0 点 = 前一天 16:00 UTC
+    assert activity.as_utc(cn) == datetime(2025, 12, 31, 16, 0, tzinfo=timezone.utc)
+
+
+from app.core.rewards import CN_TZ
+from app.models.analytics_event import AnalyticsEvent
+from app.models.coupon_state import CouponPromptEngagement
+from app.models.user import User
+from app.models.wallet import CoinAccount
+from app.repositories import wallet as wallet_repo
+
+_PHONE_SEQ = [0]
+
+
+@pytest.fixture(autouse=True)
+def _isolate_inactivity_state():
+    """本文件的测试都做全表扫描 + 全局计数,而 SQLite 测试库 session 级共享、无逐用例回滚
+    (commit 后的 rollback 是 no-op),故先把可能泄漏的余额清零 + 清掉活跃事件/审计行,
+    保证每个用例干净起步。不删 User(零余额用户不会被扫描选中,避免跨文件/外键影响)。"""
+    db = SessionLocal()
+    try:
+        db.execute(update(CoinAccount).values(
+            coin_balance=0, cash_balance_cents=0, invite_cash_balance_cents=0))
+        for model in (AnalyticsEvent, CouponPromptEngagement,
+                      InactivityResetLog, InactivityNotificationLog):
+            db.execute(delete(model))
+        db.commit()
+    finally:
+        db.close()
+    yield
+
+
+def _new_user(db, *, created_at, coin=0, cash=0, invite=0) -> int:
+    """直接建一个 User + CoinAccount,created_at 可控。返回 user_id。"""
+    _PHONE_SEQ[0] += 1
+    # 199 前缀 + 递增序号:共享测试库跨文件累积用户,别的文件用固定手机号(如 test_admin_write
+    # 的 13900000001..),这里用没人用的 199 段避免撞 user.phone / username 的 UNIQUE。
+    u = User(phone=f"199{_PHONE_SEQ[0]:08d}", created_at=created_at,
+             last_login_at=created_at, status="active",
+             username=f"inact{_PHONE_SEQ[0]}")
+    db.add(u)
+    db.flush()
+    acc = wallet_repo.get_or_create_account(db, u.id, commit=False)
+    acc.coin_balance, acc.cash_balance_cents, acc.invite_cash_balance_cents = coin, cash, invite
+    acc.total_coin_earned = coin
+    db.flush()
+    return u.id
+
+
+def _add_event(db, user_id, event, when: datetime, page=None) -> None:
+    db.add(AnalyticsEvent(event=event, device_id="d", user_id=user_id, client_ts=0,
+                          created_at=when, page=page))
+
+
+def _add_engage(db, user_id, when: datetime, engage_type="claim_started") -> None:
+    db.add(CouponPromptEngagement(device_id=f"dev{user_id}", package="p", user_id=user_id,
+                                  engage_date=when.date(), engage_type=engage_type, created_at=when))
+
+
+def test_last_active_expr_takes_max_of_baseline_and_events() -> None:
+    from sqlalchemy import select
+    db = SessionLocal()
+    try:
+        base = datetime(2026, 1, 1, tzinfo=timezone.utc)
+        uid = _new_user(db, created_at=base, coin=5)
+        _add_event(db, uid, "real_compare_start", datetime(2026, 1, 10, tzinfo=timezone.utc))
+        db.commit()
+        ev_sub, eng_sub = activity.last_active_subqueries(db)
+        dialect = db.get_bind().dialect.name
+        expr = activity.last_active_expr(User.created_at, ev_sub, eng_sub, dialect)
+        stmt = (select(expr).select_from(User)
+                .outerjoin(ev_sub, ev_sub.c.user_id == User.id)
+                .outerjoin(eng_sub, eng_sub.c.user_id == User.id)
+                .where(User.id == uid))
+        got = activity.norm_utc(db.execute(stmt).scalar_one())
+        assert got == datetime(2026, 1, 10, tzinfo=timezone.utc)  # 事件 > 基线
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_home_signal_uses_show_event_on_home_page() -> None:
+    """首页可见活跃口径 = event=show + page=home 组合;show 但非 home 页不算活跃。"""
+    from sqlalchemy import select
+    db = SessionLocal()
+    try:
+        base = datetime(2026, 1, 1, tzinfo=timezone.utc)
+        seen = _new_user(db, created_at=base)      # show/home → 活跃
+        _add_event(db, seen, "show", datetime(2026, 1, 10, tzinfo=timezone.utc), page="home")
+        other = _new_user(db, created_at=base)     # show/其他页 → 不算活跃
+        _add_event(db, other, "show", datetime(2026, 1, 10, tzinfo=timezone.utc), page="coupon")
+        db.commit()
+
+        ev_sub, eng_sub = activity.last_active_subqueries(db)
+        dialect = db.get_bind().dialect.name
+        expr = activity.last_active_expr(User.created_at, ev_sub, eng_sub, dialect)
+
+        def last_active(uid):
+            stmt = (select(expr).select_from(User)
+                    .outerjoin(ev_sub, ev_sub.c.user_id == User.id)
+                    .outerjoin(eng_sub, eng_sub.c.user_id == User.id)
+                    .where(User.id == uid))
+            return activity.norm_utc(db.execute(stmt).scalar_one())
+
+        assert last_active(seen) == datetime(2026, 1, 10, tzinfo=timezone.utc)  # show/home 算
+        assert last_active(other) == base                                       # show/其他页 不算
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_inactivity_warn_stages_parsing() -> None:
+    from app.core.config import Settings
+    s = Settings(INACTIVITY_WARN_DAYS_BEFORE="7,2", INACTIVITY_RESET_DAYS=15)
+    assert s.inactivity_warn_stages == [7, 2]        # 降序去重
+    s2 = Settings(INACTIVITY_WARN_DAYS_BEFORE="", INACTIVITY_RESET_DAYS=15)
+    assert s2.inactivity_warn_stages == []           # 空=不推
+    s3 = Settings(INACTIVITY_WARN_DAYS_BEFORE="2,20,7,2", INACTIVITY_RESET_DAYS=15)
+    assert s3.inactivity_warn_stages == [7, 2]        # 去重 + 丢弃 >=RESET_DAYS(20)
+
+
+def test_log_notifier_returns_placeholder(caplog) -> None:
+    from app.integrations.notifier import LogNotifier, get_notifier
+    n = get_notifier("log")
+    assert isinstance(n, LogNotifier) and n.channel == "log"
+    status = n.warn(user_id=1, coin=10, cash_cents=20, stage=7, days_until_reset=8)
+    assert status == "placeholder"
+    # 未实现通道回退 LogNotifier(占位)
+    assert get_notifier("jpush").channel == "log"
+
+
+def test_run_reset_clears_coin_and_cash_but_preserves_invite_cash() -> None:
+    from sqlalchemy import select
+    from app.models.wallet import CoinAccount, CoinTransaction, CashTransaction, InviteCashTransaction
+    from app.repositories import inactivity
+
+    db = SessionLocal()
+    try:
+        today = date(2026, 2, 1)
+        # 末次活跃 = created_at 基线 = 1/10(距 today 22 天 → 应清)
+        old = _new_user(db, created_at=datetime(2026, 1, 10, tzinfo=timezone.utc),
+                        coin=100, cash=200, invite=300)
+        # 活跃用户:昨天有 home_view → 不清
+        fresh = _new_user(db, created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), coin=50)
+        _add_event(db, fresh, "show", datetime(2026, 1, 31, tzinfo=timezone.utc), page="home")
+        db.commit()
+
+        stats = inactivity.run_reset_once(db, reset_days=15, today=today)
+        assert stats["cleared"] == 1 and stats["failed"] == 0
+
+        acc = db.get(CoinAccount, old)
+        # 金币 + 折算现金清零;邀请现金是产品红线,原封不动(见 wallet.CoinAccount 注释)
+        assert (acc.coin_balance, acc.cash_balance_cents) == (0, 0)
+        assert acc.invite_cash_balance_cents == 300
+        assert acc.total_coin_earned == 100  # 历史累计不动
+
+        log = db.execute(select(InactivityResetLog).where(InactivityResetLog.user_id == old)).scalar_one()
+        # 审计仍快照三桶余额(邀请现金记为"清零时仍保留"的余额,便于纠纷排查)
+        assert (log.coin_balance_before, log.cash_balance_cents_before,
+                log.invite_cash_balance_cents_before) == (100, 200, 300)
+        assert log.inactive_days == 22 and log.reason == "inactive_15d"
+
+        ct = db.execute(select(CoinTransaction).where(
+            CoinTransaction.user_id == old, CoinTransaction.biz_type == "inactivity_reset")).scalar_one()
+        assert ct.amount == -100 and ct.balance_after == 0 and ct.ref_id == str(log.id)
+        assert db.execute(select(CashTransaction).where(
+            CashTransaction.user_id == old, CashTransaction.biz_type == "inactivity_reset")).scalar_one().amount_cents == -200
+        # 关键:不写邀请现金流水(邀请现金不清)
+        assert db.execute(select(InviteCashTransaction).where(
+            InviteCashTransaction.user_id == old,
+            InviteCashTransaction.biz_type == "inactivity_reset")).first() is None
+
+        # 活跃用户不动;再跑一次幂等(coin+cash 已 0、邀请现金不算候选 → 不再匹配)
+        assert db.get(CoinAccount, fresh).coin_balance == 50
+        assert inactivity.run_reset_once(db, reset_days=15, today=today)["cleared"] == 0
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_user_with_only_invite_cash_is_not_cleared() -> None:
+    """只有邀请现金余额的久不活跃用户:邀请现金是产品红线,不清 → 根本不该被选中。"""
+    from app.models.wallet import CoinAccount
+    from app.repositories import inactivity
+
+    db = SessionLocal()
+    try:
+        today = date(2026, 2, 1)
+        uid = _new_user(db, created_at=datetime(2026, 1, 10, tzinfo=timezone.utc),
+                        coin=0, cash=0, invite=500)
+        db.commit()
+        stats = inactivity.run_reset_once(db, reset_days=15, today=today)
+        assert stats["cleared"] == 0
+        assert db.get(CoinAccount, uid).invite_cash_balance_cents == 500  # 原封不动
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_run_warn_picks_stage_and_dedups_within_streak() -> None:
+    from app.integrations.notifier import LogNotifier
+    from app.repositories import inactivity
+
+    db = SessionLocal()
+    try:
+        today = date(2026, 2, 1)
+        # 末次活跃 1/22(距 today 10 天)→ 档 7 命中(idays>=8),档 2 未到(需>=13)
+        uid = _new_user(db, created_at=datetime(2026, 1, 22, tzinfo=timezone.utc), coin=100)
+        db.commit()
+
+        stats = inactivity.run_warn_once(db, LogNotifier(), reset_days=15, warn_stages=[7, 2], today=today)
+        assert stats["warned"] == 1
+        from sqlalchemy import select
+        rows = db.execute(select(InactivityNotificationLog).where(
+            InactivityNotificationLog.user_id == uid)).scalars().all()
+        assert len(rows) == 1 and rows[0].stage == 7 and rows[0].status == "placeholder"
+        assert rows[0].inactive_days == 10 and rows[0].coin_balance == 100
+
+        # 同一 streak 再跑 → 不重推
+        assert inactivity.run_warn_once(db, LogNotifier(), reset_days=15, warn_stages=[7, 2], today=today)["warned"] == 0
+
+        # 无余额用户不预警
+        _new_user(db, created_at=datetime(2026, 1, 22, tzinfo=timezone.utc), coin=0)
+        db.commit()
+        assert inactivity.run_warn_once(db, LogNotifier(), reset_days=15, warn_stages=[7, 2], today=today)["warned"] == 0
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_run_once_warns_then_resets() -> None:
+    from app.integrations.notifier import LogNotifier
+    from app.models.wallet import CoinAccount
+    from app.repositories import inactivity
+
+    db = SessionLocal()
+    try:
+        today = date(2026, 2, 1)
+        warn_uid = _new_user(db, created_at=datetime(2026, 1, 22, tzinfo=timezone.utc), coin=10)   # 10天→预警
+        clear_uid = _new_user(db, created_at=datetime(2026, 1, 5, tzinfo=timezone.utc), coin=10)   # 27天→清零
+        db.commit()
+        stats = inactivity.run_once(db, notifier=LogNotifier(), reset_days=15, warn_stages=[7, 2], today=today)
+        assert stats["warned"] == 1 and stats["cleared"] == 1
+        assert db.get(CoinAccount, clear_uid).coin_balance == 0
+        assert db.get(CoinAccount, warn_uid).coin_balance == 10   # 预警不动钱
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_worker_run_once_entry_dry_run(monkeypatch) -> None:
+    """ENABLED=false(默认语义)→ worker 常驻但只记审计不清(dry_run = not ENABLED)。"""
+    from sqlalchemy import select
+
+    from app.core import inactivity_reset_worker as w
+    from app.core.config import settings
+    from app.models.wallet import CoinAccount
+
+    monkeypatch.setattr(settings, "INACTIVITY_RESET_ENABLED", False)  # false = 只记审计
+    monkeypatch.setattr(settings, "INACTIVITY_RESET_DAYS", 15)
+    monkeypatch.setattr(settings, "INACTIVITY_WARN_DAYS_BEFORE", "")
+    monkeypatch.setattr(w, "_cn_today", lambda: date(2026, 2, 1))
+
+    db = SessionLocal()
+    try:
+        uid = _new_user(db, created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), coin=100)
+        db.commit()
+    finally:
+        db.close()
+
+    w._run_once_entry()
+
+    db = SessionLocal()
+    try:
+        assert db.get(CoinAccount, uid).coin_balance == 100  # 没清
+        log = db.execute(select(InactivityResetLog).where(InactivityResetLog.user_id == uid)).scalar_one()
+        assert log.reason.endswith("dryrun")  # 记了审计
+    finally:
+        db.close()
+
+
+def test_worker_run_once_entry_executes(monkeypatch) -> None:
+    """_run_once_entry 用真实 SessionLocal 跑一轮,总闸开时能清掉一个不活跃用户。"""
+    from app.core import inactivity_reset_worker as w
+    from app.core.config import settings
+    from app.models.wallet import CoinAccount
+
+    monkeypatch.setattr(settings, "INACTIVITY_RESET_ENABLED", True)
+    monkeypatch.setattr(settings, "INACTIVITY_RESET_DAYS", 15)
+    monkeypatch.setattr(settings, "INACTIVITY_WARN_DAYS_BEFORE", "")  # 只测清零
+    # 固定"今天"避免依赖真实时钟
+    monkeypatch.setattr(w, "_cn_today", lambda: date(2026, 2, 1))
+
+    db = SessionLocal()
+    try:
+        uid = _new_user(db, created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), coin=100)
+        db.commit()
+    finally:
+        db.close()
+
+    stats = w._run_once_entry()
+    assert stats["cleared"] >= 1
+
+    db = SessionLocal()
+    try:
+        assert db.get(CoinAccount, uid).coin_balance == 0
+    finally:
+        db.close()
+
+
+def test_run_once_dry_run_records_audit_but_does_not_clear() -> None:
+    """dry-run:只写审计(标 dryrun)、不动钱、不预警;重复跑不重复记(streak dedup)。"""
+    from sqlalchemy import select
+
+    from app.integrations.notifier import LogNotifier
+    from app.models.wallet import CoinAccount, CoinTransaction
+    from app.repositories import inactivity
+
+    db = SessionLocal()
+    try:
+        today = date(2026, 2, 1)
+        old = _new_user(db, created_at=datetime(2026, 1, 10, tzinfo=timezone.utc), coin=100, cash=200, invite=300)
+        warn_uid = _new_user(db, created_at=datetime(2026, 1, 22, tzinfo=timezone.utc), coin=50)  # 预警窗
+        db.commit()
+
+        stats = inactivity.run_once(db, notifier=LogNotifier(), reset_days=15,
+                                    warn_stages=[7, 2], today=today, dry_run=True)
+        acc = db.get(CoinAccount, old)
+        assert (acc.coin_balance, acc.cash_balance_cents, acc.invite_cash_balance_cents) == (100, 200, 300)  # 原封
+        log = db.execute(select(InactivityResetLog).where(InactivityResetLog.user_id == old)).scalar_one()
+        assert log.coin_balance_before == 100 and log.reason.endswith("dryrun")   # 审计标 dryrun
+        assert db.execute(select(CoinTransaction).where(
+            CoinTransaction.user_id == old, CoinTransaction.biz_type == "inactivity_reset")).first() is None  # 无流水
+        assert stats["warned"] == 0                                              # dry-run 不预警
+        assert db.execute(select(InactivityNotificationLog).where(
+            InactivityNotificationLog.user_id == warn_uid)).first() is None
+        assert stats["cleared"] == 1                                             # dry-run:cleared=记了几条
+
+        # 再跑一次 → 不重复记(dedup),余额仍原封
+        inactivity.run_once(db, notifier=LogNotifier(), reset_days=15, warn_stages=[7, 2], today=today, dry_run=True)
+        assert len(db.execute(select(InactivityResetLog).where(
+            InactivityResetLog.user_id == old)).scalars().all()) == 1
+        assert db.get(CoinAccount, old).coin_balance == 100
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_admin_list_users_last_active_ignores_login() -> None:
+    """admin 用户列表 last_active_at 改用共享口径:登录不算活跃(baseline=created_at)、只认活跃事件。"""
+    from app.admin.repositories import queries
+
+    db = SessionLocal()
+    try:
+        created = datetime(2026, 1, 1, tzinfo=timezone.utc)
+        uid = _new_user(db, created_at=created)
+        u = db.get(User, uid)
+        u.last_login_at = datetime(2026, 6, 1, tzinfo=timezone.utc)  # 登录很新、但无任何活跃事件
+        db.commit()
+        phone = db.get(User, uid).phone
+        users, _cursor, _total = queries.list_users(db, phone=phone)
+        item = next(x for x in users if x.id == uid)
+        assert activity.norm_utc(item.last_active_at) == created  # 登录不算 → last_active=created_at
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_run_warn_isolates_notifier_failure_and_does_not_block_reset() -> None:
+    """单用户通知器抛错:预警计 warn_failed、不外抛,且清零(reset)照常执行。"""
+    from app.models.wallet import CoinAccount
+    from app.repositories import inactivity
+
+    class BoomNotifier:
+        channel = "log"
+
+        def warn(self, *, user_id, coin, cash_cents, stage, days_until_reset) -> str:
+            raise RuntimeError("push service down")
+
+    db = SessionLocal()
+    try:
+        today = date(2026, 2, 1)
+        warn_uid = _new_user(db, created_at=datetime(2026, 1, 22, tzinfo=timezone.utc), coin=10)   # 10天→预警
+        clear_uid = _new_user(db, created_at=datetime(2026, 1, 5, tzinfo=timezone.utc), coin=10)   # 27天→清零
+        db.commit()
+
+        stats = inactivity.run_once(db, notifier=BoomNotifier(), reset_days=15,
+                                    warn_stages=[7, 2], today=today)
+        assert stats["warned"] == 0 and stats["warn_failed"] >= 1   # 预警失败被隔离
+        assert stats["cleared"] == 1                                # 关键:清零没被阻塞
+        assert db.get(CoinAccount, clear_uid).coin_balance == 0
+        assert db.get(CoinAccount, warn_uid).coin_balance == 10     # 预警用户不动钱
+        # 预警失败已回滚,不留半条 notification_log
+        from sqlalchemy import select
+        assert db.execute(select(InactivityNotificationLog).where(
+            InactivityNotificationLog.user_id == warn_uid)).first() is None
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_run_once_reset_runs_even_if_warn_phase_throws(monkeypatch) -> None:
+    """预警整段异常(如候选查询失败)也绝不阻塞清零。"""
+    from app.integrations.notifier import LogNotifier
+    from app.models.wallet import CoinAccount
+    from app.repositories import inactivity
+
+    def boom(*a, **k):
+        raise RuntimeError("warn phase blew up")
+
+    monkeypatch.setattr(inactivity, "run_warn_once", boom)
+
+    db = SessionLocal()
+    try:
+        today = date(2026, 2, 1)
+        clear_uid = _new_user(db, created_at=datetime(2026, 1, 5, tzinfo=timezone.utc), coin=10)
+        db.commit()
+        stats = inactivity.run_once(db, notifier=LogNotifier(), reset_days=15,
+                                    warn_stages=[7, 2], today=today)
+        assert stats["cleared"] == 1
+        assert db.get(CoinAccount, clear_uid).coin_balance == 0
+    finally:
+        db.rollback()
+        db.close()

tests/test_invite_cash_withdraw.py

@@ -143,14 +143,15 @@ def test_two_accounts_withdraw_independent(client, monkeypatch) -> None:
     _reject(r1.json()["out_bill_no"])    # 退回 invite_cash + 结清活跃单
     r2 = client.post(
         "/api/v1/wallet/withdraw",
-        json={"amount_cents": 100, "source": "coin_cash"},
+        # 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位)
+        json={"amount_cents": 50, "source": "coin_cash"},
         headers=_auth(token),
     )
     assert r2.json()["status"] == "reviewing"
 
     cash, invite_cash = _balances(client, token)
     assert invite_cash == 500    # 已退回
-    assert cash == 300           # 扣了 cash 100
+    assert cash == 350           # 扣了 cash 50
 
 
 def test_invite_me_returns_reward_stats(client) -> None:
@@ -177,7 +178,8 @@ def test_withdraw_orders_source_filter(client, monkeypatch) -> None:
     _reject(r1.json()["out_bill_no"])    # 结清,才能提第二笔
     client.post(
         "/api/v1/wallet/withdraw",
-        json={"amount_cents": 100, "source": "coin_cash"},
+        # 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位)
+        json={"amount_cents": 50, "source": "coin_cash"},
         headers=_auth(token),
     )
 

tests/test_llm_cost.py

@@ -0,0 +1,184 @@
+"""LLM 调用成本计算 compute_llm_cost:按模型分桶累加 token × 单价;error/无 usage 跳过。"""
+from __future__ import annotations
+
+from app.services.llm_cost import compute_llm_cost
+
+_PRICE = {
+    "per_model": {"qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0}},
+    "default": {"input_per_1m": 3.0, "output_per_1m": 15.0},
+}
+
+
+def test_sums_per_model_single_model():
+    # 真实样本:4 次 qwen3.5-flash;Σprompt=6888、Σcompletion=337
+    calls = [
+        {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1512, "completion_tokens": 22}},
+        {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 2111, "completion_tokens": 160}},
+        {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1940, "completion_tokens": 142}},
+        {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1325, "completion_tokens": 13}},
+    ]
+    cost, snapshot = compute_llm_cost(calls, _PRICE)
+    # 6888/1e6*0.8 + 337/1e6*2.0 = 0.0055104 + 0.000674 = 0.0061844 → round(6)
+    assert cost == 0.006184
+    assert snapshot == {
+        "mode": "per_model",
+        "prices": {
+            "qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0, "_source": "per_model"},
+        },
+    }
+
+
+def test_multi_model_prices_each_bucket_separately():
+    calls = [
+        {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 0}},
+        {"model": "gpt-x", "error": None, "usage": {"prompt_tokens": 0, "completion_tokens": 1_000_000}},
+    ]
+    price = {
+        "per_model": {
+            "qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0},
+            "gpt-x": {"input_per_1m": 10.0, "output_per_1m": 30.0},
+        },
+        "default": {"input_per_1m": 3.0, "output_per_1m": 15.0},
+    }
+    cost, snap = compute_llm_cost(calls, price)
+    assert cost == 30.8  # qwen 1M入×0.8=0.8 + gpt-x 1M出×30=30.0
+    assert set(snap["prices"]) == {"qwen3.5-flash", "gpt-x"}
+
+
+def test_unknown_model_falls_back_to_default():
+    calls = [{"model": "mystery", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 0}}]
+    price = {"per_model": {}, "default": {"input_per_1m": 3.0, "output_per_1m": 15.0}}
+    cost, snap = compute_llm_cost(calls, price)
+    assert cost == 3.0
+    assert snap["prices"]["mystery"]["_source"] == "default"
+
+
+def test_unpriced_model_marked_and_zero_cost():
+    calls = [{"model": "mystery", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 999}}]
+    cost, snap = compute_llm_cost(calls, {"per_model": {}})  # 无 default
+    assert cost == 0.0
+    assert snap["prices"]["mystery"]["unpriced"] is True
+
+
+def test_error_and_missing_usage_calls_skipped():
+    calls = [
+        {"model": "qwen3.5-flash", "error": "boom", "usage": {"prompt_tokens": 9_999_999, "completion_tokens": 9_999_999}},
+        {"model": "qwen3.5-flash", "error": None, "usage": None},  # 无 usage
+        {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 0}},
+    ]
+    cost, _ = compute_llm_cost(calls, _PRICE)
+    assert cost == 0.8  # 只有第 3 条计入
+
+
+def test_empty_or_all_error_returns_none():
+    assert compute_llm_cost([], _PRICE) == (None, None)
+    assert compute_llm_cost(None, _PRICE) == (None, None)
+    all_error = [{"model": "x", "error": "boom", "usage": {"prompt_tokens": 100, "completion_tokens": 100}}]
+    assert compute_llm_cost(all_error, _PRICE) == (None, None)
+
+
+def test_malformed_price_entry_is_treated_as_unpriced_not_raised():
+    # 手改配置页可能存出残缺/非法单价(缺 output_per_1m、非 dict);不能抛异常连累 token 回填。
+    calls = [
+        {"model": "bad-a", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 5}},
+        {"model": "bad-b", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 5}},
+        {"model": "ok", "error": None, "usage": {"prompt_tokens": 1_000_000, "completion_tokens": 0}},
+    ]
+    price = {
+        "per_model": {
+            "bad-a": {"input_per_1m": 0.8},   # 缺 output_per_1m
+            "bad-b": 5,                        # 非 dict
+            "ok": {"input_per_1m": 3.0, "output_per_1m": 15.0},
+        },
+    }
+    cost, snap = compute_llm_cost(calls, price)  # 不得抛异常
+    assert cost == 3.0  # 只有 ok(1M 入 × 3.0)计入;两个残缺项按 unpriced
+    assert snap["prices"]["bad-a"].get("unpriced") is True
+    assert snap["prices"]["bad-b"].get("unpriced") is True
+
+
+def test_get_llm_prices_falls_back_to_default_then_uses_override():
+    from app.db.session import SessionLocal
+    from app.models.app_config import AppConfig
+    from app.repositories import app_config
+    from app.services.llm_cost import get_llm_prices
+
+    db = SessionLocal()
+    try:
+        # 无 override → CONFIG_DEFS 默认(含 per_model / default)
+        prices = get_llm_prices(db)
+        assert "per_model" in prices and "default" in prices
+        # 有 override → 用 DB 值
+        app_config.set_value(
+            db, "llm_token_price",
+            {"per_model": {"m": {"input_per_1m": 1.0, "output_per_1m": 2.0}},
+             "default": {"input_per_1m": 0.0, "output_per_1m": 0.0}},
+            admin_id=1,
+        )
+        assert get_llm_prices(db)["per_model"]["m"]["input_per_1m"] == 1.0
+    finally:
+        row = db.get(AppConfig, "llm_token_price")
+        if row is not None:
+            db.delete(row)
+            db.commit()
+        db.close()
+
+
+def test_backfill_llm_calls_stores_cost_and_snapshot(monkeypatch):
+    from datetime import UTC, datetime
+
+    from app.api.v1 import compare_record
+    from app.db.session import SessionLocal
+    from app.models.app_config import AppConfig
+    from app.models.comparison import ComparisonRecord
+    from app.repositories import app_config
+
+    sample = [
+        {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1512, "completion_tokens": 22}},
+        {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 2111, "completion_tokens": 160}},
+        {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1940, "completion_tokens": 142}},
+        {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1325, "completion_tokens": 13}},
+    ]
+    monkeypatch.setattr(compare_record, "fetch_llm_calls", lambda trace_id: sample)
+
+    db = SessionLocal()
+    try:
+        app_config.set_value(
+            db, "llm_token_price",
+            {"per_model": {"qwen3.5-flash": {"input_per_1m": 0.8, "output_per_1m": 2.0}},
+             "default": {"input_per_1m": 3.0, "output_per_1m": 15.0}},
+            admin_id=1,
+        )
+        rec = ComparisonRecord(
+            trace_id="llmcost-bf-1", status="success",
+            created_at=datetime.now(UTC).replace(tzinfo=None),
+        )
+        db.add(rec)
+        db.commit()
+        rid = rec.id
+    finally:
+        db.close()
+
+    compare_record._backfill_llm_calls(rid, "llmcost-bf-1")  # 独立 session 内回填
+
+    db = SessionLocal()
+    try:
+        rec = db.get(ComparisonRecord, rid)
+        assert rec.llm_cost_yuan == 0.006184
+        assert rec.llm_price_snapshot["prices"]["qwen3.5-flash"]["input_per_1m"] == 0.8
+        assert rec.input_tokens == 6888  # 现有 token 派生仍在
+    finally:
+        db.delete(db.get(ComparisonRecord, rid))
+        row = db.get(AppConfig, "llm_token_price")
+        if row is not None:
+            db.delete(row)
+        db.commit()
+        db.close()
+
+
+def test_admin_detail_schema_exposes_llm_cost_fields():
+    from app.admin.schemas.comparison import AdminComparisonDetail
+
+    fields = AdminComparisonDetail.model_fields
+    assert "llm_cost_yuan" in fields
+    assert "llm_price_snapshot" in fields

tests/test_ratelimit.py

@@ -0,0 +1,57 @@
+"""ratelimit 内存桶过期清理(GC)测试。
+
+回归重点:_buckets 是**全局共享**、混着不同窗口(60s 广告 / 3600s 登录 / 86400s 日闸)的 key。
+GC 必须按【每个 key 自己存的 window_sec】判过期,而不是当前调用方的窗口 —— 否则高频的 60s 端点
+触发 GC 时会把本该存活更久的 3600s/86400s 计数(如短信日闸)一并删掉,使其被反复清零、限流失效。
+用 monkeypatch 把 _GC_THRESHOLD 调 0 强制每次都扫,免造上万条(仿 test_auth 里对 sms._GC_THRESHOLD 的做法)。
+"""
+from __future__ import annotations
+
+from app.core import ratelimit
+
+
+def test_purge_expired_respects_each_key_own_window(monkeypatch) -> None:
+    """短窗口(60s)触发的 GC 只删真正过期的 key,不得删掉仍在自身窗口内的长窗口 key。"""
+    monkeypatch.setattr(ratelimit, "_GC_THRESHOLD", 0)  # 强制每次都扫
+    ratelimit._buckets.clear()
+
+    now = 1_000_000.0
+    # 日闸:100s 前开窗、window=86400 → 远未过期,必须保留
+    ratelimit._buckets["sms-send-device-daily:D:IP"] = (now - 100, 7, 86400.0)
+    # 登录:1800s、window=3600 → 未过期,保留
+    ratelimit._buckets["sms-login-device:D:IP"] = (now - 1800, 2, 3600.0)
+    # 广告:120s、window=60 → 已过期,应删
+    ratelimit._buckets["ad-watch-report:IP2"] = (now - 120, 3, 60.0)
+
+    ratelimit._purge_expired(now)
+
+    assert "sms-send-device-daily:D:IP" in ratelimit._buckets
+    assert "sms-login-device:D:IP" in ratelimit._buckets
+    assert "ad-watch-report:IP2" not in ratelimit._buckets
+
+
+def test_purge_expired_keeps_long_window_key_older_than_short_window(monkeypatch) -> None:
+    """反证旧 bug:日闸 key 已老于 3600s,旧代码在 60s/3600s 端点触发 GC 时会误删它;
+    现在按自身 86400s 窗口判 → 未过期 → 必须保留。"""
+    monkeypatch.setattr(ratelimit, "_GC_THRESHOLD", 0)
+    ratelimit._buckets.clear()
+
+    now = 2_000_000.0
+    # 3700s 前开窗(> 1 小时),但 window=86400 → 未过期
+    ratelimit._buckets["sms-send-device-daily:D:IP"] = (now - 3700, 20, 86400.0)
+
+    ratelimit._purge_expired(now)
+
+    assert "sms-send-device-daily:D:IP" in ratelimit._buckets
+
+
+def test_purge_expired_noop_below_threshold(monkeypatch) -> None:
+    """未超阈值时不扫(即便有过期 key 也不动),避免每次请求都 O(n) 扫全表。"""
+    monkeypatch.setattr(ratelimit, "_GC_THRESHOLD", 10)
+    ratelimit._buckets.clear()
+
+    now = 3_000_000.0
+    ratelimit._buckets["stale:IP"] = (now - 999, 1, 60.0)  # 早过期,但没超阈值
+    ratelimit._purge_expired(now)
+
+    assert "stale:IP" in ratelimit._buckets  # 桶数没超阈值 → 不清理

tests/test_wechat_conflict.py

@@ -0,0 +1,232 @@
+"""微信登录 M2 测试:conflict_ticket 令牌、继续绑定(attach/只登入)、换绑(建号+软删+30天限)。
+
+沿用 tests/test_wechat_login.py 风格:HTTP 走 client;微信 code→openid 用 monkeypatch;
+短信走 SMS_MOCK(任意 6 位过)。数据变更用"再走一遍 wechat-login 看 openid 落在哪个账号"做行为断言。
+"""
+from __future__ import annotations
+
+import pytest
+
+from app.api.v1 import auth  # noqa: F401  (后续测试打桩 verify_and_get_phone 用)
+from app.core import security
+from app.integrations import wxpay
+from app.models.phone_rebind_log import PhoneRebindLog
+
+
+def _fake_userinfo(openid: str, nickname: str | None = "微信昵称", avatar: str | None = "http://x/a.png"):
+    def _f(code: str) -> dict:
+        return {"openid": openid, "nickname": nickname, "avatar_url": avatar, "raw": {}}
+    return _f
+
+
+def _sms_occupy(client, phone: str) -> int:
+    """用普通短信登录占用一个手机号(register_channel=sms),返回该账号 id。"""
+    assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 200
+    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
+    assert r.status_code == 200, r.text
+    return r.json()["user"]["id"]
+
+
+def _occupy_via_conflict(client, monkeypatch, openid: str, phone: str, device_id: str) -> dict:
+    """微信登录(新 openid)→ 绑同一手机号 → 返回 phone_occupied 的响应体(含 conflict_ticket)。"""
+    monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo(openid))
+    ticket = client.post(
+        "/api/v1/auth/wechat-login", json={"code": "c", "device_id": device_id}
+    ).json()["bind_ticket"]
+    r = client.post(
+        "/api/v1/auth/wechat/bind-phone/sms",
+        json={"bind_ticket": ticket, "phone": phone, "code": "123456", "device_id": device_id},
+    )
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["status"] == "phone_occupied"
+    return body
+
+
+# ===== Task 1: 模型可导入(建表由 conftest 的 create_all 完成) =====
+
+def test_phone_rebind_log_model_importable() -> None:
+    assert PhoneRebindLog.__tablename__ == "phone_rebind_log"
+
+
+# ===== Task 2: conflict_ticket 令牌 =====
+
+def test_conflict_ticket_roundtrip() -> None:
+    token = security.create_conflict_ticket(
+        openid="oid1", wechat_nickname="昵", wechat_avatar_url="http://a", phone="13900139000"
+    )
+    claims = security.decode_conflict_ticket(token)
+    assert claims["openid"] == "oid1"
+    assert claims["wnk"] == "昵"
+    assert claims["wav"] == "http://a"
+    assert claims["phone"] == "13900139000"
+
+
+def test_conflict_ticket_wrong_type_rejected() -> None:
+    # bind_ticket 冒充 conflict_ticket → TokenError(typ 不匹配)
+    bind = security.create_bind_ticket(openid="oid", wechat_nickname=None, wechat_avatar_url=None)
+    with pytest.raises(security.TokenError):
+        security.decode_conflict_ticket(bind)
+
+
+def test_conflict_ticket_expired_rejected(monkeypatch) -> None:
+    monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
+    token = security.create_conflict_ticket(
+        openid="oid", wechat_nickname=None, wechat_avatar_url=None, phone="13900139000"
+    )
+    with pytest.raises(security.TokenError):
+        security.decode_conflict_ticket(token)
+
+
+# ===== Task 3: 占用响应扩展 =====
+
+def test_phone_occupied_returns_conflict_ticket_and_flags(client, monkeypatch) -> None:
+    phone = "13900139101"
+    _sms_occupy(client, phone)  # 老账号 X(sms,无微信)
+    body = _occupy_via_conflict(client, monkeypatch, "openid_occ_101", phone, "devO1")
+    assert body["conflict_ticket"]
+    assert body["rebind_available"] is True      # 首次,未换绑过
+    assert body["rebind_blocked_days"] == 0
+    assert body["occupied_account"]["has_wechat"] is False  # X 是 sms 账号
+
+
+# ===== Task 4: 继续绑定 =====
+
+def test_continue_attaches_wechat_and_logs_into_existing(client, monkeypatch) -> None:
+    """X 无微信 → 继续绑定并入 openid + 登入 X;之后同 openid 登录直接命中 X。"""
+    phone = "13900139201"
+    x_id = _sms_occupy(client, phone)  # X:sms 账号,无微信
+    body = _occupy_via_conflict(client, monkeypatch, "openid_cont_201", phone, "devC1")
+
+    r = client.post(
+        "/api/v1/auth/wechat/conflict/continue",
+        json={"conflict_ticket": body["conflict_ticket"], "device_id": "devC1"},
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["status"] == "logged_in"
+    assert r.json()["token"]["user"]["id"] == x_id  # 登入的是老账号 X
+
+    # openid 现已并入 X:再走 wechat-login 直接命中 X
+    r = client.post("/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devC1"})
+    assert r.json()["status"] == "logged_in"
+    assert r.json()["token"]["user"]["id"] == x_id
+
+
+def test_continue_when_existing_has_wechat_logs_in_and_discards_openid(client, monkeypatch) -> None:
+    """X 已绑别的微信 → 继续绑定只登入 X、丢弃本次 openid(不覆盖)。"""
+    phone = "13900139202"
+    # 先建一个已绑微信 O1 的账号 X(微信登录 O1 + 短信绑号)
+    monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_o1_202"))
+    t = client.post("/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devC2"}).json()["bind_ticket"]
+    x = client.post(
+        "/api/v1/auth/wechat/bind-phone/sms",
+        json={"bind_ticket": t, "phone": phone, "code": "123456", "device_id": "devC2"},
+    ).json()
+    x_id = x["token"]["user"]["id"]
+
+    # 新 openid O2 撞同号 → 占用(has_wechat=True)→ 继续绑定
+    body = _occupy_via_conflict(client, monkeypatch, "openid_o2_202", phone, "devC2b")
+    assert body["occupied_account"]["has_wechat"] is True
+    r = client.post(
+        "/api/v1/auth/wechat/conflict/continue",
+        json={"conflict_ticket": body["conflict_ticket"], "device_id": "devC2b"},
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["token"]["user"]["id"] == x_id  # 登入 X
+
+    # O2 被丢弃:再走 wechat-login(O2)→ 仍未命中(need_bind_phone)
+    monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_o2_202"))
+    assert client.post(
+        "/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devC2b"}
+    ).json()["status"] == "need_bind_phone"
+
+
+def test_continue_expired_ticket_returns_401(client, monkeypatch) -> None:
+    monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
+    expired = security.create_conflict_ticket(
+        openid="oid", wechat_nickname=None, wechat_avatar_url=None, phone="13900139209"
+    )
+    r = client.post(
+        "/api/v1/auth/wechat/conflict/continue",
+        json={"conflict_ticket": expired, "device_id": "devC3"},
+    )
+    assert r.status_code == 401, r.text
+
+
+# ===== Task 5: 换绑 =====
+
+def test_rebind_creates_new_account_and_binds_openid(client, monkeypatch) -> None:
+    """换绑 → 建全新微信账号 Y(≠X)+ openid 落到 Y;老账号 X 被注销(手机号归 Y)。"""
+    phone = "13900139301"
+    x_id = _sms_occupy(client, phone)
+    body = _occupy_via_conflict(client, monkeypatch, "openid_rb_301", phone, "devR1")
+
+    r = client.post(
+        "/api/v1/auth/wechat/conflict/rebind",
+        json={"conflict_ticket": body["conflict_ticket"], "device_id": "devR1"},
+    )
+    assert r.status_code == 200, r.text
+    y = r.json()["token"]["user"]
+    assert r.json()["status"] == "logged_in"
+    assert y["phone"] == phone
+    assert y["register_channel"] == "wechat"
+    assert y["id"] != x_id  # 是全新账号,不是老账号
+
+    # openid 落到 Y:再走 wechat-login 命中 Y
+    r = client.post("/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devR1"})
+    assert r.json()["status"] == "logged_in"
+    assert r.json()["token"]["user"]["id"] == y["id"]
+
+
+# ===== §10: continue 路径也应用展示身份回填规则 =====
+
+def test_continue_applies_section10(client, monkeypatch) -> None:
+    """§10 via M2 attach 路径: X 是默认昵称+null头像的 sms 账号;
+    continue 绑定微信后,展示昵称/头像应被微信值替换,并体现在响应的 token.user 中。"""
+    phone = "13900139211"
+    _sms_occupy(client, phone)  # 建 X:默认昵称, null avatar
+    body = _occupy_via_conflict(
+        client, monkeypatch, "openid_s10", phone, "devS10"
+    )  # fake_userinfo 默认 nickname="微信昵称", avatar="http://x/a.png"
+
+    r = client.post(
+        "/api/v1/auth/wechat/conflict/continue",
+        json={"conflict_ticket": body["conflict_ticket"], "device_id": "devS10"},
+    )
+    assert r.status_code == 200, r.text
+    user_out = r.json()["token"]["user"]
+    assert user_out["nickname"] == "微信昵称"
+    assert user_out["avatar_url"] == "http://x/a.png"
+
+
+def test_rebind_blocked_within_30_days(client, monkeypatch) -> None:
+    """同一手机号 30 天内二次换绑 → 409;占用响应 rebind_available=False。"""
+    phone = "13900139302"
+    _sms_occupy(client, phone)
+    body = _occupy_via_conflict(client, monkeypatch, "openid_rb_302a", phone, "devR2")
+    assert client.post(
+        "/api/v1/auth/wechat/conflict/rebind",
+        json={"conflict_ticket": body["conflict_ticket"], "device_id": "devR2"},
+    ).status_code == 200
+
+    # 第二次:新 openid 撞同号 → 占用响应此时 rebind_available=False
+    body2 = _occupy_via_conflict(client, monkeypatch, "openid_rb_302b", phone, "devR2b")
+    assert body2["rebind_available"] is False
+    assert body2["rebind_blocked_days"] >= 1
+    r = client.post(
+        "/api/v1/auth/wechat/conflict/rebind",
+        json={"conflict_ticket": body2["conflict_ticket"], "device_id": "devR2b"},
+    )
+    assert r.status_code == 409, r.text
+
+
+def test_rebind_expired_ticket_returns_401(client, monkeypatch) -> None:
+    monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
+    expired = security.create_conflict_ticket(
+        openid="oid", wechat_nickname=None, wechat_avatar_url=None, phone="13900139309"
+    )
+    r = client.post(
+        "/api/v1/auth/wechat/conflict/rebind",
+        json={"conflict_ticket": expired, "device_id": "devR3"},
+    )
+    assert r.status_code == 401, r.text

tests/test_wechat_login.py

@@ -0,0 +1,198 @@
+"""微信登录 M1 测试:bind_ticket 令牌、wechat-login(openid 命中/未命中)、
+bind-phone(建号/占用/令牌过期)。
+
+沿用 tests/test_auth.py 风格:HTTP 走 client fixture;微信 code→openid 用 monkeypatch
+拦掉(conftest 里 WECHAT_APP_ID/SECRET 是 dummy,不真连微信);短信走 SMS_MOCK(任意 6 位通过)。
+"""
+from __future__ import annotations
+
+import pytest
+
+from app.api.v1 import auth
+from app.core import security
+from app.integrations import wxpay
+
+
+def _fake_userinfo(openid: str, nickname: str | None = "微信昵称", avatar: str | None = "http://x/a.png"):
+    """返回一个可传给 monkeypatch 的假 code_to_userinfo(忽略 code,固定返回给定 openid)。"""
+    def _f(code: str) -> dict:
+        return {"openid": openid, "nickname": nickname, "avatar_url": avatar, "raw": {}}
+    return _f
+
+
+# ===== Task 1: bind_ticket 令牌 =====
+
+def test_bind_ticket_roundtrip() -> None:
+    token = security.create_bind_ticket(openid="oid1", wechat_nickname="昵", wechat_avatar_url="http://a")
+    claims = security.decode_bind_ticket(token)
+    assert claims["openid"] == "oid1"
+    assert claims["wnk"] == "昵"
+    assert claims["wav"] == "http://a"
+
+
+def test_bind_ticket_wrong_type_rejected() -> None:
+    # 用 access token 冒充 bind_ticket → TokenError(typ 不匹配)
+    access, _ = security.create_token(user_id=1, token_type="access")
+    with pytest.raises(security.TokenError):
+        security.decode_bind_ticket(access)
+
+
+def test_bind_ticket_expired_rejected(monkeypatch) -> None:
+    monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
+    token = security.create_bind_ticket(openid="oid", wechat_nickname=None, wechat_avatar_url=None)
+    with pytest.raises(security.TokenError):
+        security.decode_bind_ticket(token)
+
+
+# ===== Task 2: wechat-login =====
+
+def test_wechat_login_new_openid_returns_bind_ticket(client, monkeypatch) -> None:
+    monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_new_1", "小明", "http://x/m.png"))
+    r = client.post("/api/v1/auth/wechat-login", json={"code": "wxcode1", "device_id": "devA"})
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["status"] == "need_bind_phone"
+    assert body["bind_ticket"]
+    assert body["wechat_nickname"] == "小明"
+    assert body["wechat_avatar_url"] == "http://x/m.png"
+    assert body["token"] is None
+
+
+def test_wechat_login_invalid_code_returns_400(client, monkeypatch) -> None:
+    def _raise(code: str) -> dict:
+        raise ValueError("微信授权失败: invalid code")
+    monkeypatch.setattr(wxpay, "code_to_userinfo", _raise)
+    r = client.post("/api/v1/auth/wechat-login", json={"code": "bad", "device_id": "devA"})
+    assert r.status_code == 400, r.text
+
+
+# ===== Task 3: bind-phone/sms =====
+
+def test_wechat_bind_sms_creates_account_then_openid_logs_in(client, monkeypatch) -> None:
+    """未占用 → 建微信账号(channel=wechat,昵称头像取微信);再次同 openid 登录 → 直接登入同一账号。"""
+    monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_flow_2", "阿花", "http://x/h.png"))
+    phone = "13900139002"
+
+    # 1) 微信登录 → 未命中 → 拿 ticket
+    r = client.post("/api/v1/auth/wechat-login", json={"code": "c1", "device_id": "devB"})
+    ticket = r.json()["bind_ticket"]
+    assert ticket
+
+    # 2) 短信绑号(SMS_MOCK:任意 6 位通过)→ 建号 + 登入
+    r = client.post(
+        "/api/v1/auth/wechat/bind-phone/sms",
+        json={"bind_ticket": ticket, "phone": phone, "code": "123456", "device_id": "devB"},
+    )
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["status"] == "logged_in"
+    user = body["token"]["user"]
+    assert user["phone"] == phone
+    assert user["register_channel"] == "wechat"
+    assert user["nickname"] == "阿花"
+    assert user["avatar_url"] == "http://x/h.png"
+    uid = user["id"]
+
+    # 3) 再次微信登录(同 openid)→ 命中 → 直接登入同一账号
+    r = client.post("/api/v1/auth/wechat-login", json={"code": "c2", "device_id": "devB"})
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["status"] == "logged_in"
+    assert body["token"]["user"]["id"] == uid
+
+
+def test_wechat_bind_sms_phone_occupied(client, monkeypatch) -> None:
+    """手机号已被其他账号占用 → 返回 phone_occupied + 原账号信息(不建号)。"""
+    phone = "13900139003"
+    # 先用普通短信登录占用该手机号(register_channel=sms)
+    assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 200
+    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
+    assert r.status_code == 200, r.text
+    occupied_nickname = r.json()["user"]["nickname"]
+
+    # 微信登录(新 openid)→ 未命中 → ticket
+    monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_occ_3"))
+    ticket = client.post(
+        "/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devC"}
+    ).json()["bind_ticket"]
+
+    # 绑同一手机号 → 占用
+    r = client.post(
+        "/api/v1/auth/wechat/bind-phone/sms",
+        json={"bind_ticket": ticket, "phone": phone, "code": "123456", "device_id": "devC"},
+    )
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["status"] == "phone_occupied"
+    assert body["token"] is None
+    assert body["occupied_account"]["nickname"] == occupied_nickname
+    assert body["occupied_account"]["avatar_url"] is None  # 短信注册账号无头像 → 序列化为 null
+    assert body["occupied_account"]["created_at"]
+
+
+def test_wechat_bind_sms_expired_ticket_returns_401(client, monkeypatch) -> None:
+    """过期 bind_ticket → 401。"""
+    monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
+    expired = security.create_bind_ticket(openid="openid_exp", wechat_nickname="x", wechat_avatar_url=None)
+    r = client.post(
+        "/api/v1/auth/wechat/bind-phone/sms",
+        json={"bind_ticket": expired, "phone": "13900139009", "code": "123456", "device_id": "devD"},
+    )
+    assert r.status_code == 401, r.text
+
+
+# ===== Task 4: bind-phone/jverify =====
+
+def test_wechat_bind_jverify_creates_account(client, monkeypatch) -> None:
+    """本机号(极光)绑定路径:verify_and_get_phone 拦掉,未占用 → 建号登入。"""
+    monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_jv_5", "极光用户", None))
+    phone = "13900139005"
+    # 极光 loginToken→手机号 在 auth 模块命名空间打桩(auth.py 顶部 from ...jiguang import verify_and_get_phone)
+    monkeypatch.setattr(auth, "verify_and_get_phone", lambda token: phone)
+
+    ticket = client.post(
+        "/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devE"}
+    ).json()["bind_ticket"]
+
+    r = client.post(
+        "/api/v1/auth/wechat/bind-phone/jverify",
+        json={"bind_ticket": ticket, "login_token": "jgtoken", "device_id": "devE"},
+    )
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["status"] == "logged_in"
+    user = body["token"]["user"]
+    assert user["phone"] == phone
+    assert user["register_channel"] == "wechat"
+    assert user["nickname"] == "极光用户"
+    # 微信 userinfo 隐私脱敏 avatar=None → 头像为空(客户端兜底默认头像)
+    assert user["avatar_url"] is None
+
+
+def test_wechat_bind_jverify_expired_ticket_returns_401(client, monkeypatch) -> None:
+    """过期 bind_ticket → 401(极光绑号路径,decode 先于极光核验)。"""
+    monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
+    expired = security.create_bind_ticket(openid="openid_jv_exp", wechat_nickname="x", wechat_avatar_url=None)
+    r = client.post(
+        "/api/v1/auth/wechat/bind-phone/jverify",
+        json={"bind_ticket": expired, "login_token": "jgtoken", "device_id": "devE"},
+    )
+    assert r.status_code == 401, r.text
+
+
+def test_wechat_bind_jverify_jiguang_error_returns_502(client, monkeypatch) -> None:
+    """极光核验失败(JiguangError)→ 502。"""
+    monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_jv_err"))
+
+    def _raise(token: str) -> str:
+        raise auth.JiguangError("mock jg failure")
+
+    monkeypatch.setattr(auth, "verify_and_get_phone", _raise)
+    ticket = client.post(
+        "/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devE"}
+    ).json()["bind_ticket"]
+    r = client.post(
+        "/api/v1/auth/wechat/bind-phone/jverify",
+        json={"bind_ticket": ticket, "login_token": "badtoken", "device_id": "devE"},
+    )
+    assert r.status_code == 502, r.text

tests/test_withdraw.py

@@ -312,6 +312,94 @@ def test_bind_rejects_openid_already_bound(client, monkeypatch) -> None:
     assert r.status_code == 409, r.text
 
 
+# ===== §10 绑微信时展示身份回填规则 =====
+
+def test_bind_replaces_default_nickname_and_null_avatar(client, monkeypatch) -> None:
+    """§10-A: 全默认(昵称=系统默认,头像=null) → 绑微信后两个展示字段都替换为微信值。"""
+    monkeypatch.setattr(
+        "app.integrations.wxpay.code_to_userinfo",
+        lambda code: {"openid": "openid_s10_a", "nickname": "微信昵称A", "avatar_url": "https://x/a.png", "raw": {}},
+    )
+    token = _login(client, "13800003001")
+    r = client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+    assert r.status_code == 200, r.text
+
+    r = client.get("/api/v1/auth/me", headers=_auth(token))
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["nickname"] == "微信昵称A"
+    assert body["avatar_url"] == "https://x/a.png"
+
+
+def test_bind_keeps_customized_nickname(client, monkeypatch) -> None:
+    """§10-B: 用户已改过昵称 → 绑微信后昵称保留,但 null 头像仍替换为微信头像。"""
+    monkeypatch.setattr(
+        "app.integrations.wxpay.code_to_userinfo",
+        lambda code: {"openid": "openid_s10_b", "nickname": "微信昵称B", "avatar_url": "https://x/b.png", "raw": {}},
+    )
+    token = _login(client, "13800003002")
+    # 先把昵称改成非默认值
+    r = client.patch(
+        "/api/v1/user/profile", json={"nickname": "我的名字"}, headers=_auth(token)
+    )
+    assert r.status_code == 200, r.text
+
+    r = client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+    assert r.status_code == 200, r.text
+
+    r = client.get("/api/v1/auth/me", headers=_auth(token))
+    body = r.json()
+    assert body["nickname"] == "我的名字"       # 保留自定义昵称
+    assert body["avatar_url"] == "https://x/b.png"  # null 头像被微信头像替换
+
+
+def test_bind_keeps_customized_avatar(client, monkeypatch) -> None:
+    """§10-C: 已有自定义头像 → 绑微信后头像保留,但默认昵称替换为微信昵称。"""
+    monkeypatch.setattr(
+        "app.integrations.wxpay.code_to_userinfo",
+        lambda code: {"openid": "openid_s10_c", "nickname": "微信昵称C", "avatar_url": "https://x/c.png", "raw": {}},
+    )
+    token = _login(client, "13800003003")
+    phone = "13800003003"
+    # 直接用 DB 给用户写入自定义头像(保持默认昵称)
+    db = SessionLocal()
+    try:
+        user = db.execute(select(User).where(User.phone == phone)).scalar_one()
+        user.avatar_url = "https://custom/av.png"
+        db.commit()
+    finally:
+        db.close()
+
+    r = client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+    assert r.status_code == 200, r.text
+
+    r = client.get("/api/v1/auth/me", headers=_auth(token))
+    body = r.json()
+    assert body["nickname"] == "微信昵称C"           # 默认昵称被替换
+    assert body["avatar_url"] == "https://custom/av.png"  # 自定义头像保留
+
+
+def test_bind_wechat_empty_keeps_default(client, monkeypatch) -> None:
+    """§10-D: 微信侧 nickname/avatar 均为 None → 不覆盖,展示字段维持原默认值。"""
+    monkeypatch.setattr(
+        "app.integrations.wxpay.code_to_userinfo",
+        lambda code: {"openid": "openid_s10_d", "nickname": None, "avatar_url": None, "raw": {}},
+    )
+    token = _login(client, "13800003004")
+
+    r = client.get("/api/v1/auth/me", headers=_auth(token))
+    original_nickname = r.json()["nickname"]  # 系统分配的默认昵称
+    assert original_nickname.startswith("用户")
+
+    r = client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+    assert r.status_code == 200, r.text
+
+    r = client.get("/api/v1/auth/me", headers=_auth(token))
+    body = r.json()
+    assert body["nickname"] == original_nickname  # 默认昵称不变
+    assert body["avatar_url"] is None             # 头像仍为 null
+
+
 def test_withdraw_reject_refunds(client, monkeypatch) -> None:
     """管理员审核拒绝 → 退回现金 + 单 rejected + 理由写入 fail_reason(用户可见)。"""
     monkeypatch.setattr("app.integrations.wxpay.code_to_userinfo", lambda code: {"openid": "openid_reject", "nickname": None, "avatar_url": None, "raw": {}})

tests/test_withdraw_ledger_check.py

@@ -0,0 +1,151 @@
+"""提现现金账本校验(admin ledger-check)测试:两本物理隔离的账各自对账。
+
+历史盲区:`withdraw_ledger_check` 曾拿全部提现单去和**普通现金流水**(cash_transaction)比对,
+而 source=invite_cash 的提现单流水其实在 invite_cash_transaction 表,导致每笔邀请提现单都被
+误报「缺扣款/缺退款流水」。这里用真实提现 API 造单 + before/after 差值断言锁定修复:
+  1) 邀请提现单不再污染普通现金账的缺流水计数;
+  2) 邀请账户已被纳入对账(能抓到它自己的缺流水);
+  3) 普通现金账的原有对账未被改坏。
+
+conftest 的库是 session 级共享、测试间不清,故一律用 before/after 差值,只反映本用例造的数据。
+"""
+from __future__ import annotations
+
+from sqlalchemy import delete, select
+
+from app.admin.repositories.queries import withdraw_ledger_check
+from app.db.session import SessionLocal
+from app.models.user import User
+from app.models.wallet import CoinAccount, InviteCashTransaction
+from app.repositories import wallet as crud_wallet
+
+
+def _login(client, 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 _patch_userinfo(monkeypatch, openid: str) -> None:
+    monkeypatch.setattr(
+        "app.integrations.wxpay.code_to_userinfo",
+        lambda code: {"openid": openid, "nickname": None, "avatar_url": None, "raw": {}},
+    )
+
+
+def _seed_balances(client, token: str, phone: str, *, cash: int = 0, invite_cash: int = 0) -> None:
+    client.get("/api/v1/wallet/account", headers=_auth(token))
+    db = SessionLocal()
+    try:
+        user = db.execute(select(User).where(User.phone == phone)).scalar_one()
+        acc = db.get(CoinAccount, user.id)
+        acc.cash_balance_cents = cash
+        acc.invite_cash_balance_cents = invite_cash
+        db.commit()
+    finally:
+        db.close()
+
+
+def _reject(bill: str, reason: str = "测试拒绝") -> None:
+    db = SessionLocal()
+    try:
+        crud_wallet.reject_withdraw(db, bill, reason)
+    finally:
+        db.close()
+
+
+def _ledger() -> dict:
+    db = SessionLocal()
+    try:
+        return withdraw_ledger_check(db)
+    finally:
+        db.close()
+
+
+def test_rejected_invite_withdraw_not_flagged_missing(client, monkeypatch) -> None:
+    """核心回归:一笔被拒绝的 invite_cash 提现单,扣款/退款流水都在 invite_cash_transaction,
+    不应让普通现金账的缺扣款/缺退款计数增加(修复前每笔会各 +1)。"""
+    before = _ledger()
+
+    _patch_userinfo(monkeypatch, "openid_lc_1")
+    token = _login(client, "13800005001")
+    _seed_balances(client, token, "13800005001", cash=0, invite_cash=500)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+    r = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 200, "source": "invite_cash"},
+        headers=_auth(token),
+    )
+    assert r.status_code == 200, r.text
+    _reject(r.json()["out_bill_no"])  # rejected + 退款流水落 invite_cash_transaction
+
+    after = _ledger()
+
+    # 普通现金账不该因这笔 invite 单产生缺流水(修复前会各 +1 → 就是页面上看到的误报)
+    assert after["missing_withdraw_txn_count"] == before["missing_withdraw_txn_count"]
+    assert after["missing_refund_txn_count"] == before["missing_refund_txn_count"]
+    # 邀请账扣款 + 退款流水齐全,邀请账自身也不该缺
+    assert after["invite_missing_withdraw_txn_count"] == before["invite_missing_withdraw_txn_count"]
+    assert after["invite_missing_refund_txn_count"] == before["invite_missing_refund_txn_count"]
+
+
+def test_invite_ledger_detects_missing_withdraw_txn(client, monkeypatch) -> None:
+    """删掉一笔 invite 提现单的扣款流水 → 邀请账缺扣款计数 +1、ok=False,
+    证明邀请账户已真正纳入对账(修复前邀请账完全不校验、永远报不出问题)。"""
+    _patch_userinfo(monkeypatch, "openid_lc_2")
+    token = _login(client, "13800005002")
+    _seed_balances(client, token, "13800005002", cash=0, invite_cash=500)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+    r = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 200, "source": "invite_cash"},
+        headers=_auth(token),
+    )
+    bill = r.json()["out_bill_no"]
+
+    before = _ledger()
+    db = SessionLocal()
+    try:
+        db.execute(
+            delete(InviteCashTransaction).where(
+                InviteCashTransaction.ref_id == bill,
+                InviteCashTransaction.biz_type == "invite_withdraw",
+            )
+        )
+        db.commit()
+    finally:
+        db.close()
+    after = _ledger()
+
+    assert (
+        after["invite_missing_withdraw_txn_count"]
+        == before["invite_missing_withdraw_txn_count"] + 1
+    )
+    assert after["ok"] is False
+
+
+def test_coin_cash_withdraw_still_reconciled(client, monkeypatch) -> None:
+    """普通现金 coin_cash 提现单齐全时不新增缺流水(确保分账改造没弄坏原有普通现金对账)。"""
+    _patch_userinfo(monkeypatch, "openid_lc_3")
+    token = _login(client, "13800005003")
+    _seed_balances(client, token, "13800005003", cash=500, invite_cash=0)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    before = _ledger()
+    r = client.post(
+        "/api/v1/wallet/withdraw",
+        # 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位)
+        json={"amount_cents": 50, "source": "coin_cash"},
+        headers=_auth(token),
+    )
+    assert r.status_code == 200, r.text
+    after = _ledger()
+
+    # 普通现金提现扣款流水随单写入 cash_transaction,缺扣款计数不变;邀请账更不受影响
+    assert after["missing_withdraw_txn_count"] == before["missing_withdraw_txn_count"]
+    assert after["invite_missing_withdraw_txn_count"] == before["invite_missing_withdraw_txn_count"]

tests/test_withdraw_tiers.py

@@ -0,0 +1,186 @@
+"""福利页(coin_cash)提现档位规则测试(7-9提现ui对齐)。
+
+规则(2026-07-09 拍板):
+  - 档位 0.1/0.3(新人,历史一次性,免广告)+ 0.5(日3次)/10/20(日1次)。
+  - 计次口径"发起就算":当天创建的单不论最终状态(含被拒)都占名额;新人档任何状态都算用过。
+  - 常规三档每天只能选一个;新人档不参与该互斥,两个新人档同天可各提一次。
+  - invite_cash 无档位概念:tiers 为空、下单不走档位闸(邀请页行为不变)。
+wxpay 调用全部 monkeypatch;现金余额 DB 直灌(同 test_withdraw.py 套路)。
+"""
+from __future__ import annotations
+
+from sqlalchemy import select
+
+from app.db.session import SessionLocal
+from app.models.user import User
+from app.models.wallet import CoinAccount
+from app.repositories import wallet as crud_wallet
+
+
+def _login(client, 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 _seed_balances(client, token: str, phone: str, cash: int = 0, invite_cash: int = 0) -> None:
+    client.get("/api/v1/wallet/account", headers=_auth(token))
+    db = SessionLocal()
+    try:
+        user = db.execute(select(User).where(User.phone == phone)).scalar_one()
+        acc = db.get(CoinAccount, user.id)
+        acc.cash_balance_cents = cash
+        acc.invite_cash_balance_cents = invite_cash
+        db.commit()
+    finally:
+        db.close()
+
+
+def _patch_userinfo(monkeypatch, openid: str) -> None:
+    monkeypatch.setattr(
+        "app.integrations.wxpay.code_to_userinfo",
+        lambda code: {"openid": openid, "nickname": "昵称", "avatar_url": None, "raw": {}},
+    )
+
+
+def _reject(bill: str) -> None:
+    """管理员拒绝:退款结清活跃单(便于同用户继续发起下一笔;名额按'发起就算'仍占)。"""
+    db = SessionLocal()
+    try:
+        crud_wallet.reject_withdraw(db, bill, "test")
+    finally:
+        db.close()
+
+
+def _withdraw(client, token: str, cents: int):
+    return client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": cents, "source": "coin_cash"},
+        headers=_auth(token),
+    )
+
+
+def _tiers(client, token: str, source: str = "coin_cash") -> list[dict]:
+    r = client.get(
+        "/api/v1/wallet/withdraw-info", params={"source": source}, headers=_auth(token)
+    )
+    assert r.status_code == 200, r.text
+    return r.json()["tiers"]
+
+
+def test_withdraw_info_tiers_full_and_invite_empty(client, monkeypatch) -> None:
+    """新用户 coin_cash 下发 5 档(新人角标齐);invite_cash tiers 为空。"""
+    token = _login(client, "13800006001")
+    tiers = _tiers(client, token)
+    assert [t["amount_cents"] for t in tiers] == [10, 30, 50, 1000, 2000]
+    assert [t["label"] for t in tiers] == ["0.1", "0.3", "0.5", "10", "20"]
+    assert tiers[0]["badge"] == "新人福利" and tiers[0]["is_newbie"] is True
+    assert tiers[1]["badge"] == "新人福利" and tiers[1]["is_newbie"] is True
+    assert all(t["available"] for t in tiers)
+    assert tiers[2]["remaining_today"] == 3  # 0.5 日 3 次
+    assert _tiers(client, token, source="invite_cash") == []
+
+
+def test_newbie_tiers_independent_and_once_forever(client, monkeypatch) -> None:
+    """0.1 提过(即使被拒)→ 永久消失;同天 0.3 仍可提;新人档不锁常规档。"""
+    _patch_userinfo(monkeypatch, "openid_tier_2")
+    token = _login(client, "13800006002")
+    _seed_balances(client, token, "13800006002", cash=5000)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    r = _withdraw(client, token, 10)  # 0.1 新人档
+    assert r.status_code == 200, r.text
+    _reject(r.json()["out_bill_no"])  # 被拒也算用过("发起就算")
+
+    tiers = _tiers(client, token)
+    amounts = [t["amount_cents"] for t in tiers]
+    assert 10 not in amounts          # 0.1 消失
+    assert 30 in amounts              # 0.3 还在,同天仍可提
+    # 新人档不参与"选一个额度":常规三档全部仍可提
+    regular = {t["amount_cents"]: t for t in tiers if not t["is_newbie"]}
+    assert all(regular[a]["available"] for a in (50, 1000, 2000))
+
+    r = _withdraw(client, token, 30)  # 同天 0.3 照提
+    assert r.status_code == 200, r.text
+    _reject(r.json()["out_bill_no"])
+
+    # 0.1 已用过,重提 → 非法金额(档位已消失)
+    r = _withdraw(client, token, 10)
+    assert r.status_code == 400, r.text
+
+
+def test_regular_daily_select_one_tier(client, monkeypatch) -> None:
+    """当天提过 0.5 → 10/20 置灰 other_tier_selected,下单 409;0.5 还能继续提(3 次内)。"""
+    _patch_userinfo(monkeypatch, "openid_tier_3")
+    token = _login(client, "13800006003")
+    _seed_balances(client, token, "13800006003", cash=10_000)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    r = _withdraw(client, token, 50)
+    assert r.status_code == 200, r.text
+    _reject(r.json()["out_bill_no"])  # 结清活跃单;当天名额仍占("发起就算")
+
+    tiers = {t["amount_cents"]: t for t in _tiers(client, token)}
+    assert tiers[50]["available"] and tiers[50]["remaining_today"] == 2
+    assert not tiers[1000]["available"] and tiers[1000]["disabled_reason"] == "other_tier_selected"
+    assert not tiers[2000]["available"] and tiers[2000]["disabled_reason"] == "other_tier_selected"
+
+    r = _withdraw(client, token, 1000)  # 选一额度互斥 → 409
+    assert r.status_code == 409, r.text
+    assert "今日额度已达上限" in r.json()["detail"]
+
+    r = _withdraw(client, token, 50)  # 0.5 第 2 次照常
+    assert r.status_code == 200, r.text
+
+
+def test_regular_daily_quota_exhausted(client, monkeypatch) -> None:
+    """0.5 日 3 次:第 4 次 409;tiers 显示 quota_exhausted。"""
+    _patch_userinfo(monkeypatch, "openid_tier_4")
+    token = _login(client, "13800006004")
+    _seed_balances(client, token, "13800006004", cash=10_000)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    for _ in range(3):
+        r = _withdraw(client, token, 50)
+        assert r.status_code == 200, r.text
+        _reject(r.json()["out_bill_no"])
+
+    tiers = {t["amount_cents"]: t for t in _tiers(client, token)}
+    assert not tiers[50]["available"]
+    assert tiers[50]["disabled_reason"] == "quota_exhausted"
+    assert tiers[50]["remaining_today"] == 0
+
+    r = _withdraw(client, token, 50)
+    assert r.status_code == 409, r.text
+
+
+def test_non_tier_amount_rejected_for_coin_cash(client, monkeypatch) -> None:
+    """coin_cash 只能提预设档位:任意其他金额 400(防绕过客户端刷)。"""
+    _patch_userinfo(monkeypatch, "openid_tier_5")
+    token = _login(client, "13800006005")
+    _seed_balances(client, token, "13800006005", cash=10_000)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    r = _withdraw(client, token, 123)
+    assert r.status_code == 400, r.text
+
+
+def test_invite_cash_not_gated_by_tiers(client, monkeypatch) -> None:
+    """invite_cash 不走档位闸:非档位金额(200=2元)照常下单——邀请页行为不变。"""
+    _patch_userinfo(monkeypatch, "openid_tier_6")
+    token = _login(client, "13800006006")
+    _seed_balances(client, token, "13800006006", invite_cash=500)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    r = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 200, "source": "invite_cash"},
+        headers=_auth(token),
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["status"] == "reviewing"