shaguabijia-app-server/tests/test_inactivity_reset.py

"""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:
    assert activity.HOME_VIEW_EVENT 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
    u = User(phone=f"139{_PHONE_SEQ[0]:08d}", created_at=created_at,
             last_login_at=created_at, status="active",
             username=f"2{_PHONE_SEQ[0]:010d}")
    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) -> None:
    db.add(AnalyticsEvent(event=event, device_id="d", user_id=user_id, client_ts=0, created_at=when))


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_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, invite_cash_cents=30, stage=7, days_until_reset=8)
    assert status == "placeholder"
    # 未实现通道回退 LogNotifier(占位)
    assert get_notifier("jpush").channel == "log"


def test_run_reset_clears_three_buckets_and_writes_audit_and_flows() -> 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, "home_view", datetime(2026, 1, 31, tzinfo=timezone.utc))
        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)
        assert (acc.coin_balance, acc.cash_balance_cents, acc.invite_cash_balance_cents) == (0, 0, 0)
        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")).scalar_one().amount_cents == -300

        # 活跃用户不动;再跑一次幂等(已清零 → 不再匹配)
        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()