shaguabijia-app-server/tests/test_inactivity_reset.py

"""15 天不活跃清零:模型 / 活跃口径 / 清零 / 预警 / 配置 / worker。"""
from __future__ import annotations

from datetime import date, datetime, timedelta, timezone

from sqlalchemy import select

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]


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, update
    from app.models.wallet import CoinAccount, CoinTransaction, CashTransaction, InviteCashTransaction
    from app.repositories import inactivity

    db = SessionLocal()
    try:
        # 清掉前序测试遗留的余额(session-scoped DB,前测 commit 后 rollback 是 no-op)
        db.execute(update(CoinAccount).values(coin_balance=0, cash_balance_cents=0,
                                              invite_cash_balance_cents=0))
        db.commit()

        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()