yunke/yunke-server/tests/test_summary_daily_stats.py

"""Tests for the msg_daily_stats 汇总层(日粒度会话统计)。

三个被钉死的行为:

1. **日桶口径**:day 一律 Asia/Shanghai 自然日(Python 计算,不用 SQL 日期
   函数——旧实现 from_unixtime 跟随 MySQL 会话时区,UTC 容器下和今日数
   的上海口径不一致,这是顺带修复的旧病)。
2. **写路径钩子**::func:`app.storage.repos.upsert_messages` 落库后自动
   重算本批触及的「会话×天」行——页面抓取器/CLI 手动同步全部路径共用此
   入口,一处挂全。幂等:同一批重放不产生重复/漂移。
3. **全量重算**::func:`app.storage.summary.recompute_all` 从 message 表
   重建全部日行(夜间自愈/部署回填用),幂等。

fixtures 用内存 SQLite(测试对 message 的所有写入都走 upsert_messages,
从而同时覆盖钩子行为)。
"""

from __future__ import annotations

from datetime import datetime, timezone
from zoneinfo import ZoneInfo

import pytest
from sqlalchemy import 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 Message, MsgDailyStats
from app.storage.repos import upsert_messages
from app.storage.summary import day_key, recompute_all

_SH = ZoneInfo("Asia/Shanghai")


def _sh(y, m, d, h, mi=0, s=0) -> int:
    return int(
        datetime(y, m, d, h, mi, s, tzinfo=_SH).timestamp() * 1000
    )


def _utc(y, m, d, h, mi=0, s=0) -> int:
    return int(
        datetime(y, m, d, h, mi, s, tzinfo=timezone.utc).timestamp() * 1000
    )


@pytest.fixture
async def 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)
    async with maker() as s:
        yield s
    await engine.dispose()


# ── day_key:上海自然日 ──────────────────────────────────────────────────── #


def test_day_key_shanghai_calendar_day():
    # 上海 2026-09-03 09:30 → 20260903
    assert day_key(_sh(2026, 9, 3, 9, 30)) == 20260903


def test_day_key_uses_shanghai_not_utc():
    # UTC 2026-09-02 16:30 = 上海 2026-09-03 00:30 → 必须归入 0903(旧
    # from_unixtime 在 UTC 容器里会归入 0902——口径修复点)
    assert day_key(_utc(2026, 9, 2, 16, 30)) == 20260903


def test_day_key_midnight_boundary():
    # 上海零点整属于新的一天;前一天 23:59:59 属于前一天
    assert day_key(_sh(2026, 9, 3, 0, 0, 0)) == 20260903
    assert day_key(_sh(2026, 9, 2, 23, 59, 59)) == 20260902


# ── 写路径钩子:upsert_messages 自动维护日行 ────────────────────────────── #


def _msg(msg_id, ts, *, owner="wx_a", talker="t1", room=None, is_send=True):
    return dict(
        msg_id=msg_id,
        channel="wechat",
        owner_account_id=owner,
        talker_id=talker if room is None else "room_member",
        room_id=room,
        is_send=is_send,
        timestamp=ts,
        type="text",
        type_raw=1,
        text="x",
    )


async def _stats_rows(session):
    rows = (await session.execute(select(MsgDailyStats))).scalars().all()
    return {
        (r.peer_kind, r.peer_id, r.day): (r.send_cnt, r.recv_cnt, r.last_ts, r.last_is_send)
        for r in rows
    }


async def test_upsert_messages_maintains_daily_rows(session):
    await upsert_messages(
        session,
        [
            _msg("m1", _sh(2026, 9, 3, 10, 0), is_send=True),
            _msg("m2", _sh(2026, 9, 3, 11, 0), is_send=False),
            _msg("m3", _sh(2026, 9, 3, 11, 30), is_send=True),
        ],
    )
    await session.commit()
    rows = await _stats_rows(session)
    assert set(rows) == {("friend", "t1", 20260903)}
    assert rows[("friend", "t1", 20260903)] == (2, 1, _sh(2026, 9, 3, 11, 30), True)


async def test_upsert_messages_splits_days_and_kinds(session):
    await upsert_messages(
        session,
        [
            _msg("d1", _sh(2026, 9, 2, 23, 59), talker="t1"),
            _msg("d2", _sh(2026, 9, 3, 0, 1), talker="t1"),
            _msg("g1", _sh(2026, 9, 3, 9, 0), room="room1@chatroom", is_send=None),
        ],
    )
    await session.commit()
    rows = await _stats_rows(session)
    # 跨上海零点的两条消息分属两天;群消息单独一行(room 维度)
    assert set(rows) == {
        ("friend", "t1", 20260902),
        ("friend", "t1", 20260903),
        ("room", "room1@chatroom", 20260903),
    }
    # is_send=NULL 的消息:两个方向都不计(对齐旧聚合 case 语义),但占用行
    assert rows[("room", "room1@chatroom", 20260903)] == (0, 0, _sh(2026, 9, 3, 9, 0), None)


async def test_upsert_messages_idempotent(session):
    batch = [
        _msg("m1", _sh(2026, 9, 3, 10, 0)),
        _msg("m2", _sh(2026, 9, 3, 11, 0), is_send=False),
    ]
    await upsert_messages(session, batch)
    await session.commit()
    await upsert_messages(session, batch)  # 同批重放(跨页重复是常态)
    await session.commit()
    rows = await _stats_rows(session)
    assert len(rows) == 1
    assert rows[("friend", "t1", 20260903)] == (1, 1, _sh(2026, 9, 3, 11, 0), False)


async def test_upsert_messages_null_timestamp_skipped(session):
    await upsert_messages(session, [_msg("n1", None)])
    await session.commit()
    assert await _stats_rows(session) == {}


# ── 全量重算(夜间自愈/部署回填)────────────────────────────────────────── #


async def test_recompute_all_rebuilds_and_is_idempotent(session):
    # 先写一批消息(顺带产生日行),再篡改一条日行模拟漂移,全量重算应修复
    await upsert_messages(
        session,
        [_msg("m1", _sh(2026, 9, 3, 10, 0)), _msg("m2", _sh(2026, 9, 3, 11, 0), is_send=False)],
    )
    await session.commit()
    bad = (await session.execute(select(MsgDailyStats))).scalars().one()
    bad.send_cnt = 999  # 漂移
    await session.commit()

    n = await recompute_all(session)
    assert n >= 1
    rows = await _stats_rows(session)
    assert rows[("friend", "t1", 20260903)] == (1, 1, _sh(2026, 9, 3, 11, 0), False)

    await recompute_all(session)  # 幂等(重建后是全新 ORM 对象,比字段值不比对象)
    assert await _stats_rows(session) == rows