From e52c96fc382345e0067342214e157e32306e1ad7 Mon Sep 17 00:00:00 2001 From: Jatin K Malik Date: Sat, 2 May 2026 21:13:21 -0700 Subject: [PATCH] test: cover recognition internals and suppress desktop notifications (#414) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test(recognition): cover _perform_recognition and _enqueue_audio_segment The previous commit demoted logger.info('DEBUG: ...') to logger.debug(...) and removed a dead loop body in _perform_recognition. Codecov flagged 12 uncovered lines — all inside _perform_recognition and _enqueue_audio_segment, which existing tests always mocked out. Add direct (non-mocked) tests for both methods: - _enqueue_audio_segment: * empty buffer → early return * normal buffer → enqueues correctly * full queue → drops oldest and retries - _perform_recognition: * processes a segment then exits on None signal * drains remaining items on None signal * exits cleanly when idle with empty queue * continues looping on queue timeout while recording These tests exercise the actual threading/queue logic and cover the logger.debug / logger.warning calls as a side effect. * test(conftest): suppress real desktop notifications during test runs _show_notification shells out to notify-send, which pops up real system notifications when tests exercise the model-not-ready path. Add an autouse fixture that patches _show_notification for all tests EXCEPT those whose name contains 'show_notification' (so direct tests of the helper still verify subprocess.Popen calls). * chore(gitignore): ignore .sisyphus directory --- .gitignore | 3 + tests/conftest.py | 13 ++- .../test_recognition_manager_device_config.py | 103 ++++++++++++++++++ 3 files changed, 118 insertions(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index 95ea3c9..9e4b6ee 100644 --- a/.gitignore +++ b/.gitignore @@ -156,3 +156,6 @@ FIX_PLAN.md # opentree - local AI worktree orchestration tool (per-developer, not shared) .opentree/ opentree.toml + +# Sisyphus - local AI agent loop state (per-developer, not shared) +.sisyphus/ diff --git a/tests/conftest.py b/tests/conftest.py index ddf2785..2f11221 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,9 +4,10 @@ This file makes sure that the 'src' module can be imported in tests. """ import os +import subprocess import sys import threading -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch import pytest @@ -80,6 +81,16 @@ mock_audio_feedback.play_error_sound = MagicMock() sys.modules["vocalinux.ui.audio_feedback"] = mock_audio_feedback +@pytest.fixture(autouse=True) +def _suppress_desktop_notifications(request): + """Patch _show_notification for all tests except those testing it directly.""" + if "show_notification" in request.node.nodeid: + yield + return + with patch("vocalinux.speech_recognition.recognition_manager._show_notification"): + yield + + @pytest.fixture(autouse=True) def _clear_hardware_detection_cache(): """Clear lru_cache on hardware-detection helpers between tests. diff --git a/tests/test_recognition_manager_device_config.py b/tests/test_recognition_manager_device_config.py index 93bfacc..9471c22 100644 --- a/tests/test_recognition_manager_device_config.py +++ b/tests/test_recognition_manager_device_config.py @@ -18,6 +18,7 @@ import struct import sys import tempfile import threading +import time import unittest from unittest import mock from unittest.mock import MagicMock, Mock, patch @@ -699,5 +700,107 @@ class TestReconfiguration(unittest.TestCase): assert manager.engine == "whisper" +class TestEnqueueAudioSegment(unittest.TestCase): + """Direct tests for _enqueue_audio_segment (not mocked).""" + + def test_enqueue_empty_buffer_returns_early(self): + """Empty buffer should return without enqueuing anything.""" + manager = _make_manager() + manager._segment_queue = queue.Queue() + manager._enqueue_audio_segment([]) + assert manager._segment_queue.empty() + + def test_enqueue_puts_segment(self): + """Non-empty buffer should be copied and enqueued.""" + manager = _make_manager() + manager._segment_queue = queue.Queue() + buf = [b"\x00\x01", b"\x02\x03"] + manager._enqueue_audio_segment(buf) + assert not manager._segment_queue.empty() + assert manager._segment_queue.get_nowait() == buf + + def test_enqueue_drops_oldest_when_queue_full(self): + """When queue is full, oldest item is dropped and new one inserted.""" + manager = _make_manager() + manager._segment_queue = queue.Queue(maxsize=1) + manager._segment_queue.put_nowait([b"old"]) + manager._enqueue_audio_segment([b"new"]) + assert manager._segment_queue.get_nowait() == [b"new"] + + +class TestPerformRecognition(unittest.TestCase): + """Direct tests for _perform_recognition (not mocked).""" + + def _run_recognition(self, manager, pre_hook=None): + """Start _perform_recognition in a thread and return the thread.""" + t = threading.Thread(target=manager._perform_recognition) + t.start() + if pre_hook: + pre_hook(manager) + return t + + def test_processes_segment_and_exits_on_none_signal(self): + """Feed a segment, then a None signal; thread processes and exits.""" + manager = _make_manager() + manager._segment_queue = queue.Queue() + manager.should_record = True + + with patch.object(manager, "_process_audio_buffer") as mock_proc: + with patch.object(manager, "_update_state"): + t = self._run_recognition(manager) + manager._segment_queue.put([b"\x00\x01"]) + manager.should_record = False + manager._segment_queue.put(None) + t.join(timeout=2) + assert not t.is_alive() + assert mock_proc.called + + def test_drains_remaining_items_on_none_signal(self): + """None signal drains any remaining queued segments before exit.""" + manager = _make_manager() + manager._segment_queue = queue.Queue() + manager.should_record = True + + with patch.object(manager, "_process_audio_buffer") as mock_proc: + with patch.object(manager, "_update_state"): + t = self._run_recognition(manager) + manager._segment_queue.put([b"first"]) + manager._segment_queue.put([b"second"]) + manager.should_record = False + manager._segment_queue.put(None) + t.join(timeout=2) + assert not t.is_alive() + assert mock_proc.call_count == 2 + + def test_exits_when_idle_and_queue_empty(self): + """When not recording and queue empty, thread exits after brief wait.""" + manager = _make_manager() + manager._segment_queue = queue.Queue() + manager.should_record = False + + with patch.object(manager, "_process_audio_buffer") as mock_proc: + with patch.object(manager, "_update_state"): + t = self._run_recognition(manager) + t.join(timeout=2) + assert not t.is_alive() + assert not mock_proc.called + + def test_queue_timeout_while_recording(self): + """While recording, an empty-queue timeout causes the loop to continue.""" + manager = _make_manager() + manager._segment_queue = queue.Queue() + manager.should_record = True + + with patch.object(manager, "_process_audio_buffer") as mock_proc: + with patch.object(manager, "_update_state"): + t = self._run_recognition(manager) + # Allow at least one 0.1 s timeout to fire + time.sleep(0.25) + manager.should_record = False + manager._segment_queue.put(None) + t.join(timeout=2) + assert not t.is_alive() + + if __name__ == "__main__": pytest.main([__file__, "-v"])