blurt/tests/test_app_streaming.py

"""Tray integration without a microphone, display, network, or clipboard."""
import os
os.environ.setdefault("PYSTRAY_BACKEND", "dummy")

import threading
import unittest
from types import SimpleNamespace
from unittest.mock import Mock, patch

import blurt


class StreamingAppTests(unittest.TestCase):
    def app(self):
        app = blurt.BlurtApp.__new__(blurt.BlurtApp)
        app._live_lock = threading.RLock()
        app._delivery_lock = threading.Lock()
        app.lock = threading.Lock()
        app._live_sessions = []
        app._quitting = False
        app._recording_session = None
        app._update_icon = Mock()
        app.platform = SimpleNamespace(notify=Mock(), dispatch=Mock())
        app.transcription_model = "whisper-1"
        app.transcription_language = "en"
        app.api_keys = {"openai": "test-key", "openrouter": "router-key"}
        return app

    def session(self, text, complete=True):
        return SimpleNamespace(
            result=SimpleNamespace(session_id="test", text=text, complete=complete,
                                   errors=[] if complete else ["failed"]),
            acknowledge_delivery=Mock(),
            release_for_retry=Mock(),
        )

    @patch("blurt.pyperclip.copy")
    def test_results_delivered_in_recording_order(self, copy):
        app = self.app()
        first = self.session("first")
        final_result = first.result
        first.result = None
        second = self.session("second")
        app._live_sessions = [first, second]
        app._session_complete(second)
        copy.assert_not_called()
        first.result = final_result
        app._session_complete(first)
        self.assertEqual([c.args[0] for c in copy.call_args_list], ["first", "second"])
        self.assertEqual(app._live_sessions, [])

    @patch("blurt.pyperclip.copy")
    def test_partial_agent_command_never_dispatched(self, copy):
        app = self.app()
        active = self.session("to my agent change […]", complete=False)
        app._live_sessions = [active]
        app._session_complete(active)
        copy.assert_called_once_with("to my agent change […]")
        app.platform.dispatch.assert_not_called()

    @patch("blurt.pyperclip.copy", side_effect=RuntimeError("clipboard unavailable"))
    def test_clipboard_failure_does_not_acknowledge(self, copy):
        app = self.app()
        active = self.session("keep these words")
        app._live_sessions = [active]
        app._session_complete(active)
        active.acknowledge_delivery.assert_not_called()
        active.release_for_retry.assert_called_once()

    @patch("blurt.pyperclip.copy")
    def test_capture_failure_is_not_delivered_until_microphone_stops(self, copy):
        app = self.app()
        active = self.session("some words […]", complete=False)
        active.stop = Mock()
        app._recording_session = active
        app._live_sessions = [active]
        app.stream = Mock()
        app.recording = True
        app._session_complete(active)
        copy.assert_not_called()
        app._stop_streaming()
        copy.assert_called_once_with("some words […]")

    @patch("blurt.pyperclip.copy")
    def test_quitting_keeps_completed_recording_for_recovery(self, copy):
        app = self.app()
        app._quitting = True
        active = self.session("keep these words")
        app._live_sessions = [active]
        app._session_complete(active)
        copy.assert_not_called()
        active.acknowledge_delivery.assert_not_called()

    def test_settings_snapshot_survives_menu_changes(self):
        app = self.app()
        config = app._transcription_config()
        app.transcription_model = "another-model"
        app.transcription_language = "de"
        app.api_keys["openai"] = "replacement"
        self.assertEqual((config.model, config.language, config.api_key),
                         ("whisper-1", "en", "test-key"))

    @patch("blurt.pyperclip.copy")
    def test_retries_collect_oldest_first_and_never_dispatch(self, copy):
        app = self.app()
        app._retrying = True
        app._scheduler = Mock()
        app._retryable_files = Mock(return_value=[])
        app._recovery = Mock()
        newer = SimpleNamespace(created_at=20, config={"language": "de"})
        older = SimpleNamespace(created_at=10, config={"language": "en"})
        app._recovery.discover.return_value = [newer, older]
        first = self.session("to my agent first")
        second = self.session("second")
        first.join = Mock(return_value=first.result)
        second.join = Mock(return_value=second.result)
        app._recovery.resume.side_effect = [first, second]
        app._retry_sessions()
        copy.assert_called_once_with("to my agent first\n\nsecond")
        app.platform.dispatch.assert_not_called()
        self.assertIs(app._recovery.resume.call_args_list[0].args[0], older)
        first.acknowledge_delivery.assert_called_once()
        second.acknowledge_delivery.assert_called_once()
        self.assertFalse(app._retrying)

    @patch("blurt.pyperclip.copy", side_effect=RuntimeError("unavailable"))
    def test_retry_clipboard_failure_releases_records_without_deleting(self, copy):
        app = self.app()
        app._retrying = True
        app._scheduler = Mock()
        app._retryable_files = Mock(return_value=[])
        app._recovery = Mock()
        app._recovery.discover.return_value = [SimpleNamespace(created_at=10, config={})]
        active = self.session("recovered words")
        active.join = Mock(return_value=active.result)
        app._recovery.resume.return_value = active
        with self.assertLogs(level="ERROR"):
            app._retry_sessions()
        active.acknowledge_delivery.assert_not_called()
        active.release_for_retry.assert_called_once()

    def test_callback_routes_to_owning_session_and_stop_drains_first(self):
        app = self.app()
        order = []
        active = Mock()
        active.stop.side_effect = lambda: order.append("session-eof")
        app._recording_session = active
        app.recording = True
        app.stream = Mock()
        app.stream.stop.side_effect = lambda: order.append("stream-stop")
        app.stream.close.side_effect = lambda: order.append("stream-close")
        frames = object()
        app._audio_callback(frames, 0, None, None)
        active.feed.assert_called_once_with(frames)
        app._stop_streaming()
        self.assertEqual(order, ["stream-stop", "stream-close", "session-eof"])
        self.assertIsNone(app._recording_session)
        self.assertFalse(app.recording)


if __name__ == "__main__":
    unittest.main()