Increase test coverage

gpt3_assistant/clients/open_ai_client.py

@@ -14,9 +14,10 @@ def get_completion(self, prompt, model="text-davinci-003", max_tokens=200, tempe
         Get a completion (response) from the specified OpenAI GPT-3 model.
         Reference: https://beta.openai.com/docs/api-reference/completions
 
-        :param str prompt:      the prompt to send to the model for completion.
-        :param str model:       optionally specify the model to use.
-        :param int max_tokens:  the max number of tokens to use including the prompt and response.
+        :param str prompt:       the prompt to send to the model for completion.
+        :param str model:        optionally specify the model to use.
+        :param int max_tokens:   the max number of tokens to use including the prompt and response.
+        :param int temperature:  the temperature for the model to use.
         """
         completion = openai.Completion.create(
             model=model, prompt=prompt, max_tokens=max_tokens, temperature=temperature

gpt3_assistant/computer_voice_responder.py

@@ -1,16 +1,14 @@
 import os
 import subprocess
 import logging
-from gpt3_assistant.clients.google_text_to_speech_client import GoogleTextToSpeechClient
 from gpt3_assistant.bases.responder import Responder
+from clients.text_to_speech_client import TextToSpeechClient
 
 
 class ComputerVoiceResponder(Responder):
-    def __init__(self, mp3_filename, lang=None, tld=None):
+    def __init__(self, text_to_speech_client: TextToSpeechClient, mp3_filename):
         self._mp3_filename = mp3_filename
-        self._language = lang
-        self._top_level_domain = tld
-        self.text_to_speech_client = GoogleTextToSpeechClient(lang, tld)
+        self.text_to_speech_client: TextToSpeechClient = text_to_speech_client
 
     def respond(self, text_to_speak: str):
         """

gpt3_assistant/main.py

@@ -7,6 +7,7 @@
 from input_devices import InputDevices
 from helpers.set_log_level import set_log_level
 from helpers.set_keyboard_interrupt_handler import set_keyboard_interrupt_handler
+from clients.google_text_to_speech_client import GoogleTextToSpeechClient
 
 if __name__ == "__main__":
     options = CommandLineParser.parse()
@@ -29,11 +30,15 @@
     # service to generate text given an input
     text_generator = OpenAITextGenerator(options.open_ai_key)
 
+    google_text_to_speech_client = GoogleTextToSpeechClient(
+        options.lang,
+        options.tld
+    )
+
     # service to respond to the user the generated text
     responder = ComputerVoiceResponder(
-        "temp.mp3",
-        options.lang,
-        options.tld,
+        google_text_to_speech_client,
+        "temp.mp3"
     )
 
     # set interrupt to exit the process when Cmd+C / Ctrl+C is hit

gpt3_assistant/open_ai_text_generator.py

@@ -15,21 +15,22 @@ def __init__(self, open_ai_key: str, **kwargs):
         self._open_ai_client = OpenAIClient(open_ai_key)
         self._model = kwargs.get("model", "text-davinci-003")
         self._max_tokens = kwargs.get("max_tokens", 200)
-        self._previous_responses: list[Exchange] = []
+        self._temperature = kwargs.get("temperature", 0.7)
+        self._previous_responses: list[Exchange] = kwargs.get('previous_responses', [])
 
     def generate_text(self, input_text) -> Exchange:
         """
         Generates text based on the input and returns it.
         :param input_text: the input text.
         :return: the response.
         """
-        full_prompt_with_history: str = self._get_request_under_max_tokens(input_text)
+        full_prompt_with_history = self._get_request_under_max_tokens(input_text)
 
         exchange: Exchange = self._open_ai_client.get_completion(
             prompt=full_prompt_with_history,
             model=self._model,
             max_tokens=self._max_tokens,
-            temperature=0.7
+            temperature=self._temperature
         )
 
         self._previous_responses.append(exchange)

tests/clients/test_open_ai_client.py

@@ -0,0 +1,48 @@
+from gpt3_assistant.clients.open_ai_client import OpenAIClient
+from pytest import fixture, raises
+import mock
+
+OPEN_AI_KEY = "fake-key"
+MOCK_RESPONSES = [
+    {"finish_reason": "stop", "text": "hey there"},
+    {"finish_reason": "stop", "text": "what's up"},
+]
+
+
+@fixture
+def open_ai_client():
+    return OpenAIClient(OPEN_AI_KEY)
+
+
+def mock_create_completion_no_responses(**kwargs):
+    return {"choices": [], "usage": {"total_tokens": kwargs["max_tokens"]}}
+
+
+def mock_create_completion_multiple_responses(**kwargs):
+    return {"choices": MOCK_RESPONSES, "usage": {"total_tokens": kwargs["max_tokens"]}}
+
+
+@mock.patch("openai.Completion.create", mock_create_completion_no_responses)
+def test_get_completion_throws_exception_no_responses(open_ai_client):
+    max_tokens = 70
+
+    prompt = "Yeah do you have one in mind?"
+
+    with raises(Exception):
+        open_ai_client.get_completion(
+            prompt=prompt, max_tokens=max_tokens
+        )
+
+
+@mock.patch("openai.Completion.create", mock_create_completion_multiple_responses)
+def test_get_completion_returns_first_response(open_ai_client):
+    max_tokens = 70
+    prompt = "Yeah do you have one in mind?"
+
+    response = open_ai_client.get_completion(
+        prompt=prompt, max_tokens=max_tokens
+    )
+
+    assert response is not None
+    assert response.computer_response == MOCK_RESPONSES[0]["text"]
+    assert response.user_message == prompt

tests/test_computer_voice_responder.py

@@ -0,0 +1,89 @@
+import os.path
+from mock import mock
+from gpt3_assistant.computer_voice_responder import ComputerVoiceResponder
+from pytest import fixture
+from gpt3_assistant.clients.text_to_speech_client import TextToSpeechClient
+
+TEMP_FILE_NAME = "temp.mp3"
+FULL_MP3_PATH = os.path.join(os.getcwd(), TEMP_FILE_NAME)
+TEXT_TO_SPEAK = "hello there"
+GTTS_ERROR_TO_THROW = AssertionError
+
+
+class mock_subprocess_call():
+    _calls = 0
+    _last_inputs = None
+
+    def __init__(self, inputs: list[str]):
+        mock_subprocess_call._calls += 1
+        mock_subprocess_call._last_inputs = inputs
+
+    @staticmethod
+    def reset():
+        mock_subprocess_call._calls = 0
+        mock_subprocess_call._last_inputs = None
+
+
+class MockGoogleTextToSpeechClient(TextToSpeechClient):
+    _exception = None
+
+    def convert_text_to_mp3(self, text_to_speak, mp3_file_path):
+        if MockGoogleTextToSpeechClient._exception is not None:
+            raise MockGoogleTextToSpeechClient._exception
+
+        pass
+
+
+class mock_print:
+    _calls: int = 0
+    _last_printed: str = None
+
+    def __init__(self, text: str):
+        mock_print._calls += 1
+        mock_print._last_printed = text
+
+    @staticmethod
+    def reset():
+        mock_print._calls = 0
+        mock_print._last_printed = None
+
+
+@fixture
+def computer_voice():
+    mock_text_to_speech_client = MockGoogleTextToSpeechClient()
+
+    return ComputerVoiceResponder(
+        text_to_speech_client=mock_text_to_speech_client,
+        mp3_filename=TEMP_FILE_NAME
+    )
+
+
+@mock.patch(
+    'gpt3_assistant.clients.google_text_to_speech_client.GoogleTextToSpeechClient',
+    MockGoogleTextToSpeechClient
+)
+@mock.patch('builtins.print', mock_print)
+@mock.patch("subprocess.call", mock_subprocess_call)
+def test_respond_succeeds(computer_voice):
+    computer_voice.respond(TEXT_TO_SPEAK)
+    assert mock_print._last_printed is None or not mock_print._last_printed.startswith("Exception")
+    assert mock_subprocess_call._calls == 1
+    mock_print.reset()
+    mock_subprocess_call.reset()
+
+
+@mock.patch(
+    'gpt3_assistant.clients.google_text_to_speech_client.GoogleTextToSpeechClient',
+    MockGoogleTextToSpeechClient
+)
+@mock.patch('builtins.print', mock_print)
+@mock.patch("subprocess.call", mock_subprocess_call)
+def test_respond_throws_error(computer_voice):
+    MockGoogleTextToSpeechClient._exception = Exception("Bad news")
+    computer_voice.respond(TEXT_TO_SPEAK)
+    assert mock_print._calls == 1
+    assert mock_print._last_printed is not None
+    assert mock_print._last_printed.startswith("Exception caught")
+    assert mock_subprocess_call._calls == 0
+    mock_print.reset()
+    mock_subprocess_call.reset()

tests/test_input_devices.py

@@ -54,5 +54,5 @@ def test_get_input_device_index_throws_error_not_found():
         InputDevices.get_input_device_index(fake_input_device_name)
 
     assert (
-        str(e.value) == f"Input device with name '{fake_input_device_name}' not found"
+            str(e.value) == f"Input device with name '{fake_input_device_name}' not found"
     )