diff --git a/yunke-server/app/sync/orchestrator.py b/yunke-server/app/sync/orchestrator.py index 416ae6b..2afd60b 100644 --- a/yunke-server/app/sync/orchestrator.py +++ b/yunke-server/app/sync/orchestrator.py @@ -120,8 +120,16 @@ class SyncResult: """Outcome of one sync run (one entity / one conversation). Mirrors the essential fields of the persisted ``SyncRun`` row so callers can - react to a run without re-querying the DB. ``status`` is ``"success"`` or - ``"failed"``; on failure ``error`` carries the exception text. + react to a run without re-querying the DB. ``status`` is one of: + + * ``"success"`` — the scrape + upsert completed (the SyncRun row is + persisted with ``status='success'``). + * ``"failed"`` — the scrape/upsert raised; ``error`` carries the + exception text (the SyncRun row is persisted with ``status='failed'``). + * ``"skipped"`` — a precondition was unmet (e.g. no resolvable root + ``department_id`` for an endpoint that requires one); ``error`` carries + the reason. NO SyncRun row is recorded (a skip is neither success nor + failure), so a skipped entity does not advance the watermark. """ entity: str diff --git a/yunke-server/tests/test_orchestrator.py b/yunke-server/tests/test_orchestrator.py index b7e6bfe..c0931be 100644 --- a/yunke-server/tests/test_orchestrator.py +++ b/yunke-server/tests/test_orchestrator.py @@ -613,6 +613,185 @@ async def test_sync_wechat_accounts_passes_resolved_root_to_scraper(monkeypatch) 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) # ===========================================================================