Add support for default_volume_type per project

openstack_project_manager/manage.py

@@ -445,6 +445,75 @@ def manage_external_network_rbacs(
         del_service_network(configuration, project, public_net_name)
 
 
+def check_default_volume_type(
+    configuration: Configuration,
+    project: openstack.identity.v3.project.Project,
+    domain: openstack.identity.v3.domain.Domain,
+    classes: str,
+) -> None:
+    """Set default volume type for a project.
+
+    The default_volume_type can be specified either:
+    1. As a project property (takes precedence)
+    2. In the quotaclass configuration
+
+    Before setting, the volume type existence is verified.
+    """
+    # Determine the default_volume_type value
+    default_volume_type = None
+
+    # Check project property first (takes precedence)
+    if "default_volume_type" in project:
+        default_volume_type = project.default_volume_type
+    else:
+        # Check quotaclass
+        if "quotaclass" in project:
+            quotaclass = get_quotaclass(classes, project.quotaclass)
+        else:
+            if domain.name.startswith("ok"):
+                quotaclass = get_quotaclass(classes, "okeanos")
+            else:
+                quotaclass = get_quotaclass(classes, "basic")
+
+        if quotaclass and "default_volume_type" in quotaclass:
+            default_volume_type = quotaclass["default_volume_type"]
+
+    if not default_volume_type:
+        return
+
+    logger.info(f"{project.name} - check default volume type")
+
+    # Verify the volume type exists
+    volume_type = configuration.os_cloud.block_storage.find_type(default_volume_type)
+
+    if not volume_type:
+        logger.warning(
+            f"{project.name} - default volume type {default_volume_type} not found"
+        )
+        return
+
+    # Get current default volume type for the project
+    try:
+        current_default = configuration.os_cloud.block_storage.show_default_type(
+            project.id
+        )
+        current_type_id = (
+            current_default.get("volume_type_id") if current_default else None
+        )
+    except Exception:
+        current_type_id = None
+
+    # Set default volume type if different
+    if current_type_id != volume_type.id:
+        logger.info(
+            f"{project.name} - setting default volume type to {default_volume_type}"
+        )
+        if not configuration.dry_run:
+            configuration.os_cloud.block_storage.set_default_type(
+                project.id, volume_type.id
+            )
+
+
 def check_volume_types(
     configuration: Configuration,
     project: openstack.identity.v3.project.Project,
@@ -1271,6 +1340,7 @@ def process_project(
             create_network_resources(configuration, project, domain)
 
         check_volume_types(configuration, project, domain, classes)
+        check_default_volume_type(configuration, project, domain, classes)
 
         if manage_privatevolumetypes:
             manage_private_volumetypes(configuration, project, domain)

test/unit/test_manage.py

@@ -14,6 +14,7 @@
     check_quota,
     update_bandwidth_policy_rule,
     manage_external_network_rbacs,
+    check_default_volume_type,
     check_volume_types,
     check_bandwidth_limit,
     manage_private_volumetypes,
@@ -72,6 +73,10 @@
     item1:
       name: name
 
+default_volume_type_test:
+  parent: default
+  default_volume_type: ssd
+
 flavor_test:
   parent: default
   flavors:
@@ -620,6 +625,119 @@ def test_manage_private_volumetypes_1(self):
         self.config.os_cloud.block_storage.types.assert_not_called()
 
 
+class TestCheckDefaultVolumeType(TestBase):
+
+    def setUp(self):
+        self.select_quota_class = "default_volume_type_test"
+        super().setUp()
+
+        self.mock_project = MagicMock()
+        self.mock_project.id = 1234
+        self.mock_project.name = "test_project"
+
+        self.mock_domain = MagicMock()
+        self.mock_domain.id = 5678
+        self.mock_domain.name = "TestDomain"
+
+    def mock_volume_type(self, name, type_id):
+        vt = MagicMock()
+        vt.name = name
+        vt.id = type_id
+        return vt
+
+    def test_check_default_volume_type_from_quotaclass(self):
+        """Test setting default volume type from quotaclass."""
+        # Only quotaclass is set, not default_volume_type project property
+        self.mock_project.__contains__.side_effect = lambda x: x == "quotaclass"
+        self.mock_project.quotaclass = "default_volume_type_test"
+
+        vt = self.mock_volume_type("ssd", "vt-123")
+        self.config.os_cloud.block_storage.find_type.return_value = vt
+        self.config.os_cloud.block_storage.show_default_type.return_value = None
+
+        check_default_volume_type(
+            self.config, self.mock_project, self.mock_domain, "classes.yaml"
+        )
+
+        self.config.os_cloud.block_storage.find_type.assert_called_once_with("ssd")
+        self.config.os_cloud.block_storage.set_default_type.assert_called_once_with(
+            1234, "vt-123"
+        )
+
+    def test_check_default_volume_type_from_project_property(self):
+        """Test setting default volume type from project property (takes precedence)."""
+        self.mock_project.__contains__.side_effect = lambda x: x in [
+            "quotaclass",
+            "default_volume_type",
+        ]
+        self.mock_project.quotaclass = "default_volume_type_test"
+        self.mock_project.default_volume_type = "hdd"
+
+        vt = self.mock_volume_type("hdd", "vt-456")
+        self.config.os_cloud.block_storage.find_type.return_value = vt
+        self.config.os_cloud.block_storage.show_default_type.return_value = None
+
+        check_default_volume_type(
+            self.config, self.mock_project, self.mock_domain, "classes.yaml"
+        )
+
+        # Should use project property, not quotaclass
+        self.config.os_cloud.block_storage.find_type.assert_called_once_with("hdd")
+        self.config.os_cloud.block_storage.set_default_type.assert_called_once_with(
+            1234, "vt-456"
+        )
+
+    def test_check_default_volume_type_not_found(self):
+        """Test handling when volume type does not exist."""
+        # Only quotaclass is set, not default_volume_type project property
+        self.mock_project.__contains__.side_effect = lambda x: x == "quotaclass"
+        self.mock_project.quotaclass = "default_volume_type_test"
+
+        self.config.os_cloud.block_storage.find_type.return_value = None
+
+        check_default_volume_type(
+            self.config, self.mock_project, self.mock_domain, "classes.yaml"
+        )
+
+        self.config.os_cloud.block_storage.find_type.assert_called_once_with("ssd")
+        self.config.os_cloud.block_storage.set_default_type.assert_not_called()
+
+    def test_check_default_volume_type_already_set(self):
+        """Test skipping when default volume type is already set correctly."""
+        # Only quotaclass is set, not default_volume_type project property
+        self.mock_project.__contains__.side_effect = lambda x: x == "quotaclass"
+        self.mock_project.quotaclass = "default_volume_type_test"
+
+        vt = self.mock_volume_type("ssd", "vt-123")
+        self.config.os_cloud.block_storage.find_type.return_value = vt
+        self.config.os_cloud.block_storage.show_default_type.return_value = {
+            "volume_type_id": "vt-123"
+        }
+
+        check_default_volume_type(
+            self.config, self.mock_project, self.mock_domain, "classes.yaml"
+        )
+
+        self.config.os_cloud.block_storage.find_type.assert_called_once_with("ssd")
+        self.config.os_cloud.block_storage.set_default_type.assert_not_called()
+
+    def test_check_default_volume_type_no_config(self):
+        """Test when no default_volume_type is configured."""
+        # Only quotaclass is set, using 'default' class which has no default_volume_type
+        self.mock_project.__contains__.side_effect = lambda x: x == "quotaclass"
+        self.mock_project.quotaclass = "default"
+
+        # Override the mock to return a quotaclass without default_volume_type
+        self.mock_get_quotaclass.return_value = copy.copy(MOCK_QUOTA_CLASSES["default"])
+
+        check_default_volume_type(
+            self.config, self.mock_project, self.mock_domain, "classes.yaml"
+        )
+
+        self.config.os_cloud.block_storage.find_type.assert_not_called()
+        self.config.os_cloud.block_storage.set_default_type.assert_not_called()
+
+
 class TestCheckPrivateFlavorTypes(TestBase):
 
     def setUp(self):