Files
浅念 df29dd027a feat: scheduler + CLI + docker-compose + README
M6 — the scheduling/deployment layer. Makes the system runnable as a
scheduled service + deployable via Docker.

Part A — SyncScheduler (app/sync/scheduler.py):
- APScheduler (AsyncIOScheduler) wraps the orchestrator's sync_* functions.
- Daily full sync at FULL_SYNC_HOUR (default 02:00 local) + incremental
  message sync every MESSAGE_INTERVAL_HOURS (default 4).
- ENTITY_FUNCS registry dispatches run_once("<entity>") to the right runner
  (departments/accounts/friends/groups/voice_calls/phone_calls/sms/messages/
  all_wechat/all).
- Conversations for the message run are discovered from the local customer
  mirror (no extra upstream call).
- Failure policy: every run is wrapped — a blown entity is logged + swallowed
  so the scheduler keeps its next appointment.
- Per-run JSON-line log (level=info/warning/error, totals, failed entities).

Part B — CLI entry point (app/sync/cli.py):
- `python -m app.sync.cli sync [--entity <name>] [--quiet]` — one-shot sync.
- `python -m app.sync.cli scheduler [--full-sync-hour H] [--message-interval-hours N]`
  — long-running scheduler (blocks until SIGINT/SIGTERM).
- Uses _push_settings_into_environ + build_client_from_env (session-cookie
  mode preferred; pcLogin deferred).
- Base.metadata.create_all on every command (fresh-checkout friendly).
- Exit codes: 0 success, 1 generic failure, 2 bad CLI / auth misconfig.

Part C — Docker stack:
- yunke-server/Dockerfile: python:3.12-slim, installs deps + app, single
  image serves both API (uvicorn) + scheduler (CLI).
- yunkes/Dockerfile: multi-stage node:20 build → nginx serves dist/ +
  reverse-proxies /api → backend:8000.
- yunkes/nginx.conf: /api proxy with proxy_buffering off (streaming media),
  SPA fallback, asset caching.
- docker-compose.yml: mysql (8.0, healthcheck, volume) + redis (7-alpine) +
  backend + scheduler + frontend (port 5173 → 80). Env-driven creds + cadence.

Part D — README.md at repo root: dev mode + docker mode quick-start, config
table, architecture overview, current limitations (pcLogin deferred, WA voice/
media deferred, conservative pacing, conversation discovery is local-only).

Verification:
- pytest: 323 passed (was 294; +29 new for scheduler + CLI).
- npm run build: clean.
- python -m app.sync.cli --help: prints usage.
- docker compose config: VALID.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-05 12:11:29 +08:00

230 lines
8.4 KiB
Python

"""Tests for ``python -m app.sync.cli`` (M6 Part B).
The CLI's network / DB / scheduler-loop paths are exercised manually (they
need real env + a running DB). These tests verify the testable surface:
* ``--help`` and the subcommand parsers build without error.
* The ``sync`` command dispatches into :class:`SyncScheduler.run_once` with
the right entity and prints a summary line.
* An unknown subcommand / bad ``--entity`` exits non-zero.
* Env-var fallbacks (``FULL_SYNC_HOUR`` / ``MESSAGE_INTERVAL_HOURS``) work.
"""
from __future__ import annotations
import asyncio
from typing import Any
from unittest.mock import AsyncMock, MagicMock
import pytest
from app.sync import cli as cli_module
from app.sync.orchestrator import SyncResult
# ---------------------------------------------------------------------------
# Argument parser
# ---------------------------------------------------------------------------
def test_parser_help_lists_both_subcommands(capsys):
"""``--help`` mentions both ``sync`` and ``scheduler``."""
with pytest.raises(SystemExit) as exc:
cli_module._build_parser().parse_args(["--help"])
assert exc.value.code == 0
out = capsys.readouterr().out
assert "sync" in out
assert "scheduler" in out
def test_parser_requires_subcommand():
"""No subcommand -> argparse errors (exit code 2)."""
with pytest.raises(SystemExit) as exc:
cli_module._build_parser().parse_args([])
assert exc.value.code == 2
def test_parser_sync_entity_choices_match_registry():
"""``--entity`` choices mirror the scheduler registry."""
parser = cli_module._build_parser()
args = parser.parse_args(["sync", "--entity", "friends"])
assert args.command == "sync"
assert args.entity == "friends"
def test_parser_rejects_unknown_entity():
"""An unknown entity name is rejected by argparse (exit code 2)."""
with pytest.raises(SystemExit) as exc:
cli_module._build_parser().parse_args(["sync", "--entity", "nope"])
assert exc.value.code == 2
def test_parser_scheduler_flags_default_none():
"""``--full-sync-hour`` and ``--message-interval-hours`` default to None (env-fallback path)."""
args = cli_module._build_parser().parse_args(["scheduler"])
assert args.full_sync_hour is None
assert args.message_interval_hours is None
# ---------------------------------------------------------------------------
# main() — exit codes
# ---------------------------------------------------------------------------
def test_main_returns_2_on_unknown_subcommand(capsys):
"""An unknown subcommand surfaces as exit code 2 (argparse)."""
with pytest.raises(SystemExit) as exc:
cli_module.main(["bogus"])
assert exc.value.code == 2
def test_main_value_error_becomes_exit_2(monkeypatch, capsys):
"""A ValueError surfaced from a command (e.g. unknown entity) is exit 2."""
async def _boom(_args):
raise ValueError("unknown entity: does_not_exist")
monkeypatch.setattr(cli_module, "_cmd_sync", _boom)
rc = cli_module.main(["sync", "--entity", "friends"])
assert rc == 2
err = capsys.readouterr().err
assert "unknown entity" in err
def test_main_generic_exception_becomes_exit_1(monkeypatch):
"""Any non-ValueError exception is exit 1 (don't print misleading usage)."""
async def _boom(_args):
raise RuntimeError("kaboom")
monkeypatch.setattr(cli_module, "_cmd_sync", _boom)
rc = cli_module.main(["sync"])
assert rc == 1
# ---------------------------------------------------------------------------
# _cmd_sync — dispatch + summary printing
# ---------------------------------------------------------------------------
def test_cmd_sync_dispatches_run_once_and_prints_summary(monkeypatch, capsys):
"""``sync`` builds a SyncScheduler, calls run_once(<entity>), prints a summary."""
# ``_cmd_sync`` is async — wrap a coroutine-returning stub via patch.
fake_results = [
SyncResult(entity="wechat_friend", status="success", fetched=3, upserted=3),
SyncResult(entity="message", status="success", fetched=10, upserted=10, channel="wechat"),
]
async def _fake_init_db(_db_url):
engine = MagicMock()
engine.dispose = AsyncMock()
sm = MagicMock()
return engine, sm
scheduler_mock = MagicMock()
scheduler_mock.run_once = AsyncMock(return_value=list(fake_results))
def _ctor(*, engine, session_maker):
# Returned to _cmd_sync which then calls .run_once() on it.
return scheduler_mock
monkeypatch.setattr(cli_module, "_init_db", _fake_init_db)
monkeypatch.setattr(cli_module, "SyncScheduler", _ctor)
args = cli_module._build_parser().parse_args(["sync", "--entity", "friends"])
rc = asyncio.run(cli_module._cmd_sync(args))
assert rc == 0
scheduler_mock.run_once.assert_awaited_once_with("friends")
out = capsys.readouterr().out
assert "[sync] friends" in out
assert "fetched=13" in out
assert "upserted=13" in out
def test_cmd_sync_partial_failure_is_exit_0(monkeypatch):
"""If at least one sub-result succeeded, exit 0 (partial delivery)."""
fake_results = [
SyncResult(entity="department", status="success", fetched=2, upserted=2),
SyncResult(entity="wechat_friend", status="failed", fetched=0, upserted=0, error="x"),
]
async def _fake_init_db(_db_url):
engine = MagicMock()
engine.dispose = AsyncMock()
return engine, MagicMock()
scheduler_mock = MagicMock()
scheduler_mock.run_once = AsyncMock(return_value=list(fake_results))
monkeypatch.setattr(cli_module, "_init_db", _fake_init_db)
monkeypatch.setattr(cli_module, "SyncScheduler", lambda **kw: scheduler_mock)
args = cli_module._build_parser().parse_args(["sync"])
rc = asyncio.run(cli_module._cmd_sync(args))
assert rc == 0
def test_cmd_sync_all_failed_is_exit_1(monkeypatch):
"""Every sub-run failed -> exit 1 (nothing useful delivered)."""
fake_results = [
SyncResult(entity="department", status="failed", fetched=0, upserted=0, error="x"),
SyncResult(entity="wechat_friend", status="failed", fetched=0, upserted=0, error="y"),
]
async def _fake_init_db(_db_url):
engine = MagicMock()
engine.dispose = AsyncMock()
return engine, MagicMock()
scheduler_mock = MagicMock()
scheduler_mock.run_once = AsyncMock(return_value=list(fake_results))
monkeypatch.setattr(cli_module, "_init_db", _fake_init_db)
monkeypatch.setattr(cli_module, "SyncScheduler", lambda **kw: scheduler_mock)
args = cli_module._build_parser().parse_args(["sync"])
rc = asyncio.run(cli_module._cmd_sync(args))
assert rc == 1
# ---------------------------------------------------------------------------
# Env-var fallbacks for the scheduler command
# ---------------------------------------------------------------------------
def test_env_int_returns_default_when_missing(monkeypatch):
"""Missing env var -> default."""
monkeypatch.delenv("FULL_SYNC_HOUR", raising=False)
assert cli_module._env_int("FULL_SYNC_HOUR", default=42) == 42
def test_env_int_parses_value(monkeypatch):
"""A clean int env var is parsed."""
monkeypatch.setenv("FULL_SYNC_HOUR", "13")
assert cli_module._env_int("FULL_SYNC_HOUR", default=2) == 13
def test_env_int_falls_back_on_garbage(monkeypatch, capsys):
"""Non-int env var falls back to default + prints a warning."""
monkeypatch.setenv("FULL_SYNC_HOUR", "two")
assert cli_module._env_int("FULL_SYNC_HOUR", default=2) == 2
err = capsys.readouterr().err
assert "FULL_SYNC_HOUR" in err
# ---------------------------------------------------------------------------
# _init_db — create_all is invoked
# ---------------------------------------------------------------------------
async def test_init_db_creates_tables():
"""_init_db returns an engine + session maker and runs create_all.
Uses an in-memory SQLite engine via the real SQLAlchemy path so the
create_all call is exercised end-to-end (not mocked).
"""
from sqlalchemy import text
engine, sm = await cli_module._init_db("sqlite+aiosqlite:///:memory:")
try:
# Tables created: customer is one of the registered tables.
from app.db.base import Base
assert "customer" in Base.metadata.tables
# engine + session_maker both usable (no exception = pass).
async with sm() as s:
await s.execute(text("SELECT 1"))
finally:
await engine.dispose()