Add some tests

pyproject.toml

@@ -4,7 +4,7 @@
   license = "MIT"
   name = "pyrobbot"
   readme = "README.md"
-  version = "0.2.2"
+  version = "0.2.3"
 
 [build-system]
   build-backend = "poetry.core.masonry.api"
@@ -132,6 +132,9 @@
   log_cli_level = "INFO"
   testpaths = ["tests/smoke", "tests/unit"]
 
+[tool.coverage.run]
+  omit = ["*/app/*"]
+
   ####################################
   # Leave configs for `poe` separate #
   ####################################

pyrobbot/text_to_speech.py

@@ -13,7 +13,6 @@
 import webrtcvad
 from gtts import gTTS
 from loguru import logger
-from pygame import mixer
 
 try:
     import sounddevice as sd
@@ -44,48 +43,39 @@ def __post_init__(self):
             )
             logger.error("Cannot continue. Exiting.")
             raise SystemExit(1)
-        mixer.init()
+
+        self.mixer = pygame.mixer
         self.vad = webrtcvad.Vad(2)
 
+        self.mixer.init()
+
+    def sound_from_bytes_io(self, bytes_io):
+        """Create a pygame sound object from a BytesIO object."""
+        return self.mixer.Sound(bytes_io)
+
+    def still_talking(self):
+        """Check if the assistant is still talking."""
+        return self.mixer.get_busy()
+
     def speak(self, text):
         """Convert text to speech."""
         logger.debug("Converting text to speech...")
         # Initialize gTTS with the text to convert
-        speech = gTTS(text, lang=self.language)
+        tts = gTTS(text, lang=self.language)
 
         # Convert the recorded array to an in-memory wav file
-        byte_io = io.BytesIO()
-        speech.write_to_fp(byte_io)
-        byte_io.seek(0)
+        tts_as_bytes_io = io.BytesIO()
+        tts.write_to_fp(tts_as_bytes_io)
+        tts_as_bytes_io.seek(0)
 
         logger.debug("Done converting text to speech.")
 
         # Play the audio file
-        speech = mixer.Sound(byte_io)
-        channel = speech.play()
-        while channel.get_busy():
+        speech_sound = self.sound_from_bytes_io(bytes_io=tts_as_bytes_io)
+        _channel = speech_sound.play()
+        while self.still_talking():
             pygame.time.wait(100)
 
-    def listen_time_limited(self):
-        """Record audio from the mic, for a limited timelength, and convert it to text."""
-        n_frames = int(self.recording_duration_seconds * self.sample_rate)
-        # Record audio from the microphone
-        rec_as_array = sd.rec(
-            frames=n_frames, samplerate=self.sample_rate, channels=1, dtype="int16"
-        )
-        logger.debug("Recording Audio")
-        sd.wait()
-        logger.debug("Done Recording")
-
-        logger.debug("Converting audio to text...")
-        # Convert the recorded array to an in-memory wav file
-        byte_io = io.BytesIO()
-        wav.write(byte_io, rate=self.sample_rate, data=rec_as_array.astype(np.int16))
-        text = self._audio_buffer_to_text(self, byte_io)
-        logger.debug("Done converting audio to text.")
-
-        return text
-
     def listen(self):
         """Record audio from the microphone, until user stops, and convert it to text."""
         # Adapted from

tests/conftest.py

@@ -1,13 +1,17 @@
+from unittest.mock import MagicMock
+
 import lorem
 import numpy as np
 import openai
+import pygame
 import pytest
 from _pytest.logging import LogCaptureFixture
 from loguru import logger
 
 import pyrobbot
 from pyrobbot.chat import Chat
 from pyrobbot.chat_configs import ChatOptions
+from pyrobbot.text_to_speech import LiveAssistant
 
 
 @pytest.fixture()
@@ -133,3 +137,27 @@ def cli_args_overrides(default_chat_configs):
 @pytest.fixture()
 def default_chat(default_chat_configs):
     return Chat(configs=default_chat_configs)
+
+
+@pytest.fixture(autouse=True)
+def _text_to_speech_mockers(mocker):
+    """Mockers for the text-to-speech module."""
+    mocker.patch(
+        "pyrobbot.text_to_speech.LiveAssistant.still_talking", return_value=False
+    )
+    mocker.patch("gtts.gTTS.write_to_fp")
+
+    orig_func = LiveAssistant.sound_from_bytes_io
+
+    def mock_sound_from_bytes_io(self: LiveAssistant, bytes_io):
+        try:
+            return orig_func(self, bytes_io)
+        except pygame.error:
+            return MagicMock()
+
+    mocker.patch(
+        "pyrobbot.text_to_speech.LiveAssistant.sound_from_bytes_io",
+        mock_sound_from_bytes_io,
+    )
+
+    mocker.patch("webrtcvad.Vad.is_speech", return_value=False)

tests/smoke/test_app.py

@@ -3,6 +3,7 @@
 
 def test_app(mocker, default_chat_configs):
     mocker.patch("streamlit.session_state", {})
+    mocker.patch("streamlit.chat_input", return_value="foobar")
     mocker.patch(
         "pyrobbot.chat_configs.ChatOptions.from_file",
         return_value=default_chat_configs,

tests/smoke/test_commands.py

@@ -25,3 +25,21 @@ def _mock_subprocess_run(*args, **kwargs):  # noqa: ARG001
 
     mocker.patch("subprocess.run", new=_mock_subprocess_run)
     main(argv=[])
+
+
+def test_voice_chat(cli_args_overrides, mocker):
+    # We allow two calls in order to let the function be tested first and then terminate
+    # the chat
+    def _mock_listen(*args, **kwargs):  # noqa: ARG001
+        try:
+            _mock_listen.execution_counter += 1
+        except AttributeError:
+            _mock_listen.execution_counter = 0
+        if _mock_listen.execution_counter > 1:
+            raise KeyboardInterrupt
+        return "foobar"
+
+    mocker.patch("pyrobbot.text_to_speech.LiveAssistant.listen", _mock_listen)
+    args = ["voice", *cli_args_overrides]
+    args = list(dict.fromkeys(args))
+    main(args)

tests/unit/test_text_to_speech.py

@@ -0,0 +1,20 @@
+import contextlib
+
+from pyrobbot.text_to_speech import LiveAssistant
+
+
+def test_speak():
+    """Test the speak method."""
+    assistant = LiveAssistant()
+
+    # Call the speak method
+    assistant.speak("Hello world!")
+
+
+def test_listen():
+    """Test the listen method."""
+    assistant = LiveAssistant(inactivity_timeout_seconds=1e-5)
+
+    # Call the listen method
+    with contextlib.suppress(KeyboardInterrupt):
+        assistant.listen()