guke
120 lines
5.0 KiB
Python
120 lines
5.0 KiB
Python
"""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)
|