Merge pull request #209 from FullStackWithLawrence/next

fix: add unit tests for common

Author: lpm0073

api/terraform/python/openai_api/common/tests/mock_data/json/passthrough_openai_v2_request.json

@@ -0,0 +1,24 @@
+{
+  "model": "gpt-3.5-turbo",
+  "object": "chat.completion",
+  "temperature": 0,
+  "max_tokens": 256,
+  "messages": [
+    {
+      "role": "system",
+      "content": "you always return the integer value 42."
+    },
+    {
+      "role": "user",
+      "content": "return the integer value 42."
+    }
+  ],
+  "chat_history": [
+    {
+      "message": "Hello.",
+      "direction": "incoming",
+      "sentTime": "11/16/2023, 5:53:32 PM",
+      "sender": "system"
+    }
+  ]
+}

api/terraform/python/openai_api/common/tests/mock_data/json/passthrough_openai_v2_response.json

@@ -0,0 +1,41 @@
+{
+  "retval": {
+    "isBase64Encoded": false,
+    "statusCode": 200,
+    "headers": {
+      "Content-Type": "application/json"
+    },
+    "body": {
+      "id": "chatcmpl-8WtM7gZIqvLQKPlNi9ALm7Im6CdXa",
+      "choices": [
+        {
+          "finish_reason": "stop",
+          "index": 0,
+          "message": {
+            "content": "42",
+            "role": "assistant",
+            "function_call": null,
+            "tool_calls": null
+          },
+          "logprobs": null
+        }
+      ],
+      "created": 1702849975,
+      "model": "gpt-3.5-turbo-0613",
+      "object": "chat.completion",
+      "system_fingerprint": null,
+      "usage": {
+        "completion_tokens": 1,
+        "prompt_tokens": 23,
+        "total_tokens": 24
+      },
+      "request_meta_data": {
+        "lambda": "lambda_openai_v2",
+        "model": "gpt-3.5-turbo",
+        "object_type": "chat.completion",
+        "temperature": 0,
+        "max_tokens": 256
+      }
+    }
+  }
+}

api/terraform/python/openai_api/common/tests/test_utils.py

@@ -0,0 +1,110 @@
+# -*- coding: utf-8 -*-
+# pylint: disable=wrong-import-position
+"""Test configuration Settings class."""
+
+# python stuff
+import json
+import os
+import sys
+import unittest
+from pathlib import Path
+
+
+HERE = os.path.abspath(os.path.dirname(__file__))
+PROJECT_ROOT = str(Path(HERE).parent.parent)
+PYTHON_ROOT = str(Path(PROJECT_ROOT).parent)
+if PYTHON_ROOT not in sys.path:
+    sys.path.append(PYTHON_ROOT)  # noqa: E402
+
+from openai_api.common.const import OpenAIMessageKeys  # noqa: E402
+
+# our stuff
+from openai_api.common.tests.test_setup import get_test_file  # noqa: E402
+from openai_api.common.utils import (  # noqa: E402
+    exception_response_factory,
+    get_content_for_role,
+    get_message_history,
+    get_messages_for_role,
+    get_request_body,
+    http_response_factory,
+    parse_request,
+)
+
+
+class TestUtils(unittest.TestCase):
+    """Test utils."""
+
+    # Get the directory of the current script
+    here = HERE
+    request = get_test_file("json/passthrough_openai_v2_request.json")
+    response = get_test_file("json/passthrough_openai_v2_response.json")
+
+    def setUp(self):
+        """Set up test fixtures."""
+
+    def test_http_response_factory(self):
+        """Test test_http_response_factory."""
+        retval = http_response_factory(200, self.response)
+        self.assertEqual(retval["statusCode"], 200)
+        self.assertEqual(retval["body"], json.dumps(self.response))
+        self.assertEqual(retval["isBase64Encoded"], False)
+        self.assertEqual(retval["headers"]["Content-Type"], "application/json")
+
+    def test_exception_response_factory(self):
+        """Test exception_response_factory."""
+        try:
+            raise AssertionError("test")
+        except AssertionError as exception:
+            retval = exception_response_factory(exception)
+            self.assertIn("error", retval)
+            self.assertIn("description", retval)
+
+    def test_get_request_body(self):
+        """Test get_request_body"""
+        request_body = get_request_body(self.request)
+        self.assertEqual(request_body, self.request)
+        self.assertEqual(request_body["model"], "gpt-3.5-turbo")
+        self.assertEqual(request_body["object"], "chat.completion")
+        self.assertIn("temperature", request_body)
+        self.assertIn("max_tokens", request_body)
+        self.assertIn("messages", request_body)
+
+    def test_parse_request(self):
+        """Test parse_request"""
+        request_body = get_request_body(self.request)
+        object_type, model, messages, input_text, temperature, max_tokens = parse_request(request_body)
+        self.assertEqual(object_type, "chat.completion")
+        self.assertEqual(model, "gpt-3.5-turbo")
+        self.assertEqual(input_text, None)
+        self.assertEqual(temperature, 0)
+        self.assertEqual(max_tokens, 256)
+        self.assertEqual(len(messages), 2)
+
+    def test_get_content_for_role(self):
+        """Test get_content_for_role"""
+        request_body = get_request_body(self.request)
+        _, _, messages, _, _, _ = parse_request(request_body)
+        system_message = get_content_for_role(messages, OpenAIMessageKeys.OPENAI_SYSTEM_MESSAGE_KEY)
+        user_message = get_content_for_role(messages, OpenAIMessageKeys.OPENAI_USER_MESSAGE_KEY)
+        self.assertEqual(system_message, "you always return the integer value 42.")
+        self.assertEqual(user_message, "return the integer value 42.")
+
+    def test_get_message_history(self):
+        """test get_message_history"""
+        request_body = get_request_body(self.request)
+        _, _, messages, _, _, _ = parse_request(request_body)
+        message_history = get_message_history(messages)
+        self.assertIsInstance(message_history, list)
+        self.assertEqual(len(message_history), 1)
+        self.assertEqual(message_history[0]["role"], "user")
+        self.assertEqual(message_history[0]["content"], "return the integer value 42.")
+
+    def test_get_messages_for_role(self):
+        """test get_messages_for_role"""
+        request_body = get_request_body(self.request)
+        _, _, messages, _, _, _ = parse_request(request_body)
+        message_history = get_message_history(messages)
+        self.assertIsInstance(message_history, list)
+        user_messages = get_messages_for_role(message_history, OpenAIMessageKeys.OPENAI_USER_MESSAGE_KEY)
+        self.assertEqual(len(user_messages), 1)
+        self.assertEqual(user_messages[0], "return the integer value 42.")

api/terraform/python/openai_api/common/utils.py

@@ -174,9 +174,3 @@ def get_messages_for_role(messages: list, role: str) -> list:
     """Get the text content from the messages list for a given role"""
     retval = [d.get("content") for d in messages if d["role"] == role]
     return retval
-
-
-def get_messages_for_type(messages: list, message_type: str) -> list:
-    """Get the text content from the messages list for a given role"""
-    retval = [d.get("content") for d in messages if d["type"] == message_type]
-    return retval

api/terraform/python/openai_api/lambda_langchain/tests/test_lambda_langchain.py

@@ -31,10 +31,6 @@ class TestLambdaLangchain(unittest.TestCase):
     def setUp(self):
         """Set up test fixtures."""
 
-    def get_event(self, event):
-        """Get the event json from the mock file."""
-        return event["event"]
-
     def test_lambda_handler(self):
         """Test lambda_handler."""
         response = handler(self.event, None)

api/terraform/python/openai_api/lambda_openai_v2/tests/test_lambda_openai_v2.py

@@ -32,10 +32,6 @@ class TestLambdaOpenai(unittest.TestCase):
     def setUp(self):
         """Set up test fixtures."""
 
-    def get_event(self, event):
-        """Get the event json from the mock file."""
-        return event["event"]
-
     def test_lambda_handler(self):
         """Test lambda_handler."""
         response = handler(self.event, None)