test: cover recognition internals and suppress desktop notifications (#414)

* 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
This commit is contained in:
Jatin K Malik
2026-05-02 21:13:21 -07:00
committed by GitHub
parent 7c9ef372de
commit e52c96fc38
3 changed files with 118 additions and 1 deletions
+3
View File
@@ -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/
+12 -1
View File
@@ -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.
@@ -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"])