670 lines
24 KiB
Python
670 lines
24 KiB
Python
"""Интеграционные тесты `/api/v1/livekit/webhook`: полный цикл, дедуп, fallback,
|
||
невалидная подпись, пропуск невалидного identity, гостевой identity, жизненный
|
||
цикл закреплённой/незакреплённой конференции — на фикстурах payload'ов LiveKit.
|
||
"""
|
||
|
||
import base64
|
||
import hashlib
|
||
import uuid
|
||
from datetime import UTC, datetime, timedelta
|
||
from pathlib import Path
|
||
from unittest.mock import AsyncMock, Mock
|
||
|
||
import httpx
|
||
import jwt
|
||
import pytest
|
||
from sqlalchemy import select
|
||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
import services.webhook_handlers as webhook_handlers_module
|
||
from core.config import get_settings
|
||
from core.security import hash_password
|
||
from models.audio_track import SessionAudioTrack
|
||
from models.conference import Conference
|
||
from models.guest import GuestAccess
|
||
from models.instance_setting import InstanceSetting
|
||
from models.participant import ConferenceParticipant
|
||
from models.session import ConferenceSession
|
||
from models.user import User
|
||
from services.conference_ids import generate_number, generate_slug
|
||
from services.egress import EgressStartResult
|
||
|
||
FIXTURES_DIR = Path(__file__).parent / "fixtures" / "livekit"
|
||
WEBHOOK_URL = "/api/v1/livekit/webhook"
|
||
|
||
|
||
def _load_fixture(name: str, **placeholders: str) -> bytes:
|
||
"""Загрузить JSON-фикстуру webhook-события, подставив плейсхолдеры `{name}`."""
|
||
raw = (FIXTURES_DIR / name).read_text()
|
||
for key, value in placeholders.items():
|
||
raw = raw.replace(f"{{{key}}}", value)
|
||
return raw.encode()
|
||
|
||
|
||
def _sign(body: bytes) -> str:
|
||
"""Подписать тело webhook-запроса так же, как это делает LiveKit server."""
|
||
settings = get_settings()
|
||
sha256_b64 = base64.b64encode(hashlib.sha256(body).digest()).decode()
|
||
payload = {
|
||
"iss": settings.livekit_api_key,
|
||
"sha256": sha256_b64,
|
||
"exp": datetime.now(UTC) + timedelta(minutes=5),
|
||
}
|
||
return jwt.encode(payload, settings.livekit_api_secret, algorithm="HS256")
|
||
|
||
|
||
async def _post_webhook(client: httpx.AsyncClient, body: bytes) -> httpx.Response:
|
||
return await client.post(
|
||
WEBHOOK_URL,
|
||
content=body,
|
||
headers={"Authorization": _sign(body), "Content-Type": "application/json"},
|
||
)
|
||
|
||
|
||
async def _make_conference(
|
||
session: AsyncSession, slug: str, *, is_pinned: bool = False
|
||
) -> Conference:
|
||
conference = Conference(
|
||
number=generate_number(), slug=slug, title=slug, is_pinned=is_pinned, status="scheduled"
|
||
)
|
||
session.add(conference)
|
||
await session.flush()
|
||
return conference
|
||
|
||
|
||
async def _make_user(session: AsyncSession, email: str) -> User:
|
||
user = User(
|
||
email=email,
|
||
name_user="Participant",
|
||
password_hash=hash_password("password123"),
|
||
email_verified=True,
|
||
)
|
||
session.add(user)
|
||
await session.flush()
|
||
return user
|
||
|
||
|
||
async def _make_guest(session: AsyncSession, conference: Conference) -> GuestAccess:
|
||
guest = GuestAccess(
|
||
conference_id=conference.id, display_name="Guest Tester", email="guest-track@example.com"
|
||
)
|
||
session.add(guest)
|
||
await session.flush()
|
||
return guest
|
||
|
||
|
||
async def test_full_cycle_joined_left_finished(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
conference = await _make_conference(db_session, generate_slug())
|
||
user = await _make_user(db_session, "webhook-user-1@example.com")
|
||
await db_session.commit()
|
||
|
||
started = _load_fixture(
|
||
"room_started.json", event_id=f"evt-{uuid.uuid4()}", room_name=conference.slug
|
||
)
|
||
resp = await _post_webhook(client, started)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["status"] == "ok"
|
||
|
||
await db_session.refresh(conference)
|
||
assert conference.status == "active"
|
||
|
||
session_record = await db_session.scalar(
|
||
select(ConferenceSession).where(
|
||
ConferenceSession.conference_id == conference.id, ConferenceSession.t_end.is_(None)
|
||
)
|
||
)
|
||
assert session_record is not None
|
||
assert session_record.t_start is not None
|
||
|
||
joined = _load_fixture(
|
||
"participant_joined.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
identity=str(user.id),
|
||
)
|
||
resp = await _post_webhook(client, joined)
|
||
assert resp.status_code == 200
|
||
|
||
participant = await db_session.scalar(
|
||
select(ConferenceParticipant).where(
|
||
ConferenceParticipant.session_id == session_record.id,
|
||
ConferenceParticipant.user_id == user.id,
|
||
)
|
||
)
|
||
assert participant is not None
|
||
assert participant.joined_at is not None
|
||
assert participant.left_at is None
|
||
assert participant.guest_id is None
|
||
|
||
left = _load_fixture(
|
||
"participant_left.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
identity=str(user.id),
|
||
)
|
||
resp = await _post_webhook(client, left)
|
||
assert resp.status_code == 200
|
||
|
||
await db_session.refresh(participant)
|
||
assert participant.left_at is not None
|
||
|
||
finished = _load_fixture(
|
||
"room_finished.json", event_id=f"evt-{uuid.uuid4()}", room_name=conference.slug
|
||
)
|
||
resp = await _post_webhook(client, finished)
|
||
assert resp.status_code == 200
|
||
|
||
await db_session.refresh(session_record)
|
||
assert session_record.t_end is not None
|
||
await db_session.refresh(conference)
|
||
assert conference.status == "ended"
|
||
assert conference.ended_at is not None
|
||
|
||
|
||
async def test_pinned_conference_returns_to_scheduled_on_finish(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
conference = await _make_conference(db_session, generate_slug(), is_pinned=True)
|
||
await db_session.commit()
|
||
|
||
started = _load_fixture(
|
||
"room_started.json", event_id=f"evt-{uuid.uuid4()}", room_name=conference.slug
|
||
)
|
||
assert (await _post_webhook(client, started)).status_code == 200
|
||
|
||
finished = _load_fixture(
|
||
"room_finished.json", event_id=f"evt-{uuid.uuid4()}", room_name=conference.slug
|
||
)
|
||
resp = await _post_webhook(client, finished)
|
||
assert resp.status_code == 200
|
||
|
||
await db_session.refresh(conference)
|
||
assert conference.status == "scheduled"
|
||
assert conference.ended_at is None
|
||
|
||
|
||
async def test_guest_identity_creates_participant_with_guest_id(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
conference = await _make_conference(db_session, generate_slug())
|
||
guest = GuestAccess(
|
||
conference_id=conference.id, display_name="Guest Tester", email="guest@example.com"
|
||
)
|
||
db_session.add(guest)
|
||
await db_session.commit()
|
||
|
||
joined = _load_fixture(
|
||
"participant_joined.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
identity=f"guest:{guest.id}",
|
||
)
|
||
resp = await _post_webhook(client, joined)
|
||
assert resp.status_code == 200
|
||
|
||
session_record = await db_session.scalar(
|
||
select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
|
||
)
|
||
assert session_record is not None
|
||
participant = await db_session.scalar(
|
||
select(ConferenceParticipant).where(
|
||
ConferenceParticipant.session_id == session_record.id,
|
||
ConferenceParticipant.guest_id == guest.id,
|
||
)
|
||
)
|
||
assert participant is not None
|
||
assert participant.user_id is None
|
||
|
||
|
||
async def test_duplicate_event_is_deduplicated(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
conference = await _make_conference(db_session, generate_slug())
|
||
await db_session.commit()
|
||
|
||
event_id = f"evt-{uuid.uuid4()}"
|
||
body = _load_fixture("room_started.json", event_id=event_id, room_name=conference.slug)
|
||
|
||
first = await _post_webhook(client, body)
|
||
assert first.status_code == 200
|
||
assert first.json()["status"] == "ok"
|
||
|
||
second = await _post_webhook(client, body)
|
||
assert second.status_code == 200
|
||
assert second.json()["status"] == "duplicate"
|
||
|
||
sessions = (
|
||
await db_session.scalars(
|
||
select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
|
||
)
|
||
).all()
|
||
assert len(sessions) == 1
|
||
|
||
|
||
async def test_missing_room_started_falls_back_on_participant_joined(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
conference = await _make_conference(db_session, generate_slug())
|
||
user = await _make_user(db_session, "webhook-user-fallback@example.com")
|
||
await db_session.commit()
|
||
|
||
joined = _load_fixture(
|
||
"participant_joined.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
identity=str(user.id),
|
||
)
|
||
resp = await _post_webhook(client, joined)
|
||
assert resp.status_code == 200
|
||
|
||
session_record = await db_session.scalar(
|
||
select(ConferenceSession).where(
|
||
ConferenceSession.conference_id == conference.id, ConferenceSession.t_end.is_(None)
|
||
)
|
||
)
|
||
assert session_record is not None
|
||
|
||
participant = await db_session.scalar(
|
||
select(ConferenceParticipant).where(
|
||
ConferenceParticipant.session_id == session_record.id,
|
||
ConferenceParticipant.user_id == user.id,
|
||
)
|
||
)
|
||
assert participant is not None
|
||
|
||
|
||
async def test_invalid_signature_returns_401(client: httpx.AsyncClient) -> None:
|
||
body = _load_fixture(
|
||
"room_started.json", event_id=f"evt-{uuid.uuid4()}", room_name="does-not-matter"
|
||
)
|
||
resp = await client.post(
|
||
WEBHOOK_URL,
|
||
content=body,
|
||
headers={"Authorization": "not-a-valid-jwt", "Content-Type": "application/json"},
|
||
)
|
||
assert resp.status_code == 401
|
||
|
||
|
||
async def test_unknown_identity_is_skipped_without_error(
|
||
client: httpx.AsyncClient, db_session: AsyncSession
|
||
) -> None:
|
||
conference = await _make_conference(db_session, generate_slug())
|
||
await db_session.commit()
|
||
|
||
joined = _load_fixture(
|
||
"participant_joined.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
identity="not-a-uuid",
|
||
)
|
||
resp = await _post_webhook(client, joined)
|
||
assert resp.status_code == 200
|
||
|
||
# Невалидный identity парсится раньше get-or-create сеанса, поэтому для
|
||
# этой конференции не должно появиться ни сеанса, ни участника.
|
||
session_record = await db_session.scalar(
|
||
select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
|
||
)
|
||
assert session_record is None
|
||
|
||
|
||
async def test_track_published_by_guest_starts_egress_and_creates_track_row(
|
||
client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""№8 плана: гость (`identity="guest:<uuid>"`) публикует аудиотрек — AC-1."""
|
||
started_at = datetime.now(UTC)
|
||
mock_start = AsyncMock(
|
||
return_value=EgressStartResult(egress_id="EG_guest_track", started_at=started_at)
|
||
)
|
||
monkeypatch.setattr(webhook_handlers_module, "start_track_egress", mock_start)
|
||
|
||
conference = await _make_conference(db_session, generate_slug())
|
||
guest = await _make_guest(db_session, conference)
|
||
await db_session.commit()
|
||
|
||
joined = _load_fixture(
|
||
"participant_joined.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
identity=f"guest:{guest.id}",
|
||
)
|
||
assert (await _post_webhook(client, joined)).status_code == 200
|
||
|
||
session_record = await db_session.scalar(
|
||
select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
|
||
)
|
||
assert session_record is not None
|
||
participant = await db_session.scalar(
|
||
select(ConferenceParticipant).where(
|
||
ConferenceParticipant.session_id == session_record.id,
|
||
ConferenceParticipant.guest_id == guest.id,
|
||
)
|
||
)
|
||
assert participant is not None
|
||
|
||
track_sid = f"TR_{uuid.uuid4().hex[:8]}"
|
||
published = _load_fixture(
|
||
"track_published.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
identity=f"guest:{guest.id}",
|
||
track_sid=track_sid,
|
||
)
|
||
resp = await _post_webhook(client, published)
|
||
assert resp.status_code == 200
|
||
|
||
settings = get_settings()
|
||
expected_filepath = (
|
||
f"{settings.recordings_dir}/{session_record.id}/{participant.id}_{track_sid}.ogg"
|
||
)
|
||
mock_start.assert_awaited_once_with(conference.slug, track_sid, expected_filepath)
|
||
|
||
track_row = await db_session.scalar(
|
||
select(SessionAudioTrack).where(
|
||
SessionAudioTrack.session_id == session_record.id,
|
||
SessionAudioTrack.track_sid == track_sid,
|
||
)
|
||
)
|
||
assert track_row is not None
|
||
assert track_row.participant_id == participant.id
|
||
assert participant.guest_id == guest.id
|
||
assert participant.user_id is None
|
||
assert track_row.egress_id == "EG_guest_track"
|
||
assert track_row.file_path == expected_filepath
|
||
assert track_row.status == "recording"
|
||
|
||
|
||
async def test_track_published_survives_egress_unavailable(
|
||
client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Недоступность egress не должна ронять webhook (блок D): 200 + warning, без строки трека."""
|
||
mock_start = AsyncMock(side_effect=RuntimeError("egress service unavailable"))
|
||
monkeypatch.setattr(webhook_handlers_module, "start_track_egress", mock_start)
|
||
|
||
conference = await _make_conference(db_session, generate_slug())
|
||
user = await _make_user(db_session, "webhook-track-egress-down@example.com")
|
||
await db_session.commit()
|
||
|
||
joined = _load_fixture(
|
||
"participant_joined.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
identity=str(user.id),
|
||
)
|
||
assert (await _post_webhook(client, joined)).status_code == 200
|
||
|
||
track_sid = "TR_egress_down"
|
||
published = _load_fixture(
|
||
"track_published.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
identity=str(user.id),
|
||
track_sid=track_sid,
|
||
)
|
||
resp = await _post_webhook(client, published)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["status"] == "ok"
|
||
|
||
mock_start.assert_awaited_once()
|
||
tracks = (
|
||
await db_session.scalars(
|
||
select(SessionAudioTrack).where(SessionAudioTrack.track_sid == track_sid)
|
||
)
|
||
).all()
|
||
assert tracks == []
|
||
|
||
|
||
async def test_track_published_video_is_noop(
|
||
client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""№9 плана (часть 1): video-трек — no-op, egress не запускается."""
|
||
mock_start = AsyncMock()
|
||
monkeypatch.setattr(webhook_handlers_module, "start_track_egress", mock_start)
|
||
|
||
conference = await _make_conference(db_session, generate_slug())
|
||
user = await _make_user(db_session, "webhook-track-video@example.com")
|
||
await db_session.commit()
|
||
|
||
joined = _load_fixture(
|
||
"participant_joined.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
identity=str(user.id),
|
||
)
|
||
assert (await _post_webhook(client, joined)).status_code == 200
|
||
|
||
published = _load_fixture(
|
||
"track_published_video.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
identity=str(user.id),
|
||
track_sid="TR_video",
|
||
)
|
||
resp = await _post_webhook(client, published)
|
||
assert resp.status_code == 200
|
||
|
||
mock_start.assert_not_awaited()
|
||
tracks = (
|
||
await db_session.scalars(
|
||
select(SessionAudioTrack).where(SessionAudioTrack.track_sid == "TR_video")
|
||
)
|
||
).all()
|
||
assert tracks == []
|
||
|
||
|
||
async def test_track_published_repeated_webhook_creates_single_row(
|
||
client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""№9 плана (часть 2): повторный `track_published` того же трека → одна строка."""
|
||
mock_start = AsyncMock(
|
||
return_value=EgressStartResult(egress_id="EG_repeat", started_at=datetime.now(UTC))
|
||
)
|
||
monkeypatch.setattr(webhook_handlers_module, "start_track_egress", mock_start)
|
||
|
||
conference = await _make_conference(db_session, generate_slug())
|
||
user = await _make_user(db_session, "webhook-track-repeat@example.com")
|
||
await db_session.commit()
|
||
|
||
joined = _load_fixture(
|
||
"participant_joined.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
identity=str(user.id),
|
||
)
|
||
assert (await _post_webhook(client, joined)).status_code == 200
|
||
|
||
track_sid = "TR_repeat"
|
||
for _ in range(2):
|
||
published = _load_fixture(
|
||
"track_published.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
identity=str(user.id),
|
||
track_sid=track_sid,
|
||
)
|
||
resp = await _post_webhook(client, published)
|
||
assert resp.status_code == 200
|
||
|
||
session_record = await db_session.scalar(
|
||
select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
|
||
)
|
||
assert session_record is not None
|
||
tracks = (
|
||
await db_session.scalars(
|
||
select(SessionAudioTrack).where(
|
||
SessionAudioTrack.session_id == session_record.id,
|
||
SessionAudioTrack.track_sid == track_sid,
|
||
)
|
||
)
|
||
).all()
|
||
assert len(tracks) == 1
|
||
mock_start.assert_awaited_once()
|
||
|
||
|
||
async def test_egress_ended_finalizes_track_success_and_failure(
|
||
client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""№10 плана (часть 1): `egress_ended` — 'recorded' на успехе, 'failed' на ошибке."""
|
||
conference = await _make_conference(db_session, generate_slug())
|
||
user = await _make_user(db_session, "webhook-egress-ended@example.com")
|
||
await db_session.commit()
|
||
|
||
joined = _load_fixture(
|
||
"participant_joined.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
identity=str(user.id),
|
||
)
|
||
assert (await _post_webhook(client, joined)).status_code == 200
|
||
session_record = await db_session.scalar(
|
||
select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
|
||
)
|
||
assert session_record is not None
|
||
|
||
# Успешная запись.
|
||
ok_started = datetime.now(UTC)
|
||
monkeypatch.setattr(
|
||
webhook_handlers_module,
|
||
"start_track_egress",
|
||
AsyncMock(return_value=EgressStartResult(egress_id="EG_ok", started_at=ok_started)),
|
||
)
|
||
published_ok = _load_fixture(
|
||
"track_published.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
identity=str(user.id),
|
||
track_sid="TR_ok",
|
||
)
|
||
assert (await _post_webhook(client, published_ok)).status_code == 200
|
||
|
||
ended_ok = _load_fixture(
|
||
"egress_ended.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
egress_id="EG_ok",
|
||
file_path=f"/recordings/{session_record.id}/track-ok.ogg",
|
||
)
|
||
assert (await _post_webhook(client, ended_ok)).status_code == 200
|
||
|
||
ok_track = await db_session.scalar(
|
||
select(SessionAudioTrack).where(SessionAudioTrack.egress_id == "EG_ok")
|
||
)
|
||
assert ok_track is not None
|
||
assert ok_track.status == "recorded"
|
||
assert ok_track.ended_at is not None
|
||
assert ok_track.file_path == f"/recordings/{session_record.id}/track-ok.ogg"
|
||
|
||
# Ошибка записи.
|
||
monkeypatch.setattr(
|
||
webhook_handlers_module,
|
||
"start_track_egress",
|
||
AsyncMock(
|
||
return_value=EgressStartResult(egress_id="EG_fail", started_at=datetime.now(UTC))
|
||
),
|
||
)
|
||
published_fail = _load_fixture(
|
||
"track_published.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
identity=str(user.id),
|
||
track_sid="TR_fail",
|
||
)
|
||
assert (await _post_webhook(client, published_fail)).status_code == 200
|
||
|
||
ended_failed = _load_fixture(
|
||
"egress_ended_failed.json",
|
||
event_id=f"evt-{uuid.uuid4()}",
|
||
room_name=conference.slug,
|
||
egress_id="EG_fail",
|
||
)
|
||
assert (await _post_webhook(client, ended_failed)).status_code == 200
|
||
|
||
failed_track = await db_session.scalar(
|
||
select(SessionAudioTrack).where(SessionAudioTrack.egress_id == "EG_fail")
|
||
)
|
||
assert failed_track is not None
|
||
assert failed_track.status == "failed"
|
||
assert failed_track.ended_at is not None
|
||
|
||
|
||
async def test_room_finished_enqueues_pipeline(
|
||
client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""№10 плана (часть 2): `room_finished` ставит `run_pipeline` в очередь (мок)."""
|
||
mock_enqueue = Mock()
|
||
monkeypatch.setattr(webhook_handlers_module, "enqueue_pipeline", mock_enqueue)
|
||
|
||
conference = await _make_conference(db_session, generate_slug())
|
||
await db_session.commit()
|
||
|
||
started = _load_fixture(
|
||
"room_started.json", event_id=f"evt-{uuid.uuid4()}", room_name=conference.slug
|
||
)
|
||
assert (await _post_webhook(client, started)).status_code == 200
|
||
|
||
session_record = await db_session.scalar(
|
||
select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
|
||
)
|
||
assert session_record is not None
|
||
|
||
finished = _load_fixture(
|
||
"room_finished.json", event_id=f"evt-{uuid.uuid4()}", room_name=conference.slug
|
||
)
|
||
assert (await _post_webhook(client, finished)).status_code == 200
|
||
|
||
mock_enqueue.assert_called_once_with(session_record.id)
|
||
|
||
|
||
async def test_room_finished_skips_enqueue_when_transcription_disabled(
|
||
client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Guard от зависания сеанса: транскрибация
|
||
выключена в настройках инстанса (пресеты 1/2 инсталлятора — без AI) →
|
||
`room_finished` НЕ ставит `run_pipeline` в очередь `transcription` (некому
|
||
обслужить), сеанс сразу переводится в терминальный `pipeline_status='notified'`
|
||
вместо зависания в `recording` навсегда."""
|
||
mock_enqueue = Mock()
|
||
monkeypatch.setattr(webhook_handlers_module, "enqueue_pipeline", mock_enqueue)
|
||
|
||
# Настройка `instance_settings.transcriber` пишется тем же `db_session`
|
||
# (savepoint), что и обработчик webhook (подмена `get_session` в фикстуре
|
||
# `app`) — видна обработчику без реального коммита в dev-БД.
|
||
disabled_transcriber = {
|
||
"enabled": False,
|
||
"provider": "null",
|
||
"model": None,
|
||
"language": "ru",
|
||
"options": {},
|
||
}
|
||
stmt = (
|
||
pg_insert(InstanceSetting)
|
||
.values(key="transcriber", value=disabled_transcriber)
|
||
.on_conflict_do_update(index_elements=["key"], set_={"value": disabled_transcriber})
|
||
)
|
||
await db_session.execute(stmt)
|
||
|
||
conference = await _make_conference(db_session, generate_slug())
|
||
await db_session.commit()
|
||
|
||
started = _load_fixture(
|
||
"room_started.json", event_id=f"evt-{uuid.uuid4()}", room_name=conference.slug
|
||
)
|
||
assert (await _post_webhook(client, started)).status_code == 200
|
||
|
||
session_record = await db_session.scalar(
|
||
select(ConferenceSession).where(ConferenceSession.conference_id == conference.id)
|
||
)
|
||
assert session_record is not None
|
||
|
||
finished = _load_fixture(
|
||
"room_finished.json", event_id=f"evt-{uuid.uuid4()}", room_name=conference.slug
|
||
)
|
||
assert (await _post_webhook(client, finished)).status_code == 200
|
||
|
||
mock_enqueue.assert_not_called()
|
||
await db_session.refresh(session_record)
|
||
assert session_record.pipeline_status == "notified"
|