Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 46 additions & 1 deletion src/openstack_mcp_server/tools/compute_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,8 @@ def register_tools(self, mcp: FastMCP):
mcp.tool()(self.create_server)
mcp.tool()(self.get_flavors)
mcp.tool()(self.action_server)
mcp.tool()(self.update_server)
mcp.tool()(self.delete_server)

def get_servers(self) -> list[Server]:
"""
Expand Down Expand Up @@ -123,7 +125,7 @@ def get_flavors(self) -> list[Flavor]:
flavor_list.append(Flavor(**flavor))
return flavor_list

def action_server(self, id: str, action: ServerActionEnum):
def action_server(self, id: str, action: ServerActionEnum) -> None:
"""
Perform an action on a Compute server.

Expand Down Expand Up @@ -169,3 +171,46 @@ def action_server(self, id: str, action: ServerActionEnum):

action_methods[action](id)
return None

def update_server(
self,
id: str,
accessIPv4: str | None = None,
accessIPv6: str | None = None,
name: str | None = None,
hostname: str | None = None,
description: str | None = None,
) -> Server:
"""
Update a Compute server's name, hostname, or description.

:param id: The UUID of the server.
:param accessIPv4: IPv4 address that should be used to access this server.
:param accessIPv6: IPv6 address that should be used to access this server.
:param name: The server name.
:param hostname: The hostname to configure for the instance in the metadata service.
:param description: A free form description of the server.
:return: The updated Server object.
"""
conn = get_openstack_conn()
server_params = {
"accessIPv4": accessIPv4,
"accessIPv6": accessIPv6,
"name": name,
"hostname": hostname,
"description": description,
}
server_params = {
k: v for k, v in server_params.items() if v is not None
}
server = conn.compute.update_server(id, **server_params)
return Server(**server)

def delete_server(self, id: str) -> None:
"""
Delete a Compute server.

:param id: The UUID of the server.
"""
conn = get_openstack_conn()
conn.compute.delete_server(id)
4 changes: 4 additions & 0 deletions src/openstack_mcp_server/tools/response/compute.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,12 +25,16 @@ class SecurityGroup(BaseModel):

id: str
name: str
hostname: str | None = None
description: str | None = None
status: str | None = None
flavor: Flavor | None = None
image: Image | None = None
addresses: dict[str, list[IPAddress]] | None = None
key_name: str | None = None
security_groups: list[SecurityGroup] | None = None
accessIPv4: str | None = None
accessIPv6: str | None = None


class Flavor(BaseModel):
Expand Down
125 changes: 124 additions & 1 deletion tests/tools/test_compute_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -266,9 +266,11 @@ def test_register_tools(self):
call(compute_tools.create_server),
call(compute_tools.get_flavors),
call(compute_tools.action_server),
call(compute_tools.update_server),
call(compute_tools.delete_server),
],
)
assert mock_tool_decorator.call_count == 5
assert mock_tool_decorator.call_count == 7

def test_compute_tools_instantiation(self):
"""Test ComputeTools can be instantiated."""
Expand Down Expand Up @@ -432,3 +434,124 @@ def test_action_server_conflict_exception(self, mock_get_openstack_conn):
compute_tools.action_server(server_id, action)

mock_conn.compute.start_server.assert_called_once_with(server_id)

def test_update_server_success(self, mock_get_openstack_conn):
"""Test updating a server successfully with all parameters."""
mock_conn = mock_get_openstack_conn
server_id = "test-server-id"

mock_server = {
"name": "updated-server",
"id": server_id,
"status": "ACTIVE",
"hostname": "updated-hostname",
"description": "Updated server description",
"accessIPv4": "192.168.1.100",
"accessIPv6": "2001:db8::1",
}

mock_conn.compute.update_server.return_value = mock_server

compute_tools = ComputeTools()
server_params = mock_server.copy()
server_params.pop("status")
result = compute_tools.update_server(**server_params)

expected_output = Server(**mock_server)
assert result == expected_output

expected_params = {
"accessIPv4": "192.168.1.100",
"accessIPv6": "2001:db8::1",
"name": "updated-server",
"hostname": "updated-hostname",
"description": "Updated server description",
}
mock_conn.compute.update_server.assert_called_once_with(
server_id, **expected_params
)

@pytest.mark.parametrize(
"params",
[
{"param_key": "name", "value": "new-name"},
{"param_key": "hostname", "value": "new-hostname"},
{"param_key": "description", "value": "New description"},
{"param_key": "accessIPv4", "value": "192.168.1.100"},
{"param_key": "accessIPv6", "value": "2001:db8::1"},
],
)
def test_update_server_optional_params(
self, mock_get_openstack_conn, params
):
"""Test updating a server with optional parameters."""
mock_conn = mock_get_openstack_conn
server_id = "test-server-id"

mock_server = {
"id": server_id,
"name": "original-name",
"description": "Original description",
"hostname": "original-hostname",
"accessIPv4": "1.1.1.1",
"accessIPv6": "::",
"status": "ACTIVE",
**{params["param_key"]: params["value"]},
}

mock_conn.compute.update_server.return_value = mock_server

compute_tools = ComputeTools()
result = compute_tools.update_server(
id=server_id,
**{params["param_key"]: params["value"]},
)
assert result == Server(**mock_server)

expected_params = {params["param_key"]: params["value"]}
mock_conn.compute.update_server.assert_called_once_with(
server_id, **expected_params
)

def test_update_server_not_found(self, mock_get_openstack_conn):
"""Test updating a server that does not exist."""
mock_conn = mock_get_openstack_conn
server_id = "non-existent-server-id"

# Mock the update_server method to raise NotFoundException
mock_conn.compute.update_server.side_effect = NotFoundException()

compute_tools = ComputeTools()

with pytest.raises(NotFoundException):
compute_tools.update_server(id=server_id)

mock_conn.compute.update_server.assert_called_once_with(server_id)

def test_delete_server_success(self, mock_get_openstack_conn):
"""Test deleting a server successfully."""
mock_conn = mock_get_openstack_conn
server_id = "test-server-id"

mock_conn.compute.delete_server.return_value = None

compute_tools = ComputeTools()
result = compute_tools.delete_server(server_id)

assert result is None
mock_conn.compute.delete_server.assert_called_once_with(server_id)

def test_delete_server_not_found(self, mock_get_openstack_conn):
"""Test deleting a server that does not exist."""
mock_conn = mock_get_openstack_conn
server_id = "non-existent-server-id"

# Mock the delete_server method to raise NotFoundException
mock_conn.compute.delete_server.side_effect = NotFoundException()

compute_tools = ComputeTools()

with pytest.raises(NotFoundException):
compute_tools.delete_server(server_id)

mock_conn.compute.delete_server.assert_called_once_with(server_id)