Auto setup server for tests

tests/api/test_genai_server/__init__.py

@@ -0,0 +1,59 @@
+import unittest
+import threading
+import time
+import os
+import signal
+import socket
+import psutil
+from setup import stack
+
+
+def run_server(port):
+    stack.run_server(host="127.0.0.1", port=port)
+
+
+def find_free_port(initial_port):
+    port = initial_port
+    while True:
+        try:
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+                s.bind(("127.0.0.1", port))
+            return port
+        except OSError:
+            port += 1
+
+
+def start_server():
+    free_port = find_free_port(5600)
+    server_thread = threading.Thread(target=run_server, args=(free_port,))
+    server_thread.start()
+    time.sleep(6)
+    return free_port
+
+
+def stop_server(port):
+    try:
+        for conn in psutil.net_connections():
+            if conn.laddr.port == port:
+                os.kill(conn.pid, signal.SIGTERM)
+                break
+    except (ProcessLookupError, psutil.NoSuchProcess):
+        pass
+    time.sleep(2)
+
+
+class TestCaseServer(unittest.TestCase):
+
+    host = None
+    server_port = None
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.server_port = start_server()
+        cls.host = "127.0.0.1"
+        cls.base_url = f"http://{cls.host}:{cls.server_port}/api/"
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        stop_server(cls.server_port)
+

tests/api/test_genai_server/setup/__init__.py

@@ -0,0 +1 @@
+from .main import stack

tests/api/test_genai_server/setup/main.py

@@ -0,0 +1,7 @@
+from genai_stack.genai_server.settings.config import read_configurations
+from genai_stack.genai_server.utils import get_current_stack
+import pathlib
+
+path = f"{str(pathlib.Path().resolve())}/setup"
+stack_configurations = read_configurations(path)[1]
+stack = get_current_stack(config=stack_configurations)

tests/api/test_genai_server/setup/server.conf

@@ -0,0 +1,3 @@
+[database]
+database_name = db
+database_driver = sqlite

tests/api/test_genai_server/setup/stack_config.json

@@ -0,0 +1,46 @@
+{
+    "components":{
+        "vectordb":{
+            "name":"chromadb",
+            "config":{
+            }
+        },
+        "memory":{
+            "name":"langchain",
+            "config":{
+
+            }
+        },
+        "llm_cache":{
+            "name":"cache",
+            "config":{
+
+            }
+        },
+        "model":{
+            "name":"gpt3.5",
+            "config":{
+                "parameters":{"openai_api_key": "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"}
+            }
+        },
+        "embedding":{
+            "name":"langchain",
+            "config":{
+                "name":"HuggingFaceEmbeddings",
+                "fields":{
+                    "model_name": "sentence-transformers/all-mpnet-base-v2",
+                    "model_kwargs": {"device": "cpu"},
+                    "encode_kwargs": {"normalize_embeddings": false}
+                }
+            }
+        },
+        "prompt_engine":{
+            "name": "engine",
+            "config": {}
+        },
+        "retriever": {
+            "name": "langchain",
+            "config": {}
+        }
+    }
+}

tests/api/test_genai_server/test_etl.py

@@ -2,14 +2,16 @@
 
 """Tests for `genai_server`."""
 
-import unittest
 import requests
 
+from tests.api.test_genai_server import TestCaseServer
 
-class TestETLServerAPIs(unittest.TestCase):
+
+class TestETLServerAPIs(TestCaseServer):
 
     def setUp(self) -> None:
-        self.base_url = "http://127.0.0.1:8080/api/etl"
+        super().setUp()
+        self.base_url = self.base_url + "etl"
 
     def test_submit_job(self):
         response = requests.get(

tests/api/test_genai_server/test_retrieve_from_vector_db.py

@@ -2,15 +2,15 @@
 
 """Tests for `genai_server`."""
 import json
-import unittest
 import requests
 
+from tests.api.test_genai_server import TestCaseServer
 
-class TestRetrieverDataFromDBAPIs(unittest.TestCase):
 
-    def setUp(self) -> None:
-        self.base_url = "http://127.0.0.1:5000/api"
+class TestRetrieverDataFromDBAPIs(TestCaseServer):
 
+    def setUp(self) -> None:
+        super().setUp()
         self.data = [
             {
                 "id": self.create_session(),
@@ -42,7 +42,7 @@ def setUp(self) -> None:
         ]
 
     def create_session(self):
-        response = requests.post(url=self.base_url + "/session")
+        response = requests.post(url=self.base_url + "session")
         assert response.status_code == 200
         assert response.json()
         data = response.json()
@@ -51,7 +51,7 @@ def create_session(self):
 
     def add_document_to_session(self, session_id, data):
         response = requests.post(
-            url=self.base_url + "/vectordb/add-documents",
+            url=self.base_url + "vectordb/add-documents",
             data=json.dumps(
                 {
                     "session_id": session_id,
@@ -63,7 +63,7 @@ def add_document_to_session(self, session_id, data):
 
     def search_for_doc(self, session_id, query):
         response = requests.get(
-            url=self.base_url + "/retriever/retrieve",
+            url=self.base_url + "retriever/retrieve",
             params={"session_id": session_id, "query": query},
         )
         assert response.status_code == 200

tests/api/test_genai_server/test_retriever.py

@@ -2,14 +2,16 @@
 
 """Tests for `genai_server`."""
 
-import unittest
 import requests
 
+from tests.api.test_genai_server import TestCaseServer
 
-class TestRetrieverServerAPIs(unittest.TestCase):
+
+class TestRetrieverServerAPIs(TestCaseServer):
 
     def setUp(self) -> None:
-        self.base_url = "http://127.0.0.1:5000/api/retriever"
+        super().setUp()
+        self.base_url = self.base_url + "retriever"
 
     def test_retrieve(self):
         response = requests.get(

tests/api/test_genai_server/test_session.py

@@ -2,18 +2,19 @@
 
 """Tests for `genai_server`."""
 
-import unittest
 import requests
 
+from tests.api.test_genai_server import TestCaseServer
 
-class TestSessionServerAPIs(unittest.TestCase):
+
+class TestSessionServerAPIs(TestCaseServer):
 
     def setUp(self) -> None:
-        self.base_url = "http://127.0.0.1:5000/api/session"
+        super().setUp()
+        self.base_url = self.base_url + "session"
 
     def test_create_session(self):
         response = requests.post(url=self.base_url)
-        print(response.json())
         assert response.status_code == 200
 
     def test_sessions_list(self):

tests/api/test_genai_server/test_vectordb.py

@@ -2,14 +2,16 @@
 
 """Tests for `genai_server`."""
 import json
-import unittest
 import requests
 
+from tests.api.test_genai_server import TestCaseServer
 
-class TestVectorDBServerAPIs(unittest.TestCase):
+
+class TestVectorDBServerAPIs(TestCaseServer):
 
     def setUp(self) -> None:
-        self.base_url = "http://127.0.0.1:5000/api/vectordb"
+        super().setUp()
+        self.base_url = self.base_url + "vectordb"
 
     def test_add_document(self):
         response = requests.post(