openstack/ironic
Code Issues Proposed changes

Add networking driver framework infrastructure

Browse Source 
Implements the foundational driver framework for the networking service,
providing abstraction and loading mechanisms for network switch drivers:

- Driver factory for loading and managing switch driver plugins using
  stevedore, with support for multiple concurrent drivers
- Driver adapter for preprocessing switch configuration files and
  managing driver lifecycle
- Driver translators for converting between ironic network data formats
  and driver-specific configuration formats
- Utility functions for network configuration validation, VLAN range
  parsing, and RPC transport detection

This framework provides the foundation for integrating various network
switch drivers (e.g., networking-generic-switch) with the ironic
networking service. The framework is used by the manager implementation
added in the subsequent commit.

Related-Bug: 2113769
Assisted-by: Claude/sonnet-4.5
Change-Id: Ifb6e662ef59f9e12aad7c34356d2e78c3ebb4143
Signed-off-by: Allain Legacy <alegacy@redhat.com>
This commit is contained in:
Allain Legacy
2025-11-05 14:42:32 -05:00
parent 7495f77258
commit 05b9dc22c7
11 changed files with 2208 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
ironic/networking/switch_config.py Normal file
View File

@@ -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),
}
)

0
ironic/networking/switch_drivers/__init__.py Normal file
View File

293
ironic/networking/switch_drivers/driver_adapter.py Normal file
View File

@@ -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
)

175
ironic/networking/switch_drivers/driver_factory.py Normal file
View File

@@ -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())

85
ironic/networking/switch_drivers/driver_translators.py Normal file
View File

@@ -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"
)

54
ironic/networking/utils.py
View File

@@ -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

642
ironic/tests/unit/networking/test_driver_adapter.py Normal file
View File

@@ -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)

176
ironic/tests/unit/networking/test_driver_factory.py Normal file
View File

@@ -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)

132
ironic/tests/unit/networking/test_driver_translators.py Normal file
View File

@@ -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)

481
ironic/tests/unit/networking/test_switch_config.py Normal file
View File

@@ -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'
)

223
ironic/tests/unit/networking/test_utils.py
View File

@@ -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):