diff --git a/ironic/networking/switch_config.py b/ironic/networking/switch_config.py new file mode 100644 index 0000000000..5da3001bd3 --- /dev/null +++ b/ironic/networking/switch_config.py @@ -0,0 +1,140 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +""" +Utilities for switch configuration and VLAN management. +""" + +from oslo_config import cfg +from oslo_log import log + +from ironic.common import exception +from ironic.common.i18n import _ +from ironic.networking import utils + +LOG = log.getLogger(__name__) +CONF = cfg.CONF + + +def validate_vlan_allowed(vlan_id, allowed_vlans_config=None, + switch_config=None): + """Validate that a VLAN ID is allowed. + + :param vlan_id: The VLAN ID to validate + :param allowed_vlans_config: Global list of allowed vlans from config + :param switch_config: Optional switch-specific configuration dict that + may contain an 'allowed_vlans' key + :returns: True if the VLAN is allowed + :raises: InvalidParameterValue if the VLAN is not allowed + """ + # Check switch-specific configuration first (if provided) + if switch_config and 'allowed_vlans' in switch_config: + allowed_spec = switch_config['allowed_vlans'] + else: + # Fall back to global configuration + if allowed_vlans_config is not None: + allowed_spec = allowed_vlans_config + else: + allowed_spec = CONF.ironic_networking.allowed_vlans + + # None means all VLANs are allowed + if allowed_spec is None: + return True + + # Empty list means no VLANs are allowed + if isinstance(allowed_spec, list) and not allowed_spec: + raise exception.InvalidParameterValue( + _('VLAN %(vlan)s is not allowed: no VLANs are permitted by ' + 'configuration') % {'vlan': vlan_id}) + + # Parse and check against allowed VLANs + allowed_vlans = utils.parse_vlan_ranges(allowed_spec) + if vlan_id not in allowed_vlans: + raise exception.InvalidParameterValue( + _('VLAN %(vlan)s is not in the list of allowed VLANs') % + {'vlan': vlan_id}) + + return True + + +def get_switch_vlan_config(switch_driver, switch_id): + """Get VLAN configuration for a switch. + + Retrieves switch-specific VLAN configuration from the driver, with + fallback to global configuration options. + + :param switch_driver: The switch driver instance. + :param switch_id: Identifier of the switch. + :returns: Dictionary containing allowed_vlans + :raises: Any exceptions from switch_driver.get_switch_info() + """ + config = { + "allowed_vlans": set(), + } + + # Get switch-specific configuration from driver + switch_info = switch_driver.get_switch_info(switch_id) + # Process switch-specific allowed_vlans + if switch_info and "allowed_vlans" in switch_info: + switch_allowed = switch_info["allowed_vlans"] + config["allowed_vlans"] = ( + utils.parse_vlan_ranges(switch_allowed)) + + # Use global config if switch-specific config is not available + if not config["allowed_vlans"] and CONF.ironic_networking.allowed_vlans: + config["allowed_vlans"] = utils.parse_vlan_ranges( + CONF.ironic_networking.allowed_vlans + ) + + return config + + +def validate_vlan_configuration( + vlans_to_check, switch_driver, switch_id, operation_description="operation" +): + """Validate VLAN configuration against allowed lists. + + Checks if the specified VLANs are allowed according to the switch-specific + or global configuration. + + :param vlans_to_check: List of VLAN IDs to validate. + :param switch_driver: The switch driver instance. + :param switch_id: Identifier of the switch. + :param operation_description: Description of the operation for error + messages. + :raises: InvalidParameterValue if any VLAN is not allowed. + """ + if not vlans_to_check: + return + + # Get switch-specific configuration + config = get_switch_vlan_config(switch_driver, switch_id) + allowed_vlans = config["allowed_vlans"] + + # Convert to set for easier processing + vlans_set = set(vlans_to_check) + + # Check allowed VLANs if specified + if allowed_vlans: + disallowed_requested = vlans_set - allowed_vlans + if disallowed_requested: + raise exception.InvalidParameterValue( + _("VLANs %(vlans)s are not allowed for %(operation)s on " + "switch %(switch)s. Allowed VLANs: %(allowed)s" + ) % { + "vlans": sorted(disallowed_requested), + "operation": operation_description, + "switch": switch_id, + "allowed": sorted(allowed_vlans), + } + ) diff --git a/ironic/networking/switch_drivers/__init__.py b/ironic/networking/switch_drivers/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/ironic/networking/switch_drivers/driver_adapter.py b/ironic/networking/switch_drivers/driver_adapter.py new file mode 100644 index 0000000000..af5ac9083e --- /dev/null +++ b/ironic/networking/switch_drivers/driver_adapter.py @@ -0,0 +1,293 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +""" +Driver Configuration Adapter + +This module provides functionality to translate user-friendly switch +configuration into driver-specific configuration formats. It allows users +to configure switches using a generic format while supporting multiple +switch driver implementations. +""" + +import configparser +import glob +import os +import tempfile + +from oslo_config import cfg +from oslo_log import log + +from ironic.common import exception +from ironic.common.i18n import _ + +LOG = log.getLogger(__name__) + +CONF = cfg.CONF + + +class NetworkingDriverAdapter: + """Adapter for translating switch config to driver-specific format.""" + + def __init__(self): + """Initialize the driver adapter.""" + self.driver_translators = {} + self._register_translators() + + def _register_translators(self): + """Register available driver translators.""" + LOG.debug( + "Registered translators for drivers: %s", + list(self.driver_translators.keys()), + ) + + def register_translator(self, driver_type, translator_instance): + """Register a custom translator for a driver type. + + :param driver_type: String identifier for the driver type + :param translator_instance: Instance of a translator class + """ + self.driver_translators[driver_type] = translator_instance + LOG.info( + "Registered custom translator for driver type: %s", driver_type + ) + + def _validate_switch_config(self, switch_name, config): + """Validate switch configuration has required fields. + + :param switch_name: Name of the switch + :param config: Dictionary of configuration options + :raises: NetworkError if validation fails + """ + required_fields = [ + 'driver_type', + 'device_type', + 'address', + 'username', + 'mac_address', + ] + missing_fields = [f for f in required_fields if f not in config] + + # Check for authentication: must have either password or key_file + has_auth = 'password' in config or 'key_file' in config + + if missing_fields or not has_auth: + error_parts = [] + if missing_fields: + error_parts.append( + "missing required fields: %s" % ', '.join(missing_fields) + ) + if not has_auth: + error_parts.append( + "must specify either 'password' or 'key_file'" + ) + + raise exception.NetworkError( + _("Invalid configuration for switch '%(switch)s': %(errors)s") + % {'switch': switch_name, 'errors': '; '.join(error_parts)} + ) + + def preprocess_config(self, output_file): + """Transform user config into driver-specific config files. + + Scans oslo.config for switch configurations and generates + driver-specific config files that then get written to a driver-specific + config file. + + :returns: Number of translations generated + """ + try: + if not os.path.exists(CONF.ironic_networking.switch_config_file): + raise exception.NetworkError( + _("Switch configuration file %s does not exist") + % CONF.ironic_networking.switch_config_file + ) + + # Extract generic switch sections from config + switch_sections = self._extract_switch_sections( + CONF.ironic_networking.switch_config_file + ) + + if not switch_sections: + LOG.debug( + "No user defined switch sections found in %s", + CONF.ironic_networking.switch_config_file, + ) + return 0 + + # Generate driver-specific configs + translations = {} + for switch_name, config in switch_sections.items(): + # Validate configuration before processing + self._validate_switch_config(switch_name, config) + + driver_type = config["driver_type"] + LOG.debug( + "Translating switch %s with driver type %s", + switch_name, + driver_type, + ) + if driver_type in self.driver_translators: + translator = self.driver_translators[driver_type] + else: + error_msg = (_("No driver translator registered for " + "switch: %(switch_name)s, with driver type: " + "%(driver_type)s") % + {"switch_name": switch_name, + "driver_type": driver_type}) + raise exception.ConfigInvalid(error_msg=error_msg) + + translation = translator.translate_config(switch_name, config) + if translation: + translations.update(translation) + + if translations: + self._write_config_file(output_file, translations) + CONF.reload_config_files() + + return len(translations) + except Exception as e: + LOG.exception("Failed to preprocess switch configuration: %s", e) + raise exception.NetworkError( + _("Configuration preprocessing failed: %s") % e + ) + + def _config_files(self): + """Generate which yields all config files in the required order""" + for config_file in CONF.config_file: + yield config_file + for config_dir in CONF.config_dir: + config_dir_glob = os.path.join(config_dir, "*.conf") + for config_file in sorted(glob.glob(config_dir_glob)): + yield config_file + + def _extract_switch_sections(self, config_file): + """Extract switch configuration sections from oslo.config. + + Looks for sections with names like: + - [switch:switch_name] + + :returns: Dictionary of section_name -> config_dict + """ + switch_sections = {} + + sections = {} + parser = cfg.ConfigParser(config_file, sections) + try: + parser.parse() + except Exception as e: + LOG.warning("Failed to parse config file %s: %s", config_file, e) + return {} + + for section_name, section_config in sections.items(): + if section_name.startswith("switch:"): + switch_name = section_name[7:] + # Get all key/value pairs in this section + switch_sections[switch_name] = { + k: v[0] for k, v in section_config.items() + } + + LOG.debug("Found %d switch sections", len(switch_sections)) + return switch_sections + + def _write_config_file(self, output_file, switch_configs): + """Generate driver-specific configuration file. + + :param output_file: Path to the output file + :param switch_configs: Dictionary of switch_name -> config_dict + """ + # Create temp file in same directory as output file for atomic rename + output_dir = os.path.dirname(output_file) + temp_fd = None + temp_path = None + + try: + config = configparser.ConfigParser() + + # Add all sections and their key-value pairs + for section_name, section_config in switch_configs.items(): + config.add_section(section_name) + for key, value in section_config.items(): + config.set(section_name, key, str(value)) + + # Write to temporary file first + temp_fd, temp_path = tempfile.mkstemp( + dir=output_dir, prefix='.tmp_driver_config_', text=True + ) + with os.fdopen(temp_fd, 'w') as f: + temp_fd = None # fdopen takes ownership + f.write( + "# Auto-generated config for driver-specific switch " + "configurations\n" + ) + f.write( + "# Generated from user defined switch configuration\n\n" + ) + config.write(f) + + # Atomically move temp file to final location + os.replace(temp_path, output_file) + temp_path = None # Successfully moved + + LOG.info("Generated driver config file: %s", output_file) + + except Exception as e: + LOG.error("Failed to generate config file: %s", e) + # Clean up temp file if it still exists + if temp_fd is not None: + try: + os.close(temp_fd) + except OSError as cleanup_error: + LOG.debug("Failed to close temp file descriptor: %s", + cleanup_error) + if temp_path is not None: + try: + os.unlink(temp_path) + except OSError as cleanup_error: + LOG.debug("Failed to remove temp file %s: %s", + temp_path, cleanup_error) + # Re-raise the original exception + raise e + + def reload_configuration(self, output_file): + """Reload and regenerate switch configuration files. + + This method re-extracts switch configurations from the config files + and regenerates the driver-specific configuration files. It should + be called when the switch configuration file has been modified. + + :param output_file: Path to the output file for driver-specific configs + :returns: Number of translations generated + :raises: NetworkError if configuration reload fails + """ + LOG.info("Reloading switch configuration from config files") + + try: + # Force oslo.config to reload configuration files + CONF.reload_config_files() + + # Re-run the preprocessing steps + count = self.preprocess_config(output_file) + + LOG.info( + "Successfully reloaded switch configuration. " + "Generated %d driver-specific config sections", + count, + ) + return count + + except Exception as e: + LOG.error("Failed to reload switch configuration: %s", e) + raise exception.NetworkError( + _("Configuration reload failed: %s") % e + ) diff --git a/ironic/networking/switch_drivers/driver_factory.py b/ironic/networking/switch_drivers/driver_factory.py new file mode 100644 index 0000000000..d394bcdb74 --- /dev/null +++ b/ironic/networking/switch_drivers/driver_factory.py @@ -0,0 +1,175 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +""" +Networking service driver factory for loading switch drivers. + +This module provides a driver factory system for the networking service, +allowing dynamic loading of switch drivers from external projects via +entry points. +""" + +import collections + +from oslo_log import log + +from ironic.common import driver_factory +from ironic.common import exception +from ironic.conf import CONF + +LOG = log.getLogger(__name__) + + +class BaseSwitchDriverFactory(driver_factory.BaseDriverFactory): + """Base factory for discovering, loading and managing switch drivers. + + This factory loads switch drivers from entry points and manages their + lifecycle. Switch drivers are loaded from external projects and provide + vendor-specific implementations for network switch management. + + Inherits from common.BaseDriverFactory to ensure consistency with + Ironic's standard driver factory pattern. + """ + + # Entry point namespace for switch drivers + _entrypoint_name = "ironic.networking.switch_drivers" + + # Configuration option containing enabled drivers list + _enabled_driver_list_config_option = "enabled_switch_drivers" + + # Template for logging loaded drivers + _logging_template = "Loaded the following switch drivers: %s" + + @classmethod + def _set_enabled_drivers(cls): + """Set the list of enabled drivers from configuration. + + Overrides the base implementation to read from the networking + configuration group instead of the default group. + """ + enabled_drivers = getattr( + CONF.ironic_networking, cls._enabled_driver_list_config_option, [] + ) + + # Check for duplicated driver entries and warn about them + counter = collections.Counter(enabled_drivers).items() + duplicated_drivers = [] + cls._enabled_driver_list = [] + + for item, cnt in counter: + if not item: + raise exception.ConfigInvalid( + error_msg=( + 'An empty switch driver was specified in the "%s" ' + "configuration option. Please fix your ironic.conf " + "file." % cls._enabled_driver_list_config_option + ) + ) + if cnt > 1: + duplicated_drivers.append(item) + cls._enabled_driver_list.append(item) + + if duplicated_drivers: + raise exception.ConfigInvalid( + error_msg=( + 'The switch driver(s) "%s" is/are duplicated in the ' + "list of enabled drivers. Please check your " + "configuration file." % ", ".join(duplicated_drivers) + ) + ) + + @classmethod + def _init_extension_manager(cls): + """Initialize the extension manager for loading switch drivers. + + Extends the base implementation to handle the case where no + switch drivers are enabled. + """ + # Set enabled drivers first + cls._set_enabled_drivers() + + # Only proceed if we have enabled drivers + if not cls._enabled_driver_list: + LOG.info("No switch drivers enabled in configuration") + return + + # Call parent implementation + try: + super()._init_extension_manager() + except RuntimeError as e: + if "No suitable drivers found" in str(e): + LOG.warning( + "No switch drivers could be loaded. Check that " + "the specified drivers are installed and their " + "entry points are correctly defined." + ) + cls._extension_manager = None + else: + raise + + +def _warn_if_unsupported(ext): + """Warn if a driver is marked as unsupported.""" + if hasattr(ext.obj, "supported") and not ext.obj.supported: + LOG.warning( + 'Switch driver "%s" is UNSUPPORTED. It has been ' + "deprecated and may be removed in a future release.", + ext.name, + ) + + +class SwitchDriverFactory(BaseSwitchDriverFactory): + """Factory for loading switch drivers from entry points.""" + + pass + + +# Global factory instance +_switch_driver_factory = None + + +def get_switch_driver_factory(): + """Get the global switch driver factory instance.""" + global _switch_driver_factory + if _switch_driver_factory is None: + _switch_driver_factory = SwitchDriverFactory() + return _switch_driver_factory + + +def get_switch_driver(driver_name): + """Get a switch driver instance by name. + + :param driver_name: Name of the switch driver to retrieve. + :returns: Instance of the switch driver. + :raises: DriverNotFound if the driver is not found. + """ + factory = get_switch_driver_factory() + return factory.get_driver(driver_name) + + +def list_switch_drivers(): + """Get a list of all available switch driver names. + + :returns: List of switch driver names. + """ + factory = get_switch_driver_factory() + return factory.names + + +def switch_drivers(): + """Get all switch drivers as a dictionary. + + :returns: Dictionary mapping driver name to driver instance. + """ + factory = get_switch_driver_factory() + return dict(factory.items()) diff --git a/ironic/networking/switch_drivers/driver_translators.py b/ironic/networking/switch_drivers/driver_translators.py new file mode 100644 index 0000000000..6d8d0a3b76 --- /dev/null +++ b/ironic/networking/switch_drivers/driver_translators.py @@ -0,0 +1,85 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +""" +Driver Configuration Translators + +This module contains translator classes that convert generic switch +configuration into driver-specific configuration formats. Each translator +handles the specifics of how a particular driver expects its configuration +to be structured. +""" +import abc + +from oslo_log import log + +LOG = log.getLogger(__name__) + + +class BaseTranslator(metaclass=abc.ABCMeta): + """Base class for configuration translators.""" + + def translate_configs(self, switch_configs): + """Translate all switch configurations. + + :param switch_configs: Dictionary of switch_name -> config_dict + :returns: Dictionary of section_name -> translated_config_dict + """ + translated = {} + + for switch_name, config in switch_configs.items(): + translated.update(self.translate_config(switch_name, config)) + + return translated + + def translate_config(self, switch_name, config): + """Translate a single switch configuration. + + :param switch_name: Name of the switch + :param config: Dictionary of configuration options for the switch + :returns: Dictionary of section_name -> translated_config_dict + """ + section_name = self._get_section_name(switch_name) + translated_config = self._translate_switch_config(config) + + if translated_config: + LOG.debug( + "Translated config for switch %s to section %s", + switch_name, + section_name, + ) + return {section_name: translated_config} + + return {} + + @abc.abstractmethod + def _get_section_name(self, switch_name): + """Get the section name for a switch in driver-specific format. + + :param switch_name: Name of the switch + :returns: Section name string + """ + raise NotImplementedError( + "Subclasses must implement _get_section_name" + ) + + @abc.abstractmethod + def _translate_switch_config(self, config): + """Translate a single switch configuration. + + :param config: Dictionary of configuration options + :returns: Dictionary of translated configuration options + """ + raise NotImplementedError( + "Subclasses must implement _translate_switch_config" + ) diff --git a/ironic/networking/utils.py b/ironic/networking/utils.py index 509f1f39e5..81036ef118 100644 --- a/ironic/networking/utils.py +++ b/ironic/networking/utils.py @@ -36,16 +36,23 @@ def rpc_transport(): def parse_vlan_ranges(vlan_spec): """Parse VLAN specification into a set of VLAN IDs. - :param vlan_spec: List of VLAN IDs or ranges (e.g., ['100', '102-104']) + :param vlan_spec: List of VLAN IDs or ranges (e.g., ['100', '102-104']) or + a string representing a list of VLAN IDs (e.g., '100,102-104'). :returns: Set of integer VLAN IDs :raises: InvalidParameterValue if the specification is invalid """ if vlan_spec is None: return None + if isinstance(vlan_spec, str): + vlan_spec = vlan_spec.split(',') + vlan_set = set() for item in vlan_spec: item = item.strip() + if not item: + # Skip empty elements (e.g., from "100,,200" or trailing commas) + continue if '-' in item: # Handle range (e.g., "102-104") try: @@ -82,48 +89,3 @@ def parse_vlan_ranges(vlan_spec): {'item': item, 'error': str(e)}) return vlan_set - - -def validate_vlan_allowed(vlan_id, allowed_vlans_config=None, - switch_config=None): - """Validate that a VLAN ID is allowed. - - :param vlan_id: The VLAN ID to validate - :param allowed_vlans_config: Global list of allowed vlans from config - :param switch_config: Optional switch-specific configuration dict that - may contain an 'allowed_vlans' key - :returns: True if the VLAN is allowed - :raises: InvalidParameterValue if the VLAN is not allowed - """ - # Check switch-specific configuration first (if provided) - if switch_config and 'allowed_vlans' in switch_config: - allowed_spec = switch_config['allowed_vlans'] - else: - # Fall back to global configuration - if allowed_vlans_config is not None: - allowed_spec = allowed_vlans_config - else: - allowed_spec = CONF.ironic_networking.allowed_vlans - - # None means all VLANs are allowed - if allowed_spec is None: - return True - - # Empty list means no VLANs are allowed - if isinstance(allowed_spec, list) and len(allowed_spec) == 0: - raise exception.InvalidParameterValue( - _('VLAN %(vlan)s is not allowed: no VLANs are permitted by ' - 'configuration') % {'vlan': vlan_id}) - - # Parse and check against allowed VLANs - try: - allowed_vlans = parse_vlan_ranges(allowed_spec) - if vlan_id not in allowed_vlans: - raise exception.InvalidParameterValue( - _('VLAN %(vlan)s is not in the list of allowed VLANs') % - {'vlan': vlan_id}) - except exception.InvalidParameterValue: - # Re-raise validation errors from parse_vlan_ranges - raise - - return True diff --git a/ironic/tests/unit/networking/test_driver_adapter.py b/ironic/tests/unit/networking/test_driver_adapter.py new file mode 100644 index 0000000000..98d05fe7ac --- /dev/null +++ b/ironic/tests/unit/networking/test_driver_adapter.py @@ -0,0 +1,642 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +"""Unit tests for ironic.networking.switch_drivers.driver_adapter""" + +import os +from unittest import mock + +import fixtures +from oslo_config import cfg + +from ironic.common import exception +from ironic.networking.switch_drivers import driver_adapter +from ironic import tests as tests_root + + +CONF = cfg.CONF + + +class NetworkingDriverAdapterTestCase(tests_root.base.TestCase): + """Test cases for NetworkingDriverAdapter class.""" + + def setUp(self): + super(NetworkingDriverAdapterTestCase, self).setUp() + temp_dir = self.useFixture(fixtures.TempDir()).path + self.switch_config_path = os.path.join(temp_dir, 'switch.conf') + with open(self.switch_config_path, 'w', encoding='utf-8') as config_fp: + config_fp.write('[DEFAULT]\n') + self.output_config_path = os.path.join(temp_dir, 'driver.conf') + self.config(group='ironic_networking', + switch_config_file=self.switch_config_path) + self.config(group='ironic_networking', driver_config_dir=temp_dir) + self.adapter = driver_adapter.NetworkingDriverAdapter() + + def test_register_translator(self): + """Test register_translator method.""" + mock_translator = mock.Mock() + + self.adapter.register_translator('test_driver', mock_translator) + + self.assertEqual(mock_translator, + self.adapter.driver_translators['test_driver']) + + @mock.patch.object(driver_adapter.NetworkingDriverAdapter, + '_write_config_file', autospec=True) + @mock.patch.object(driver_adapter.NetworkingDriverAdapter, + '_extract_switch_sections', autospec=True) + @mock.patch.object(CONF, 'reload_config_files', autospec=True) + def test_preprocess_config_success(self, mock_reload, mock_extract, + mock_write): + """Test preprocess_config method with successful translation.""" + # Setup mock data + mock_extract.return_value = { + 'switch1': { + 'driver_type': 'generic-switch', + 'device_type': 'netmiko_cisco_ios', + 'address': '192.168.1.1', + 'username': 'admin', + 'password': 'secret', + 'mac_address': '00:11:22:33:44:55' + } + } + + mock_translator = mock.Mock() + mock_translator.translate_config.return_value = { + 'genericswitch:switch1': { + 'ip': '192.168.1.1', + 'username': 'admin' + } + } + self.adapter.driver_translators['generic-switch'] = mock_translator + + result = self.adapter.preprocess_config(self.output_config_path) + + self.assertEqual(1, result) + mock_extract.assert_called_once() + mock_translator.translate_config.assert_called_once_with( + 'switch1', { + 'driver_type': 'generic-switch', + 'device_type': 'netmiko_cisco_ios', + 'address': '192.168.1.1', + 'username': 'admin', + 'password': 'secret', + 'mac_address': '00:11:22:33:44:55' + }) + mock_write.assert_called_once_with( + self.adapter, + self.output_config_path, + {'genericswitch:switch1': { + 'ip': '192.168.1.1', + 'username': 'admin' + }} + ) + mock_reload.assert_called_once() + + @mock.patch( + 'ironic.networking.switch_drivers.driver_adapter.os.path.exists', + autospec=True) + def test_preprocess_config_missing_config_file(self, mock_exists): + """Raise NetworkError when switch config file is absent.""" + mock_exists.return_value = False + + self.assertRaises( + exception.NetworkError, + self.adapter.preprocess_config, + self.output_config_path, + ) + mock_exists.assert_called_once_with(self.switch_config_path) + + @mock.patch.object(driver_adapter.NetworkingDriverAdapter, + '_extract_switch_sections', autospec=True) + def test_preprocess_config_no_switches(self, mock_extract): + """Test preprocess_config method with no switch sections.""" + mock_extract.return_value = {} + + result = self.adapter.preprocess_config(self.output_config_path) + + self.assertEqual(0, result) + + @mock.patch.object(driver_adapter.NetworkingDriverAdapter, + '_extract_switch_sections', autospec=True) + def test_preprocess_config_translator_error(self, mock_extract): + """Translator failure should bubble up as NetworkError.""" + mock_extract.return_value = { + 'switch1': { + 'driver_type': 'generic-switch', + 'device_type': 'netmiko_cisco_ios', + 'address': '10.0.0.1', + 'username': 'admin', + 'password': 'secret', + 'mac_address': '00:11:22:33:44:55' + } + } + broken = mock.Mock() + broken.translate_config.side_effect = RuntimeError('boom') + self.adapter.driver_translators['generic-switch'] = broken + + with mock.patch.object(CONF, 'reload_config_files', autospec=True): + self.assertRaises( + exception.NetworkError, + self.adapter.preprocess_config, + self.output_config_path, + ) + mock_extract.assert_called_once() + + def test__validate_switch_config_valid_with_password(self): + """Test _validate_switch_config with valid config using password.""" + config = { + 'driver_type': 'generic-switch', + 'device_type': 'netmiko_cisco_ios', + 'address': '192.168.1.1', + 'username': 'admin', + 'password': 'secret', + 'mac_address': '00:11:22:33:44:55' + } + # Should not raise + self.adapter._validate_switch_config('switch1', config) + + def test__validate_switch_config_valid_with_key_file(self): + """Test _validate_switch_config with valid config using key_file.""" + config = { + 'driver_type': 'generic-switch', + 'device_type': 'netmiko_cisco_ios', + 'address': '192.168.1.1', + 'username': 'admin', + 'key_file': '/path/to/key', + 'mac_address': '00:11:22:33:44:55' + } + # Should not raise + self.adapter._validate_switch_config('switch1', config) + + def test__validate_switch_config_missing_driver_type(self): + """Test _validate_switch_config with missing driver_type.""" + config = { + 'device_type': 'netmiko_cisco_ios', + 'address': '192.168.1.1', + 'username': 'admin', + 'password': 'secret', + 'mac_address': '00:11:22:33:44:55' + } + exc = self.assertRaises( + exception.NetworkError, + self.adapter._validate_switch_config, + 'switch1', + config + ) + self.assertIn('driver_type', str(exc)) + self.assertIn('switch1', str(exc)) + + def test__validate_switch_config_missing_device_type(self): + """Test _validate_switch_config with missing device_type.""" + config = { + 'driver_type': 'generic-switch', + 'address': '192.168.1.1', + 'username': 'admin', + 'password': 'secret', + 'mac_address': '00:11:22:33:44:55' + } + exc = self.assertRaises( + exception.NetworkError, + self.adapter._validate_switch_config, + 'switch1', + config + ) + self.assertIn('device_type', str(exc)) + + def test__validate_switch_config_missing_address(self): + """Test _validate_switch_config with missing address.""" + config = { + 'driver_type': 'generic-switch', + 'device_type': 'netmiko_cisco_ios', + 'username': 'admin', + 'password': 'secret', + 'mac_address': '00:11:22:33:44:55' + } + exc = self.assertRaises( + exception.NetworkError, + self.adapter._validate_switch_config, + 'switch1', + config + ) + self.assertIn('address', str(exc)) + + def test__validate_switch_config_missing_username(self): + """Test _validate_switch_config with missing username.""" + config = { + 'driver_type': 'generic-switch', + 'device_type': 'netmiko_cisco_ios', + 'address': '192.168.1.1', + 'password': 'secret', + 'mac_address': '00:11:22:33:44:55' + } + exc = self.assertRaises( + exception.NetworkError, + self.adapter._validate_switch_config, + 'switch1', + config + ) + self.assertIn('username', str(exc)) + + def test__validate_switch_config_missing_mac_address(self): + """Test _validate_switch_config with missing mac_address.""" + config = { + 'driver_type': 'generic-switch', + 'device_type': 'netmiko_cisco_ios', + 'address': '192.168.1.1', + 'username': 'admin', + 'password': 'secret' + } + exc = self.assertRaises( + exception.NetworkError, + self.adapter._validate_switch_config, + 'switch1', + config + ) + self.assertIn('mac_address', str(exc)) + + def test__validate_switch_config_missing_authentication(self): + """Test _validate_switch_config with no password or key_file.""" + config = { + 'driver_type': 'generic-switch', + 'device_type': 'netmiko_cisco_ios', + 'address': '192.168.1.1', + 'username': 'admin', + 'mac_address': '00:11:22:33:44:55' + } + exc = self.assertRaises( + exception.NetworkError, + self.adapter._validate_switch_config, + 'switch1', + config + ) + self.assertIn('password', str(exc)) + self.assertIn('key_file', str(exc)) + + def test__validate_switch_config_multiple_missing_fields(self): + """Test _validate_switch_config with multiple missing fields.""" + config = { + 'driver_type': 'generic-switch', + } + exc = self.assertRaises( + exception.NetworkError, + self.adapter._validate_switch_config, + 'switch1', + config + ) + error_msg = str(exc) + self.assertIn('device_type', error_msg) + self.assertIn('address', error_msg) + self.assertIn('username', error_msg) + self.assertIn('mac_address', error_msg) + self.assertIn('password', error_msg) + self.assertIn('key_file', error_msg) + + @mock.patch.object(driver_adapter.NetworkingDriverAdapter, + '_extract_switch_sections', autospec=True) + def test_preprocess_config_empty_translation(self, mock_extract): + """Test preprocess_config with empty translation result.""" + mock_extract.return_value = { + 'switch1': { + 'driver_type': 'generic-switch', + 'device_type': 'netmiko_cisco_ios', + 'address': '192.168.1.1', + 'username': 'admin', + 'password': 'secret', + 'mac_address': '00:11:22:33:44:55' + } + } + + mock_translator = mock.Mock() + mock_translator.translate_config.return_value = {} + self.adapter.driver_translators['generic-switch'] = mock_translator + + result = self.adapter.preprocess_config(self.output_config_path) + + self.assertEqual(0, result) + + @mock.patch.object(driver_adapter.NetworkingDriverAdapter, + '_extract_switch_sections', autospec=True) + def test_preprocess_config_exception(self, mock_extract): + """Test preprocess_config method with exception.""" + mock_extract.side_effect = Exception("Test error") + + self.assertRaises(exception.NetworkError, + self.adapter.preprocess_config, + self.output_config_path) + + @mock.patch.object(driver_adapter.NetworkingDriverAdapter, + '_extract_switch_sections', autospec=True) + def test_preprocess_config_unknown_driver_type(self, mock_extract): + """Test preprocess_config with unknown driver type.""" + mock_extract.return_value = { + 'switch1': { + 'driver_type': 'unknown_driver', + 'device_type': 'netmiko_cisco_ios', + 'address': '192.168.1.1', + 'username': 'admin', + 'password': 'secret', + 'mac_address': '00:11:22:33:44:55' + } + } + + # No translator registered for 'unknown_driver' + # Should fail validation before reaching translator + self.assertRaises(exception.NetworkError, + self.adapter.preprocess_config, + self.output_config_path) + + @mock.patch.object(driver_adapter.NetworkingDriverAdapter, + '_extract_switch_sections', autospec=True) + def test_preprocess_config_validation_fails(self, mock_extract): + """Test preprocess_config with validation failure.""" + mock_extract.return_value = { + 'switch1': { + 'driver_type': 'generic-switch', + # Missing required fields + 'address': '192.168.1.1' + } + } + + # Should fail validation before reaching translator + self.assertRaises(exception.NetworkError, + self.adapter.preprocess_config, + self.output_config_path) + + @mock.patch('glob.glob', autospec=True) + def test__config_files_with_config_dir(self, mock_glob): + """Test _config_files method with config directories.""" + # Setup CONF mock + CONF.config_file = ['/etc/ironic/ironic.conf'] + CONF.config_dir = ['/etc/ironic/conf.d'] + mock_glob.return_value = ['/etc/ironic/conf.d/test.conf'] + + result = list(self.adapter._config_files()) + + expected = ['/etc/ironic/ironic.conf', '/etc/ironic/conf.d/test.conf'] + self.assertEqual(expected, result) + mock_glob.assert_called_once_with('/etc/ironic/conf.d/*.conf') + + def test__config_files_only_config_file(self): + """Test _config_files method with only config files.""" + CONF.config_file = ['/etc/ironic/ironic.conf', + '/etc/ironic/other.conf'] + CONF.config_dir = [] + + result = list(self.adapter._config_files()) + + expected = ['/etc/ironic/ironic.conf', '/etc/ironic/other.conf'] + self.assertEqual(expected, result) + + @mock.patch( + 'ironic.networking.switch_drivers.driver_adapter.cfg.ConfigParser', + autospec=True) + def test__extract_switch_sections(self, mock_parser_class): + """Test _extract_switch_sections method.""" + # Mock the sections dict that will be populated + sections = { + 'switch:switch1': { + 'address': ['192.168.1.1'], + 'username': ['admin'], + 'password': ['secret'] + }, + 'switch:switch2': { + 'address': ['192.168.1.2'], + 'device_type': ['cisco_ios'] + }, + 'regular_section': { + 'key': ['value'] + } + } + + # Create a mock parser that populates the sections dict when called + def mock_parser_init(config_file, sections_dict): + # Populate the sections dict that was passed in + sections_dict.update(sections) + # Return a mock parser instance + parser = mock.Mock() + parser.parse.return_value = None + return parser + + mock_parser_class.side_effect = mock_parser_init + + result = self.adapter._extract_switch_sections( + '/etc/ironic/ironic.conf' + ) + + expected = { + 'switch1': { + 'address': '192.168.1.1', + 'username': 'admin', + 'password': 'secret' + }, + 'switch2': { + 'address': '192.168.1.2', + 'device_type': 'cisco_ios' + } + } + self.assertEqual(expected, result) + + @mock.patch( + 'ironic.networking.switch_drivers.driver_adapter.cfg.ConfigParser', + autospec=True) + def test__extract_switch_sections_parse_error(self, mock_parser_class): + """Test _extract_switch_sections with config parse error.""" + # Create a mock parser that raises an exception when parse() is called + def mock_parser_init(config_file, sections_dict): + parser = mock.Mock() + parser.parse.side_effect = Exception("Parse error") + return parser + + mock_parser_class.side_effect = mock_parser_init + + result = self.adapter._extract_switch_sections( + '/etc/ironic/ironic.conf' + ) + + self.assertEqual({}, result) + + @mock.patch('os.replace', autospec=True) + @mock.patch('os.fdopen', autospec=True) + @mock.patch('tempfile.mkstemp', autospec=True) + def test__write_config_file_success(self, mock_mkstemp, mock_fdopen, + mock_replace): + """Test _write_config_file method successful write.""" + switch_configs = { + 'genericswitch:switch1': { + 'ip': '192.168.1.1', + 'username': 'admin', + 'device_type': 'netmiko_cisco_ios' + }, + 'genericswitch:switch2': { + 'ip': '192.168.1.2', + 'device_type': 'netmiko_ovs_linux' + } + } + + # Mock tempfile.mkstemp to return a fake fd and path + mock_mkstemp.return_value = (42, '/tmp/.tmp_driver_config_xyz') + + # Mock fdopen to return a mock file object + mock_file = mock.mock_open()() + mock_fdopen.return_value.__enter__.return_value = mock_file + + self.adapter._write_config_file('/tmp/test.conf', switch_configs) + + mock_mkstemp.assert_called_once_with( + dir='/tmp', prefix='.tmp_driver_config_', text=True) + mock_fdopen.assert_called_once_with(42, 'w') + mock_replace.assert_called_once_with( + '/tmp/.tmp_driver_config_xyz', '/tmp/test.conf') + + # Verify the content written + write_calls = mock_file.write.call_args_list + written_content = ''.join(call[0][0] for call in write_calls) + + self.assertIn('# Auto-generated config', written_content) + self.assertIn('[genericswitch:switch1]', written_content) + self.assertIn('ip = 192.168.1.1', written_content) + self.assertIn('username = admin', written_content) + self.assertIn('device_type = netmiko_cisco_ios', written_content) + self.assertIn('[genericswitch:switch2]', written_content) + self.assertIn('ip = 192.168.1.2', written_content) + self.assertIn('device_type = netmiko_ovs_linux', written_content) + + @mock.patch('os.unlink', autospec=True) + @mock.patch('os.close', autospec=True) + @mock.patch('os.fdopen', autospec=True) + @mock.patch('tempfile.mkstemp', autospec=True) + def test__write_config_file_error(self, mock_mkstemp, mock_fdopen, + mock_close, mock_unlink): + """Test _write_config_file method with write error.""" + switch_configs = {'test': {'key': 'value'}} + + # Mock tempfile.mkstemp to return a fake fd and path + mock_mkstemp.return_value = (42, '/tmp/.tmp_driver_config_xyz') + + # Mock fdopen to raise an error during writing + mock_fdopen.return_value.__enter__.side_effect = IOError( + "Permission denied") + + # Should raise the IOError after cleanup + self.assertRaises(IOError, + self.adapter._write_config_file, + '/tmp/test.conf', switch_configs) + + # Verify cleanup was attempted + mock_close.assert_called_once_with(42) + mock_unlink.assert_called_once_with('/tmp/.tmp_driver_config_xyz') + + @mock.patch('os.unlink', autospec=True) + @mock.patch('os.replace', autospec=True) + @mock.patch('os.fdopen', autospec=True) + @mock.patch('tempfile.mkstemp', autospec=True) + def test__write_config_file_write_error_with_cleanup( + self, mock_mkstemp, mock_fdopen, mock_replace, mock_unlink): + """Test _write_config_file cleans up temp file on write error.""" + switch_configs = {'test': {'key': 'value'}} + + # Mock tempfile.mkstemp to return a fake fd and path + mock_mkstemp.return_value = (42, '/tmp/.tmp_driver_config_xyz') + + # Mock file write to succeed but replace to fail + mock_file = mock.mock_open()() + mock_fdopen.return_value.__enter__.return_value = mock_file + mock_replace.side_effect = OSError("Rename failed") + + # Should raise the OSError after cleanup + self.assertRaises(OSError, + self.adapter._write_config_file, + '/tmp/test.conf', switch_configs) + + # Verify temp file cleanup was attempted + mock_unlink.assert_called_once_with('/tmp/.tmp_driver_config_xyz') + + +class NetworkingDriverAdapterReloadTestCase(tests_root.base.TestCase): + """Tests for reload_configuration helper.""" + + def setUp(self): + super(NetworkingDriverAdapterReloadTestCase, self).setUp() + self.adapter = driver_adapter.NetworkingDriverAdapter() + + def test_reload_configuration_success(self): + output_file = '/tmp/switches.conf' + + with mock.patch.object( + self.adapter, 'preprocess_config', autospec=True + ) as mock_preprocess, mock.patch( + 'ironic.networking.switch_drivers.driver_adapter.CONF', + autospec=True + ) as mock_conf: + mock_preprocess.return_value = 3 + + result = self.adapter.reload_configuration(output_file) + + mock_conf.reload_config_files.assert_called_once() + mock_preprocess.assert_called_once_with(output_file) + self.assertEqual(3, result) + + def test_reload_configuration_preprocess_failure(self): + output_file = '/tmp/switches.conf' + + with mock.patch.object( + self.adapter, 'preprocess_config', autospec=True + ) as mock_preprocess, mock.patch( + 'ironic.networking.switch_drivers.driver_adapter.CONF', + autospec=True + ) as mock_conf: + mock_preprocess.side_effect = Exception('fail') + + self.assertRaises( + exception.NetworkError, + self.adapter.reload_configuration, + output_file, + ) + + mock_conf.reload_config_files.assert_called_once() + mock_preprocess.assert_called_once_with(output_file) + + def test_reload_configuration_conf_reload_failure(self): + output_file = '/tmp/switches.conf' + + with mock.patch( + 'ironic.networking.switch_drivers.driver_adapter.CONF', + autospec=True + ) as mock_conf: + mock_conf.reload_config_files.side_effect = Exception('boom') + + self.assertRaises( + exception.NetworkError, + self.adapter.reload_configuration, + output_file, + ) + + mock_conf.reload_config_files.assert_called_once() + + def test_reload_configuration_zero_translations(self): + output_file = '/tmp/switches.conf' + + with mock.patch.object( + self.adapter, 'preprocess_config', autospec=True + ) as mock_preprocess, mock.patch( + 'ironic.networking.switch_drivers.driver_adapter.CONF', + autospec=True + ) as mock_conf: + mock_preprocess.return_value = 0 + + result = self.adapter.reload_configuration(output_file) + + mock_conf.reload_config_files.assert_called_once() + mock_preprocess.assert_called_once_with(output_file) + self.assertEqual(0, result) diff --git a/ironic/tests/unit/networking/test_driver_factory.py b/ironic/tests/unit/networking/test_driver_factory.py new file mode 100644 index 0000000000..e2511f8b67 --- /dev/null +++ b/ironic/tests/unit/networking/test_driver_factory.py @@ -0,0 +1,176 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +"""Unit tests for ``ironic.networking.driver_factory``.""" + +from unittest import mock + +from oslo_config import cfg + +from ironic.common import exception +from ironic.networking.switch_drivers import driver_factory +from ironic import tests as tests_root + + +CONF = cfg.CONF + + +class BaseSwitchDriverFactoryTestCase(tests_root.base.TestCase): + """Test behaviour common to switch driver factory classes.""" + + def setUp(self): + super(BaseSwitchDriverFactoryTestCase, self).setUp() + self.factory = driver_factory.BaseSwitchDriverFactory() + + def test_set_enabled_drivers(self): + self.config(group='ironic_networking', enabled_switch_drivers=['noop']) + + self.factory._set_enabled_drivers() + + self.assertEqual(['noop'], self.factory._enabled_driver_list) + + def test_set_enabled_drivers_with_duplicates(self): + self.config(group='ironic_networking', + enabled_switch_drivers=['noop', 'noop']) + + exc = self.assertRaises(exception.ConfigInvalid, + self.factory._set_enabled_drivers) + self.assertIn('noop', str(exc)) + self.assertIn('duplicated', str(exc)) + + def test_set_enabled_drivers_with_empty_value(self): + self.config(group='ironic_networking', + enabled_switch_drivers=['', 'noop']) + + exc = self.assertRaises(exception.ConfigInvalid, + self.factory._set_enabled_drivers) + self.assertIn('empty', str(exc)) + self.assertIn('enabled_switch_drivers', str(exc)) + + def test_init_extension_manager_no_enabled_drivers(self): + self.factory._enabled_driver_list = [] + + with mock.patch.object( + driver_factory.LOG, 'info', autospec=True + ) as mock_info: + self.factory._init_extension_manager() + + mock_info.assert_called_once() + + @mock.patch('stevedore.NamedExtensionManager', autospec=True) + def test_init_extension_manager_handles_runtime_error( + self, mock_named_ext_mgr): + # Reset extension manager state + type(self.factory)._extension_manager = None + type(self.factory)._enabled_driver_list = None + self.config(group='ironic_networking', enabled_switch_drivers=['noop']) + + mock_named_ext_mgr.side_effect = RuntimeError( + 'No suitable drivers found') + + with mock.patch.object( + driver_factory.LOG, 'warning', autospec=True + ) as mock_warn: + type(self.factory)._init_extension_manager() + + mock_warn.assert_called_once() + self.assertIsNone(type(self.factory)._extension_manager) + + @mock.patch('stevedore.NamedExtensionManager', autospec=True) + def test_init_extension_manager_unexpected_runtime_error( + self, mock_named_ext_mgr): + # Reset extension manager state + type(self.factory)._extension_manager = None + type(self.factory)._enabled_driver_list = None + self.config(group='ironic_networking', enabled_switch_drivers=['noop']) + + mock_named_ext_mgr.side_effect = RuntimeError('boom') + + self.assertRaises(RuntimeError, + type(self.factory)._init_extension_manager) + + +class FactoryHelpersTestCase(tests_root.base.TestCase): + """Tests for module-level helper functions.""" + + def test_warn_if_unsupported(self): + fake_extension = mock.Mock() + fake_extension.obj.supported = False + fake_extension.name = 'unsupported' + + with mock.patch.object( + driver_factory.LOG, 'warning', autospec=True + ) as mock_warn: + driver_factory._warn_if_unsupported(fake_extension) + + mock_warn.assert_called_once() + + def test_warn_if_supported(self): + fake_extension = mock.Mock() + fake_extension.obj.supported = True + + with mock.patch.object( + driver_factory.LOG, 'warning', autospec=True + ) as mock_warn: + driver_factory._warn_if_unsupported(fake_extension) + + mock_warn.assert_not_called() + + +class GlobalFactoryHelpersTestCase(tests_root.base.TestCase): + """Tests for global helper functions providing factory access.""" + + def setUp(self): + super(GlobalFactoryHelpersTestCase, self).setUp() + # Save and restore the global factory singleton + original_factory = driver_factory._switch_driver_factory + self.addCleanup(setattr, driver_factory, '_switch_driver_factory', + original_factory) + driver_factory._switch_driver_factory = None + + def test_get_switch_driver_factory_singleton(self): + factory1 = driver_factory.get_switch_driver_factory() + factory2 = driver_factory.get_switch_driver_factory() + self.assertIs(factory1, factory2) + + def test_get_switch_driver(self): + factory = driver_factory.get_switch_driver_factory() + + with mock.patch.object( + factory, 'get_driver', return_value='driver', autospec=True + ) as mock_get: + result = driver_factory.get_switch_driver('noop') + + mock_get.assert_called_once_with('noop') + self.assertEqual('driver', result) + + def test_list_switch_drivers(self): + factory = driver_factory.get_switch_driver_factory() + + with mock.patch.object( + type(factory), 'names', new_callable=mock.PropertyMock, + return_value=['noop'] + ): + result = driver_factory.list_switch_drivers() + + self.assertEqual(['noop'], result) + + def test_switch_drivers(self): + factory = driver_factory.get_switch_driver_factory() + + with mock.patch.object( + factory, 'items', return_value=[('noop', 'driver')], + autospec=True + ): + result = driver_factory.switch_drivers() + + self.assertEqual({'noop': 'driver'}, result) diff --git a/ironic/tests/unit/networking/test_driver_translators.py b/ironic/tests/unit/networking/test_driver_translators.py new file mode 100644 index 0000000000..6146646c32 --- /dev/null +++ b/ironic/tests/unit/networking/test_driver_translators.py @@ -0,0 +1,132 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +"""Unit tests for ironic.networking.driver_translators""" + +import unittest +from unittest import mock + +from ironic.networking.switch_drivers import driver_translators + + +class ConcreteTranslatorForTesting(driver_translators.BaseTranslator): + """Concrete implementation of BaseTranslator for testing purposes.""" + + def _get_section_name(self, switch_name): + """Return a test section name.""" + return f"section_{switch_name}" + + def _translate_switch_config(self, config): + """Return a test translated config.""" + return {'translated': True, **config} + + +class BaseTranslatorTestCase(unittest.TestCase): + """Test cases for BaseTranslator class.""" + + def setUp(self): + super(BaseTranslatorTestCase, self).setUp() + self.translator = ConcreteTranslatorForTesting() + + def test_translate_configs(self): + """Test translate_configs method.""" + switch_configs = { + 'switch1': {'address': '192.168.1.1', 'username': 'admin'}, + 'switch2': {'address': '192.168.1.2', 'device_type': 'cisco_ios'} + } + + with mock.patch.object(self.translator, + 'translate_config', + autospec=True) as mock_translate: + mock_translate.side_effect = [ + {'section1': {'config1': 'value1'}}, + {'section2': {'config2': 'value2'}}, + ] + + result = self.translator.translate_configs(switch_configs) + + expected = { + 'section1': {'config1': 'value1'}, + 'section2': {'config2': 'value2'} + } + self.assertEqual(expected, result) + + mock_translate.assert_has_calls([ + mock.call('switch1', + {'address': '192.168.1.1', + 'username': 'admin'}), + mock.call('switch2', + {'address': '192.168.1.2', + 'device_type': 'cisco_ios'}) + ]) + + def test_translate_config_success(self): + """Test translate_config method with successful translation.""" + config = {'address': '192.168.1.1', 'username': 'admin'} + + with mock.patch.object(self.translator, + '_get_section_name', + autospec=True) as mock_section: + with mock.patch.object( + self.translator, + '_translate_switch_config', + autospec=True) as mock_translate: + mock_section.return_value = 'test_section' + mock_translate.return_value = {'translated': 'config'} + + result = self.translator.translate_config( + 'test_switch', config) + + expected = {'test_section': {'translated': 'config'}} + self.assertEqual(expected, result) + + mock_section.assert_called_once_with('test_switch') + mock_translate.assert_called_once_with(config) + + def test_translate_config_empty_translation(self): + """Test translate_config method with empty translation.""" + config = {'address': '192.168.1.1'} + + with mock.patch.object(self.translator, + '_get_section_name', + autospec=True) as mock_section: + with mock.patch.object( + self.translator, + '_translate_switch_config', + autospec=True) as mock_translate: + mock_section.return_value = 'test_section' + mock_translate.return_value = {} + + result = self.translator.translate_config( + 'test_switch', config) + + self.assertEqual({}, result) + + def test_translate_config_none_translation(self): + """Test translate_config method with None translation.""" + config = {'address': '192.168.1.1'} + + with mock.patch.object(self.translator, + '_get_section_name', + autospec=True) as mock_section: + with mock.patch.object( + self.translator, + '_translate_switch_config', + autospec=True) as mock_translate: + mock_section.return_value = 'test_section' + mock_translate.return_value = None + + result = self.translator.translate_config( + 'test_switch', config) + + self.assertEqual({}, result) diff --git a/ironic/tests/unit/networking/test_switch_config.py b/ironic/tests/unit/networking/test_switch_config.py new file mode 100644 index 0000000000..3cb9570aae --- /dev/null +++ b/ironic/tests/unit/networking/test_switch_config.py @@ -0,0 +1,481 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +"""Unit tests for ``ironic.networking.switch_config``.""" + +from unittest import mock + +from ironic.common import exception +from ironic.networking import switch_config +from ironic.tests import base + + +class ValidateVlanAllowedTestCase(base.TestCase): + """Test cases for validate_vlan_allowed function.""" + + def test_validate_vlan_allowed_none_config(self): + """Test that None config allows all VLANs.""" + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = None + result = switch_config.validate_vlan_allowed(100) + self.assertTrue(result) + + def test_validate_vlan_allowed_empty_list_config(self): + """Test that empty list config denies all VLANs.""" + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = [] + self.assertRaises( + exception.InvalidParameterValue, + switch_config.validate_vlan_allowed, + 100 + ) + + def test_validate_vlan_allowed_vlan_in_list(self): + """Test that VLAN in allowed list is accepted.""" + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = ['100', '200', '300'] + result = switch_config.validate_vlan_allowed(100) + self.assertTrue(result) + + def test_validate_vlan_allowed_vlan_not_in_list(self): + """Test that VLAN not in allowed list is rejected.""" + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = ['100', '200', '300'] + self.assertRaises( + exception.InvalidParameterValue, + switch_config.validate_vlan_allowed, + 150 + ) + + def test_validate_vlan_allowed_vlan_in_range(self): + """Test that VLAN in allowed range is accepted.""" + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = ['100-200'] + result = switch_config.validate_vlan_allowed(150) + self.assertTrue(result) + + def test_validate_vlan_allowed_vlan_not_in_range(self): + """Test that VLAN not in allowed range is rejected.""" + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = ['100-200'] + self.assertRaises( + exception.InvalidParameterValue, + switch_config.validate_vlan_allowed, + 250 + ) + + def test_validate_vlan_allowed_complex_spec(self): + """Test validation with complex allowed VLAN specification.""" + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = [ + '100', '101', '102-104', '106' + ] + # Test allowed VLANs + self.assertTrue(switch_config.validate_vlan_allowed(100)) + self.assertTrue(switch_config.validate_vlan_allowed(101)) + self.assertTrue(switch_config.validate_vlan_allowed(102)) + self.assertTrue(switch_config.validate_vlan_allowed(103)) + self.assertTrue(switch_config.validate_vlan_allowed(104)) + self.assertTrue(switch_config.validate_vlan_allowed(106)) + # Test disallowed VLAN + self.assertRaises( + exception.InvalidParameterValue, + switch_config.validate_vlan_allowed, + 105 + ) + + def test_validate_vlan_allowed_override_config(self): + """Test that allowed_vlans_config parameter overrides CONF.""" + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = ['100'] + # Override should allow 200, not 100 + result = switch_config.validate_vlan_allowed( + 200, + allowed_vlans_config=['200'] + ) + self.assertTrue(result) + # Should reject 100 when using override + self.assertRaises( + exception.InvalidParameterValue, + switch_config.validate_vlan_allowed, + 100, + allowed_vlans_config=['200'] + ) + + def test_validate_vlan_allowed_switch_config_override(self): + """Test that switch config overrides global config.""" + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = ['100'] + switch_cfg = {'allowed_vlans': ['200']} + # Switch config should allow 200, not 100 + result = switch_config.validate_vlan_allowed( + 200, + switch_config=switch_cfg + ) + self.assertTrue(result) + # Should reject 100 when using switch config + self.assertRaises( + exception.InvalidParameterValue, + switch_config.validate_vlan_allowed, + 100, + switch_config=switch_cfg + ) + + def test_validate_vlan_allowed_switch_config_no_allowed_vlans(self): + """Test that switch config without allowed_vlans uses global.""" + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = ['100'] + switch_cfg = {'some_other_key': 'value'} + # Should fall back to global config + result = switch_config.validate_vlan_allowed( + 100, + switch_config=switch_cfg + ) + self.assertTrue(result) + + def test_validate_vlan_allowed_switch_config_empty_list(self): + """Test that switch config with empty list denies all VLANs.""" + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = ['100'] + switch_cfg = {'allowed_vlans': []} + # Switch config empty list should deny even though global allows + self.assertRaises( + exception.InvalidParameterValue, + switch_config.validate_vlan_allowed, + 100, + switch_config=switch_cfg + ) + + def test_validate_vlan_allowed_switch_config_none(self): + """Test that switch config with None allows all VLANs.""" + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = ['100'] + switch_cfg = {'allowed_vlans': None} + # Switch config None should allow all, even though global restricts + result = switch_config.validate_vlan_allowed( + 200, + switch_config=switch_cfg + ) + self.assertTrue(result) + + +class GetSwitchVlanConfigTestCase(base.TestCase): + """Test cases for get_switch_vlan_config function.""" + + def test_get_switch_vlan_config_with_switch_allowed_vlans(self): + """Test getting config when switch has allowed_vlans.""" + mock_driver = mock.Mock() + mock_driver.get_switch_info.return_value = { + 'allowed_vlans': ['100', '200-300'] + } + + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = ['400'] + + result = switch_config.get_switch_vlan_config( + mock_driver, 'switch1' + ) + + # Should use switch-specific config, not global + self.assertEqual({100, 200, 201, 202, 203, 204, 205, 206, 207, + 208, 209, 210, 211, 212, 213, 214, 215, 216, + 217, 218, 219, 220, 221, 222, 223, 224, 225, + 226, 227, 228, 229, 230, 231, 232, 233, 234, + 235, 236, 237, 238, 239, 240, 241, 242, 243, + 244, 245, 246, 247, 248, 249, 250, 251, 252, + 253, 254, 255, 256, 257, 258, 259, 260, 261, + 262, 263, 264, 265, 266, 267, 268, 269, 270, + 271, 272, 273, 274, 275, 276, 277, 278, 279, + 280, 281, 282, 283, 284, 285, 286, 287, 288, + 289, 290, 291, 292, 293, 294, 295, 296, 297, + 298, 299, 300}, + result['allowed_vlans']) + mock_driver.get_switch_info.assert_called_once_with('switch1') + + def test_get_switch_vlan_config_fallback_to_global(self): + """Test fallback to global config when switch has no allowed_vlans.""" + mock_driver = mock.Mock() + mock_driver.get_switch_info.return_value = {} + + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = ['100', '200-202'] + + result = switch_config.get_switch_vlan_config( + mock_driver, 'switch1' + ) + + # Should use global config + self.assertEqual({100, 200, 201, 202}, result['allowed_vlans']) + mock_driver.get_switch_info.assert_called_once_with('switch1') + + def test_get_switch_vlan_config_switch_info_none(self): + """Test when get_switch_info returns None.""" + mock_driver = mock.Mock() + mock_driver.get_switch_info.return_value = None + + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = ['500'] + + result = switch_config.get_switch_vlan_config( + mock_driver, 'switch1' + ) + + # Should use global config + self.assertEqual({500}, result['allowed_vlans']) + mock_driver.get_switch_info.assert_called_once_with('switch1') + + def test_get_switch_vlan_config_empty_switch_allowed_vlans(self): + """Test when switch has empty allowed_vlans.""" + mock_driver = mock.Mock() + mock_driver.get_switch_info.return_value = { + 'allowed_vlans': [] + } + + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = ['100'] + + result = switch_config.get_switch_vlan_config( + mock_driver, 'switch1' + ) + + # Empty switch config should fallback to global + self.assertEqual({100}, result['allowed_vlans']) + mock_driver.get_switch_info.assert_called_once_with('switch1') + + def test_get_switch_vlan_config_no_global_config(self): + """Test when there's no global config.""" + mock_driver = mock.Mock() + mock_driver.get_switch_info.return_value = {} + + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = None + + result = switch_config.get_switch_vlan_config( + mock_driver, 'switch1' + ) + + # Should return empty set + self.assertEqual(set(), result['allowed_vlans']) + mock_driver.get_switch_info.assert_called_once_with('switch1') + + def test_get_switch_vlan_config_switch_specific_priority(self): + """Test that switch-specific config takes priority over global.""" + mock_driver = mock.Mock() + mock_driver.get_switch_info.return_value = { + 'allowed_vlans': ['300-302'] + } + + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = ['100-200'] + + result = switch_config.get_switch_vlan_config( + mock_driver, 'switch1' + ) + + # Should use switch-specific, not global + self.assertEqual({300, 301, 302}, result['allowed_vlans']) + mock_driver.get_switch_info.assert_called_once_with('switch1') + + +class ValidateVlanConfigurationTestCase(base.TestCase): + """Test cases for validate_vlan_configuration function.""" + + def test_validate_vlan_configuration_empty_vlans(self): + """Test that empty VLAN list returns without validation.""" + mock_driver = mock.Mock() + + # Should not call get_switch_info if vlans_to_check is empty + switch_config.validate_vlan_configuration( + [], mock_driver, 'switch1' + ) + mock_driver.get_switch_info.assert_not_called() + + def test_validate_vlan_configuration_none_vlans(self): + """Test that None VLAN list returns without validation.""" + mock_driver = mock.Mock() + + # Should not call get_switch_info if vlans_to_check is None + switch_config.validate_vlan_configuration( + None, mock_driver, 'switch1' + ) + mock_driver.get_switch_info.assert_not_called() + + def test_validate_vlan_configuration_all_allowed(self): + """Test when all VLANs are allowed.""" + mock_driver = mock.Mock() + mock_driver.get_switch_info.return_value = { + 'allowed_vlans': ['100-200'] + } + + # Should not raise exception + switch_config.validate_vlan_configuration( + [100, 150, 200], mock_driver, 'switch1' + ) + mock_driver.get_switch_info.assert_called_once_with('switch1') + + def test_validate_vlan_configuration_some_disallowed(self): + """Test when some VLANs are not allowed.""" + mock_driver = mock.Mock() + mock_driver.get_switch_info.return_value = { + 'allowed_vlans': ['100-200'] + } + + # Should raise exception for VLANs outside allowed range + exc = self.assertRaises( + exception.InvalidParameterValue, + switch_config.validate_vlan_configuration, + [100, 250, 300], + mock_driver, + 'switch1', + 'port configuration' + ) + + # Check exception message contains expected info + self.assertIn('250', str(exc)) + self.assertIn('300', str(exc)) + self.assertIn('switch1', str(exc)) + self.assertIn('port configuration', str(exc)) + mock_driver.get_switch_info.assert_called_once_with('switch1') + + def test_validate_vlan_configuration_no_allowed_vlans_set(self): + """Test when no allowed VLANs are configured (all allowed).""" + mock_driver = mock.Mock() + mock_driver.get_switch_info.return_value = {} + + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = None + + # Should not raise exception when no restrictions + switch_config.validate_vlan_configuration( + [100, 200, 300], mock_driver, 'switch1' + ) + mock_driver.get_switch_info.assert_called_once_with('switch1') + + def test_validate_vlan_configuration_single_vlan_allowed(self): + """Test validation with a single VLAN.""" + mock_driver = mock.Mock() + mock_driver.get_switch_info.return_value = { + 'allowed_vlans': ['100'] + } + + # Should not raise exception for allowed VLAN + switch_config.validate_vlan_configuration( + [100], mock_driver, 'switch1' + ) + mock_driver.get_switch_info.assert_called_once_with('switch1') + + def test_validate_vlan_configuration_single_vlan_disallowed(self): + """Test validation with a single disallowed VLAN.""" + mock_driver = mock.Mock() + mock_driver.get_switch_info.return_value = { + 'allowed_vlans': ['100'] + } + + # Should raise exception for disallowed VLAN + exc = self.assertRaises( + exception.InvalidParameterValue, + switch_config.validate_vlan_configuration, + [200], + mock_driver, + 'switch1', + 'VLAN assignment' + ) + + self.assertIn('200', str(exc)) + self.assertIn('switch1', str(exc)) + self.assertIn('VLAN assignment', str(exc)) + mock_driver.get_switch_info.assert_called_once_with('switch1') + + def test_validate_vlan_configuration_complex_ranges(self): + """Test validation with complex VLAN ranges.""" + mock_driver = mock.Mock() + mock_driver.get_switch_info.return_value = { + 'allowed_vlans': ['100', '200-202', '300-310'] + } + + # Should not raise for VLANs in allowed list/ranges + switch_config.validate_vlan_configuration( + [100, 200, 201, 202, 305], + mock_driver, + 'switch1' + ) + + # Should raise for VLANs outside allowed list/ranges + self.assertRaises( + exception.InvalidParameterValue, + switch_config.validate_vlan_configuration, + [100, 150, 200], # 150 is not in allowed list + mock_driver, + 'switch1' + ) + + def test_validate_vlan_configuration_custom_operation_description(self): + """Test that custom operation description appears in error.""" + mock_driver = mock.Mock() + mock_driver.get_switch_info.return_value = { + 'allowed_vlans': ['100'] + } + + exc = self.assertRaises( + exception.InvalidParameterValue, + switch_config.validate_vlan_configuration, + [200], + mock_driver, + 'switch-xyz', + 'trunk port configuration' + ) + + # Verify custom operation description is in error message + self.assertIn('trunk port configuration', str(exc)) + self.assertIn('switch-xyz', str(exc)) + + def test_validate_vlan_configuration_fallback_to_global(self): + """Test validation falls back to global config.""" + mock_driver = mock.Mock() + mock_driver.get_switch_info.return_value = {} + + with mock.patch('ironic.networking.switch_config.CONF', + spec_set=['ironic_networking']) as mock_conf: + mock_conf.ironic_networking.allowed_vlans = ['100-105'] + + # Should use global config and allow VLANs in range + switch_config.validate_vlan_configuration( + [100, 105], mock_driver, 'switch1' + ) + + # Should raise for VLANs outside global range + self.assertRaises( + exception.InvalidParameterValue, + switch_config.validate_vlan_configuration, + [200], + mock_driver, + 'switch1' + ) diff --git a/ironic/tests/unit/networking/test_utils.py b/ironic/tests/unit/networking/test_utils.py index 2a05afe28c..832ab6a5ea 100644 --- a/ironic/tests/unit/networking/test_utils.py +++ b/ironic/tests/unit/networking/test_utils.py @@ -123,166 +123,95 @@ class ParseVlanRangesTestCase(base.TestCase): result = utils.parse_vlan_ranges(['1', '4094']) self.assertEqual({1, 4094}, result) + def test_parse_vlan_ranges_string_single_vlan(self): + """Test parsing a string with single VLAN ID.""" + result = utils.parse_vlan_ranges('100') + self.assertEqual({100}, result) -class ValidateVlanAllowedTestCase(base.TestCase): - """Test cases for validate_vlan_allowed function.""" + def test_parse_vlan_ranges_string_multiple_vlans(self): + """Test parsing a string with multiple VLAN IDs.""" + result = utils.parse_vlan_ranges('100,200,300') + self.assertEqual({100, 200, 300}, result) - def test_validate_vlan_allowed_none_config(self): - """Test that None config allows all VLANs.""" - with mock.patch('ironic.networking.utils.CONF', - spec_set=['ironic_networking']) as mock_conf: - mock_conf.ironic_networking.allowed_vlans = None - result = utils.validate_vlan_allowed(100) - self.assertTrue(result) + def test_parse_vlan_ranges_string_simple_range(self): + """Test parsing a string with a simple VLAN range.""" + result = utils.parse_vlan_ranges('100-103') + self.assertEqual({100, 101, 102, 103}, result) - def test_validate_vlan_allowed_empty_list_config(self): - """Test that empty list config denies all VLANs.""" - with mock.patch('ironic.networking.utils.CONF', - spec_set=['ironic_networking']) as mock_conf: - mock_conf.ironic_networking.allowed_vlans = [] - self.assertRaises( - exception.InvalidParameterValue, - utils.validate_vlan_allowed, - 100 - ) + def test_parse_vlan_ranges_string_complex_spec(self): + """Test parsing a string with ranges and singles.""" + result = utils.parse_vlan_ranges('100,101,102-104,106') + self.assertEqual({100, 101, 102, 103, 104, 106}, result) - def test_validate_vlan_allowed_vlan_in_list(self): - """Test that VLAN in allowed list is accepted.""" - with mock.patch('ironic.networking.utils.CONF', - spec_set=['ironic_networking']) as mock_conf: - mock_conf.ironic_networking.allowed_vlans = ['100', '200', '300'] - result = utils.validate_vlan_allowed(100) - self.assertTrue(result) + def test_parse_vlan_ranges_string_with_spaces(self): + """Test parsing a string with spaces.""" + result = utils.parse_vlan_ranges(' 100 , 102 - 104 , 106 ') + self.assertEqual({100, 102, 103, 104, 106}, result) - def test_validate_vlan_allowed_vlan_not_in_list(self): - """Test that VLAN not in allowed list is rejected.""" - with mock.patch('ironic.networking.utils.CONF', - spec_set=['ironic_networking']) as mock_conf: - mock_conf.ironic_networking.allowed_vlans = ['100', '200', '300'] - self.assertRaises( - exception.InvalidParameterValue, - utils.validate_vlan_allowed, - 150 - ) + def test_parse_vlan_ranges_string_with_trailing_comma(self): + """Test parsing a string with trailing comma.""" + result = utils.parse_vlan_ranges('100,200,') + self.assertEqual({100, 200}, result) - def test_validate_vlan_allowed_vlan_in_range(self): - """Test that VLAN in allowed range is accepted.""" - with mock.patch('ironic.networking.utils.CONF', - spec_set=['ironic_networking']) as mock_conf: - mock_conf.ironic_networking.allowed_vlans = ['100-200'] - result = utils.validate_vlan_allowed(150) - self.assertTrue(result) + def test_parse_vlan_ranges_string_with_leading_comma(self): + """Test parsing a string with leading comma.""" + result = utils.parse_vlan_ranges(',100,200') + self.assertEqual({100, 200}, result) - def test_validate_vlan_allowed_vlan_not_in_range(self): - """Test that VLAN not in allowed range is rejected.""" - with mock.patch('ironic.networking.utils.CONF', - spec_set=['ironic_networking']) as mock_conf: - mock_conf.ironic_networking.allowed_vlans = ['100-200'] - self.assertRaises( - exception.InvalidParameterValue, - utils.validate_vlan_allowed, - 250 - ) + def test_parse_vlan_ranges_string_with_double_comma(self): + """Test parsing a string with double commas (empty elements).""" + result = utils.parse_vlan_ranges('100,,200') + self.assertEqual({100, 200}, result) - def test_validate_vlan_allowed_complex_spec(self): - """Test validation with complex allowed VLAN specification.""" - with mock.patch('ironic.networking.utils.CONF', - spec_set=['ironic_networking']) as mock_conf: - mock_conf.ironic_networking.allowed_vlans = [ - '100', '101', '102-104', '106' - ] - # Test allowed VLANs - self.assertTrue(utils.validate_vlan_allowed(100)) - self.assertTrue(utils.validate_vlan_allowed(101)) - self.assertTrue(utils.validate_vlan_allowed(102)) - self.assertTrue(utils.validate_vlan_allowed(103)) - self.assertTrue(utils.validate_vlan_allowed(104)) - self.assertTrue(utils.validate_vlan_allowed(106)) - # Test disallowed VLAN - self.assertRaises( - exception.InvalidParameterValue, - utils.validate_vlan_allowed, - 105 - ) + def test_parse_vlan_ranges_string_empty(self): + """Test parsing an empty string returns empty set.""" + result = utils.parse_vlan_ranges('') + self.assertEqual(set(), result) - def test_validate_vlan_allowed_override_config(self): - """Test that allowed_vlans_config parameter overrides CONF.""" - with mock.patch('ironic.networking.utils.CONF', - spec_set=['ironic_networking']) as mock_conf: - mock_conf.ironic_networking.allowed_vlans = ['100'] - # Override should allow 200, not 100 - result = utils.validate_vlan_allowed( - 200, - allowed_vlans_config=['200'] - ) - self.assertTrue(result) - # Should reject 100 when using override - self.assertRaises( - exception.InvalidParameterValue, - utils.validate_vlan_allowed, - 100, - allowed_vlans_config=['200'] - ) + def test_parse_vlan_ranges_string_whitespace_only(self): + """Test parsing a whitespace-only string returns empty set.""" + result = utils.parse_vlan_ranges(' ') + self.assertEqual(set(), result) - def test_validate_vlan_allowed_switch_config_override(self): - """Test that switch config overrides global config.""" - with mock.patch('ironic.networking.utils.CONF', - spec_set=['ironic_networking']) as mock_conf: - mock_conf.ironic_networking.allowed_vlans = ['100'] - switch_config = {'allowed_vlans': ['200']} - # Switch config should allow 200, not 100 - result = utils.validate_vlan_allowed( - 200, - switch_config=switch_config - ) - self.assertTrue(result) - # Should reject 100 when using switch config - self.assertRaises( - exception.InvalidParameterValue, - utils.validate_vlan_allowed, - 100, - switch_config=switch_config - ) + def test_parse_vlan_ranges_string_invalid_vlan_too_low(self): + """Test that string with VLAN ID 0 raises an error.""" + self.assertRaises( + exception.InvalidParameterValue, + utils.parse_vlan_ranges, + '0' + ) - def test_validate_vlan_allowed_switch_config_no_allowed_vlans(self): - """Test that switch config without allowed_vlans uses global.""" - with mock.patch('ironic.networking.utils.CONF', - spec_set=['ironic_networking']) as mock_conf: - mock_conf.ironic_networking.allowed_vlans = ['100'] - switch_config = {'some_other_key': 'value'} - # Should fall back to global config - result = utils.validate_vlan_allowed( - 100, - switch_config=switch_config - ) - self.assertTrue(result) + def test_parse_vlan_ranges_string_invalid_vlan_too_high(self): + """Test that string with VLAN ID 4095 raises an error.""" + self.assertRaises( + exception.InvalidParameterValue, + utils.parse_vlan_ranges, + '4095' + ) - def test_validate_vlan_allowed_switch_config_empty_list(self): - """Test that switch config with empty list denies all VLANs.""" - with mock.patch('ironic.networking.utils.CONF', - spec_set=['ironic_networking']) as mock_conf: - mock_conf.ironic_networking.allowed_vlans = ['100'] - switch_config = {'allowed_vlans': []} - # Switch config empty list should deny even though global allows - self.assertRaises( - exception.InvalidParameterValue, - utils.validate_vlan_allowed, - 100, - switch_config=switch_config - ) + def test_parse_vlan_ranges_string_invalid_format_not_a_number(self): + """Test that string with non-numeric VLAN raises an error.""" + self.assertRaises( + exception.InvalidParameterValue, + utils.parse_vlan_ranges, + 'abc' + ) - def test_validate_vlan_allowed_switch_config_none(self): - """Test that switch config with None allows all VLANs.""" - with mock.patch('ironic.networking.utils.CONF', - spec_set=['ironic_networking']) as mock_conf: - mock_conf.ironic_networking.allowed_vlans = ['100'] - switch_config = {'allowed_vlans': None} - # Switch config None should allow all, even though global restricts - result = utils.validate_vlan_allowed( - 200, - switch_config=switch_config - ) - self.assertTrue(result) + def test_parse_vlan_ranges_string_invalid_format_bad_range(self): + """Test that string with malformed range raises an error.""" + self.assertRaises( + exception.InvalidParameterValue, + utils.parse_vlan_ranges, + '100-200-300' + ) + + def test_parse_vlan_ranges_string_mixed_valid_invalid(self): + """Test that string with mixed valid and invalid raises an error.""" + self.assertRaises( + exception.InvalidParameterValue, + utils.parse_vlan_ranges, + '100,abc,200' + ) class RpcTransportTestCase(unittest.TestCase):