Refactor: connection reuse (#12)

Author: paikend

src/openstack_mcp_server/config.py

@@ -3,7 +3,11 @@
 
 
 # Transport protocol
-MCP_TRANSPORT = os.environ.get("TRANSPORT", "stdio")
+MCP_TRANSPORT: str = os.environ.get("TRANSPORT", "stdio")
+
+# Openstack client settings
+MCP_CLOUD_NAME: str = os.environ.get("CLOUD_NAME", "openstack")
+MCP_DEBUG_MODE: bool = os.environ.get("DEBUG_MODE", "true").lower() == "true"
 
 # Application paths
 BASE_DIR = Path(__file__).parent.parent.parent

src/openstack_mcp_server/tools/__init__.py

@@ -5,6 +5,8 @@ def register_tool(mcp: FastMCP):
     """
     Register Openstack MCP tools.
     """
+    from .glance_tools import GlanceTools
     from .nova_tools import NovaTools
 
     NovaTools().register_tools(mcp)
+    GlanceTools.register_tools(mcp)

src/openstack_mcp_server/tools/base.py

@@ -0,0 +1,26 @@
+import openstack
+from openstack import connection
+from openstack_mcp_server import config
+
+
+class OpenStackConnectionManager:
+    """OpenStack Connection Manager"""
+
+    _connection: connection.Connection | None = None
+
+    # TODO: Try/Catch disconnection by token expired case
+    @classmethod
+    def get_connection(cls) -> connection.Connection:
+        """OpenStack Connection)"""
+        if cls._connection is None:
+            openstack.enable_logging(debug=config.MCP_DEBUG_MODE)
+            cls._connection = openstack.connect(cloud=config.MCP_CLOUD_NAME)
+        return cls._connection
+
+
+    # TODO: Close connection
+
+
+def get_openstack_conn():
+    """Get OpenStack Connection"""
+    return OpenStackConnectionManager.get_connection()

src/openstack_mcp_server/tools/glance_tools.py

@@ -0,0 +1,33 @@
+from .base import get_openstack_conn
+from mcp.server.fastmcp import FastMCP
+
+
+class GlanceTools:
+    """
+    A class to encapsulate Nova-related tools and utilities.
+    """
+
+    def register_tools(self, mcp: FastMCP):
+        """
+        Register Glance-related tools with the FastMCP instance.
+        """
+
+        mcp.tool()(self.get_glance_images)
+
+    def get_glance_images(self) -> str:
+        """
+        Get the list of Glance images by invoking the registered tool.
+
+        :return: A string containing the names, IDs, and statuses of the images.
+        """
+        # Initialize connection
+        conn = get_openstack_conn()
+
+        # List the servers
+        image_list = []
+        for image in conn.image.images():
+            image_list.append(
+                f"{image.name} ({image.id}) - Status: {image.status}"
+            )
+
+        return "\n".join(image_list)

src/openstack_mcp_server/tools/nova_tools.py

@@ -1,4 +1,4 @@
-import openstack
+from .base import get_openstack_conn
 from mcp.server.fastmcp import FastMCP
 
 
@@ -7,11 +7,6 @@ class NovaTools:
     A class to encapsulate Nova-related tools and utilities.
     """
 
-    def __init__(self):
-        """
-        Initialize the NovaTools with a FastMCP instance.
-        """
-
     def register_tools(self, mcp: FastMCP):
         """
         Register Nova-related tools with the FastMCP instance.
@@ -25,16 +20,12 @@ def get_nova_servers(self) -> str:
 
         :return: A string containing the names, IDs, and statuses of the servers.
         """
-
-        # Initialize and turn on debug logging
-        openstack.enable_logging(debug=True)
-
         # Initialize connection
-        conn = openstack.connect(cloud="openstack")
+        conn = get_openstack_conn()
 
         # List the servers
         server_list = []
-        for server in conn.list_servers():
+        for server in conn.compute.list_servers():
             server_list.append(
                 f"{server.name} ({server.id}) - Status: {server.status}"
             )

tests/conftest.py

@@ -3,60 +3,36 @@
 
 
 @pytest.fixture
-def mock_openstack_logging():
-    """Mock openstack.enable_logging function."""
-    with patch(
-        "openstack_mcp_server.tools.nova_tools.openstack.enable_logging"
-    ) as mock:
-        yield mock
-
+def mock_get_openstack_conn():
+    """Mock get_openstack_conn function for nova_tools."""
+    mock_conn = Mock()
 
-@pytest.fixture
-def mock_openstack_connect():
-    """Mock openstack.connect function."""
     with patch(
-        "openstack_mcp_server.tools.nova_tools.openstack.connect"
-    ) as mock:
-        yield mock
+        "openstack_mcp_server.tools.nova_tools.get_openstack_conn",
+        return_value=mock_conn
+    ) as mock_func:
+        yield mock_conn
 
 
 @pytest.fixture
-def mock_openstack_connection():
-    """Mock openstack connection object."""
+def mock_get_openstack_conn_glance():
+    """Mock get_openstack_conn function for glance_tools."""
     mock_conn = Mock()
-    with patch(
-        "openstack_mcp_server.tools.nova_tools.openstack.connect",
-        return_value=mock_conn,
-    ) as mock_connect:
-        yield mock_conn, mock_connect
-
-
-@pytest.fixture
-def mock_server():
-    """Create a mock server object."""
-
-    def _create_server(
-        name="test-server", server_id="test-id", status="ACTIVE"
-    ):
-        mock = Mock()
-        mock.name = name
-        mock.id = server_id
-        mock.status = status
-        return mock
 
-    return _create_server
+    with patch(
+        "openstack_mcp_server.tools.glance_tools.get_openstack_conn",
+        return_value=mock_conn
+    ) as mock_func:
+        yield mock_conn
 
 
 @pytest.fixture
-def sample_servers(mock_server):
-    """Create sample server list."""
-    return [
-        mock_server("server1", "id1", "ACTIVE"),
-        mock_server("server2", "id2", "STOPPED"),
-    ]
-
+def mock_openstack_base():
+    """Mock base module functions."""
+    mock_conn = Mock()
 
-@pytest.fixture
-def mock_fastmcp():
-    """Mock FastMCP instance."""
-    return Mock()
+    with patch(
+        "openstack_mcp_server.tools.base.get_openstack_conn",
+        return_value=mock_conn
+    ) as mock_func:
+        yield mock_conn

tests/tools/test_glance_tools.py

@@ -0,0 +1,79 @@
+import pytest
+from unittest.mock import Mock
+from openstack_mcp_server.tools.glance_tools import GlanceTools
+
+
+class TestGlanceTools:
+    """Test cases for GlanceTools class."""
+
+    def test_get_glance_images_success(self, mock_get_openstack_conn_glance):
+        """Test getting glance images successfully."""
+        mock_conn = mock_get_openstack_conn_glance
+
+        # Create mock image objects
+        mock_image1 = Mock()
+        mock_image1.name = "ubuntu-20.04-server"
+        mock_image1.id = "img-123-abc-def"
+        mock_image1.status = "active"
+
+        mock_image2 = Mock()
+        mock_image2.name = "centos-8-stream"
+        mock_image2.id = "img-456-ghi-jkl"
+        mock_image2.status = "active"
+
+        # Configure mock image.images()
+        mock_conn.image.images.return_value = [mock_image1, mock_image2]
+
+        # Test GlanceTools
+        glance_tools = GlanceTools()
+        result = glance_tools.get_glance_images()
+
+        # Verify results
+        expected_output = (
+            "ubuntu-20.04-server (img-123-abc-def) - Status: active\n"
+            "centos-8-stream (img-456-ghi-jkl) - Status: active"
+        )
+        assert result == expected_output
+
+        # Verify mock calls
+        mock_conn.image.images.assert_called_once()
+
+    def test_get_glance_images_empty_list(self, mock_get_openstack_conn_glance):
+        """Test getting glance images when no images exist."""
+        mock_conn = mock_get_openstack_conn_glance
+
+        # Empty image list
+        mock_conn.image.images.return_value = []
+
+        glance_tools = GlanceTools()
+        result = glance_tools.get_glance_images()
+
+        # Verify empty string
+        assert result == ""
+
+        mock_conn.image.images.assert_called_once()
+
+    def test_get_glance_images_with_empty_name(self, mock_get_openstack_conn_glance):
+        """Test images with empty or None names."""
+        mock_conn = mock_get_openstack_conn_glance
+
+        # Images with empty name (edge case)
+        mock_image1 = Mock()
+        mock_image1.name = "normal-image"
+        mock_image1.id = "img-normal"
+        mock_image1.status = "active"
+
+        mock_image2 = Mock()
+        mock_image2.name = ""  # Empty name
+        mock_image2.id = "img-empty-name"
+        mock_image2.status = "active"
+
+        mock_conn.image.images.return_value = [mock_image1, mock_image2]
+
+        glance_tools = GlanceTools()
+        result = glance_tools.get_glance_images()
+
+        assert "normal-image (img-normal) - Status: active" in result
+        assert " (img-empty-name) - Status: active" in result  # Empty name
+
+        mock_conn.image.images.assert_called_once()