feat : create image tool

src/openstack_mcp_server/tools/image_tools.py

@@ -1,11 +1,14 @@
 from fastmcp import FastMCP
 
+from openstack_mcp_server.tools.request.image import CreateImage
+from openstack_mcp_server.tools.response.image import Image
+
 from .base import get_openstack_conn
 
 
 class ImageTools:
     """
-    A class to encapsulate Compute-related tools and utilities.
+    A class to encapsulate Image-related tools and utilities.
     """
 
     def register_tools(self, mcp: FastMCP):
@@ -14,6 +17,7 @@ def register_tools(self, mcp: FastMCP):
         """
 
         mcp.tool()(self.get_image_images)
+        mcp.tool()(self.create_image)
 
     def get_image_images(self) -> str:
         """
@@ -32,3 +36,62 @@ def get_image_images(self) -> str:
             )
 
         return "\n".join(image_list)
+
+    def create_image(self, image_data: CreateImage) -> Image:
+        """Create a new Openstack image.
+        This method handles both cases of image creation:
+        1. If a volume is provided, it creates an image from the volume.
+        2. If no volume is provided, it creates an image using the Image imports method
+            import_options field is required for this method.
+        Following import methods are supported:
+        - glance-direct: The image data is made available to the Image service via the Stage binary
+        - web-download: The image data is made available to the Image service by being posted to an accessible location with a URL that you know.
+            - must provide a URI to the image data.
+        - copy-image: The image data is made available to the Image service by copying existing image
+        - glance-download: The image data is made available to the Image service by fetching an image accessible from another glance service specified by a region name and an image id that you know.
+            - must provide a glance_region and glance_image_id.
+
+        :param image_data: An instance of CreateImage containing the image details.
+        :return: An Image object representing the created image.
+        """
+        conn = get_openstack_conn()
+
+        if image_data.volume:
+            created_image = conn.block_storage.create_image(
+                name=image_data.name,
+                volume=image_data.volume,
+                allow_duplicates=image_data.allow_duplicates,
+                container_format=image_data.container_format,
+                disk_format=image_data.disk_format,
+                wait=False,
+                timeout=3600,
+            )
+        else:
+            # Create an image with Image imports
+            # First, Creates a catalog record for an operating system disk image.
+            created_image = conn.image.create_image(
+                name=image_data.name,
+                container=image_data.container,
+                container_format=image_data.container_format,
+                disk_format=image_data.disk_format,
+                min_disk=image_data.min_disk,
+                min_ram=image_data.min_ram,
+                tags=image_data.tags,
+                protected=image_data.protected,
+                visibility=image_data.visibility,
+                allow_duplicates=image_data.allow_duplicates,
+            )
+
+            # Then, import the image data
+            conn.image.import_image(
+                image=created_image,
+                method=image_data.import_options.import_method,
+                uri=image_data.import_options.uri,
+                stores=image_data.import_options.stores,
+                remote_region=image_data.import_options.glance_region,
+                remote_image_id=image_data.import_options.glance_image_id,
+                remote_service_interface=image_data.import_options.glance_service_interface,
+            )
+
+        image = conn.get_image(created_image.id)
+        return Image(**image)

src/openstack_mcp_server/tools/request/image.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class CreateImage(BaseModel):
+    """OpenStack Glance Image Creation Request Pydantic Model"""
+
+    id: str | None = Field(default=None)
+    volume: str | None = Field(default=None)
+    name: str | None = Field(default=None)
+    container: str | None = Field(default=None)
+    container_format: str | None = Field(default=None)
+    allow_duplicates: bool = Field(default=False)
+    disk_format: str | None = Field(default=None)
+    min_disk: int | None = Field(default=None)
+    min_ram: int | None = Field(default=None)
+    tags: list[str] | None = Field(default=[])
+    protected: bool | None = Field(default=False)
+    visibility: str | None = Field(default="public")
+    import_options: ImportOptions | None = Field(default=None)
+
+    class ImportOptions(BaseModel):
+        """Options for image import"""
+
+        import_method: str | None = Field(default=None)
+        stores: list[str] | None = Field(default=None)
+        uri: str | None = Field(default=None)
+        glance_region: str | None = Field(default=None)
+        glance_image_id: str | None = Field(default=None)
+        glance_service_interface: str | None = Field(default=None)

src/openstack_mcp_server/tools/response/image.py

@@ -0,0 +1,50 @@
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class OwnerSpecified(BaseModel):
+    """Owner specified metadata for OpenStack images"""
+
+    openstack_object: str | None = Field(
+        default=None,
+        alias="owner_specified.openstack.object",
+    )
+    openstack_sha256: str | None = Field(
+        default=None,
+        alias="'owner_specified.openstack.sha256'",
+    )
+    openstack_md5: str | None = Field(
+        default=None,
+        alias="'owner_specified.openstack.md5'",
+    )
+
+    model_config = ConfigDict(validate_by_name=True)
+
+
+class Image(BaseModel):
+    """OpenStack Glance Image Pydantic Model"""
+
+    id: str
+    name: str | None = Field(default=None)
+    checksum: str | None = Field(default=None)
+    container_format: str | None = Field(default=None)
+    disk_format: str | None = Field(default=None)
+    file: str | None = Field(default=None)
+    min_disk: int | None = Field(default=None)
+    min_ram: int | None = Field(default=None)
+    os_hash_algo: str | None = Field(default=None)
+    os_hash_value: str | None = Field(default=None)
+    size: int | None = Field(default=None)
+    virtual_size: int | None = Field(default=None)
+    owner: str | None = Field(default=None)
+    visibility: str | None = Field(default=None)
+    hw_rng_model: str | None = Field(default=None)
+    status: str | None = Field(default=None)
+    schema_: str | None = Field(default=None, alias="schema")
+    protected: bool | None = Field(default=None)
+    os_hidden: bool | None = Field(default=None)
+    tags: list[str] | None = Field(default=None)
+    properties: OwnerSpecified | None = Field(default=None)
+    model_config = ConfigDict(validate_by_name=True)
+
+    created_at: str | None = Field(default=None)
+    updated_at: str | None = Field(default=None)

tests/tools/test_image_tools.py

@@ -1,11 +1,51 @@
+import uuid
+
 from unittest.mock import Mock
 
 from openstack_mcp_server.tools.image_tools import ImageTools
+from openstack_mcp_server.tools.request.image import CreateImage
+from openstack_mcp_server.tools.response.image import Image
 
 
 class TestImageTools:
     """Test cases for ImageTools class."""
 
+    @staticmethod
+    def image_factory(**overrides):
+        defaults = {
+            "id": str(uuid.uuid4()),
+            "name": "test-image",
+            "checksum": "abc123",
+            "container_format": "bare",
+            "disk_format": "qcow2",
+            "file": None,
+            "min_disk": 1,
+            "min_ram": 512,
+            "os_hash_algo": "sha512",
+            "os_hash_value": "hash123",
+            "size": 1073741824,
+            "virtual_size": None,
+            "owner": str(uuid.uuid4()),
+            "visibility": "public",
+            "hw_rng_model": None,
+            "status": "active",
+            "schema": "/v2/schemas/image",
+            "protected": False,
+            "os_hidden": False,
+            "tags": [],
+            "properties": None,
+            "created_at": "2025-01-01T00:00:00Z",
+            "updated_at": "2025-01-01T00:00:00Z",
+            "owner_specified.openstack.md5": "a1b2c3d4e5f6",
+            "owner_specified.openstack.sha256": "a1b2c3d",
+            "owner_specified.openstack.object": "image",
+        }
+        for key, value in overrides.items():
+            if value is not None:
+                defaults[key] = value
+
+        return defaults
+
     def test_get_image_images_success(self, mock_get_openstack_conn_image):
         """Test getting image images successfully."""
         mock_conn = mock_get_openstack_conn_image
@@ -80,3 +120,113 @@ def test_get_image_images_with_empty_name(
         assert " (img-empty-name) - Status: active" in result  # Empty name
 
         mock_conn.image.images.assert_called_once()
+
+    def test_create_image_success_with_volume_id(
+        self,
+        mock_get_openstack_conn_image,
+    ):
+        """Test creating an image from a volume ID."""
+        volume_id = "6cf57d8d-00ca-43ff-ae6f-56912b69528a"  # Example volume ID
+
+        mock_image = self.image_factory()
+        mock_get_openstack_conn_image.block_storage.create_image.return_value = Mock(
+            id=mock_image["id"],
+        )
+        mock_get_openstack_conn_image.get_image.return_value = mock_image
+
+        # Create an instance with volume ID
+        image_tools = ImageTools()
+        image_data = CreateImage(
+            name=mock_image["name"],
+            volume=volume_id,
+            allow_duplicates=False,
+            container=mock_image["container_format"],
+            disk_format=mock_image["disk_format"],
+            container_format=mock_image["container_format"],
+            min_disk=mock_image["min_disk"],
+        )
+
+        expected_output = Image(**mock_image)
+
+        created_image = image_tools.create_image(image_data)
+
+        # Verify the created image
+        assert created_image == expected_output
+        assert mock_get_openstack_conn_image.block_storage.create_image.called_once_with(
+            name=mock_image["name"],
+            volume=volume_id,
+            allow_duplicates=False,
+            container=mock_image["container_format"],
+            disk_format=mock_image["disk_format"],
+            wait=False,
+            timeout=3600,
+        )
+
+        assert mock_get_openstack_conn_image.get_image.called_once_with(
+            mock_image["id"],
+        )
+
+    def test_create_image_success_with_import_options(
+        self,
+        mock_get_openstack_conn_image,
+    ):
+        """Test creating an image with import options."""
+        create_image_data = CreateImage(
+            name="example_image",
+            container="bare",
+            disk_format="qcow2",
+            container_format="bare",
+            min_disk=10,
+            min_ram=512,
+            tags=["example", "test"],
+            import_options=CreateImage.ImportOptions(
+                import_method="web-download",
+                uri="https://example.com/image.qcow2",
+            ),
+            allow_duplicates=False,
+        )
+
+        mock_image = self.image_factory(**create_image_data.__dict__)
+        mock_create_image = Mock(id=mock_image["id"])
+
+        mock_get_openstack_conn_image.image.create_image.return_value = (
+            mock_create_image
+        )
+        mock_get_openstack_conn_image.image.import_image.return_value = None
+        mock_get_openstack_conn_image.get_image.return_value = mock_image
+
+        # Create an instance with import options
+        image_tools = ImageTools()
+
+        expected_output = Image(**mock_image)
+
+        created_image = image_tools.create_image(create_image_data)
+
+        # Verify the created image
+        assert created_image == expected_output
+        assert (
+            mock_get_openstack_conn_image.image.create_image.called_once_with(
+                name=create_image_data.name,
+                container=create_image_data.container,
+                container_format=create_image_data.container_format,
+                disk_format=create_image_data.disk_format,
+                min_disk=create_image_data.min_disk,
+                min_ram=create_image_data.min_ram,
+                tags=create_image_data.tags,
+                protected=create_image_data.protected,
+                visibility=create_image_data.visibility,
+                allow_duplicates=create_image_data.allow_duplicates,
+            )
+        )
+        assert mock_get_openstack_conn_image.image.import_image.called_once_with(
+            image=mock_create_image,
+            method=create_image_data.import_options.import_method,
+            uri=create_image_data.import_options.uri,
+            stores=create_image_data.import_options.stores,
+            remote_region=create_image_data.import_options.glance_region,
+            remote_image_id=create_image_data.import_options.glance_image_id,
+            remote_service_interface=create_image_data.import_options.glance_service_interface,
+        )
+        assert mock_get_openstack_conn_image.get_image.called_once_with(
+            mock_image["id"],
+        )