浅念
202 lines
7.5 KiB
Python
202 lines
7.5 KiB
Python
"""Tests for :func:`app.sync.orchestrator.sync_call_recordings`.
|
|
|
|
Covers the orchestrator's call-recording localization pass: it is NOT a scrape
|
|
(it reads already-mirrored ``phone_call`` rows), so the contract under test is
|
|
that it:
|
|
|
|
* opens a ``SyncRun`` (``entity_type='call_recordings'``), calls
|
|
:func:`download_call_recordings`, then closes the run with the returned
|
|
counts and returns a matching :class:`SyncResult`;
|
|
* on success, the ``SyncRun`` row is ``status='success'`` and the underlying
|
|
call's ``record_media_hash`` is populated (the download actually happened);
|
|
* on a downloader failure, the run is recorded ``status='failed'`` and a
|
|
failed :class:`SyncResult` is returned without re-raising.
|
|
|
|
The happy path exercises the real :func:`download_call_recordings` against a
|
|
respx-mocked OSS URL and a temp-dir :class:`ObjectStore`, mirroring
|
|
``test_recording_downloader.py``. The failure path monkeypatches the
|
|
downloader so the orchestrator's exception handling is tested in isolation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
|
|
import httpx
|
|
import respx
|
|
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 PhoneCall, SyncRun
|
|
from app.session.yunke_client import YunkeClient
|
|
from app.storage.object_store import ObjectStore
|
|
from app.sync import orchestrator
|
|
from app.sync.orchestrator import (
|
|
ENTITY_CALL_RECORDINGS,
|
|
SyncResult,
|
|
sync_call_recordings,
|
|
)
|
|
|
|
BASE = "https://crm2.yunkecn.com"
|
|
# Realistic OSS recording URL (same shape as the yunke-pcfile call
|
|
# recordings): public GET, .mp3 suffix.
|
|
OSS_MP3 = (
|
|
"https://yunke-pcfile.oss-cn-beijing.aliyuncs.com/wechat-call/8m0atd/"
|
|
"x/ossCallback/a_b.mp3"
|
|
)
|
|
|
|
USERID = "a" * 32
|
|
COMPANY = "8m0atd"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared helpers (mirror tests/test_recording_downloader.py)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _primed_client() -> YunkeClient:
|
|
"""A YunkeClient already in a logged-in state (login flow skipped)."""
|
|
return YunkeClient(base_url=BASE, userid=USERID, company=COMPANY)
|
|
|
|
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# sync_call_recordings — happy path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@respx.mock
|
|
async def test_sync_call_recordings_localizes_and_logs_run(tmp_path):
|
|
"""One call with an OSS mp3 record_file: download happens, a success
|
|
``SyncRun`` (``entity_type='call_recordings'``) is recorded, and a matching
|
|
:class:`SyncResult` is returned with ``fetched == upserted == 1``. The
|
|
call's ``record_media_hash`` is populated.
|
|
|
|
``fetched`` here means "calls needing download" and ``upserted`` means
|
|
"calls localized" — the recording downloader returns a ``(attempted,
|
|
localized)`` tuple, which the orchestrator surfaces verbatim.
|
|
"""
|
|
store = ObjectStore(tmp_path)
|
|
engine = await _make_engine()
|
|
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
|
|
|
|
blob = b"ID3\x04\x00\x00\x00\x00\x00\x00mp3-payload"
|
|
respx.get(OSS_MP3).mock(return_value=httpx.Response(200, content=blob))
|
|
|
|
client = _primed_client()
|
|
try:
|
|
# Seed one call whose record_file is an OSS URL and no hash yet.
|
|
async with SessionLocal() as session:
|
|
session.add(
|
|
PhoneCall(
|
|
call_id="c1", channel="voice_call", record_file=OSS_MP3
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
async with SessionLocal() as session:
|
|
result = await sync_call_recordings(session, client, store)
|
|
# The function commits its own SyncRun row (per the plan's
|
|
# contract — differs from the other sync_* funcs that defer commit
|
|
# to the caller); no extra commit needed here.
|
|
|
|
# Returned the right SyncResult shape.
|
|
assert isinstance(result, SyncResult)
|
|
assert result.entity == ENTITY_CALL_RECORDINGS
|
|
assert result.status == "success"
|
|
assert result.fetched == 1
|
|
assert result.upserted == 1
|
|
assert result.error is None
|
|
|
|
expected_sha = hashlib.sha256(blob).hexdigest()
|
|
|
|
# A SyncRun row exists with entity_type == 'call_recordings'.
|
|
async with SessionLocal() as session:
|
|
run = (
|
|
await session.execute(
|
|
sa.select(SyncRun).where(
|
|
SyncRun.entity_type == ENTITY_CALL_RECORDINGS
|
|
)
|
|
)
|
|
).scalar_one()
|
|
assert run.status == "success"
|
|
assert run.records_fetched == 1
|
|
assert run.records_upserted == 1
|
|
assert run.error is None
|
|
|
|
# The call's record_media_hash got set (download happened).
|
|
call = (
|
|
await session.execute(
|
|
sa.select(PhoneCall).where(PhoneCall.call_id == "c1")
|
|
)
|
|
).scalar_one()
|
|
assert call.record_media_hash == expected_sha
|
|
finally:
|
|
await client.aclose()
|
|
await engine.dispose()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# sync_call_recordings — failure path (downloader raises)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_sync_call_recordings_records_failed_run_and_does_not_raise(
|
|
monkeypatch, tmp_path
|
|
):
|
|
"""When the downloader raises, the orchestrator records a failed SyncRun
|
|
(``entity_type='call_recordings'``) and returns a failed SyncResult —
|
|
matching the failure policy of every other ``sync_*`` function. The
|
|
exception does NOT escape, so a multi-entity batch keeps going.
|
|
"""
|
|
store = ObjectStore(tmp_path)
|
|
engine = await _make_engine()
|
|
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
|
|
|
|
async def boom(client, store, session, *, batch_size=50):
|
|
raise RuntimeError("oss exploded")
|
|
|
|
monkeypatch.setattr(orchestrator, "download_call_recordings", boom)
|
|
|
|
async with SessionLocal() as session:
|
|
# Seed a candidate so the downloader WOULD have work (it raises before
|
|
# touching rows, but its presence makes the test realistic).
|
|
session.add(
|
|
PhoneCall(call_id="c2", channel="voice_call", record_file=OSS_MP3)
|
|
)
|
|
await session.commit()
|
|
|
|
async with SessionLocal() as session:
|
|
result = await sync_call_recordings(
|
|
session, client=_primed_client(), store=store
|
|
)
|
|
|
|
assert isinstance(result, SyncResult)
|
|
assert result.entity == ENTITY_CALL_RECORDINGS
|
|
assert result.status == "failed"
|
|
assert result.fetched == 0
|
|
assert result.upserted == 0
|
|
assert "oss exploded" in (result.error or "")
|
|
|
|
async with SessionLocal() as session:
|
|
run = (
|
|
await session.execute(
|
|
sa.select(SyncRun).where(
|
|
SyncRun.entity_type == ENTITY_CALL_RECORDINGS
|
|
)
|
|
)
|
|
).scalar_one()
|
|
assert run.status == "failed"
|
|
assert run.records_fetched == 0
|
|
assert run.records_upserted == 0
|
|
assert "oss exploded" in (run.error or "")
|
|
assert run.finished_at is not None
|