"""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()