vidconf/backend/tests/test_mute_participant_api.py

"""Тесты `POST /api/v1/conferences/{id}/mute-participant` (задача B2).

`mute_participant_track` (реальный вызов LiveKit `RoomServiceClient`) мокается
на уровне `services.conferences` — тот же паттерн, что и `start_track_egress`
в `tests/test_livekit_webhook.py`: сетевой вызов к LiveKit в тестах не нужен,
важна только бизнес-логика (права, маршрутизация ошибок, broadcast).
"""

import asyncio
import uuid
from typing import Any
from unittest.mock import AsyncMock

import httpx
import pytest
from redis.asyncio.client import PubSub
from sqlalchemy.ext.asyncio import AsyncSession

import services.conferences as conferences_module
from core.redis import redis_client
from core.security import create_access_token, hash_password
from models.conference import Conference
from models.user import User
from services.conference_ids import generate_number, generate_slug
from services.hand_queue import hand_queue_channel
from services.room_control import ParticipantNotInRoomError

MUTE_URL = "{base}/mute-participant"


async def _receive_within(pubsub: PubSub, *, max_wait: float) -> dict[str, Any] | None:
    """Дождаться СОДЕРЖАТЕЛЬНОГО сообщения канала в пределах `max_wait` секунд.

    `ignore_subscribe_messages=True` у `get_message` фильтрует служебное
    подтверждение подписки, но при этом всё равно может вернуть `None` для
    ЭТОГО конкретного вызова (см. `api/chat.py::_pump_pubsub_to_websocket`,
    ровно поэтому там `while True: ... if raw is None: continue`) — здесь тот
    же цикл, но с общим дедлайном вместо бесконечного ожидания.
    """
    deadline = asyncio.get_event_loop().time() + max_wait
    while True:
        remaining = deadline - asyncio.get_event_loop().time()
        if remaining <= 0:
            return None
        raw: dict[str, Any] | None = await pubsub.get_message(
            ignore_subscribe_messages=True, timeout=remaining
        )
        if raw is not None:
            return raw


async def _make_user(session: AsyncSession, *, role: str = "user") -> User:
    user = User(
        email=f"{uuid.uuid4()}@example.com",
        name_user="Mute Tester",
        password_hash=await hash_password("password123"),
        email_verified=True,
        role=role,
    )
    session.add(user)
    await session.flush()
    return user


async def _make_conference(
    session: AsyncSession, *, owner_id: uuid.UUID | None = None
) -> Conference:
    conference = Conference(
        number=generate_number(),
        slug=generate_slug(),
        title="Mute Test",
        status="active",
        owner_id=owner_id,
    )
    session.add(conference)
    await session.flush()
    return conference


def _auth_headers(user: User) -> dict[str, str]:
    return {"Authorization": f"Bearer {create_access_token(user.id, user.role)}"}


def _url(conference_id: uuid.UUID) -> str:
    return MUTE_URL.format(base=f"/api/v1/conferences/{conference_id}")


async def test_owner_can_mute_participant_and_broadcast_is_published(
    client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    owner = await _make_user(db_session)
    conference = await _make_conference(db_session, owner_id=owner.id)
    await db_session.commit()

    mock_mute = AsyncMock(return_value=True)
    monkeypatch.setattr(conferences_module, "mute_participant_track", mock_mute)

    pubsub = redis_client.pubsub()
    channel = hand_queue_channel(conference.id)
    await pubsub.subscribe(channel)
    try:
        response = await client.post(
            _url(conference.id),
            json={"identity": "some-identity", "source": "microphone"},
            headers=_auth_headers(owner),
        )
        assert response.status_code == 200, response.text
        assert response.json() == {"muted": True}
        mock_mute.assert_awaited_once_with(
            conference.slug, identity="some-identity", source="microphone"
        )

        raw = await _receive_within(pubsub, max_wait=2)
        assert raw is not None
        assert raw["data"] == (
            '{"type":"forced_mute","identity":"some-identity","source":"microphone"}'
        )
    finally:
        await pubsub.unsubscribe(channel)
        await pubsub.aclose()  # type: ignore[no-untyped-call]


async def test_mute_already_off_does_not_broadcast(
    client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Трек не был опубликован (камера/мьют и так выключены, задача A1) — не ошибка.

    Ответ `muted=false` (искомое состояние уже достигнуто), без broadcast'а.
    """
    owner = await _make_user(db_session)
    conference = await _make_conference(db_session, owner_id=owner.id)
    await db_session.commit()

    monkeypatch.setattr(
        conferences_module, "mute_participant_track", AsyncMock(return_value=False)
    )

    pubsub = redis_client.pubsub()
    channel = hand_queue_channel(conference.id)
    await pubsub.subscribe(channel)
    try:
        response = await client.post(
            _url(conference.id),
            json={"identity": "some-identity", "source": "camera"},
            headers=_auth_headers(owner),
        )
        assert response.status_code == 200, response.text
        assert response.json() == {"muted": False}

        raw = await _receive_within(pubsub, max_wait=0.5)
        assert raw is None
    finally:
        await pubsub.unsubscribe(channel)
        await pubsub.aclose()  # type: ignore[no-untyped-call]


async def test_admin_can_mute_participant_of_someone_elses_conference(
    client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    owner = await _make_user(db_session)
    admin = await _make_user(db_session, role="admin")
    conference = await _make_conference(db_session, owner_id=owner.id)
    await db_session.commit()

    monkeypatch.setattr(conferences_module, "mute_participant_track", AsyncMock(return_value=True))

    response = await client.post(
        _url(conference.id),
        json={"identity": "some-identity", "source": "microphone"},
        headers=_auth_headers(admin),
    )
    assert response.status_code == 200, response.text


async def test_regular_participant_cannot_mute_someone_else(
    client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    owner = await _make_user(db_session)
    other = await _make_user(db_session)
    conference = await _make_conference(db_session, owner_id=owner.id)
    await db_session.commit()

    mock_mute = AsyncMock()
    monkeypatch.setattr(conferences_module, "mute_participant_track", mock_mute)

    response = await client.post(
        _url(conference.id),
        json={"identity": str(owner.id), "source": "microphone"},
        headers=_auth_headers(other),
    )
    assert response.status_code == 403
    assert response.json()["detail"] == "not_owner"
    mock_mute.assert_not_awaited()


async def test_mute_conference_not_found(
    client: httpx.AsyncClient, db_session: AsyncSession
) -> None:
    user = await _make_user(db_session)
    await db_session.commit()

    response = await client.post(
        _url(uuid.uuid4()),
        json={"identity": "some-identity", "source": "microphone"},
        headers=_auth_headers(user),
    )
    assert response.status_code == 404
    assert response.json()["detail"] == "conference_not_found"


async def test_mute_participant_not_in_room_returns_404(
    client: httpx.AsyncClient, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    owner = await _make_user(db_session)
    conference = await _make_conference(db_session, owner_id=owner.id)
    await db_session.commit()

    monkeypatch.setattr(
        conferences_module,
        "mute_participant_track",
        AsyncMock(side_effect=ParticipantNotInRoomError()),
    )

    response = await client.post(
        _url(conference.id),
        json={"identity": "ghost", "source": "microphone"},
        headers=_auth_headers(owner),
    )
    assert response.status_code == 404
    assert response.json()["detail"] == "participant_not_in_room"


async def test_mute_rejects_invalid_source(
    client: httpx.AsyncClient, db_session: AsyncSession
) -> None:
    owner = await _make_user(db_session)
    conference = await _make_conference(db_session, owner_id=owner.id)
    await db_session.commit()

    response = await client.post(
        _url(conference.id),
        json={"identity": "some-identity", "source": "screen_share"},
        headers=_auth_headers(owner),
    )
    assert response.status_code == 422