feat: generate inspurinstorage charm

sunbeam-python/sunbeam/storage/backends/inspurinstorage/__init__.py

@@ -0,0 +1,4 @@
+# SPDX-FileCopyrightText: 2026 - Canonical Ltd
+# SPDX-License-Identifier: Apache-2.0
+
+"""Inspur InStorage backend for Sunbeam storage."""

sunbeam-python/sunbeam/storage/backends/inspurinstorage/backend.py

@@ -0,0 +1,182 @@
+# SPDX-FileCopyrightText: 2026 - Canonical Ltd
+# SPDX-License-Identifier: Apache-2.0
+
+"""Inspur InStorageMCS backend implementation using base step classes."""
+
+import logging
+from enum import StrEnum
+from typing import Annotated
+
+from pydantic import Field
+from rich.console import Console
+
+from sunbeam.core.manifest import StorageBackendConfig
+from sunbeam.storage.base import StorageBackendBase
+from sunbeam.storage.models import SecretDictField
+
+LOG = logging.getLogger(__name__)
+console = Console()
+
+
+class Protocol(StrEnum):
+    """Enumeration of valid protocol types."""
+
+    FC = "fc"
+    ISCSI = "iscsi"
+
+
+class InspurinstorageConfig(StorageBackendConfig):
+    """Configuration model for Inspur InStorageMCS backend.
+
+    This model includes ALL configuration options for the backend.
+    Additional configuration can be managed dynamically through the charm.
+    """
+
+    # Mandatory connection parameters
+    san_ip: Annotated[str, Field(description="IP address of SAN controller")]
+    san_login: Annotated[
+        str,
+        Field(description="Username for SAN controller"),
+        SecretDictField(field="san-login"),
+    ]
+    san_password: Annotated[
+        str,
+        Field(description="Password for SAN controller"),
+        SecretDictField(field="san-password"),
+    ]
+
+    # Optional backend configuration
+    protocol: Annotated[
+        Protocol | None,
+        Field(description="Protocol selector: fc, iscsi."),
+    ] = None
+
+    instorage_mcs_vol_autoexpand: Annotated[
+        bool | None,
+        Field(
+            description=("Storage system autoexpand parameter for volumes (True/False)")
+        ),
+    ] = None
+    instorage_mcs_vol_compression: Annotated[
+        bool | None, Field(description="Storage system compression option for volumes")
+    ] = None
+    instorage_mcs_vol_intier: Annotated[
+        bool | None, Field(description="Enable InTier for volumes")
+    ] = None
+    instorage_mcs_allow_tenant_qos: Annotated[
+        bool | None, Field(description="Allow tenants to specify QOS on create")
+    ] = None
+    instorage_mcs_vol_grainsize: Annotated[
+        int | None,
+        Field(
+            description=(
+                "Storage system grain size parameter for volumes (32/64/128/256)"
+            )
+        ),
+    ] = None
+    instorage_mcs_vol_rsize: Annotated[
+        int | None,
+        Field(
+            description=(
+                "Storage system space-efficiency parameter for volumes (percentage)"
+            )
+        ),
+    ] = None
+    instorage_mcs_vol_warning: Annotated[
+        int | None,
+        Field(
+            description=(
+                "Storage system threshold for volume capacity warnings (percentage)"
+            )
+        ),
+    ] = None
+    instorage_mcs_localcopy_timeout: Annotated[
+        int | None,
+        Field(
+            description=(
+                "Maximum number of seconds to wait for LocalCopy to be prepared."
+            )
+        ),
+    ] = None
+    instorage_mcs_localcopy_rate: Annotated[
+        int | None,
+        Field(
+            description=(
+                "Specifies the InStorage LocalCopy copy rate "
+                "used when creating a full volume copy."
+            )
+        ),
+    ] = None
+    instorage_mcs_vol_iogrp: Annotated[
+        str | None, Field(description="The I/O group in which to allocate volumes.")
+    ] = None
+    instorage_san_secondary_ip: Annotated[
+        str | None,
+        Field(
+            description=(
+                "Specifies secondary management IP or hostname "
+                "used if san_ip is invalid or inaccessible."
+            )
+        ),
+    ] = None
+    instorage_mcs_volpool_name: Annotated[
+        str | None,
+        Field(
+            description=(
+                "Comma-separated list of storage system storage pools for volumes."
+            )
+        ),
+    ] = None
+    instorage_mcs_iscsi_chap_enabled: Annotated[
+        bool | None,
+        Field(
+            description=(
+                "Configure CHAP authentication for iSCSI "
+                "connections (default: enabled)."
+            )
+        ),
+    ] = None
+    san_thin_provision: Annotated[
+        bool | None, Field(description="Use thin provisioning for SAN volumes?")
+    ] = None
+    use_multipath_for_image_xfer: Annotated[
+        bool | None,
+        Field(description="Enable multipathing for image transfer operations."),
+    ] = None
+
+
+class InspurinstorageBackend(StorageBackendBase):
+    """Inspur InStorageMCS backend implementation."""
+
+    backend_type = "inspurinstorage"
+    display_name = "InStorageMCS"
+    generally_available = True
+
+    @property
+    def charm_name(self) -> str:
+        """Return the charm application name."""
+        return "cinder-volume-inspurinstorage"
+
+    @property
+    def charm_channel(self) -> str:
+        """Return the default charm channel."""
+        return "latest/edge"
+
+    @property
+    def charm_revision(self) -> str | None:
+        """Return a pinned charm revision, if any."""
+        return None
+
+    @property
+    def charm_base(self) -> str:
+        """Return the target base for this charm."""
+        return "ubuntu@24.04"
+
+    @property
+    def supports_ha(self) -> bool:
+        """Whether this backend supports HA deployments."""
+        return False
+
+    def config_type(self) -> type[StorageBackendConfig]:
+        """Return the configuration model type for this backend."""
+        return InspurinstorageConfig

sunbeam-python/tests/unit/sunbeam/storage/backends/conftest.py

@@ -8,6 +8,7 @@
 from sunbeam.storage.backends.dellpowerstore.backend import DellPowerstoreBackend
 from sunbeam.storage.backends.dellsc.backend import DellSCBackend
 from sunbeam.storage.backends.hitachi.backend import HitachiBackend
+from sunbeam.storage.backends.inspurinstorage.backend import InspurinstorageBackend
 from sunbeam.storage.backends.purestorage.backend import PureStorageBackend
 
 
@@ -29,19 +30,28 @@ def dellsc_backend():
     return DellSCBackend()
 
 
+@pytest.fixture
+def inspurinstorage_backend():
+    """Provide an Inspur InStorage backend instance."""
+    return InspurinstorageBackend()
+
+
 @pytest.fixture
 def dellpowerstore_backend():
     """Provide a Dell PowerStore backend instance."""
     return DellPowerstoreBackend()
 
 
-@pytest.fixture(params=["hitachi", "purestorage", "dellsc", "dellpowerstore"])
+@pytest.fixture(
+    params=["hitachi", "purestorage", "dellsc", "inspurinstorage", "dellpowerstore"]
+)
 def any_backend(request):
     """Parametrized fixture that provides each backend type."""
     backends = {
         "hitachi": HitachiBackend(),
         "purestorage": PureStorageBackend(),
         "dellsc": DellSCBackend(),
+        "inspurinstorage": InspurinstorageBackend(),
         "dellpowerstore": DellPowerstoreBackend(),
     }
     return backends[request.param]

sunbeam-python/tests/unit/sunbeam/storage/backends/test_common.py

@@ -157,13 +157,18 @@ def backend(self, any_backend):
 
 
 def test_all_backends_have_unique_types(
-    hitachi_backend, purestorage_backend, dellsc_backend, dellpowerstore_backend
+    hitachi_backend,
+    purestorage_backend,
+    dellsc_backend,
+    inspurinstorage_backend,
+    dellpowerstore_backend,
 ):
     """Test that all backends have unique type identifiers."""
     backends = [
         hitachi_backend,
         purestorage_backend,
         dellsc_backend,
+        inspurinstorage_backend,
         dellpowerstore_backend,
     ]
     types = [b.backend_type for b in backends]
@@ -173,13 +178,18 @@ def test_all_backends_have_unique_types(
 
 
 def test_all_backends_have_unique_charm_names(
-    hitachi_backend, purestorage_backend, dellsc_backend, dellpowerstore_backend
+    hitachi_backend,
+    purestorage_backend,
+    dellsc_backend,
+    inspurinstorage_backend,
+    dellpowerstore_backend,
 ):
     """Test that all backends have unique charm names."""
     backends = [
         hitachi_backend,
         purestorage_backend,
         dellsc_backend,
+        inspurinstorage_backend,
         dellpowerstore_backend,
     ]
     charm_names = [b.charm_name for b in backends]
@@ -196,6 +206,7 @@ def test_all_backends_have_unique_charm_names(
         ("hitachi", "hitachi"),
         ("purestorage", "purestorage"),
         ("dellsc", "dellsc"),
+        ("inspurinstorage", "inspurinstorage"),
         ("dellpowerstore", "dellpowerstore"),
     ],
 )
@@ -211,6 +222,7 @@ def test_backend_types_match_expected(any_backend, backend_type, expected_type):
         ("hitachi", "cinder-volume-hitachi"),
         ("purestorage", "cinder-volume-purestorage"),
         ("dellsc", "cinder-volume-dellsc"),
+        ("inspurinstorage", "cinder-volume-inspurinstorage"),
         ("dellpowerstore", "cinder-volume-dellpowerstore"),
     ],
 )

sunbeam-python/tests/unit/sunbeam/storage/backends/test_inspurinstorage.py

@@ -0,0 +1,86 @@
+# SPDX-FileCopyrightText: 2026 - Canonical Ltd
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for Inspur InStorage backend."""
+
+import pytest
+from pydantic import ValidationError
+
+from sunbeam.storage.models import SecretDictField
+from tests.unit.sunbeam.storage.backends.test_common import BaseBackendTests
+
+
+class TestInspurinstorageBackend(BaseBackendTests):
+    """Tests for Inspur InStorage backend."""
+
+    @pytest.fixture
+    def backend(self, inspurinstorage_backend):
+        """Provide Inspur InStorage backend instance."""
+        return inspurinstorage_backend
+
+    def test_backend_type_is_inspurinstorage(self, backend):
+        """Test that backend type is 'inspurinstorage'."""
+        assert backend.backend_type == "inspurinstorage"
+
+    def test_charm_name_is_inspurinstorage_charm(self, backend):
+        """Test that charm name is cinder-volume-inspurinstorage."""
+        assert backend.charm_name == "cinder-volume-inspurinstorage"
+
+    def test_config_has_expected_fields(self, backend):
+        """Test that InStorage config has expected fields."""
+        fields = backend.config_type().model_fields
+        for field in ("san_ip", "san_login", "san_password", "protocol"):
+            assert field in fields, f"Expected field {field} not found in config"
+
+    def test_san_credentials_are_secret(self, backend):
+        """Test that SAN login and password are marked as secrets."""
+        config_class = backend.config_type()
+        for field_name in ("san_login", "san_password"):
+            field = config_class.model_fields.get(field_name)
+            assert field is not None
+            assert any(isinstance(m, SecretDictField) for m in field.metadata), (
+                f"{field_name} should be marked as secret"
+            )
+
+
+class TestInspurinstorageConfigValidation:
+    """Test Inspur InStorage config validation behavior."""
+
+    def test_protocol_rejects_invalid_value(self, inspurinstorage_backend):
+        """Test that protocol rejects values other than fc/iscsi."""
+        config_class = inspurinstorage_backend.config_type()
+        with pytest.raises(ValidationError):
+            config_class.model_validate(
+                {
+                    "san-ip": "192.168.1.1",
+                    "san-login": "admin",
+                    "san-password": "secret",
+                    "protocol": "nvme",
+                }
+            )
+
+    def test_protocol_accepts_fc(self, inspurinstorage_backend):
+        """Test that protocol accepts fc."""
+        config_class = inspurinstorage_backend.config_type()
+        config = config_class.model_validate(
+            {
+                "san-ip": "192.168.1.1",
+                "san-login": "admin",
+                "san-password": "secret",
+                "protocol": "fc",
+            }
+        )
+        assert config.protocol == "fc"
+
+    def test_protocol_accepts_iscsi(self, inspurinstorage_backend):
+        """Test that protocol accepts iscsi."""
+        config_class = inspurinstorage_backend.config_type()
+        config = config_class.model_validate(
+            {
+                "san-ip": "192.168.1.1",
+                "san-login": "admin",
+                "san-password": "secret",
+                "protocol": "iscsi",
+            }
+        )
+        assert config.protocol == "iscsi"