"""_analytics_tick:滚动刷新 + 增量扫描 + 管理域滚动段 + 异常吞掉(v1 T11;批次F F1)。 行为测试是底线——start() 的 job 注册断言在 test_sync_scheduler.py 的 注册测试里同步更新(那里 spy 断言 add_job 全集)。 """ from __future__ import annotations import logging import time import pytest from sqlalchemy import func, select from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from sqlalchemy.pool import StaticPool from app.db.base import Base from app.db.models import MsgDailyStats, SensitiveWord, StatDailyStaff from app.storage.staff_summary import rolling_days from app.storage.summary import day_key from app.sync.scheduler import SyncScheduler @pytest.fixture async def sched_engine(): # StaticPool:内存库单连接共享,同 engine 的多 session 看到同一份数据 engine = create_async_engine( "sqlite+aiosqlite:///:memory:", poolclass=StaticPool, connect_args={"check_same_thread": False}, ) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) maker = async_sessionmaker(engine, expire_on_commit=False) yield engine, maker await engine.dispose() async def test_analytics_tick_runs(sched_engine): """滚动 3 天窗覆盖今日 seed → stat_daily_staff 出行;tick 不抛。 _analytics_tick 用真实「今天」算滚动窗(time.time() 喂 day_key,上海 自然日)——seed 的 day 必须相对当日,否则窗口错过。 """ engine, maker = sched_engine today = day_key(int(time.time() * 1000)) async with maker() as s: s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", peer_kind="friend", peer_id="t1", day=today, send_cnt=1, recv_cnt=0, last_ts=1756950000000, last_is_send=True)) s.add(SensitiveWord(word="私单")) await s.commit() sch = SyncScheduler(engine=engine, session_maker=maker) await sch._analytics_tick() # 不抛即过 async with maker() as s: n = (await s.execute( select(func.count()).select_from(StatDailyStaff) )).scalar_one() assert n >= 1 async def test_analytics_tick_swallows_errors(sched_engine, monkeypatch, caplog): """内部异常只记日志不抛(保住下一个调度周期)。 直接让 recompute_days 抛错——_analytics_tick 内部的 ``from app.storage.staff_summary import recompute_days`` 在调用时求值, monkeypatch 模块属性即可命中。 """ engine, maker = sched_engine async def _boom(*args, **kwargs): raise RuntimeError("boom") monkeypatch.setattr("app.storage.staff_summary.recompute_days", _boom) sch = SyncScheduler(engine=engine, session_maker=maker) with caplog.at_level(logging.ERROR, logger="app.sync.scheduler"): await sch._analytics_tick() # 不抛 assert any("analytics tick failed" in r.getMessage() for r in caplog.records) async def _seed_today(sched_engine) -> int: """seed 当日一条 wechat friend 会话 + 敏感词,返回 seed 用的 day 键。""" _, maker = sched_engine today = day_key(int(time.time() * 1000)) async with maker() as s: s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1", peer_kind="friend", peer_id="t1", day=today, send_cnt=1, recv_cnt=0, last_ts=1756950000000, last_is_send=True)) s.add(SensitiveWord(word="私单")) await s.commit() return today async def test_tick_calls_management_chain_in_order(sched_engine, monkeypatch): """两段链完整顺序:recompute_days → scan → pairs_active_in_days → recompute_pairs → recompute_metric_days → recompute_call_days。 记录桩替换全部六个上游——_analytics_tick 段内函数延迟 import, monkeypatch 模块属性即命中(惯例见 test_analytics_tick_swallows_errors)。 """ engine, maker = sched_engine calls: list[str] = [] def rec(name: str, ret): async def _stub(*args, **kwargs): calls.append(name) return ret return _stub monkeypatch.setattr("app.storage.staff_summary.recompute_days", rec("recompute_days", 0)) monkeypatch.setattr("app.analytics.sensitive.scanner.run_incremental", rec("scan", {"scanned": 0, "hits": 0})) monkeypatch.setattr("app.storage.first_reply_facts.pairs_active_in_days", rec("pairs_active_in_days", [])) monkeypatch.setattr("app.storage.first_reply_facts.recompute_pairs", rec("recompute_pairs", 0)) monkeypatch.setattr("app.storage.employee_metrics.recompute_metric_days", rec("recompute_metric_days", 0)) monkeypatch.setattr("app.storage.employee_metrics.recompute_call_days", rec("recompute_call_days", 0)) sch = SyncScheduler(engine=engine, session_maker=maker) await sch._analytics_tick() assert calls == [ "recompute_days", "scan", # 段1 "pairs_active_in_days", "recompute_pairs", # 段2 "recompute_metric_days", "recompute_call_days", ] async def test_tick_management_failure_isolated(sched_engine, monkeypatch, caplog): """段2 失败隔离:recompute_metric_days 抛 RuntimeError → ①tick 不抛; ②记 "management metrics tick failed";③段1 已提交的 StatDailyStaff 不被段2 回滚(两段独立事务)。 """ engine, maker = sched_engine await _seed_today(sched_engine) async def _boom(*args, **kwargs): raise RuntimeError("boom") monkeypatch.setattr("app.storage.employee_metrics.recompute_metric_days", _boom) sch = SyncScheduler(engine=engine, session_maker=maker) with caplog.at_level(logging.ERROR, logger="app.sync.scheduler"): await sch._analytics_tick() # ①不抛 assert any( # ②段2 独立记日志 "management metrics tick failed" in r.getMessage() for r in caplog.records ) async with maker() as s: # ③段1 结果已提交 n = (await s.execute( select(func.count()).select_from(StatDailyStaff) )).scalar_one() assert n >= 1 async def test_tick_management_rolling_window(sched_engine, monkeypatch): """管理域三函数收到的 days == rolling_days(today, 3)(长度 3,当日+前 两天),且与 staff 段捕获的窗口一致(两段共用函数顶提取的同一 today)。 """ engine, maker = sched_engine got: dict[str, list[int]] = {} def cap(name: str, ret): async def _stub(session, days, *args, **kwargs): got[name] = list(days) return ret return _stub monkeypatch.setattr("app.storage.staff_summary.recompute_days", cap("staff", 0)) monkeypatch.setattr("app.storage.first_reply_facts.pairs_active_in_days", cap("pairs", [])) monkeypatch.setattr("app.storage.employee_metrics.recompute_metric_days", cap("wide", 0)) monkeypatch.setattr("app.storage.employee_metrics.recompute_call_days", cap("call", 0)) sch = SyncScheduler(engine=engine, session_maker=maker) await sch._analytics_tick() rd = rolling_days(day_key(int(time.time() * 1000)), 3) assert len(rd) == 3 # 当日 + 前两天 assert got["staff"] == got["pairs"] == got["wide"] == got["call"] == rd