yunke/yunke-server/tests/test_sync_scheduler.py

"""Tests for app.sync.scheduler — periodic-sync wrapper (M6 Part A).

These tests do NOT hit the network and do NOT start a real APScheduler loop
(that path is exercised manually via ``python -m app.sync.cli scheduler``).
They verify the testable surface of :class:`SyncScheduler`:

* ``run_once`` dispatches by entity name to the right orchestrator function.
* ``run_once("<unknown>")`` raises ``ValueError`` with the known-entity list.
* ``_discover_conversations`` reads distinct WeChat pairs from the DB.
* ``_log_run`` emits one JSON line per run with sensible totals + level.
* The daily vs. message-every-N-hours schedule is registered with the right
  triggers when :meth:`start` is called (using a fake scheduler fixture).
* Failures inside a runner are swallowed and logged, not re-raised.
"""

from __future__ import annotations

import datetime as dt
import json
import logging
from collections.abc import AsyncIterator
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
import sqlalchemy as sa
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.db.base import Base
from app.db.models import Customer
from app.sync import scheduler as scheduler_module
from app.sync.scheduler import (
    ENTITY_FUNCS,
    SyncScheduler,
    _discover_conversations,
    _log_run,
)


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


async def _make_engine():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    return engine


@pytest.fixture
async def engine():
    eng = await _make_engine()
    try:
        yield eng
    finally:
        await eng.dispose()


@pytest.fixture
async def session_maker(engine):
    return async_sessionmaker(engine, expire_on_commit=False)


# ---------------------------------------------------------------------------
# ENTITY_FUNCS registry
# ---------------------------------------------------------------------------


def test_entity_funcs_has_expected_names():
    """The registry exposes the documented manual-trigger entity names."""
    expected = {
        "departments",
        "accounts",
        "friends",
        "groups",
        "voice_calls",
        "wa_voice_calls",
        "phone_calls",
        "sms",
        "messages",
        "all_wechat",
        "all",
    }
    assert expected <= set(ENTITY_FUNCS)


# ---------------------------------------------------------------------------
# _discover_conversations
# ---------------------------------------------------------------------------


async def test_discover_conversations_returns_distinct_wechat_pairs(session_maker):
    """Distinct (owner, friend_id) for channel='wechat' only; non-wechat + nulls dropped."""
    async with session_maker() as s:
        s.add_all(
            [
                Customer(
                    friend_id="f1",
                    channel="wechat",
                    owner_account_id="wx_a",
                ),
                Customer(
                    friend_id="f2",
                    channel="wechat",
                    owner_account_id="wx_a",
                ),
                Customer(
                    friend_id="f3",
                    channel="wechat",
                    owner_account_id="wx_b",
                ),
                # Different channel — should be excluded.
                Customer(
                    friend_id="p1@s.whatsapp.net",
                    channel="whatsapp",
                    owner_account_id="wx_a",
                ),
                # Null owner — should be excluded.
                Customer(
                    friend_id="orphan",
                    channel="wechat",
                    owner_account_id=None,
                ),
            ]
        )
        await s.commit()

    pairs = await _discover_conversations(s)
    assert sorted(pairs) == sorted([("wx_a", "f1"), ("wx_a", "f2"), ("wx_b", "f3")])


async def test_discover_conversations_respects_limit(session_maker):
    """``limit`` caps the work per message-sync run."""
    async with session_maker() as s:
        for i in range(10):
            s.add(
                Customer(
                    friend_id=f"f{i}",
                    channel="wechat",
                    owner_account_id="wx_a",
                )
            )
        await s.commit()

    pairs = await _discover_conversations(s, limit=3)
    assert len(pairs) == 3


# ---------------------------------------------------------------------------
# _log_run
# ---------------------------------------------------------------------------


def test_log_run_emits_one_json_line(capsys, caplog):
    """One JSON line, level=info when all sub-runs succeed."""
    results = [
        _mk_result("department", "success", 5, 5),
        _mk_result("wechat_friend", "success", 3, 3),
    ]
    with caplog.at_level(logging.INFO, logger="app.sync.scheduler"):
        _log_run("all", results, started_at=dt.datetime.now(dt.UTC))
    out = capsys.readouterr().out.strip().splitlines()
    assert len(out) == 1
    payload = json.loads(out[0])
    assert payload["event"] == "sync_run"
    assert payload["entity"] == "all"
    assert payload["level"] == "info"
    assert payload["totals"] == {"fetched": 8, "upserted": 8, "count": 2}
    assert payload["failed_entities"] == []
    assert len(payload["results"]) == 2


def test_log_run_level_warning_when_any_failed(capsys):
    """A failed sub-result bumps the level to warning + lists the failed entity."""
    results = [
        _mk_result("department", "success", 5, 5),
        _mk_result("wechat_friend", "failed", 0, 0, error="boom"),
    ]
    _log_run("all", results, started_at=dt.datetime.now(dt.UTC))
    payload = json.loads(capsys.readouterr().out.strip())
    assert payload["level"] == "warning"
    assert payload["failed_entities"] == ["wechat_friend"]
    assert payload["totals"]["fetched"] == 5


def test_log_run_level_error_on_top_level_failure(capsys):
    """A top-level error (no results at all) -> level=error + error field."""
    _log_run("all", [], started_at=dt.datetime.now(dt.UTC), error="AuthFailed")
    payload = json.loads(capsys.readouterr().out.strip())
    assert payload["level"] == "error"
    assert payload["error"] == "AuthFailed"
    assert payload["totals"]["count"] == 0


# ---------------------------------------------------------------------------
# SyncScheduler.run_once — dispatch
# ---------------------------------------------------------------------------


async def test_run_once_unknown_entity_raises(engine, session_maker):
    """An unknown entity name raises ValueError listing the known entities."""
    sched = SyncScheduler(engine=engine, session_maker=session_maker)
    with pytest.raises(ValueError, match="unknown entity"):
        await sched.run_once("does_not_exist")


async def test_run_once_dispatches_to_entity_runner(engine, session_maker, monkeypatch):
    """run_once('friends') calls _run_friends (which calls sync_wechat_friends)."""
    # Stub _build_client_from_env so no network / login happens.
    fake_client = MagicMock()
    monkeypatch.setattr(
        scheduler_module,
        "_build_client_from_env",
        AsyncMock(return_value=fake_client),
    )

    captured: dict[str, Any] = {}

    async def fake_runner(session, client, store):
        captured["session"] = session
        captured["client"] = client
        captured["store"] = store
        return [_mk_result("wechat_friend", "success", 7, 7)]

    # Patch the registry entry, not the orchestrator function.
    monkeypatch.setitem(ENTITY_FUNCS, "friends", fake_runner)

    sched = SyncScheduler(engine=engine, session_maker=session_maker)
    results = await sched.run_once("friends")

    assert len(results) == 1
    assert results[0].entity == "wechat_friend"
    assert results[0].fetched == 7
    assert captured["client"] is fake_client
    assert captured["store"] is not None


async def test_run_once_swallows_runner_exception(engine, session_maker, monkeypatch, capsys):
    """A blown-up runner is logged + swallowed (returns []; scheduler keeps going)."""
    fake_client = MagicMock()
    monkeypatch.setattr(
        scheduler_module,
        "_build_client_from_env",
        AsyncMock(return_value=fake_client),
    )

    async def boom(session, client, store):
        raise RuntimeError("upstream 500")

    monkeypatch.setitem(ENTITY_FUNCS, "groups", boom)

    sched = SyncScheduler(engine=engine, session_maker=session_maker)
    # Must NOT raise.
    results = await sched.run_once("groups")
    assert results == []
    out = capsys.readouterr().out
    assert "sync_run" in out
    payload = json.loads(out.strip().splitlines()[-1])
    assert payload["level"] == "error"
    assert "upstream 500" in payload["error"]


async def test_run_once_swallows_client_build_failure(engine, session_maker, monkeypatch):
    """If auth env is missing -> _build_client_from_env raises -> logged + empty list."""
    monkeypatch.setattr(
        scheduler_module,
        "_build_client_from_env",
        AsyncMock(side_effect=RuntimeError("no auth configured")),
    )
    sched = SyncScheduler(engine=engine, session_maker=session_maker)
    results = await sched.run_once("all")
    assert results == []


# ---------------------------------------------------------------------------
# Advisory lock — concurrent-invocation protection (Fix 3)
# ---------------------------------------------------------------------------


def _utcnow_naive() -> dt.datetime:
    """Naive UTC 'now', matching how SQLAlchemy reads back ``SyncRun.started_at``.

    The column is a non-tz-aware ``DateTime``; an aware ``datetime`` would
    raise TypeError when compared against DB-read values inside ``_safe_run``.
    """
    return dt.datetime.now(dt.UTC).replace(tzinfo=None)


async def _make_static_engine():
    """In-memory engine with ``StaticPool`` so cross-session rows are visible.

    The default ``:memory:`` pool gives each checkout a fresh DB; the advisory
    lock test seeds a row in one session and needs ``_safe_run``'s own session
    to see it, so the engine must pin a single connection.
    """
    from sqlalchemy.pool import StaticPool

    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)
    return engine


async def test_run_once_skips_when_another_run_is_active(monkeypatch):
    """A 'running' SyncRun for the same entity started < TTL ago → skip."""
    from app.db.models import SyncRun

    engine = await _make_static_engine()
    session_maker = async_sessionmaker(engine, expire_on_commit=False)

    # Seed a 'running' SyncRun for "friends" started just now.
    async with session_maker() as s:
        s.add(
            SyncRun(
                entity_type="friends",
                status="running",
                started_at=_utcnow_naive(),
            )
        )
        await s.commit()

    fake_client = MagicMock()
    monkeypatch.setattr(
        scheduler_module,
        "_build_client_from_env",
        AsyncMock(return_value=fake_client),
    )

    called = {"count": 0}

    async def fake_runner(session, client, store):
        called["count"] += 1
        return [_mk_result("wechat_friend", "success", 1, 1)]

    monkeypatch.setitem(ENTITY_FUNCS, "friends", fake_runner)

    try:
        sched = SyncScheduler(engine=engine, session_maker=session_maker)
        results = await sched.run_once("friends")
    finally:
        await engine.dispose()

    # Skipped: the runner must NOT have run.
    assert called["count"] == 0
    # The skip surfaces as a single 'skipped' SyncResult with a reason.
    assert len(results) == 1
    assert results[0].status == "skipped"
    assert "another sync is running" in (results[0].error or "")


async def test_run_once_proceeds_after_stale_run_expires(monkeypatch):
    """A 'running' SyncRun older than the TTL does NOT block — auto-recover."""
    from app.db.models import SyncRun

    engine = await _make_static_engine()
    session_maker = async_sessionmaker(engine, expire_on_commit=False)

    # Seed a 'running' SyncRun for "friends" 2 hours old — past the 60-min TTL.
    async with session_maker() as s:
        s.add(
            SyncRun(
                entity_type="friends",
                status="running",
                started_at=_utcnow_naive() - dt.timedelta(hours=2),
            )
        )
        await s.commit()

    fake_client = MagicMock()
    monkeypatch.setattr(
        scheduler_module,
        "_build_client_from_env",
        AsyncMock(return_value=fake_client),
    )

    async def fake_runner(session, client, store):
        return [_mk_result("wechat_friend", "success", 3, 3)]

    monkeypatch.setitem(ENTITY_FUNCS, "friends", fake_runner)

    try:
        sched = SyncScheduler(engine=engine, session_maker=session_maker)
        results = await sched.run_once("friends")
    finally:
        await engine.dispose()

    # Not skipped: the runner ran and returned its success result.
    assert len(results) == 1
    assert results[0].status == "success"
    assert results[0].fetched == 3


# ---------------------------------------------------------------------------
# SyncScheduler.start — job registration
# ---------------------------------------------------------------------------


async def test_start_registers_all_jobs_with_expected_triggers(engine, session_maker):
    """start() wires a daily CronTrigger (full) + an IntervalTrigger (messages)."""
    sched = SyncScheduler(
        full_sync_hour=3,
        message_interval_hours=6,
        engine=engine,
        session_maker=session_maker,
    )
    # Replace the real AsyncIOScheduler with a spy before start() creates it,
    # so we can assert on add_job without running the loop.
    spy = MagicMock()
    spy.add_job = MagicMock()
    with patch.object(scheduler_module, "AsyncIOScheduler", return_value=spy):
        await sched.start()

    # Four jobs registered (2026-08-21 加每周全量对齐;2026-09-04 加 4h 分析链).
    assert spy.add_job.call_count == 4
    job_ids = {c.kwargs["id"] for c in spy.add_job.call_args_list}
    assert job_ids == {
        "yunke_full_sync", "yunke_message_sync", "yunke_weekly_full_history",
        "yunke_analytics_tick",
    }

    # Full sync -> CronTrigger(hour=3); messages -> IntervalTrigger(hours=6).
    # add_job was called as (func, trigger, kwargs...) so trigger is args[1].
    full_call = next(
        c for c in spy.add_job.call_args_list if c.kwargs["id"] == "yunke_full_sync"
    )
    full_trigger = full_call.args[1]
    # CronTrigger exposes the parsed schedule via str().
    assert str(full_trigger) == "cron[hour='3']"

    msg_call = next(
        c for c in spy.add_job.call_args_list if c.kwargs["id"] == "yunke_message_sync"
    )
    interval = msg_call.args[1]
    # IntervalTrigger exposes its interval as a timedelta on .interval`.
    assert interval.interval.total_seconds() == 6 * 3600

    # 2026-08-20 起:两个 job 都打到 _safe_page_run(页面方式子进程),
    # full 传 "full"、messages 传 "messages"。
    assert full_call.args[0].__name__ == "_safe_page_run"
    assert full_call.kwargs["args"] == ["full"]
    assert msg_call.args[0].__name__ == "_safe_page_run"
    assert msg_call.kwargs["args"] == ["messages"]

    # Weekly full-history: Sunday 05:00 CronTrigger.
    weekly_call = next(
        c for c in spy.add_job.call_args_list if c.kwargs["id"] == "yunke_weekly_full_history"
    )
    weekly_trigger = weekly_call.args[1]
    assert "day_of_week='sun'" in str(weekly_trigger)
    assert "hour='5'" in str(weekly_trigger) or "hour='5'" in repr(weekly_trigger)
    assert weekly_call.args[0].__name__ == "_safe_page_run"
    assert weekly_call.kwargs["args"] == ["full-history"]

    # 2026-09-04 加 4h 分析链:纯 DB(非 _safe_page_run,无子进程),
    # IntervalTrigger 与消息增量同节奏,不带 args。
    tick_call = next(
        c for c in spy.add_job.call_args_list if c.kwargs["id"] == "yunke_analytics_tick"
    )
    assert tick_call.args[0].__name__ == "_analytics_tick"
    tick_interval = tick_call.args[1]
    assert tick_interval.interval.total_seconds() == 6 * 3600
    assert "args" not in tick_call.kwargs

    # coalesce=True + max_instances=1 protects against overlap.
    for c in spy.add_job.call_args_list:
        assert c.kwargs["coalesce"] is True
        assert c.kwargs["max_instances"] == 1


async def test_start_is_idempotent(engine, session_maker):
    """A second start() is a no-op (does not register a second scheduler)."""
    sched = SyncScheduler(engine=engine, session_maker=session_maker)
    with patch.object(scheduler_module, "AsyncIOScheduler", return_value=MagicMock()):
        await sched.start()
        first = sched._scheduler
        await sched.start()  # second call — ignored.
        assert sched._scheduler is first


# ---------------------------------------------------------------------------
# Stop / teardown
# ---------------------------------------------------------------------------


async def test_stop_disposes_owned_engine():
    """When the scheduler owns its engine, stop() disposes it."""
    # Patch create_async_engine so we don't try to connect to a real DB.
    fake_engine = MagicMock()
    fake_engine.dispose = AsyncMock()
    fake_engine.begin = MagicMock()
    # start() will call engine.begin() to create tables; fake it.
    class _Ctx:
        async def __aenter__(self):
            conn = MagicMock()
            conn.run_sync = AsyncMock()
            return conn

        async def __aexit__(self, *a):
            return None

    fake_engine.begin.return_value = _Ctx()

    with patch.object(scheduler_module, "create_async_engine", return_value=fake_engine):
        sched = SyncScheduler()
        assert sched._owns_engine is True
        await sched.stop()
    fake_engine.dispose.assert_awaited_once()


async def test_stop_does_not_dispose_injected_engine(engine, session_maker):
    """An engine injected by the caller is NOT disposed on stop()."""
    sched = SyncScheduler(engine=engine, session_maker=session_maker)
    assert sched._owns_engine is False
    await sched.stop()  # must not dispose — caller (test fixture) owns it.


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _mk_result(
    entity: str, status: str, fetched: int, upserted: int, *, error: str | None = None):
    from app.sync.orchestrator import SyncResult

    return SyncResult(
        entity=entity,
        status=status,
        fetched=fetched,
        upserted=upserted,
        error=error,
    )


# ---------------------------------------------------------------------------
# Page-based scheduled jobs (2026-08-20 — API 直调退出定时任务，全部页面方式防风控)
# ---------------------------------------------------------------------------

class _FakeProc:
    def __init__(self, rc: int = 0) -> None:
        self._rc = rc

    async def wait(self) -> int:
        return self._rc


async def test_safe_page_run_full_invokes_bash_full_sync(engine, monkeypatch):
    """full job = bash scripts/full_sync_all.sh（页面方式全量序列，子进程隔离）。"""
    sched = SyncScheduler(engine=engine)
    captured: dict = {}

    async def fake_exec(*args, cwd=None, env=None):
        captured["args"] = list(args)
        captured["cwd"] = cwd
        return _FakeProc(0)

    monkeypatch.setattr(scheduler_module.asyncio, "create_subprocess_exec", fake_exec)
    monkeypatch.setattr(scheduler_module.shutil, "which", lambda name: "/usr/bin/bash")

    await sched._safe_page_run("full")

    assert captured["args"][0] == "/usr/bin/bash"
    assert captured["args"][1].endswith("full_sync_all.sh")
    # cwd 指向 yunke-server（脚本的相对路径基准）
    assert str(captured["cwd"]).replace("\\", "/").endswith("yunke-server")


async def test_safe_page_run_messages_builds_scrape_command(engine, monkeypatch):
    """messages job = accounts→friends→四类消息(活跃窗)→moments 浅扫的命令序列。

    2026-09-10 调度根治:头部刷员工账号表(员工列表时间最多滞后 4h);
    消息实体只扫近 24h 活跃对话(--active-window-hours 24,防序列 15h+
    挤死 full);陈旧对话每日 full 兜底。
    """
    sched = SyncScheduler(engine=engine, max_conversations_per_run=123)
    commands: list[list[str]] = []

    async def fake_exec(*args, cwd=None, env=None):
        commands.append(list(args))
        return _FakeProc(0)

    monkeypatch.setattr(scheduler_module.asyncio, "create_subprocess_exec", fake_exec)

    await sched._safe_page_run("messages")

    assert len(commands) == 10
    for args in commands:
        assert args[1].endswith("scrape_via_page.py")

    def entity_of(args: list[str]) -> str:
        return args[args.index("--entity") + 1]

    # 头部:账号刷新(员工列表「数据更新/最新消息时间」来源)
    assert entity_of(commands[0]) == "accounts"
    assert entity_of(commands[1]) == "wa-accounts"
    assert entity_of(commands[2]) == "friends"
    assert entity_of(commands[3]) == "wa-friends"
    # 群消息前刷群活跃(chatroom-stats 全窗——群活跃窗判定源,2026-09-10)
    entities_seq = [entity_of(a) for a in commands]
    assert entities_seq.index("chatroom-stats") < entities_seq.index("group-messages")
    # 四类消息:24h 活跃窗(2026-09-10 根治调度堵塞)
    for args in commands:
        if entity_of(args) in ("messages", "wa-messages", "group-messages", "wa-group-messages"):
            assert args[args.index("--max-conversations") + 1] == "99999"
            assert args[args.index("--active-window-hours") + 1] == "24"
    # moments 浅扫在序列末尾:每员工最多 3 页
    moments_cmd = commands[-1]
    assert entity_of(moments_cmd) == "moments"
    assert moments_cmd[moments_cmd.index("--max-pages") + 1] == "3"


async def test_safe_page_run_nonzero_rc_is_logged_not_raised(engine, monkeypatch, capsys):
    """子进程非零退出只记日志不抛（调度器保住下一个周期）。"""
    sched = SyncScheduler(engine=engine)
    monkeypatch.setattr(scheduler_module.shutil, "which", lambda name: "/usr/bin/bash")

    async def fake_exec(*args, cwd=None, env=None):
        return _FakeProc(3)

    monkeypatch.setattr(scheduler_module.asyncio, "create_subprocess_exec", fake_exec)

    await sched._safe_page_run("full")  # 不应抛异常
    out = capsys.readouterr().out
    assert '"rc": 3' in out or "'rc': 3" in out or "rc" in out