guke
40c99afd92
- 清零改两桶(金币 + 折算现金);邀请奖励金物理隔离、不清(产品红线),仅快照入审计 - 预警编排 select_warn_candidates / run_warn_once / run_once(分档 + streak 去重); notifier.warn 去掉 invite 参数(邀请金不清、不预警) - 预警逐用户 try/except 隔离 + run_once 保证预警故障绝不阻塞清零(接真推送前必备) - 每日 worker inactivity_reset_worker(文件锁 + 北京日守卫 + RUN_HOUR 门槛 + 总闸默认关) + main.py lifespan 接线 - admin 最近活跃口径改用共享 activity(移除 last_login_at、纳入 home_view 占位) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
384 lines
17 KiB
Python
384 lines
17 KiB
Python
"""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
|
|
# 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) -> 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, 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, "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)
|
|
# 金币 + 折算现金清零;邀请现金是产品红线,原封不动(见 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_disabled_returns_none(monkeypatch) -> None:
|
|
from app.core import inactivity_reset_worker as w
|
|
from app.core.config import settings
|
|
monkeypatch.setattr(settings, "INACTIVITY_RESET_ENABLED", False)
|
|
assert w.start_inactivity_reset_worker() is None
|
|
|
|
|
|
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_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()
|