浅念
9cbd796975
Task 2.7 — the coordination layer tying scrapers + repos together with
resumable, incremental sync.
app/sync/watermarks.py — SyncState CRUD keyed on (entity_type, channel,
scope_id): get_watermark / get_last_timestamp / get_last_version /
set_watermark (upsert). A partial advance (timestamp-only) does not clobber
the other cursor with NULL.
app/sync/orchestrator.py — the sync brain:
* sync_wechat_messages: time-window incremental. Reads last_timestamp for
(message, wechat, "<wechat_id>:<friend_id>"), queries [last_ts, now] via
scrape_wechat_messages, upserts each, best-effort downloads media, advances
the watermark to max(timestamp). A re-run with no new messages queries an
empty window and fetches nothing — the watermark advanced past.
* sync_{departments,wechat_accounts,wechat_friends,wechat_groups,
wechat_voice_calls}: full re-scrape + idempotent upsert (these list APIs
have no since-param; 'incremental' = re-scrape all, rely on PK idempotency).
Watermark records last_synced_at only.
* sync_all_wechat: coordinator running departments -> accounts -> friends ->
groups -> voice_calls, then per-conversation messages for an explicit
(wechat_id, friend_id) list. Auto-discovery deferred.
Each run logs a SyncRun row (started -> success/failed, counts, error).
Failure policy: a scraper that raises is caught, recorded as failed, and a
failed SyncResult is returned — the orchestrator does not re-raise, so one
broken entity does not abort the batch. Media-download failures are
best-effort. The orchestrator never commits; callers own the transaction.
Tests (TDD red -> green; in-memory SQLite; scrapers/repos monkeypatched):
- watermarks: set/get, upsert (same key twice -> 1 row), partial-advance
preserves the other cursor, defaults when absent.
- orchestrator messages: first run upserts 2 + advances watermark to T2;
second run scrape called with start==T2 and upserts 0 (incremental proof);
media failure does not abort; SyncRun success recorded.
- orchestrator friends: 3 records upserted, SyncRun success, watermark
last_synced_at set.
- failure policy: scraper raises -> SyncRun failed + failed result, no raise
(both friends and message paths).
- sync_all_wechat: 5 entities + 2 message runs in order, all success.
Suite: 149 -> 163 passing. ruff clean.
Co-Authored-By: Claude <noreply@anthropic.com>
245 lines
8.3 KiB
Python
245 lines
8.3 KiB
Python
"""Tests for app.sync.watermarks — SyncState CRUD (watermark get/set).
|
|
|
|
Exercises the watermark helpers against an isolated in-memory SQLite engine
|
|
(same pattern as ``tests/test_repos.py``). Covers the three guarantees the
|
|
orchestrator relies on:
|
|
|
|
* ``set_watermark`` then ``get_watermark`` returns the just-written row.
|
|
* ``set_watermark`` is an **upsert** keyed on ``(entity_type, channel, scope_id)``:
|
|
the same key twice produces exactly ONE row, with the cursor fields refreshed
|
|
from the later write.
|
|
* The convenience accessors (``get_last_timestamp`` / ``get_last_version``)
|
|
return the stored value, or the documented default (``0`` / ``None``) when no
|
|
row exists yet (first sync of a brand-new scope).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
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 SyncState
|
|
from app.sync.watermarks import (
|
|
get_last_timestamp,
|
|
get_last_version,
|
|
get_watermark,
|
|
set_watermark,
|
|
)
|
|
|
|
|
|
async def _make_engine():
|
|
"""Create an in-memory async SQLite engine with all tables created."""
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
return engine
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# set then get
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_set_then_get_returns_row_with_values():
|
|
"""A fresh set creates a row; get returns it carrying the cursors."""
|
|
engine = await _make_engine()
|
|
try:
|
|
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with SessionLocal() as session:
|
|
await set_watermark(
|
|
session,
|
|
"message",
|
|
channel="wechat",
|
|
scope_id="wx_a:friend_b",
|
|
last_version="v1",
|
|
last_timestamp=1_700_000_000_000,
|
|
)
|
|
await session.commit()
|
|
|
|
async with SessionLocal() as session:
|
|
row = await get_watermark(
|
|
session, "message", channel="wechat", scope_id="wx_a:friend_b"
|
|
)
|
|
assert row is not None
|
|
assert row.entity_type == "message"
|
|
assert row.channel == "wechat"
|
|
assert row.scope_id == "wx_a:friend_b"
|
|
assert row.last_version == "v1"
|
|
assert row.last_timestamp == 1_700_000_000_000
|
|
assert row.last_synced_at is not None
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# upsert: same key twice -> 1 row, updated
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_set_watermark_is_upsert_one_row_updated():
|
|
"""Two sets on the same (entity_type, channel, scope_id) -> exactly one row.
|
|
|
|
The later write refreshes ``last_timestamp`` / ``last_version`` /
|
|
``last_synced_at``. This is the resumability guarantee the orchestrator's
|
|
incremental loop leans on.
|
|
"""
|
|
engine = await _make_engine()
|
|
try:
|
|
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with SessionLocal() as session:
|
|
await set_watermark(
|
|
session,
|
|
"message",
|
|
channel="wechat",
|
|
scope_id="s1",
|
|
last_timestamp=100,
|
|
last_version="v1",
|
|
)
|
|
await session.commit()
|
|
async with SessionLocal() as session:
|
|
await set_watermark(
|
|
session,
|
|
"message",
|
|
channel="wechat",
|
|
scope_id="s1",
|
|
last_timestamp=200,
|
|
last_version="v2",
|
|
)
|
|
await session.commit()
|
|
|
|
async with SessionLocal() as session:
|
|
count = (
|
|
await session.execute(
|
|
sa.select(sa.func.count())
|
|
.select_from(SyncState)
|
|
.where(SyncState.scope_id == "s1")
|
|
)
|
|
).scalar_one()
|
|
assert count == 1
|
|
|
|
row = await get_watermark(
|
|
session, "message", channel="wechat", scope_id="s1"
|
|
)
|
|
assert row is not None
|
|
assert row.last_timestamp == 200
|
|
assert row.last_version == "v2"
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
async def test_set_watermark_partial_advance_does_not_clobber_other_cursor():
|
|
"""A timestamp-only advance leaves the previously-set version intact.
|
|
|
|
The orchestrator advances one cursor at a time (e.g. the message loop
|
|
advances ``last_timestamp``); a partial set must NOT wipe ``last_version``
|
|
with None.
|
|
"""
|
|
engine = await _make_engine()
|
|
try:
|
|
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with SessionLocal() as session:
|
|
await set_watermark(
|
|
session,
|
|
"wechat_friend",
|
|
channel="wechat",
|
|
scope_id=None,
|
|
last_version="v1",
|
|
)
|
|
await session.commit()
|
|
async with SessionLocal() as session:
|
|
# Advance only the sync time (no new cursors supplied).
|
|
await set_watermark(
|
|
session, "wechat_friend", channel="wechat", scope_id=None
|
|
)
|
|
await session.commit()
|
|
|
|
async with SessionLocal() as session:
|
|
row = await get_watermark(
|
|
session, "wechat_friend", channel="wechat", scope_id=None
|
|
)
|
|
assert row is not None
|
|
assert row.last_version == "v1" # preserved, not clobbered to None
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# convenience accessors: defaults when row absent
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_get_last_timestamp_returns_default_when_absent():
|
|
"""No row yet -> default (0) returned, not None."""
|
|
engine = await _make_engine()
|
|
try:
|
|
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with SessionLocal() as session:
|
|
assert (
|
|
await get_last_timestamp(
|
|
session, "message", channel="wechat", scope_id="never"
|
|
)
|
|
== 0
|
|
)
|
|
# Custom default is honored.
|
|
assert (
|
|
await get_last_timestamp(
|
|
session,
|
|
"message",
|
|
channel="wechat",
|
|
scope_id="never",
|
|
default=42,
|
|
)
|
|
== 42
|
|
)
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
async def test_get_last_version_returns_none_when_absent():
|
|
"""No row yet -> None (no version cursor yet)."""
|
|
engine = await _make_engine()
|
|
try:
|
|
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with SessionLocal() as session:
|
|
assert (
|
|
await get_last_version(
|
|
session, "message", channel="wechat", scope_id="never"
|
|
)
|
|
is None
|
|
)
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
async def test_get_last_timestamp_returns_stored_value_after_set():
|
|
"""After a set, the accessor returns the stored timestamp (not default)."""
|
|
engine = await _make_engine()
|
|
try:
|
|
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with SessionLocal() as session:
|
|
await set_watermark(
|
|
session,
|
|
"message",
|
|
channel="wechat",
|
|
scope_id="s2",
|
|
last_timestamp=999,
|
|
)
|
|
await session.commit()
|
|
|
|
async with SessionLocal() as session:
|
|
assert (
|
|
await get_last_timestamp(
|
|
session, "message", channel="wechat", scope_id="s2"
|
|
)
|
|
== 999
|
|
)
|
|
assert (
|
|
await get_last_version(
|
|
session, "message", channel="wechat", scope_id="s2"
|
|
)
|
|
is None
|
|
)
|
|
finally:
|
|
await engine.dispose()
|