echo-core/tests/test_voice_session_cleanup.py

"""Cleanup-path tests for ``src/voice/pipeline.py::VoiceSession``.

Pins the centralized ``cleanup()`` contract from the voice plan
(Engineering decision #5): every one of the FIVE exit paths must drain
state cleanly and idempotently — lock released, JSONL flushed or
discarded, presence cleared, ``voice_client.cleanup()`` invoked,
``ttsq.stop()`` invoked, and a second call to ``cleanup()`` MUST be a
no-op (side effects happen exactly once).

The 5 paths under test:
    1. ``test_cleanup_on_voice_leave``           — explicit ``/voice leave``
    2. ``test_cleanup_on_disconnect``            — Discord-level disconnect
    3. ``test_cleanup_on_crash``                 — exception via ``__exit__``
    4. ``test_cleanup_on_auto_leave``            — 5-min inactivity timer
    5. ``test_cleanup_on_user_leaves_channel``   — user leaves voice channel
"""
from __future__ import annotations

import json
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock

import pytest

from src.voice.pipeline import VoiceSession


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture
def mock_bot():
    bot = MagicMock(name="bot")
    bot.user = MagicMock()
    bot.user.id = 999_999
    bot.change_presence = AsyncMock(name="change_presence")
    bot.get_user = MagicMock(return_value=None)
    return bot


@pytest.fixture
def mock_voice_client():
    vc = MagicMock(name="voice_client")
    vc.cleanup = MagicMock(name="vc_cleanup")
    return vc


@pytest.fixture
def mock_ttsq():
    ttsq = MagicMock(name="ttsq")
    ttsq.stop = MagicMock(name="ttsq_stop")
    return ttsq


@pytest.fixture
def mock_text_channel():
    tc = MagicMock(name="text_channel")
    tc.send = AsyncMock(name="text_send")
    return tc


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _make_session(
    tmp_path: Path,
    mock_bot,
    mock_voice_client,
    mock_ttsq,
    mock_text_channel,
    *,
    record_enabled: bool = True,
) -> VoiceSession:
    jsonl = tmp_path / ("transcripts.jsonl" if record_enabled else "noop.jsonl")
    # mock_text_channel kept resolvable via bot.get_channel for any future
    # send invocations; cleanup tests don't exercise mirror, just attribute.
    mock_bot.get_channel = MagicMock(return_value=mock_text_channel)
    return VoiceSession(
        text_channel_id=1001,
        voice_channel_id=2002,
        guild_id=42,
        voice_client=mock_voice_client,
        bot=mock_bot,
        ttsq=mock_ttsq,
        whitelist={1234},
        record_enabled=record_enabled,
        mirror_enabled=True,
        transcripts_jsonl_path=jsonl,
        loop=None,
        router_route_message=MagicMock(name="route_message"),
    )


def _assert_clean_post_cleanup(
    session: VoiceSession,
    voice_client,
    ttsq,
    bot,
    jsonl_path: Path,
    record_enabled: bool,
) -> None:
    """Assertions shared across all five cleanup-path tests."""
    # 1. Lock released — non-blocking acquire from this thread returns True.
    acquired = session._lock.acquire(blocking=False)
    assert acquired, "session._lock must be released after cleanup()"
    session._lock.release()

    # 2. voice_client.cleanup() called exactly once.
    assert voice_client.cleanup.call_count == 1, (
        f"voice_client.cleanup() called {voice_client.cleanup.call_count}x, "
        f"expected 1"
    )

    # 3. ttsq.stop() called exactly once.
    assert ttsq.stop.call_count == 1, (
        f"ttsq.stop() called {ttsq.stop.call_count}x, expected 1"
    )

    # 4. bot.change_presence(activity=None) called at least once with that kwarg.
    assert bot.change_presence.call_count >= 1, (
        "bot.change_presence was never called — presence not restored"
    )
    bot.change_presence.assert_called_with(activity=None)

    # 5. JSONL flushed (record=on) OR absent (record=off).
    if record_enabled:
        assert jsonl_path.exists(), (
            "record=on: JSONL file must exist (was created by __enter__ and "
            "left in place by cleanup so transcript can be persisted)"
        )
    else:
        # record=off: cleanup unlinks the file if it ever existed.
        assert not jsonl_path.exists() or jsonl_path.stat().st_size == 0


# ---------------------------------------------------------------------------
# Scenario 1 — explicit /voice leave
# ---------------------------------------------------------------------------


class TestCleanupOnVoiceLeave:
    def test_cleanup_on_voice_leave(
        self, tmp_path, mock_bot, mock_voice_client, mock_ttsq, mock_text_channel,
    ):
        session = _make_session(
            tmp_path, mock_bot, mock_voice_client, mock_ttsq, mock_text_channel,
            record_enabled=True,
        )
        jsonl_path = session.transcripts_jsonl_path

        with session:
            # Simulate one transcript line.
            session._jsonl_fh.write(json.dumps({"text": "salut"}) + "\n")
            session.cleanup("voice_leave")
            assert session._cleaned_up is True

        # __exit__ called cleanup("exit") — must be a no-op the second time.
        _assert_clean_post_cleanup(
            session, mock_voice_client, mock_ttsq, mock_bot,
            jsonl_path, record_enabled=True,
        )

        # Idempotency: a third explicit call still doesn't bump counts.
        session.cleanup("redundant")
        assert mock_voice_client.cleanup.call_count == 1
        assert mock_ttsq.stop.call_count == 1


# ---------------------------------------------------------------------------
# Scenario 2 — Discord-level voice disconnect
# ---------------------------------------------------------------------------


class TestCleanupOnDisconnect:
    def test_cleanup_on_disconnect(
        self, tmp_path, mock_bot, mock_voice_client, mock_ttsq, mock_text_channel,
    ):
        session = _make_session(
            tmp_path, mock_bot, mock_voice_client, mock_ttsq, mock_text_channel,
            record_enabled=False,
        )
        jsonl_path = session.transcripts_jsonl_path

        session.__enter__()
        # Network drop arrives outside the with-block.
        session.cleanup("disconnect")
        _assert_clean_post_cleanup(
            session, mock_voice_client, mock_ttsq, mock_bot,
            jsonl_path, record_enabled=False,
        )

        # Idempotency.
        session.cleanup("disconnect-again")
        assert mock_voice_client.cleanup.call_count == 1
        assert mock_ttsq.stop.call_count == 1


# ---------------------------------------------------------------------------
# Scenario 3 — crash / exception via __exit__
# ---------------------------------------------------------------------------


class TestCleanupOnCrash:
    def test_cleanup_on_crash(
        self, tmp_path, mock_bot, mock_voice_client, mock_ttsq, mock_text_channel,
    ):
        session = _make_session(
            tmp_path, mock_bot, mock_voice_client, mock_ttsq, mock_text_channel,
            record_enabled=True,
        )
        jsonl_path = session.transcripts_jsonl_path

        with pytest.raises(RuntimeError, match="simulated crash"):
            with session:
                # Pipeline raises mid-call.
                raise RuntimeError("simulated crash")

        # __exit__ must have driven cleanup — every side effect happened once.
        _assert_clean_post_cleanup(
            session, mock_voice_client, mock_ttsq, mock_bot,
            jsonl_path, record_enabled=True,
        )

        # Idempotency: explicit follow-up call (e.g. an outer error handler
        # also tries to cleanup) MUST be a no-op.
        session.cleanup("post-crash")
        assert mock_voice_client.cleanup.call_count == 1
        assert mock_ttsq.stop.call_count == 1


# ---------------------------------------------------------------------------
# Scenario 4 — auto-leave timer fires after 5 min inactivity
# ---------------------------------------------------------------------------


class TestCleanupOnAutoLeave:
    def test_cleanup_on_auto_leave(
        self, tmp_path, mock_bot, mock_voice_client, mock_ttsq, mock_text_channel,
    ):
        session = _make_session(
            tmp_path, mock_bot, mock_voice_client, mock_ttsq, mock_text_channel,
            record_enabled=True,
        )
        jsonl_path = session.transcripts_jsonl_path

        session.__enter__()
        # The auto-leave timer trips outside the with-block.
        session.cleanup("auto_leave")

        _assert_clean_post_cleanup(
            session, mock_voice_client, mock_ttsq, mock_bot,
            jsonl_path, record_enabled=True,
        )

        # Idempotency.
        session.cleanup("auto_leave_redundant")
        assert mock_voice_client.cleanup.call_count == 1
        assert mock_ttsq.stop.call_count == 1


# ---------------------------------------------------------------------------
# Scenario 5 — user leaves voice channel themselves
# ---------------------------------------------------------------------------


class TestCleanupOnUserLeaves:
    def test_cleanup_on_user_leaves_channel(
        self, tmp_path, mock_bot, mock_voice_client, mock_ttsq, mock_text_channel,
    ):
        session = _make_session(
            tmp_path, mock_bot, mock_voice_client, mock_ttsq, mock_text_channel,
            record_enabled=False,
        )
        jsonl_path = session.transcripts_jsonl_path

        session.__enter__()
        # voice_state_update event handler invokes cleanup directly.
        session.cleanup("user_left_channel")

        _assert_clean_post_cleanup(
            session, mock_voice_client, mock_ttsq, mock_bot,
            jsonl_path, record_enabled=False,
        )

        # Idempotency.
        session.cleanup("user_left_again")
        assert mock_voice_client.cleanup.call_count == 1
        assert mock_ttsq.stop.call_count == 1


# ---------------------------------------------------------------------------
# Cross-cutting: failures inside cleanup don't propagate
# ---------------------------------------------------------------------------


class TestCleanupRobustness:
    def test_cleanup_swallows_voice_client_errors(
        self, tmp_path, mock_bot, mock_voice_client, mock_ttsq, mock_text_channel,
    ):
        """If voice_client.cleanup() raises, ttsq.stop() must still run and
        the lock must still release — otherwise a broken Discord state would
        deadlock the channel forever."""
        mock_voice_client.cleanup.side_effect = RuntimeError("vc died")

        session = _make_session(
            tmp_path, mock_bot, mock_voice_client, mock_ttsq, mock_text_channel,
            record_enabled=False,
        )

        with session:
            session.cleanup("voice_leave")

        # ttsq.stop still ran exactly once.
        assert mock_ttsq.stop.call_count == 1
        # Lock released.
        acquired = session._lock.acquire(blocking=False)
        assert acquired, "lock must release even when voice_client.cleanup raises"
        session._lock.release()