voculinux/tests/test_speech_recognition.py

"""
Tests for speech recognition functionality.
"""

import sys  # noqa: F401
import unittest
from unittest.mock import MagicMock, patch

# Create proper mock responses that can be safely JSON serialized
MOCK_VOSK_RESULT = '{"text": "test transcription"}'


# We need to patch the entire vosk Model class and KaldiRecognizer
class MockModel:
    def __init__(self, path):
        pass


class MockKaldiRecognizer:
    def __init__(self, model, rate):
        pass

    def AcceptWaveform(self, data):
        return True

    def FinalResult(self):
        return MOCK_VOSK_RESULT

    def PartialResult(self):
        return '{"partial": "test"}'


# Create a proper mock vosk module that uses our custom classes
class MockVoskModule:
    Model = MockModel
    KaldiRecognizer = MockKaldiRecognizer


# Path modules before importing the recognition_manager
sys.modules["vosk"] = MockVoskModule

# Mock other required modules
sys.modules["pyaudio"] = MagicMock()
sys.modules["numpy"] = MagicMock()
sys.modules["whisper"] = MagicMock()
sys.modules["whisper"].load_model = MagicMock(return_value=MagicMock())
sys.modules["torch"] = MagicMock()
sys.modules["torch"].cuda.is_available = MagicMock(return_value=False)

# Mock tempfile and wave modules to avoid file system issues
mock_tempfile = MagicMock()
mock_temp_file = MagicMock()
mock_temp_file.name = "/tmp/tmpfile.wav"
mock_temp_file.__enter__ = MagicMock(return_value=mock_temp_file)
mock_temp_file.__exit__ = MagicMock(return_value=None)
mock_tempfile.NamedTemporaryFile = MagicMock(return_value=mock_temp_file)
sys.modules["tempfile"] = mock_tempfile

mock_wave = MagicMock()
mock_wave_file = MagicMock()
mock_wave_file.__enter__ = MagicMock(return_value=mock_wave_file)
mock_wave_file.__exit__ = MagicMock(return_value=None)
mock_wave.open = MagicMock(return_value=mock_wave_file)
sys.modules["wave"] = mock_wave

# Import the shared mock from conftest
from conftest import mock_audio_feedback  # noqa: E402

# Update import paths to use the new package structure
from vocalinux.common_types import RecognitionState  # noqa: E402
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager  # noqa: E402
from vocalinux.utils.vosk_model_info import VOSK_MODEL_INFO  # noqa: E402


class TestSpeechRecognition(unittest.TestCase):
    """Test cases for the speech recognition functionality."""

    def setUp(self):
        """Set up for tests."""
        # Patch os.makedirs to avoid creating directories
        self.patcher_makedirs = patch("os.makedirs")
        self.mock_makedirs = self.patcher_makedirs.start()

        # Patch os.path.exists to return True for any path
        self.patcher_exists = patch("os.path.exists", return_value=True)
        self.mock_exists = self.patcher_exists.start()

        # Mock the command processor
        self.patcher_cmd = patch(
            "vocalinux.speech_recognition.recognition_manager.CommandProcessor"
        )
        self.mock_cmd_class = self.patcher_cmd.start()
        self.mock_cmd = MagicMock()
        self.mock_cmd_class.return_value = self.mock_cmd

        # Mock threading to avoid thread creation
        self.patcher_thread = patch(
            "vocalinux.speech_recognition.recognition_manager.threading.Thread"
        )
        self.mock_thread_class = self.patcher_thread.start()
        self.mock_thread = MagicMock()
        self.mock_thread_class.return_value = self.mock_thread

        # Use the module-level audio feedback mock
        self.mock_play_start = mock_audio_feedback.play_start_sound
        self.mock_play_stop = mock_audio_feedback.play_stop_sound
        self.mock_play_error = mock_audio_feedback.play_error_sound

        # Patch the download method for vosk models
        self.patcher_download = patch.object(SpeechRecognitionManager, "_download_vosk_model")
        self.mock_download = self.patcher_download.start()

        # Patch os.unlink to avoid file removal errors
        self.patcher_unlink = patch("os.unlink")
        self.mock_unlink = self.patcher_unlink.start()

    def tearDown(self):
        """Clean up after tests."""
        self.patcher_makedirs.stop()
        self.patcher_exists.stop()
        self.patcher_cmd.stop()
        self.patcher_thread.stop()
        self.patcher_download.stop()
        self.patcher_unlink.stop()

        # Reset the audio feedback mocks for the next test
        mock_audio_feedback.reset_mock()

    def test_init_state(self):
        """Test the initial state of the speech recognition manager."""
        # Test VOSK initialization
        manager = SpeechRecognitionManager(engine="vosk", model_size="small")

        # Verify initial state
        self.assertEqual(manager.state, RecognitionState.IDLE)
        self.assertEqual(manager.engine, "vosk")
        self.assertEqual(manager.model_size, "small")
        self.assertFalse(manager.should_record)
        self.assertEqual(manager.audio_buffer, [])
        self.assertEqual(manager.text_callbacks, [])
        self.assertEqual(manager.state_callbacks, [])
        self.assertEqual(manager.action_callbacks, [])

        # Test Whisper initialization with different model size
        manager = SpeechRecognitionManager(engine="whisper", model_size="medium")

        # Verify Whisper initialization
        self.assertEqual(manager.engine, "whisper")
        self.assertEqual(manager.model_size, "medium")

        # Test invalid engine
        with self.assertRaises(ValueError):
            SpeechRecognitionManager(engine="invalid")

    def test_voice_commands_auto_mode_follows_engine(self):
        vosk_manager = SpeechRecognitionManager(
            engine="vosk", model_size="small", voice_commands_enabled=None
        )
        self.assertTrue(getattr(vosk_manager, "_voice_commands_enabled"))

        whisper_manager = SpeechRecognitionManager(
            engine="whisper", model_size="small", voice_commands_enabled=None
        )
        self.assertFalse(getattr(whisper_manager, "_voice_commands_enabled"))

    def test_voice_commands_auto_mode_updates_on_engine_reconfigure(self):
        manager = SpeechRecognitionManager(engine="whisper", voice_commands_enabled=None)
        self.assertFalse(getattr(manager, "_voice_commands_enabled"))

        manager.reconfigure(engine="vosk", force_download=False)
        self.assertTrue(getattr(manager, "_voice_commands_enabled"))

    def test_voice_commands_toggle_updates_on_reconfigure(self):
        manager = SpeechRecognitionManager(engine="vosk", voice_commands_enabled=True)
        self.assertTrue(getattr(manager, "_voice_commands_enabled"))

        manager.reconfigure(voice_commands_enabled=False, force_download=False)
        self.assertFalse(getattr(manager, "_voice_commands_enabled"))

    def test_callbacks(self):
        """Test registering and using callbacks."""
        manager = SpeechRecognitionManager(engine="vosk")

        # Replace recognizer with a mock that returns proper JSON
        mock_recognizer = MagicMock()
        mock_recognizer.FinalResult.return_value = MOCK_VOSK_RESULT
        mock_recognizer.AcceptWaveform.return_value = True
        manager.recognizer = mock_recognizer

        # Create mock callbacks
        text_callback = MagicMock()
        state_callback = MagicMock()
        action_callback = MagicMock()

        # Register callbacks
        manager.register_text_callback(text_callback)
        manager.register_state_callback(state_callback)
        manager.register_action_callback(action_callback)

        # Verify callbacks were registered
        self.assertEqual(manager.text_callbacks, [text_callback])
        self.assertEqual(manager.state_callbacks, [state_callback])
        self.assertEqual(manager.action_callbacks, [action_callback])

        # Test state update
        manager._update_state(RecognitionState.LISTENING)
        self.assertEqual(manager.state, RecognitionState.LISTENING)
        state_callback.assert_called_once_with(RecognitionState.LISTENING)

        # Test text processing with callbacks
        # Setup command processor mock return values
        self.mock_cmd.process_text.return_value = ("processed text", ["action1"])

        # Prepare audio buffer for processing
        manager.audio_buffer = [b"audio_data"]

        # Process the buffer
        manager._process_final_buffer()

        # Verify command processor was called
        self.mock_cmd.process_text.assert_called_once_with("test transcription")

        # Verify text callback was called
        text_callback.assert_called_once_with("processed text")

        # Verify action callback was called
        action_callback.assert_called_once_with("action1")

    def test_vosk_model_path(self):
        """Test getting the VOSK model path based on size."""
        with patch.object(SpeechRecognitionManager, "_get_vosk_model_path") as mock_get_path:
            # Test small model size
            mock_get_path.return_value = "/path/to/vosk-model-small-en-us-0.15"
            _ = SpeechRecognitionManager(
                engine="vosk", model_size="small", voice_commands_enabled=True
            )

            # Verify the small model path is constructed correctly
            mock_get_path.assert_called_with()

            # Test medium model size
            mock_get_path.return_value = "/path/to/vosk-model-en-us-0.22"
            _ = SpeechRecognitionManager(engine="vosk", model_size="medium")

            # Verify the medium model path is constructed correctly
            mock_get_path.assert_called_with()

            # Test large model size
            mock_get_path.return_value = "/path/to/vosk-model-en-us-0.42"
            _ = SpeechRecognitionManager(engine="vosk", model_size="large")

            # Verify the large model path is constructed correctly
            mock_get_path.assert_called_with()

    def test_vosk_medium_and_large_cover_all_small_languages(self):
        """Regression test for issue #550.

        The 'medium' and 'large' VOSK model tables must map every language
        that 'small' supports. Italian ('it') and English-India ('en-in')
        were previously missing from 'medium'/'large', which caused
        vosk_model_map[size] to resolve to None and crash later with
        "join() argument must be str, bytes, or os.PathLike object, not
        'NoneType'" instead of downloading the (existing) big model.
        """
        small_languages = set(VOSK_MODEL_INFO["small"]["languages"])
        for size in ("medium", "large"):
            size_languages = set(VOSK_MODEL_INFO[size]["languages"])
            missing = small_languages - size_languages
            self.assertEqual(
                missing,
                set(),
                f"VOSK_MODEL_INFO['{size}']['languages'] is missing languages "
                f"that 'small' supports: {missing}. Any language present in "
                "'small' must also resolve to a model for 'medium'/'large', "
                "otherwise initialization crashes with a NoneType path error.",
            )

    def test_supported_languages_vosk_entries_match_model_tables(self):
        """Every catalog language with a VOSK model must appear in size tables."""
        from vocalinux.utils.vosk_model_info import SUPPORTED_LANGUAGES

        for lang_code, lang_info in SUPPORTED_LANGUAGES.items():
            vosk_model = lang_info.get("vosk")
            if not vosk_model:
                continue
            self.assertEqual(
                VOSK_MODEL_INFO["small"]["languages"].get(lang_code),
                vosk_model,
                f"SUPPORTED_LANGUAGES['{lang_code}']['vosk'] must match "
                f"VOSK_MODEL_INFO['small']['languages']['{lang_code}']",
            )
            for size in ("medium", "large"):
                self.assertIn(
                    lang_code,
                    VOSK_MODEL_INFO[size]["languages"],
                    f"VOSK language '{lang_code}' missing from '{size}' table",
                )

    def test_hungarian_is_whisper_only_in_catalog(self):
        """Hungarian (issue #565) is exposed for Whisper engines, not VOSK."""
        from vocalinux.utils.vosk_model_info import SUPPORTED_LANGUAGES

        hu = SUPPORTED_LANGUAGES["hu"]
        self.assertEqual(hu["whisper"], "hu")
        self.assertIsNone(hu["vosk"])
        self.assertNotIn("hu", VOSK_MODEL_INFO["small"]["languages"])

    def test_vosk_init_italian_medium_model_resolves_correctly(self):
        """Selecting Italian + the 'medium' VOSK model must not crash (issue #550)."""
        manager = SpeechRecognitionManager(engine="vosk", language="it", model_size="medium")

        self.assertEqual(
            manager.vosk_model_map["medium"],
            VOSK_MODEL_INFO["medium"]["languages"]["it"],
        )
        self.assertIn("vosk-model-it-0.22", manager.vosk_model_path)

    def test_get_vosk_model_path_raises_clear_error_for_unmapped_language(self):
        """An unmapped (size, language) combination should raise a clear,
        actionable error instead of crashing deep inside os.path.join with a
        cryptic TypeError (the actual failure mode reported in issue #550,
        before the VOSK_MODEL_INFO data gap was fixed)."""
        manager = SpeechRecognitionManager(engine="vosk", language="en-us", model_size="small")

        # Simulate a language that has no mapping for the requested size,
        # mirroring what happened for "it"/"medium" prior to the data fix.
        manager.model_size = "medium"
        manager.language = "it"
        manager.vosk_model_map = {
            "small": "vosk-model-small-it-0.22",
            "medium": None,
            "large": None,
        }

        with self.assertRaises(ValueError) as ctx:
            manager._get_vosk_model_path()

        self.assertIn("medium", str(ctx.exception))
        self.assertIn("it", str(ctx.exception))

    def test_download_vosk_model(self):
        """Test downloading VOSK model."""
        # Make os.path.exists return False to trigger download
        with patch("os.path.exists", return_value=False):
            # Instantiate manager with defer_download=False to trigger download
            _ = SpeechRecognitionManager(engine="vosk", model_size="small", defer_download=False)

            # Verify download was attempted
            self.mock_download.assert_called_once()

    def test_start_recognition(self):
        """Test starting speech recognition."""
        # Create a manager
        manager = SpeechRecognitionManager(engine="vosk")

        # Reset mock for clean test
        self.mock_thread_class.reset_mock()
        self.mock_play_start.reset_mock()

        # Start recognition
        manager.start_recognition()

        # Verify state changed
        self.assertEqual(manager.state, RecognitionState.LISTENING)

        # Verify start sound was played
        self.mock_play_start.assert_called_once()

        # Verify recording flag was set
        self.assertTrue(manager.should_record)

        # Verify threads were started
        self.assertEqual(self.mock_thread_class.call_count, 2)
        self.mock_thread.start.assert_called()

        # Try starting when already listening
        self.mock_thread_class.reset_mock()
        manager.start_recognition()

        # Should not start again
        self.mock_thread_class.assert_not_called()

    def test_stop_recognition(self):
        """Test stopping speech recognition."""
        # Create a manager
        manager = SpeechRecognitionManager(engine="vosk")

        # Setup manager state
        manager.state = RecognitionState.LISTENING
        manager.should_record = True
        manager.audio_thread = self.mock_thread
        manager.recognition_thread = self.mock_thread

        # Reset mocks
        self.mock_play_stop.reset_mock()

        # Stop recognition
        manager.stop_recognition()

        # Verify state changed
        self.assertEqual(manager.state, RecognitionState.IDLE)

        # Verify stop sound was played
        self.mock_play_stop.assert_called_once()

        # Verify recording flag was cleared
        self.assertFalse(manager.should_record)

        # Verify threads were joined
        self.mock_thread.join.assert_called()

    def test_record_audio(self):
        """Test recording audio with simulated voice activity."""
        # This test would be complex due to audio processing, just test basic setup
        manager = SpeechRecognitionManager(engine="vosk")
        manager.should_record = True

        # Skip detailed testing of the audio recording function
        self.assertTrue(hasattr(manager, "_record_audio"))

    def test_process_final_buffer_vosk(self):
        """Test processing the final audio buffer with VOSK."""
        # Setup manager
        manager = SpeechRecognitionManager(engine="vosk")

        # Replace recognizer with a mock that returns proper JSON
        mock_recognizer = MagicMock()
        mock_recognizer.FinalResult.return_value = MOCK_VOSK_RESULT
        mock_recognizer.AcceptWaveform.return_value = True
        manager.recognizer = mock_recognizer

        # Setup command processor mock
        self.mock_cmd.process_text.return_value = ("processed text", ["action1"])

        # Register mock callbacks
        text_callback = MagicMock()
        action_callback = MagicMock()
        manager.register_text_callback(text_callback)
        manager.register_action_callback(action_callback)

        # Prepare audio buffer
        manager.audio_buffer = [b"audio_data1", b"audio_data2"]

        # Process the buffer
        manager._process_final_buffer()

        # Verify command processor was called
        self.mock_cmd.process_text.assert_called_once_with("test transcription")

        # Verify callbacks were called
        text_callback.assert_called_once_with("processed text")
        action_callback.assert_called_once_with("action1")

    def test_process_final_buffer_whisper(self):
        """Test processing the final audio buffer with Whisper."""
        # Skip the problematic file operations by creating a mock implementation
        # of the _process_final_buffer method
        with patch.object(SpeechRecognitionManager, "_process_final_buffer") as mock_process:
            # Setup manager with whisper engine (this should work now with mocked torch)
            manager = SpeechRecognitionManager(engine="whisper")

            # Setup side effect function that will be called when _process_final_buffer is called
            def process_side_effect():
                # Simulate processing the result through command processor
                processed_text, actions = self.mock_cmd.process_text("whisper transcription")
                # Call callbacks as the real method would
                for callback in manager.text_callbacks:
                    callback(processed_text)
                for callback in manager.action_callbacks:
                    for action in actions:
                        callback(action)

            # Set the side effect
            mock_process.side_effect = process_side_effect

            # Set up a mock model just for completeness
            whisper_model = MagicMock()
            whisper_model.transcribe.return_value = {"text": "whisper transcription"}
            manager.model = whisper_model

            # Setup command processor mock
            self.mock_cmd.process_text.return_value = ("processed whisper", ["action2"])

            # Register mock callbacks
            text_callback = MagicMock()
            action_callback = MagicMock()
            manager.register_text_callback(text_callback)
            manager.register_action_callback(action_callback)

            # Prepare audio buffer (not actually used because we're mocking the method)
            manager.audio_buffer = [b"audio_data1", b"audio_data2"]

            # Call the mocked process method
            manager._process_final_buffer()

            # Verify command processor was called with expected text
            self.mock_cmd.process_text.assert_called_once_with("whisper transcription")

            # Verify callbacks were called with expected values
            text_callback.assert_called_once_with("processed whisper")
            action_callback.assert_called_once_with("action2")

    def test_empty_buffer(self):
        """Test processing an empty buffer."""
        # Setup manager
        manager = SpeechRecognitionManager(engine="vosk")

        # Process empty buffer
        manager.audio_buffer = []
        manager._process_final_buffer()

        # Verify command processor was not called
        self.mock_cmd.process_text.assert_not_called()

    def test_configure(self):
        """Test configuring recognition parameters."""
        # Setup manager
        manager = SpeechRecognitionManager(engine="vosk")

        # Default values
        self.assertEqual(manager.vad_sensitivity, 3)
        self.assertEqual(manager.silence_timeout, 2.0)

        # Configure with valid values
        manager.reconfigure(vad_sensitivity=4, silence_timeout=1.5)
        self.assertEqual(manager.vad_sensitivity, 4)
        self.assertEqual(manager.silence_timeout, 1.5)

        # Configure with out-of-range values
        manager.reconfigure(vad_sensitivity=10, silence_timeout=10.0)
        self.assertEqual(manager.vad_sensitivity, 5)  # Clamped to max
        self.assertEqual(manager.silence_timeout, 5.0)  # Clamped to max

        manager.reconfigure(vad_sensitivity=0, silence_timeout=0.2)
        self.assertEqual(manager.vad_sensitivity, 1)  # Clamped to min
        self.assertEqual(manager.silence_timeout, 0.5)  # Clamped to min