"""Tests for app.sync.orchestrator — incremental sync coordination. The orchestrator is the integration seam that ties scrapers + repos together with resumable, incremental sync. These tests monkeypatch the scraper + repo functions (so no network / no real upsert SQL beyond what we assert) and drive the orchestrator against an in-memory SQLite engine to verify the documented behaviors: * **Messages (time-window incremental)**: first run upserts 2 messages and advances the watermark to ``max(timestamp)``; the second run's scrape is called with ``start == last_watermark`` and finds nothing new (watermark advanced past) — the proof of incremental correctness. * **Full re-scrape entity (friends)**: scraper yields 3 records; upsert is called once per record; a ``SyncRun`` row with ``status='success'`` and the right counts is recorded; the watermark's ``last_synced_at`` is set. * **Failure policy**: when the scraper raises, the orchestrator records a ``SyncRun`` row with ``status='failed'`` + the error string and returns a failed result WITHOUT re-raising (the rest of the batch keeps going). * **Media coupling**: media download is best-effort — a download that raises does NOT abort the message sync; the message is still upserted and the watermark still advances. All tests build their own in-memory SQLite engine and never touch production. """ 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 Department, SyncRun, SyncState from app.sync import orchestrator from app.sync.watermarks import get_last_timestamp T1 = 1_700_000_000_000 T2 = 1_700_000_001_000 # --------------------------------------------------------------------------- # Shared helpers # --------------------------------------------------------------------------- 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 def _msg(msg_id: str, ts: int, *, channel: str = "wechat") -> dict: """Build a scraper-shaped message dict (snake_case keys).""" return { "msg_id": msg_id, "channel": channel, "owner_account_id": "wx_staff", "talker_id": "friend_a", "timestamp": ts, "type": "text", "type_raw": 1, "raw_json": {"msgSvrId": msg_id, "timestamp": ts}, } # =========================================================================== # Messages — time-window incremental # =========================================================================== async def test_messages_first_run_upserts_and_advances_watermark(monkeypatch): """First run: scrape [m1(T1), m2(T2)] -> upsert 2, watermark -> T2.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) scrape_calls: list[dict] = [] async def fake_scrape(client, wechat_id, friend_id, start, end, *, types=""): scrape_calls.append( { "wechat_id": wechat_id, "friend_id": friend_id, "start": start, "end": end, "types": types, } ) for m in [_msg("m1", T1), _msg("m2", T2)]: yield m upsert_calls: list[list[dict]] = [] async def fake_upsert(session, recs): upsert_calls.append(list(recs)) return len(recs) async def fake_media(client, msg, store, session): return None monkeypatch.setattr(orchestrator, "scrape_wechat_messages", fake_scrape) monkeypatch.setattr(orchestrator, "upsert_messages", fake_upsert) monkeypatch.setattr(orchestrator, "download_message_media", fake_media) async with SessionLocal() as session: result = await orchestrator.sync_wechat_messages( session, client=object(), store=object(), wechat_id="wx_staff", friend_id="friend_a", ) await session.commit() # First call always starts from the default watermark 0. assert scrape_calls[0]["start"] == 0 assert scrape_calls[0]["wechat_id"] == "wx_staff" assert scrape_calls[0]["friend_id"] == "friend_a" # Result reports the fetched / upserted counts. assert result.status == "success" assert result.fetched == 2 assert result.upserted == 2 # upsert called once with both records. assert len(upsert_calls) == 1 assert len(upsert_calls[0]) == 2 # Watermark advanced to max(timestamp) == T2. async with SessionLocal() as session: assert ( await get_last_timestamp( session, "message", channel="wechat", scope_id="wx_staff:friend_a", ) == T2 ) finally: await engine.dispose() async def test_messages_second_run_scrape_starts_at_watermark_and_fetches_nothing( monkeypatch, ): """Two consecutive runs: the second scrape is called with start==T2 and upserts 0 records — the watermark advanced PAST the previously-seen messages. This is the core incremental-correctness proof.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) # A queue of batches: first call yields [m1, m2], second call yields []. batches: list[list[dict]] = [[_msg("m1", T1), _msg("m2", T2)], []] scrape_calls: list[int] = [] async def fake_scrape(client, wechat_id, friend_id, start, end, *, types=""): scrape_calls.append(start) for m in batches.pop(0): yield m upsert_calls: list[list[dict]] = [] async def fake_upsert(session, recs): upsert_calls.append(list(recs)) return len(recs) async def fake_media(client, msg, store, session): return None monkeypatch.setattr(orchestrator, "scrape_wechat_messages", fake_scrape) monkeypatch.setattr(orchestrator, "upsert_messages", fake_upsert) monkeypatch.setattr(orchestrator, "download_message_media", fake_media) # --- Run 1 --- async with SessionLocal() as session: r1 = await orchestrator.sync_wechat_messages( session, client=object(), store=object(), wechat_id="wx_staff", friend_id="friend_a", ) await session.commit() assert r1.upserted == 2 assert scrape_calls[0] == 0 # first run starts at default watermark # --- Run 2 (no new messages upstream) --- async with SessionLocal() as session: r2 = await orchestrator.sync_wechat_messages( session, client=object(), store=object(), wechat_id="wx_staff", friend_id="friend_a", ) await session.commit() # The second scrape MUST be called with start == T2 (the watermark we # advanced to in run 1) — proving we did not re-query [0, now] again. assert scrape_calls[1] == T2 assert r2.status == "success" assert r2.fetched == 0 assert r2.upserted == 0 # No new messages -> the orchestrator does not even call upsert on run 2 # (the empty-batch round-trip is skipped). The single recorded upsert # call is from run 1. assert len(upsert_calls) == 1 # Watermark unchanged (no new max to advance to). async with SessionLocal() as session: assert ( await get_last_timestamp( session, "message", channel="wechat", scope_id="wx_staff:friend_a", ) == T2 ) finally: await engine.dispose() async def test_messages_media_download_failure_does_not_abort_sync(monkeypatch): """If a media download raises, the message is still upserted and the watermark still advances — best-effort coupling by design.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) async def fake_scrape(client, wechat_id, friend_id, start, end, *, types=""): yield _msg("m1", T1) yield _msg("m2", T2) upserted: list[dict] = [] async def fake_upsert(session, recs): upserted.extend(recs) return len(recs) async def failing_media(client, msg, store, session): raise RuntimeError("oss exploded") monkeypatch.setattr(orchestrator, "scrape_wechat_messages", fake_scrape) monkeypatch.setattr(orchestrator, "upsert_messages", fake_upsert) monkeypatch.setattr(orchestrator, "download_message_media", failing_media) async with SessionLocal() as session: result = await orchestrator.sync_wechat_messages( session, client=object(), store=object(), wechat_id="wx_staff", friend_id="friend_a", ) await session.commit() # Media failure did NOT abort: both messages upserted, success recorded. assert result.status == "success" assert result.upserted == 2 assert len(upserted) == 2 async with SessionLocal() as session: assert ( await get_last_timestamp( session, "message", channel="wechat", scope_id="wx_staff:friend_a", ) == T2 ) finally: await engine.dispose() async def test_messages_records_sync_run_success(monkeypatch): """A successful message run records a SyncRun row with the right shape.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) async def fake_scrape(client, wechat_id, friend_id, start, end, *, types=""): yield _msg("m1", T1) async def fake_upsert(session, recs): return len(recs) async def fake_media(client, msg, store, session): return None monkeypatch.setattr(orchestrator, "scrape_wechat_messages", fake_scrape) monkeypatch.setattr(orchestrator, "upsert_messages", fake_upsert) monkeypatch.setattr(orchestrator, "download_message_media", fake_media) async with SessionLocal() as session: await orchestrator.sync_wechat_messages( session, client=object(), store=object(), wechat_id="wx_staff", friend_id="friend_a", ) await session.commit() async with SessionLocal() as session: runs = ( await session.execute( sa.select(SyncRun).where(SyncRun.entity_type == "message") ) ).scalars().all() assert len(runs) == 1 run = runs[0] assert run.status == "success" assert run.channel == "wechat" assert run.scope_id == "wx_staff:friend_a" assert run.records_fetched == 1 assert run.records_upserted == 1 assert run.error is None assert run.started_at is not None assert run.finished_at is not None finally: await engine.dispose() # =========================================================================== # Full re-scrape entity (friends) — idempotent upsert, no since-param # =========================================================================== async def test_friends_full_rescrape_upserts_all_records_sync_run_and_watermark( monkeypatch, ): """Friends list has no `since` param: 'incremental' = re-scrape all and rely on upsert idempotency. The orchestrator upserts every record, records a success SyncRun, and stamps the watermark's last_synced_at.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) friend_recs = [ {"friend_id": "f1", "channel": "wechat", "version": "v1"}, {"friend_id": "f2", "channel": "wechat", "version": "v1"}, {"friend_id": "f3", "channel": "wechat", "version": "v1"}, ] async def fake_scrape_friends(client): for r in friend_recs: yield r upserted: list[dict] = [] async def fake_upsert_customer(session, rec): upserted.append(rec) monkeypatch.setattr(orchestrator, "scrape_wechat_friends", fake_scrape_friends) monkeypatch.setattr(orchestrator, "upsert_customer", fake_upsert_customer) async with SessionLocal() as session: result = await orchestrator.sync_wechat_friends(session, client=object()) await session.commit() assert result.status == "success" assert result.fetched == 3 assert result.upserted == 3 assert [r["friend_id"] for r in upserted] == ["f1", "f2", "f3"] # SyncRun recorded with success + the right counts. async with SessionLocal() as session: run = ( await session.execute( sa.select(SyncRun).where(SyncRun.entity_type == "wechat_friend") ) ).scalar_one() assert run.status == "success" assert run.records_fetched == 3 assert run.records_upserted == 3 assert run.error is None # Watermark row exists with last_synced_at set (no version/timestamp # cursor advance — these entities have no since-param). state = ( await session.execute( sa.select(SyncState).where(SyncState.entity_type == "wechat_friend") ) ).scalar_one() assert state.last_synced_at is not None finally: await engine.dispose() # =========================================================================== # Failure policy — scraper raises # =========================================================================== async def test_failing_scraper_records_failed_sync_run_and_does_not_raise( monkeypatch, ): """A scraper that raises must NOT crash the batch. The orchestrator records a SyncRun with status='failed' + the error string and returns a failed result.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) async def fake_scrape_friends(client): raise RuntimeError("upstream 500") yield {} # pragma: no cover - make this an async generator monkeypatch.setattr(orchestrator, "scrape_wechat_friends", fake_scrape_friends) async with SessionLocal() as session: result = await orchestrator.sync_wechat_friends(session, client=object()) await session.commit() # Did not raise; returned a failed result. assert result.status == "failed" assert result.fetched == 0 assert result.upserted == 0 assert "upstream 500" in (result.error or "") async with SessionLocal() as session: run = ( await session.execute( sa.select(SyncRun).where(SyncRun.entity_type == "wechat_friend") ) ).scalar_one() assert run.status == "failed" assert run.records_fetched == 0 assert run.records_upserted == 0 assert "upstream 500" in (run.error or "") assert run.finished_at is not None finally: await engine.dispose() async def test_failing_message_scraper_records_failed_run(monkeypatch): """Same failure policy for the message sync path.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) async def fake_scrape(client, wechat_id, friend_id, start, end, *, types=""): raise RuntimeError("msg endpoint 500") yield {} # pragma: no cover - make this an async generator async def fake_upsert(session, recs): return 0 async def fake_media(client, msg, store, session): return None monkeypatch.setattr(orchestrator, "scrape_wechat_messages", fake_scrape) monkeypatch.setattr(orchestrator, "upsert_messages", fake_upsert) monkeypatch.setattr(orchestrator, "download_message_media", fake_media) async with SessionLocal() as session: result = await orchestrator.sync_wechat_messages( session, client=object(), store=object(), wechat_id="wx_staff", friend_id="friend_a", ) await session.commit() assert result.status == "failed" assert "msg endpoint 500" in (result.error or "") async with SessionLocal() as session: run = ( await session.execute( sa.select(SyncRun).where(SyncRun.entity_type == "message") ) ).scalar_one() assert run.status == "failed" assert "msg endpoint 500" in (run.error or "") finally: await engine.dispose() # =========================================================================== # sync_all_wechat — coordinator # =========================================================================== async def test_sync_all_wechat_runs_each_entity_then_messages(monkeypatch): """The coordinator runs departments → accounts → friends → groups → voice_calls → wa_voice_calls, then per-conversation messages for an explicit list. Returns one result per entity/conversation in order.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) async def empty_async_gen(_client=None, **_kwargs): if False: # pragma: no cover yield {} async def empty_messages(client, wechat_id, friend_id, start, end, *, types=""): if False: # pragma: no cover yield {} monkeypatch.setattr(orchestrator, "scrape_departments", empty_async_gen) monkeypatch.setattr(orchestrator, "scrape_wechat_accounts", empty_async_gen) monkeypatch.setattr(orchestrator, "scrape_wechat_friends", empty_async_gen) monkeypatch.setattr(orchestrator, "scrape_wechat_groups", empty_async_gen) monkeypatch.setattr(orchestrator, "scrape_wechat_voice_calls", empty_async_gen) monkeypatch.setattr(orchestrator, "scrape_wa_voice_calls", empty_async_gen) monkeypatch.setattr(orchestrator, "scrape_wechat_messages", empty_messages) # Scrapers yield nothing -> upsert / media loops never execute, so the # real upsert_messages / download_message_media (also imported into the # orchestrator namespace) are never called. No need to patch them. async with SessionLocal() as session: # sync_wechat_accounts resolves the company-root dept id required # by the wxUser/list form body from the Department table, so seed # a root (parent_id=None) before running the coordinator. session.add(Department(dept_id="ROOT_DEPT", parent_id=None, name="Root")) await session.flush() results = await orchestrator.sync_all_wechat( session, client=object(), store=object(), conversations=[("wx_staff", "friend_a"), ("wx_staff", "friend_b")], ) await session.commit() entities = [r.entity for r in results] # Six full-re-scrape entities in order, then two message runs. assert entities == [ "department", "wechat_account", "wechat_friend", "wechat_group", "wechat_voice_call", "wa_voice_call", "message", "message", ] assert all(r.status == "success" for r in results) finally: await engine.dispose() async def test_sync_wechat_accounts_skips_when_no_root_department(monkeypatch): """``wxUser/list`` REQUIRES a ``departmentId`` filter. When the org tree is empty (no root resolvable from the Department table) the orchestrator must NOT send a known-bad empty filter; it skips with a logged warning and returns a ``status='skipped'`` result without ever calling the scraper.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) called = False async def fake_scrape_accounts(client, *, department_id, **_kwargs): nonlocal called called = True if False: # pragma: no cover yield {} monkeypatch.setattr( orchestrator, "scrape_wechat_accounts", fake_scrape_accounts ) async with SessionLocal() as session: result = await orchestrator.sync_wechat_accounts( session, client=object() ) await session.commit() assert result.status == "skipped" assert result.fetched == 0 assert result.upserted == 0 assert not called # No SyncRun recorded for a skip (it is not a success or a failure). async with SessionLocal() as session: rows = ( await session.execute( sa.select(SyncRun).where( SyncRun.entity_type == "wechat_account" ) ) ).scalars().all() assert rows == [] finally: await engine.dispose() async def test_sync_wechat_accounts_passes_resolved_root_to_scraper(monkeypatch): """When a root department exists in the DB, the orchestrator resolves its id and passes it as ``department_id`` to the scraper.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) captured: dict = {} async def fake_scrape_accounts(client, *, department_id, **_kwargs): captured["department_id"] = department_id if False: # pragma: no cover yield {} async def fake_upsert(session, rec): return None monkeypatch.setattr( orchestrator, "scrape_wechat_accounts", fake_scrape_accounts ) monkeypatch.setattr(orchestrator, "upsert_wechat_account", fake_upsert) async with SessionLocal() as session: session.add( Department( dept_id="ROOT_DEPT", parent_id=None, name="Root" ) ) await session.flush() result = await orchestrator.sync_wechat_accounts( session, client=object() ) await session.commit() assert result.status == "success" assert captured["department_id"] == "ROOT_DEPT" finally: await engine.dispose() async def test_sync_wechat_voice_calls_skips_when_no_root_department(monkeypatch): """``queryWeChatVoiceList`` REQUIRES a ``departmentId`` filter (without it + the accompanying callSource/isRoom/time window the endpoint returns 200 with an empty ``data.data``, 0 rows). When the org tree is empty (no root resolvable from the Department table) the orchestrator must NOT send a known-bad empty filter; it skips with a logged warning and returns a ``status='skipped'`` result without ever calling the scraper.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) called = False async def fake_scrape_wechat_voice_calls(client, *, department_id, **_kwargs): nonlocal called called = True if False: # pragma: no cover yield {} monkeypatch.setattr( orchestrator, "scrape_wechat_voice_calls", fake_scrape_wechat_voice_calls, ) async with SessionLocal() as session: result = await orchestrator.sync_wechat_voice_calls( session, client=object() ) await session.commit() assert result.status == "skipped" assert result.fetched == 0 assert result.upserted == 0 assert not called # No SyncRun recorded for a skip (it is not a success or a failure). async with SessionLocal() as session: rows = ( await session.execute( sa.select(SyncRun).where( SyncRun.entity_type == "wechat_voice_call" ) ) ).scalars().all() assert rows == [] finally: await engine.dispose() async def test_sync_wechat_voice_calls_passes_resolved_root_to_scraper(monkeypatch): """When a root department exists in the DB, the orchestrator resolves its id and passes it as ``department_id`` to the WeChat voice-call scraper.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) captured: dict = {} async def fake_scrape_wechat_voice_calls(client, *, department_id, **_kwargs): captured["department_id"] = department_id if False: # pragma: no cover yield {} async def fake_upsert(session, rec): return None monkeypatch.setattr( orchestrator, "scrape_wechat_voice_calls", fake_scrape_wechat_voice_calls, ) monkeypatch.setattr(orchestrator, "upsert_voice_call", fake_upsert) async with SessionLocal() as session: session.add( Department( dept_id="ROOT_DEPT", parent_id=None, name="Root" ) ) await session.flush() result = await orchestrator.sync_wechat_voice_calls( session, client=object() ) await session.commit() assert result.status == "success" assert captured["department_id"] == "ROOT_DEPT" finally: await engine.dispose() async def test_sync_wa_voice_calls_skips_when_no_root_department(monkeypatch): """``queryWeChatVoiceList`` (WA variant, ``callSource=4``) REQUIRES a ``departmentId`` filter just like the WeChat variant. When the org tree is empty (no root resolvable from the Department table) the orchestrator skips with a logged warning and returns a ``status='skipped'`` result without ever calling the scraper.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) called = False async def fake_scrape_wa_voice_calls(client, *, department_id, **_kwargs): nonlocal called called = True if False: # pragma: no cover yield {} monkeypatch.setattr( orchestrator, "scrape_wa_voice_calls", fake_scrape_wa_voice_calls ) async with SessionLocal() as session: result = await orchestrator.sync_wa_voice_calls( session, client=object() ) await session.commit() assert result.status == "skipped" assert result.fetched == 0 assert result.upserted == 0 assert not called # No SyncRun recorded for a skip (it is not a success or a failure). async with SessionLocal() as session: rows = ( await session.execute( sa.select(SyncRun).where( SyncRun.entity_type == "wa_voice_call" ) ) ).scalars().all() assert rows == [] finally: await engine.dispose() async def test_sync_wa_voice_calls_passes_resolved_root_to_scraper(monkeypatch): """When a root department exists in the DB, the orchestrator resolves its id and passes it as ``department_id`` to the WA voice-call scraper.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) captured: dict = {} async def fake_scrape_wa_voice_calls(client, *, department_id, **_kwargs): captured["department_id"] = department_id if False: # pragma: no cover yield {} async def fake_upsert(session, rec): return None monkeypatch.setattr( orchestrator, "scrape_wa_voice_calls", fake_scrape_wa_voice_calls ) monkeypatch.setattr(orchestrator, "upsert_voice_call", fake_upsert) async with SessionLocal() as session: session.add( Department( dept_id="ROOT_DEPT", parent_id=None, name="Root" ) ) await session.flush() result = await orchestrator.sync_wa_voice_calls( session, client=object() ) await session.commit() assert result.status == "success" assert captured["department_id"] == "ROOT_DEPT" finally: await engine.dispose() # =========================================================================== # Phone calls + SMS — full re-scrape entities (M4) # =========================================================================== async def test_phone_calls_full_rescrape_upserts_all_records_sync_run_and_watermark( monkeypatch, ): """Phone-call list has no `since` param: 'incremental' = re-scrape all and rely on upsert idempotency. The orchestrator upserts every record, records a success SyncRun (entity_type='phone_call'), and stamps the watermark's last_synced_at.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) call_recs = [ {"call_id": "pc1", "channel": "phone_call", "version": "v1"}, {"call_id": "pc2", "channel": "phone_call", "version": "v1"}, ] async def fake_scrape_phone_calls(client): for r in call_recs: yield r upserted: list[dict] = [] async def fake_upsert_phone_call(session, rec): upserted.append(rec) monkeypatch.setattr( orchestrator, "scrape_phone_calls", fake_scrape_phone_calls ) monkeypatch.setattr( orchestrator, "upsert_phone_call", fake_upsert_phone_call ) async with SessionLocal() as session: result = await orchestrator.sync_phone_calls(session, client=object()) await session.commit() assert result.status == "success" assert result.fetched == 2 assert result.upserted == 2 assert [r["call_id"] for r in upserted] == ["pc1", "pc2"] async with SessionLocal() as session: run = ( await session.execute( sa.select(SyncRun).where(SyncRun.entity_type == "phone_call") ) ).scalar_one() assert run.status == "success" assert run.records_fetched == 2 assert run.records_upserted == 2 assert run.error is None state = ( await session.execute( sa.select(SyncState).where( SyncState.entity_type == "phone_call" ) ) ).scalar_one() assert state.last_synced_at is not None finally: await engine.dispose() async def test_sms_full_rescrape_upserts_all_records_sync_run_and_watermark( monkeypatch, ): """SMS list has no `since` param: same full-re-scrape shape as phone calls. Records a success SyncRun (entity_type='sms') and stamps the watermark.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) sms_recs = [ {"sms_id": "sms1", "channel": "sms", "content": "a"}, {"sms_id": "sms2", "channel": "sms", "content": "b"}, {"sms_id": "sms3", "channel": "sms", "content": "c"}, ] async def fake_scrape_sms(client): for r in sms_recs: yield r upserted: list[dict] = [] async def fake_upsert_sms(session, rec): upserted.append(rec) monkeypatch.setattr(orchestrator, "scrape_sms", fake_scrape_sms) monkeypatch.setattr(orchestrator, "upsert_sms", fake_upsert_sms) async with SessionLocal() as session: result = await orchestrator.sync_sms(session, client=object()) await session.commit() assert result.status == "success" assert result.fetched == 3 assert result.upserted == 3 assert [r["sms_id"] for r in upserted] == ["sms1", "sms2", "sms3"] async with SessionLocal() as session: run = ( await session.execute( sa.select(SyncRun).where(SyncRun.entity_type == "sms") ) ).scalar_one() assert run.status == "success" assert run.records_fetched == 3 assert run.records_upserted == 3 assert run.error is None state = ( await session.execute( sa.select(SyncState).where(SyncState.entity_type == "sms") ) ).scalar_one() assert state.last_synced_at is not None finally: await engine.dispose() async def test_failing_phone_call_scraper_records_failed_run(monkeypatch): """Same failure policy as the wechat entities: a scraper that raises records a failed SyncRun (entity_type='phone_call') and returns a failed result without re-raising.""" engine = await _make_engine() try: SessionLocal = async_sessionmaker(engine, expire_on_commit=False) async def fake_scrape_phone_calls(client): raise RuntimeError("callLog 500") yield {} # pragma: no cover - make this an async generator monkeypatch.setattr( orchestrator, "scrape_phone_calls", fake_scrape_phone_calls ) async with SessionLocal() as session: result = await orchestrator.sync_phone_calls(session, client=object()) await session.commit() assert result.status == "failed" assert "callLog 500" in (result.error or "") async with SessionLocal() as session: run = ( await session.execute( sa.select(SyncRun).where(SyncRun.entity_type == "phone_call") ) ).scalar_one() assert run.status == "failed" assert "callLog 500" in (run.error or "") finally: await engine.dispose()