feat: server action

src/openstack_mcp_server/tools/compute_tools.py

@@ -1,3 +1,4 @@
+from enum import Enum
 from typing import Any
 
 from fastmcp import FastMCP
@@ -10,6 +11,24 @@
 from .base import get_openstack_conn
 
 
+class ServerActionEnum(str, Enum):
+    """available actions without parameter for compute tools"""
+
+    PAUSE = "pause"
+    UNPAUSE = "unpause"
+    SUSPEND = "suspend"
+    RESUME = "resume"
+    LOCK = "lock"
+    UNLOCK = "unlock"
+    RESCUE = "rescue"
+    UNRESCUE = "unrescue"
+    START = "start"
+    STOP = "stop"
+    SHELVE = "shelve"
+    SHELVE_OFFLOAD = "shelve_offload"
+    UNSHELVE = "unshelve"
+
+
 class ComputeTools:
     """
     A class to encapsulate Compute-related tools and utilities.
@@ -23,6 +42,7 @@ def register_tools(self, mcp: FastMCP):
         mcp.tool()(self.get_server)
         mcp.tool()(self.create_server)
         mcp.tool()(self.get_flavors)
+        mcp.tool()(self.action_server)
 
     def get_servers(self) -> list[Server]:
         """
@@ -102,3 +122,50 @@ def get_flavors(self) -> list[Flavor]:
         for flavor in conn.compute.flavors():
             flavor_list.append(Flavor(**flavor))
         return flavor_list
+
+    def action_server(self, id: str, action: ServerActionEnum):
+        """
+        Perform an action on a Compute server.
+
+        :param id: The ID of the server.
+        :param action: The action to perform.
+                      Available actions:
+                      - pause: Pauses a server. Changes its status to PAUSED
+                      - unpause: Unpauses a paused server and changes its status to ACTIVE
+                      - suspend: Suspends a server and changes its status to SUSPENDED
+                      - resume: Resumes a suspended server and changes its status to ACTIVE
+                      - lock: Locks a server
+                      - unlock: Unlocks a locked server
+                      - rescue: Puts a server in rescue mode and changes its status to RESCUE
+                      - unrescue: Unrescues a server. Changes status to ACTIVE
+                      - start: Starts a stopped server and changes its status to ACTIVE
+                      - stop: Stops a running server and changes its status to SHUTOFF
+                      - shelve: Shelves a server
+                      - shelve_offload: Shelf-offloads, or removes, a shelved server
+                      - unshelve: Unshelves, or restores, a shelved server
+                      Only above actions are currently supported
+        :raises ValueError: If the action is not supported or invalid(ConflictException).
+        """
+        conn = get_openstack_conn()
+
+        action_methods = {
+            ServerActionEnum.PAUSE: conn.compute.pause_server,
+            ServerActionEnum.UNPAUSE: conn.compute.unpause_server,
+            ServerActionEnum.SUSPEND: conn.compute.suspend_server,
+            ServerActionEnum.RESUME: conn.compute.resume_server,
+            ServerActionEnum.LOCK: conn.compute.lock_server,
+            ServerActionEnum.UNLOCK: conn.compute.unlock_server,
+            ServerActionEnum.RESCUE: conn.compute.rescue_server,
+            ServerActionEnum.UNRESCUE: conn.compute.unrescue_server,
+            ServerActionEnum.START: conn.compute.start_server,
+            ServerActionEnum.STOP: conn.compute.stop_server,
+            ServerActionEnum.SHELVE: conn.compute.shelve_server,
+            ServerActionEnum.SHELVE_OFFLOAD: conn.compute.shelve_offload_server,
+            ServerActionEnum.UNSHELVE: conn.compute.unshelve_server,
+        }
+
+        if action not in action_methods:
+            raise ValueError(f"Unsupported action: {action}")
+
+        action_methods[action](id)
+        return None

tests/tools/test_compute_tools.py

@@ -1,5 +1,9 @@
 from unittest.mock import Mock, call
 
+import pytest
+
+from openstack.exceptions import ConflictException, NotFoundException
+
 from openstack_mcp_server.tools.compute_tools import ComputeTools
 from openstack_mcp_server.tools.response.compute import Flavor, Server
 
@@ -261,9 +265,10 @@ def test_register_tools(self):
                 call(compute_tools.get_server),
                 call(compute_tools.create_server),
                 call(compute_tools.get_flavors),
+                call(compute_tools.action_server),
             ],
         )
-        assert mock_tool_decorator.call_count == 4
+        assert mock_tool_decorator.call_count == 5
 
     def test_compute_tools_instantiation(self):
         """Test ComputeTools can be instantiated."""
@@ -346,3 +351,84 @@ def test_get_flavors_empty_list(self, mock_get_openstack_conn):
 
         assert result == []
         mock_conn.compute.flavors.assert_called_once()
+
+    @pytest.mark.parametrize(
+        "action",
+        [
+            "pause",
+            "unpause",
+            "suspend",
+            "resume",
+            "lock",
+            "unlock",
+            "rescue",
+            "unrescue",
+            "start",
+            "stop",
+            "shelve",
+            "shelve_offload",
+            "unshelve",
+        ],
+    )
+    def test_action_server_success(self, mock_get_openstack_conn, action):
+        """Test action_server with all supported actions."""
+        mock_conn = mock_get_openstack_conn
+        server_id = "test-server-id"
+
+        # Mock the action method to avoid calling actual methods
+        action_method = getattr(mock_conn.compute, f"{action}_server")
+        action_method.return_value = None
+
+        compute_tools = ComputeTools()
+        result = compute_tools.action_server(server_id, action)
+
+        # Verify the result is None (void function)
+        assert result is None
+
+        # Verify the correct method was called with server ID
+        action_method.assert_called_once_with(server_id)
+
+    def test_action_server_unsupported_action(self, mock_get_openstack_conn):
+        """Test action_server with unsupported action raises ValueError."""
+        server_id = "test-server-id"
+        unsupported_action = "invalid_action"
+
+        compute_tools = ComputeTools()
+
+        with pytest.raises(
+            ValueError,
+            match=f"Unsupported action: {unsupported_action}",
+        ):
+            compute_tools.action_server(server_id, unsupported_action)
+
+    def test_action_server_not_found(self, mock_get_openstack_conn):
+        """Test action_server when server does not exist."""
+        mock_conn = mock_get_openstack_conn
+        server_id = "non-existent-server-id"
+        action = "pause"
+
+        # Mock the action method to raise NotFoundException
+        mock_conn.compute.pause_server.side_effect = NotFoundException()
+
+        compute_tools = ComputeTools()
+
+        with pytest.raises(NotFoundException):
+            compute_tools.action_server(server_id, action)
+
+        mock_conn.compute.pause_server.assert_called_once_with(server_id)
+
+    def test_action_server_conflict_exception(self, mock_get_openstack_conn):
+        """Test action_server when action cannot be performed due to Conflict Exception."""
+        mock_conn = mock_get_openstack_conn
+        server_id = "test-server-id"
+        action = "start"
+
+        # Mock the action method to raise ConflictException
+        mock_conn.compute.start_server.side_effect = ConflictException()
+
+        compute_tools = ComputeTools()
+
+        with pytest.raises(ConflictException):
+            compute_tools.action_server(server_id, action)
+
+        mock_conn.compute.start_server.assert_called_once_with(server_id)