diff --git a/src/openstack_mcp_server/tools/block_storage_tools.py b/src/openstack_mcp_server/tools/block_storage_tools.py index 5397c16..5c558bd 100644 --- a/src/openstack_mcp_server/tools/block_storage_tools.py +++ b/src/openstack_mcp_server/tools/block_storage_tools.py @@ -25,6 +25,7 @@ def register_tools(self, mcp: FastMCP): mcp.tool()(self.extend_volume) mcp.tool()(self.get_attachment_details) + mcp.tool()(self.get_attachments) def get_volumes(self) -> list[Volume]: """ @@ -225,3 +226,41 @@ def get_attachment_details(self, attachment_id: str) -> Attachment: } return Attachment(**params) + + def get_attachments( + self, + volume_id: str | None = None, + instance: str | None = None, + ) -> list[Attachment]: + """ + Get the list of attachments. + + :param volume_id: The ID of the volume. + :param instance: The ID of the instance. + :return: A list of Attachment objects. + """ + conn = get_openstack_conn() + + filter = {} + if volume_id: + filter["volume_id"] = volume_id + if instance: + filter["instance"] = instance + + attachments = [] + for attachment in conn.block_storage.attachments(**filter): + attachments.append( + Attachment( + id=attachment.id, + instance=attachment.instance, + volume_id=attachment.volume_id, + status=attachment.status, + connection_info=attachment.connection_info, + attach_mode=attachment.attach_mode, + connector=attachment.connector, + attached_at=attachment.attached_at, + detached_at=attachment.detached_at, + ) + ) + + return attachments diff --git a/tests/tools/test_block_storage_tools.py b/tests/tools/test_block_storage_tools.py index aa5cb84..09b57e0 100644 --- a/tests/tools/test_block_storage_tools.py +++ b/tests/tools/test_block_storage_tools.py @@ -741,3 +741,45 @@ def test_get_attachment_details( mock_conn.block_storage.get_attachment.assert_called_once_with( "attach-123" ) + + def test_get_attachments(self, mock_get_openstack_conn_block_storage): + """Test getting attachments.""" + mock_conn = mock_get_openstack_conn_block_storage + + # Create mock attachment object + mock_attachment = Mock() + mock_attachment.id = "attach-123" + mock_attachment.instance = "server-123" + mock_attachment.volume_id = "vol-123" + mock_attachment.status = "attached" + mock_attachment.connection_info = None + mock_attachment.connector = None + mock_attachment.attach_mode = None + mock_attachment.attached_at = None + mock_attachment.detached_at = None + + mock_conn.block_storage.attachments.return_value = [mock_attachment] + + # Test attachments + block_storage_tools = BlockStorageTools() + + filter = { + "volume_id": "vol-123", + "instance": "server-123", + } + result = block_storage_tools.get_attachments(**filter) + + # Verify the result + assert isinstance(result, list) + assert len(result) == 1 + assert result[0].id == "attach-123" + assert result[0].instance == "server-123" + assert result[0].volume_id == "vol-123" + assert result[0].attached_at is None + assert result[0].detached_at is None + assert result[0].attach_mode is None + assert result[0].connection_info is None + assert result[0].connector is None + + # Verify the mock calls + mock_conn.block_storage.attachments.assert_called_once_with(**filter)