From a92e19fc6a1b4bb73828765b76e64954efed2a06 Mon Sep 17 00:00:00 2001 From: Christian Berendt Date: Tue, 25 Nov 2025 08:58:06 +0100 Subject: [PATCH] Add support for default_volume_type per project Allow setting a default volume type for projects via: - Project property (default_volume_type) - takes precedence - Quotaclass configuration (default_volume_type key) The volume type existence is verified before setting. AI-assisted: Claude Code Signed-off-by: Christian Berendt --- openstack_project_manager/manage.py | 70 +++++++++++++++++ test/unit/test_manage.py | 118 ++++++++++++++++++++++++++++ 2 files changed, 188 insertions(+) diff --git a/openstack_project_manager/manage.py b/openstack_project_manager/manage.py index 6838c09..d804942 100755 --- a/openstack_project_manager/manage.py +++ b/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) diff --git a/test/unit/test_manage.py b/test/unit/test_manage.py index a560c87..41267bb 100644 --- a/test/unit/test_manage.py +++ b/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):