"Fossies" - the Fresh Open Source Software Archive

Member "ironic-16.0.3/ironic/tests/unit/drivers/modules/test_ipmitool.py" (18 Jan 2021, 145862 Bytes) of package /linux/misc/openstack/ironic-16.0.3.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_ipmitool.py": 16.0.2_vs_16.0.3.

    1 # coding=utf-8
    2 
    3 # Copyright 2012 Hewlett-Packard Development Company, L.P.
    4 # Copyright (c) 2012 NTT DOCOMO, INC.
    5 # Copyright 2014 International Business Machines Corporation
    6 # All Rights Reserved.
    7 #
    8 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    9 #    not use this file except in compliance with the License. You may obtain
   10 #    a copy of the License at
   11 #
   12 #         http://www.apache.org/licenses/LICENSE-2.0
   13 #
   14 #    Unless required by applicable law or agreed to in writing, software
   15 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   16 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   17 #    License for the specific language governing permissions and limitations
   18 #    under the License.
   19 #
   20 
   21 """Test class for IPMITool driver module."""
   22 
   23 import contextlib
   24 import os
   25 import random
   26 import stat
   27 import subprocess
   28 import tempfile
   29 import time
   30 import types
   31 from unittest import mock
   32 
   33 import fixtures
   34 from ironic_lib import utils as ironic_utils
   35 from oslo_concurrency import processutils
   36 from oslo_utils import uuidutils
   37 
   38 from ironic.common import boot_devices
   39 from ironic.common import exception
   40 from ironic.common import states
   41 from ironic.common import utils
   42 from ironic.conductor import task_manager
   43 import ironic.conf
   44 from ironic.drivers.modules import boot_mode_utils
   45 from ironic.drivers.modules import console_utils
   46 from ironic.drivers.modules import ipmitool as ipmi
   47 from ironic.drivers import utils as driver_utils
   48 from ironic.tests import base
   49 from ironic.tests.unit.db import base as db_base
   50 from ironic.tests.unit.db import utils as db_utils
   51 from ironic.tests.unit.objects import utils as obj_utils
   52 
   53 CONF = ironic.conf.CONF
   54 
   55 INFO_DICT = db_utils.get_test_ipmi_info()
   56 
   57 # BRIDGE_INFO_DICT will have all the bridging parameters appended
   58 BRIDGE_INFO_DICT = INFO_DICT.copy()
   59 BRIDGE_INFO_DICT.update(db_utils.get_test_ipmi_bridging_parameters())
   60 
   61 MC_INFO = """Device ID                 : 32
   62 Device Revision           : 1
   63 Firmware Revision         : 0.99
   64 IPMI Version              : 2.0
   65 Manufacturer ID           : 1234
   66 Manufacturer Name         : {vendor}
   67 Product ID                : 2001 (0x0801)
   68 Product Name              : Unknown (0x101)
   69 Device Available          : yes
   70 Provides Device SDRs      : no
   71 Additional Device Support :
   72     Sensor Device
   73     SDR Repository Device
   74     SEL Device
   75     FRU Inventory Device
   76     IPMB Event Receiver
   77     IPMB Event Generator
   78     Chassis Device
   79 Aux Firmware Rev Info     :
   80     0x00
   81     0x00
   82     0x00
   83     0x00
   84 """
   85 
   86 
   87 class IPMIToolCheckInitTestCase(base.TestCase):
   88 
   89     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
   90     @mock.patch.object(utils, 'check_dir', autospec=True)
   91     def test_power_init_calls(self, mock_check_dir, mock_support):
   92         mock_support.return_value = True
   93         ipmi.TMP_DIR_CHECKED = None
   94         ipmi.IPMIPower()
   95         mock_support.assert_called_with(mock.ANY)
   96         mock_check_dir.assert_called_once_with()
   97 
   98     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
   99     @mock.patch.object(utils, 'check_dir', autospec=True)
  100     def test_power_init_calls_raises_1(self, mock_check_dir, mock_support):
  101         mock_support.return_value = True
  102         ipmi.TMP_DIR_CHECKED = None
  103         mock_check_dir.side_effect = exception.PathNotFound(dir="foo_dir")
  104         self.assertRaises(exception.PathNotFound, ipmi.IPMIPower)
  105 
  106     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  107     @mock.patch.object(utils, 'check_dir', autospec=True)
  108     def test_power_init_calls_raises_2(self, mock_check_dir, mock_support):
  109         mock_support.return_value = True
  110         ipmi.TMP_DIR_CHECKED = None
  111         mock_check_dir.side_effect = exception.DirectoryNotWritable(
  112             dir="foo_dir")
  113         self.assertRaises(exception.DirectoryNotWritable, ipmi.IPMIPower)
  114 
  115     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  116     @mock.patch.object(utils, 'check_dir', autospec=True)
  117     def test_power_init_calls_raises_3(self, mock_check_dir, mock_support):
  118         mock_support.return_value = True
  119         ipmi.TMP_DIR_CHECKED = None
  120         mock_check_dir.side_effect = exception.InsufficientDiskSpace(
  121             path="foo_dir", required=1, actual=0)
  122         self.assertRaises(exception.InsufficientDiskSpace, ipmi.IPMIPower)
  123 
  124     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  125     @mock.patch.object(utils, 'check_dir', autospec=True)
  126     def test_power_init_calls_already_checked(self,
  127                                               mock_check_dir,
  128                                               mock_support):
  129         mock_support.return_value = True
  130         ipmi.TMP_DIR_CHECKED = True
  131         ipmi.IPMIPower()
  132         mock_support.assert_called_with(mock.ANY)
  133         self.assertFalse(mock_check_dir.called)
  134 
  135     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  136     @mock.patch.object(utils, 'check_dir', autospec=True)
  137     def test_management_init_calls(self, mock_check_dir, mock_support):
  138         mock_support.return_value = True
  139         ipmi.TMP_DIR_CHECKED = None
  140 
  141         ipmi.IPMIManagement()
  142         mock_support.assert_called_with(mock.ANY)
  143         mock_check_dir.assert_called_once_with()
  144 
  145     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  146     @mock.patch.object(utils, 'check_dir', autospec=True)
  147     def test_management_init_calls_already_checked(self,
  148                                                    mock_check_dir,
  149                                                    mock_support):
  150         mock_support.return_value = True
  151         ipmi.TMP_DIR_CHECKED = False
  152 
  153         ipmi.IPMIManagement()
  154         mock_support.assert_called_with(mock.ANY)
  155         self.assertFalse(mock_check_dir.called)
  156 
  157     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  158     @mock.patch.object(utils, 'check_dir', autospec=True)
  159     def test_vendor_passthru_init_calls(self, mock_check_dir, mock_support):
  160         mock_support.return_value = True
  161         ipmi.TMP_DIR_CHECKED = None
  162         ipmi.VendorPassthru()
  163         mock_support.assert_called_with(mock.ANY)
  164         mock_check_dir.assert_called_once_with()
  165 
  166     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  167     @mock.patch.object(utils, 'check_dir', autospec=True)
  168     def test_vendor_passthru_init_calls_already_checked(self,
  169                                                         mock_check_dir,
  170                                                         mock_support):
  171         mock_support.return_value = True
  172         ipmi.TMP_DIR_CHECKED = True
  173         ipmi.VendorPassthru()
  174         mock_support.assert_called_with(mock.ANY)
  175         self.assertFalse(mock_check_dir.called)
  176 
  177     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  178     @mock.patch.object(utils, 'check_dir', autospec=True)
  179     def test_console_init_calls(self, mock_check_dir, mock_support):
  180         mock_support.return_value = True
  181         ipmi.TMP_DIR_CHECKED = None
  182         ipmi.IPMIShellinaboxConsole()
  183         mock_support.assert_called_with(mock.ANY)
  184         mock_check_dir.assert_called_once_with()
  185 
  186     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  187     @mock.patch.object(utils, 'check_dir', autospec=True)
  188     def test_console_init_calls_already_checked(self,
  189                                                 mock_check_dir,
  190                                                 mock_support):
  191         mock_support.return_value = True
  192         ipmi.TMP_DIR_CHECKED = True
  193         ipmi.IPMIShellinaboxConsole()
  194         mock_support.assert_called_with(mock.ANY)
  195         self.assertFalse(mock_check_dir.called)
  196 
  197     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  198     @mock.patch.object(utils, 'check_dir', autospec=True)
  199     def test_console_init_calls_for_socat(self, mock_check_dir, mock_support):
  200         with mock.patch.object(ipmi, 'TMP_DIR_CHECKED', autospec=True):
  201             mock_support.return_value = True
  202             ipmi.TMP_DIR_CHECKED = None
  203             ipmi.IPMISocatConsole()
  204             mock_support.assert_called_with(mock.ANY)
  205             mock_check_dir.assert_called_once_with()
  206 
  207     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  208     @mock.patch.object(utils, 'check_dir', autospec=True)
  209     def test_console_init_calls_for_socat_already_checked(self,
  210                                                           mock_check_dir,
  211                                                           mock_support):
  212         with mock.patch.object(ipmi, 'TMP_DIR_CHECKED', autospec=True):
  213             mock_support.return_value = True
  214             ipmi.TMP_DIR_CHECKED = True
  215             ipmi.IPMISocatConsole()
  216             mock_support.assert_called_with(mock.ANY)
  217             self.assertFalse(mock_check_dir.called)
  218 
  219 
  220 @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  221 @mock.patch.object(subprocess, 'check_call', autospec=True)
  222 class IPMIToolCheckOptionSupportedTestCase(base.TestCase):
  223 
  224     def test_check_timing_pass(self, mock_chkcall, mock_support):
  225         mock_chkcall.return_value = (None, None)
  226         mock_support.return_value = None
  227         expected = [mock.call('timing'),
  228                     mock.call('timing', True)]
  229 
  230         ipmi._check_option_support(['timing'])
  231         self.assertTrue(mock_chkcall.called)
  232         self.assertEqual(expected, mock_support.call_args_list)
  233 
  234     def test_check_timing_fail(self, mock_chkcall, mock_support):
  235         mock_chkcall.side_effect = subprocess.CalledProcessError(1, 'ipmitool')
  236         mock_support.return_value = None
  237         expected = [mock.call('timing'),
  238                     mock.call('timing', False)]
  239 
  240         ipmi._check_option_support(['timing'])
  241         self.assertTrue(mock_chkcall.called)
  242         self.assertEqual(expected, mock_support.call_args_list)
  243 
  244     def test_check_timing_no_ipmitool(self, mock_chkcall, mock_support):
  245         mock_chkcall.side_effect = OSError()
  246         mock_support.return_value = None
  247         expected = [mock.call('timing')]
  248 
  249         self.assertRaises(OSError, ipmi._check_option_support, ['timing'])
  250         self.assertTrue(mock_chkcall.called)
  251         self.assertEqual(expected, mock_support.call_args_list)
  252 
  253     def test_check_single_bridge_pass(self, mock_chkcall, mock_support):
  254         mock_chkcall.return_value = (None, None)
  255         mock_support.return_value = None
  256         expected = [mock.call('single_bridge'),
  257                     mock.call('single_bridge', True)]
  258 
  259         ipmi._check_option_support(['single_bridge'])
  260         self.assertTrue(mock_chkcall.called)
  261         self.assertEqual(expected, mock_support.call_args_list)
  262 
  263     def test_check_single_bridge_fail(self, mock_chkcall, mock_support):
  264         mock_chkcall.side_effect = subprocess.CalledProcessError(1, 'ipmitool')
  265         mock_support.return_value = None
  266         expected = [mock.call('single_bridge'),
  267                     mock.call('single_bridge', False)]
  268 
  269         ipmi._check_option_support(['single_bridge'])
  270         self.assertTrue(mock_chkcall.called)
  271         self.assertEqual(expected, mock_support.call_args_list)
  272 
  273     def test_check_single_bridge_no_ipmitool(self, mock_chkcall,
  274                                              mock_support):
  275         mock_chkcall.side_effect = OSError()
  276         mock_support.return_value = None
  277         expected = [mock.call('single_bridge')]
  278 
  279         self.assertRaises(OSError, ipmi._check_option_support,
  280                           ['single_bridge'])
  281         self.assertTrue(mock_chkcall.called)
  282         self.assertEqual(expected, mock_support.call_args_list)
  283 
  284     def test_check_dual_bridge_pass(self, mock_chkcall, mock_support):
  285         mock_chkcall.return_value = (None, None)
  286         mock_support.return_value = None
  287         expected = [mock.call('dual_bridge'),
  288                     mock.call('dual_bridge', True)]
  289 
  290         ipmi._check_option_support(['dual_bridge'])
  291         self.assertTrue(mock_chkcall.called)
  292         self.assertEqual(expected, mock_support.call_args_list)
  293 
  294     def test_check_dual_bridge_fail(self, mock_chkcall, mock_support):
  295         mock_chkcall.side_effect = subprocess.CalledProcessError(1, 'ipmitool')
  296         mock_support.return_value = None
  297         expected = [mock.call('dual_bridge'),
  298                     mock.call('dual_bridge', False)]
  299 
  300         ipmi._check_option_support(['dual_bridge'])
  301         self.assertTrue(mock_chkcall.called)
  302         self.assertEqual(expected, mock_support.call_args_list)
  303 
  304     def test_check_dual_bridge_no_ipmitool(self, mock_chkcall, mock_support):
  305         mock_chkcall.side_effect = OSError()
  306         mock_support.return_value = None
  307         expected = [mock.call('dual_bridge')]
  308 
  309         self.assertRaises(OSError, ipmi._check_option_support,
  310                           ['dual_bridge'])
  311         self.assertTrue(mock_chkcall.called)
  312         self.assertEqual(expected, mock_support.call_args_list)
  313 
  314     def test_check_all_options_pass(self, mock_chkcall, mock_support):
  315         mock_chkcall.return_value = (None, None)
  316         mock_support.return_value = None
  317         expected = [
  318             mock.call('timing'), mock.call('timing', True),
  319             mock.call('single_bridge'),
  320             mock.call('single_bridge', True),
  321             mock.call('dual_bridge'), mock.call('dual_bridge', True)]
  322 
  323         ipmi._check_option_support(['timing', 'single_bridge', 'dual_bridge'])
  324         self.assertTrue(mock_chkcall.called)
  325         self.assertEqual(expected, mock_support.call_args_list)
  326 
  327     def test_check_all_options_fail(self, mock_chkcall, mock_support):
  328         options = ['timing', 'single_bridge', 'dual_bridge']
  329         mock_chkcall.side_effect = [subprocess.CalledProcessError(
  330             1, 'ipmitool')] * len(options)
  331         mock_support.return_value = None
  332         expected = [
  333             mock.call('timing'), mock.call('timing', False),
  334             mock.call('single_bridge'),
  335             mock.call('single_bridge', False),
  336             mock.call('dual_bridge'),
  337             mock.call('dual_bridge', False)]
  338 
  339         ipmi._check_option_support(options)
  340         self.assertTrue(mock_chkcall.called)
  341         self.assertEqual(expected, mock_support.call_args_list)
  342 
  343     def test_check_all_options_no_ipmitool(self, mock_chkcall, mock_support):
  344         mock_chkcall.side_effect = OSError()
  345         mock_support.return_value = None
  346         # exception is raised once ipmitool was not found for an command
  347         expected = [mock.call('timing')]
  348 
  349         self.assertRaises(OSError, ipmi._check_option_support,
  350                           ['timing', 'single_bridge', 'dual_bridge'])
  351         self.assertTrue(mock_chkcall.called)
  352         self.assertEqual(expected, mock_support.call_args_list)
  353 
  354 
  355 awesome_password_filename = 'awesome_password_filename'
  356 
  357 
  358 @contextlib.contextmanager
  359 def _make_password_file_stub(password):
  360     yield awesome_password_filename
  361 
  362 
  363 class IPMIToolPrivateMethodTestCaseMeta(type):
  364     """Generate and inject parametrized test cases"""
  365 
  366     ipmitool_errors = [
  367         'insufficient resources for session',
  368         'Node busy',
  369         'Timeout',
  370         'Out of space',
  371         'BMC initialization in progress'
  372     ]
  373 
  374     def __new__(mcs, name, bases, attrs):
  375 
  376         def gen_test_methods(message):
  377 
  378             @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  379             @mock.patch.object(utils, 'execute', autospec=True)
  380             def exec_ipmitool_exception_retry(
  381                     self, mock_exec, mock_support):
  382                 ipmi.LAST_CMD_TIME = {}
  383                 mock_support.return_value = False
  384                 mock_exec.side_effect = [
  385                     processutils.ProcessExecutionError(
  386                         stderr=message
  387                     ),
  388                     (None, None)
  389                 ]
  390 
  391                 # Directly set the configuration values such that
  392                 # the logic will cause _exec_ipmitool to retry twice.
  393                 self.config(min_command_interval=1, group='ipmi')
  394                 self.config(command_retry_timeout=2, group='ipmi')
  395 
  396                 ipmi._exec_ipmitool(self.info, 'A B C')
  397 
  398                 mock_support.assert_called_once_with('timing')
  399                 self.assertEqual(2, mock_exec.call_count)
  400 
  401             @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  402             @mock.patch.object(utils, 'execute', autospec=True)
  403             def exec_ipmitool_exception_retries_exceeded(
  404                     self, mock_exec, mock_support):
  405                 ipmi.LAST_CMD_TIME = {}
  406                 mock_support.return_value = False
  407 
  408                 mock_exec.side_effect = [processutils.ProcessExecutionError(
  409                     stderr=message
  410                 )]
  411 
  412                 # Directly set the configuration values such that
  413                 # the logic will cause _exec_ipmitool to timeout.
  414                 self.config(min_command_interval=1, group='ipmi')
  415                 self.config(command_retry_timeout=1, group='ipmi')
  416 
  417                 self.assertRaises(processutils.ProcessExecutionError,
  418                                   ipmi._exec_ipmitool,
  419                                   self.info, 'A B C')
  420                 mock_support.assert_called_once_with('timing')
  421                 self.assertEqual(1, mock_exec.call_count)
  422 
  423             @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  424             @mock.patch.object(utils, 'execute', autospec=True)
  425             def exec_ipmitool_exception_non_retryable_failure(
  426                     self, mock_exec, mock_support):
  427                 ipmi.LAST_CMD_TIME = {}
  428                 mock_support.return_value = False
  429                 additional_msg = "RAKP 2 HMAC is invalid"
  430 
  431                 # Return a retryable error, then an error that cannot
  432                 # be retried thus resulting in a single retry
  433                 # attempt by _exec_ipmitool.
  434                 mock_exec.side_effect = [
  435                     processutils.ProcessExecutionError(
  436                         stderr=message
  437                     ),
  438                     processutils.ProcessExecutionError(
  439                         stderr="Some more info: %s" % additional_msg
  440                     ),
  441                     processutils.ProcessExecutionError(
  442                         stderr="Unknown"
  443                     ),
  444                 ]
  445 
  446                 # Directly set the configuration values such that
  447                 # the logic will cause _exec_ipmitool to retry up
  448                 # to 3 times.
  449                 self.config(min_command_interval=1, group='ipmi')
  450                 self.config(command_retry_timeout=3, group='ipmi')
  451                 self.config(additional_retryable_ipmi_errors=[additional_msg],
  452                             group='ipmi')
  453 
  454                 self.assertRaises(processutils.ProcessExecutionError,
  455                                   ipmi._exec_ipmitool,
  456                                   self.info, 'A B C')
  457                 mock_support.assert_called_once_with('timing')
  458                 self.assertEqual(3, mock_exec.call_count)
  459 
  460             return (exec_ipmitool_exception_retry,
  461                     exec_ipmitool_exception_retries_exceeded,
  462                     exec_ipmitool_exception_non_retryable_failure)
  463 
  464         # NOTE(etingof): this loop will inject some methods into the
  465         # class being built to be then picked up by unittest to test
  466         # ironic's handling of specific `ipmitool` errors
  467         for ipmi_message in mcs.ipmitool_errors:
  468             for fun in gen_test_methods(ipmi_message):
  469                 suffix = ipmi_message.lower().replace(' ', '_')
  470                 test_name = "test_%s_%s" % (fun.__name__, suffix)
  471                 attrs[test_name] = fun
  472 
  473         return type.__new__(mcs, name, bases, attrs)
  474 
  475 
  476 class Base(db_base.DbTestCase):
  477 
  478     def setUp(self):
  479         super(Base, self).setUp()
  480         self.config(enabled_power_interfaces=['fake', 'ipmitool'],
  481                     enabled_management_interfaces=['fake', 'ipmitool'],
  482                     enabled_vendor_interfaces=['fake', 'ipmitool',
  483                                                'no-vendor'],
  484                     enabled_console_interfaces=['fake', 'ipmitool-socat',
  485                                                 'ipmitool-shellinabox',
  486                                                 'no-console'])
  487         self.config(debug=True, group="ipmi")
  488         self.node = obj_utils.create_test_node(
  489             self.context,
  490             console_interface='ipmitool-socat',
  491             management_interface='ipmitool',
  492             power_interface='ipmitool',
  493             vendor_interface='ipmitool',
  494             driver_info=INFO_DICT)
  495         self.info = ipmi._parse_driver_info(self.node)
  496         self.console = ipmi.IPMISocatConsole()
  497         self.management = ipmi.IPMIManagement()
  498         self.power = ipmi.IPMIPower()
  499         self.vendor = ipmi.VendorPassthru()
  500 
  501 
  502 class IPMIToolPrivateMethodTestCase(
  503         Base,
  504         metaclass=IPMIToolPrivateMethodTestCaseMeta):
  505 
  506     def setUp(self):
  507         super(IPMIToolPrivateMethodTestCase, self).setUp()
  508 
  509         # power actions use oslo_service.BackoffLoopingCall,
  510         # mock random.SystemRandom gauss distribution
  511         self._mock_system_random_distribution()
  512 
  513         mock_sleep_fixture = self.useFixture(
  514             fixtures.MockPatchObject(time, 'sleep', autospec=True))
  515         self.mock_sleep = mock_sleep_fixture.mock
  516 
  517     # NOTE(etingof): besides the conventional unittest methods that follow,
  518     # the metaclass will inject some more `test_` methods aimed at testing
  519     # the handling of specific errors potentially returned by the `ipmitool`
  520 
  521     def _mock_system_random_distribution(self):
  522         # random.SystemRandom with gauss distribution is used by oslo_service's
  523         # BackoffLoopingCall, it multiplies default interval (equals to 1) by
  524         # 2 * return_value, so if you want BackoffLoopingCall to "sleep" for
  525         # 1 second, return_value should be 0.5.
  526         m = mock.patch.object(random.SystemRandom, 'gauss', return_value=0.5,
  527                               autospec=True)
  528         m.start()
  529         self.addCleanup(m.stop)
  530 
  531     def _test__make_password_file(self, input_password,
  532                                   exception_to_raise=None):
  533         pw_file = None
  534         try:
  535             with ipmi._make_password_file(input_password) as pw_file:
  536                 if exception_to_raise is not None:
  537                     raise exception_to_raise
  538                 self.assertTrue(os.path.isfile(pw_file))
  539                 self.assertEqual(0o600, os.stat(pw_file)[stat.ST_MODE] & 0o777)
  540                 with open(pw_file, "r") as f:
  541                     password = f.read()
  542                 self.assertEqual(str(input_password), password)
  543         finally:
  544             if pw_file is not None:
  545                 self.assertFalse(os.path.isfile(pw_file))
  546 
  547     def test__make_password_file_str_password(self):
  548         self._test__make_password_file(self.info['password'])
  549 
  550     def test__make_password_file_with_numeric_password(self):
  551         self._test__make_password_file(12345)
  552 
  553     def test__make_password_file_caller_exception(self):
  554         # Test caller raising exception
  555         result = self.assertRaises(
  556             ValueError,
  557             self._test__make_password_file,
  558             12345, ValueError('we should fail'))
  559         self.assertEqual('we should fail', str(result))
  560 
  561     @mock.patch.object(tempfile, 'NamedTemporaryFile',
  562                        new=mock.MagicMock(side_effect=OSError('Test Error')))
  563     def test__make_password_file_tempfile_known_exception(self):
  564         # Test OSError exception in _make_password_file for
  565         # tempfile.NamedTemporaryFile
  566         self.assertRaises(
  567             exception.PasswordFileFailedToCreate,
  568             self._test__make_password_file, 12345)
  569 
  570     @mock.patch.object(
  571         tempfile, 'NamedTemporaryFile',
  572         new=mock.MagicMock(side_effect=OverflowError('Test Error')))
  573     def test__make_password_file_tempfile_unknown_exception(self):
  574         # Test exception in _make_password_file for tempfile.NamedTemporaryFile
  575         result = self.assertRaises(
  576             OverflowError,
  577             self._test__make_password_file, 12345)
  578         self.assertEqual('Test Error', str(result))
  579 
  580     def test__make_password_file_write_exception(self):
  581         # Test exception in _make_password_file for write()
  582         mock_namedtemp = mock.mock_open(mock.MagicMock(name='JLV'))
  583         with mock.patch('tempfile.NamedTemporaryFile', mock_namedtemp):
  584             mock_filehandle = mock_namedtemp.return_value
  585             mock_write = mock_filehandle.write
  586             mock_write.side_effect = OSError('Test 2 Error')
  587             self.assertRaises(
  588                 exception.PasswordFileFailedToCreate,
  589                 self._test__make_password_file, 12345)
  590 
  591     def test__parse_driver_info(self):
  592         # make sure we get back the expected things
  593         _OPTIONS = ['address', 'username', 'password', 'uuid']
  594         for option in _OPTIONS:
  595             self.assertIsNotNone(self.info[option])
  596 
  597         info = dict(INFO_DICT)
  598 
  599         # test the default value for 'priv_level'
  600         node = obj_utils.get_test_node(self.context, driver_info=info)
  601         ret = ipmi._parse_driver_info(node)
  602         self.assertEqual('ADMINISTRATOR', ret['priv_level'])
  603 
  604         # ipmi_username / ipmi_password are not mandatory
  605         del info['ipmi_username']
  606         node = obj_utils.get_test_node(self.context, driver_info=info)
  607         ipmi._parse_driver_info(node)
  608         del info['ipmi_password']
  609         node = obj_utils.get_test_node(self.context, driver_info=info)
  610         ipmi._parse_driver_info(node)
  611 
  612         # make sure error is raised when ipmi_address is missing
  613         info = dict(INFO_DICT)
  614         del info['ipmi_address']
  615         node = obj_utils.get_test_node(self.context, driver_info=info)
  616         self.assertRaises(exception.MissingParameterValue,
  617                           ipmi._parse_driver_info,
  618                           node)
  619 
  620         # test the invalid priv_level value
  621         info = dict(INFO_DICT)
  622         info['ipmi_priv_level'] = 'ABCD'
  623         node = obj_utils.get_test_node(self.context, driver_info=info)
  624         self.assertRaises(exception.InvalidParameterValue,
  625                           ipmi._parse_driver_info,
  626                           node)
  627 
  628     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  629     def test__parse_driver_info_with_invalid_bridging_type(
  630             self, mock_support):
  631         info = BRIDGE_INFO_DICT.copy()
  632         # make sure error is raised when ipmi_bridging has unexpected value
  633         info['ipmi_bridging'] = 'junk'
  634         node = obj_utils.get_test_node(self.context, driver_info=info)
  635         self.assertRaises(exception.InvalidParameterValue,
  636                           ipmi._parse_driver_info,
  637                           node)
  638         self.assertFalse(mock_support.called)
  639 
  640     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  641     def test__parse_driver_info_with_no_bridging(
  642             self, mock_support):
  643         _OPTIONS = ['address', 'username', 'password', 'uuid']
  644         _BRIDGING_OPTIONS = ['local_address', 'transit_channel',
  645                              'transit_address',
  646                              'target_channel', 'target_address']
  647         info = BRIDGE_INFO_DICT.copy()
  648         info['ipmi_bridging'] = 'no'
  649         node = obj_utils.get_test_node(self.context, driver_info=info)
  650         ret = ipmi._parse_driver_info(node)
  651 
  652         # ensure that _is_option_supported was not called
  653         self.assertFalse(mock_support.called)
  654         # check if we got all the required options
  655         for option in _OPTIONS:
  656             self.assertIsNotNone(ret[option])
  657         # test the default value for 'priv_level'
  658         self.assertEqual('ADMINISTRATOR', ret['priv_level'])
  659 
  660         # check if bridging parameters were set to None
  661         for option in _BRIDGING_OPTIONS:
  662             self.assertIsNone(ret[option])
  663 
  664     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  665     def test__parse_driver_info_with_dual_bridging_pass(
  666             self, mock_support):
  667         _OPTIONS = ['address', 'username', 'password', 'uuid',
  668                     'local_address', 'transit_channel', 'transit_address',
  669                     'target_channel', 'target_address']
  670         node = obj_utils.get_test_node(self.context,
  671                                        driver_info=BRIDGE_INFO_DICT)
  672 
  673         expected = [mock.call('dual_bridge')]
  674 
  675         # test double bridging and make sure we get back expected result
  676         mock_support.return_value = True
  677         ret = ipmi._parse_driver_info(node)
  678         self.assertEqual(expected, mock_support.call_args_list)
  679         for option in _OPTIONS:
  680             self.assertIsNotNone(ret[option])
  681         # test the default value for 'priv_level'
  682         self.assertEqual('ADMINISTRATOR', ret['priv_level'])
  683 
  684         info = BRIDGE_INFO_DICT.copy()
  685         # ipmi_local_address / ipmi_username / ipmi_password are not mandatory
  686         for optional_arg in ['ipmi_local_address', 'ipmi_username',
  687                              'ipmi_password']:
  688             del info[optional_arg]
  689             node = obj_utils.get_test_node(self.context, driver_info=info)
  690             ipmi._parse_driver_info(node)
  691             self.assertEqual(mock.call('dual_bridge'), mock_support.call_args)
  692 
  693     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  694     def test__parse_driver_info_with_dual_bridging_not_supported(
  695             self, mock_support):
  696         node = obj_utils.get_test_node(self.context,
  697                                        driver_info=BRIDGE_INFO_DICT)
  698         # if dual bridge is not supported then check if error is raised
  699         mock_support.return_value = False
  700         self.assertRaises(exception.InvalidParameterValue,
  701                           ipmi._parse_driver_info, node)
  702         mock_support.assert_called_once_with('dual_bridge')
  703 
  704     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  705     def test__parse_driver_info_with_dual_bridging_missing_parameters(
  706             self, mock_support):
  707         info = BRIDGE_INFO_DICT.copy()
  708         mock_support.return_value = True
  709         # make sure error is raised when dual bridging is selected and the
  710         # required parameters for dual bridging are not provided
  711         for param in ['ipmi_transit_channel', 'ipmi_target_address',
  712                       'ipmi_transit_address', 'ipmi_target_channel']:
  713             del info[param]
  714             node = obj_utils.get_test_node(self.context, driver_info=info)
  715             self.assertRaises(exception.MissingParameterValue,
  716                               ipmi._parse_driver_info, node)
  717             self.assertEqual(mock.call('dual_bridge'),
  718                              mock_support.call_args)
  719 
  720     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  721     def test__parse_driver_info_with_single_bridging_pass(
  722             self, mock_support):
  723         _OPTIONS = ['address', 'username', 'password', 'uuid',
  724                     'local_address', 'target_channel', 'target_address']
  725 
  726         info = BRIDGE_INFO_DICT.copy()
  727         info['ipmi_bridging'] = 'single'
  728         node = obj_utils.get_test_node(self.context, driver_info=info)
  729 
  730         expected = [mock.call('single_bridge')]
  731 
  732         # test single bridging and make sure we get back expected things
  733         mock_support.return_value = True
  734         ret = ipmi._parse_driver_info(node)
  735         self.assertEqual(expected, mock_support.call_args_list)
  736         for option in _OPTIONS:
  737             self.assertIsNotNone(ret[option])
  738         # test the default value for 'priv_level'
  739         self.assertEqual('ADMINISTRATOR', ret['priv_level'])
  740 
  741         # check if dual bridge params are set to None
  742         self.assertIsNone(ret['transit_channel'])
  743         self.assertIsNone(ret['transit_address'])
  744 
  745         # ipmi_local_address / ipmi_username / ipmi_password are not mandatory
  746         for optional_arg in ['ipmi_local_address', 'ipmi_username',
  747                              'ipmi_password']:
  748             del info[optional_arg]
  749             node = obj_utils.get_test_node(self.context, driver_info=info)
  750             ipmi._parse_driver_info(node)
  751             self.assertEqual(mock.call('single_bridge'),
  752                              mock_support.call_args)
  753 
  754     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  755     def test__parse_driver_info_with_single_bridging_not_supported(
  756             self, mock_support):
  757         info = BRIDGE_INFO_DICT.copy()
  758         info['ipmi_bridging'] = 'single'
  759         node = obj_utils.get_test_node(self.context, driver_info=info)
  760 
  761         # if single bridge is not supported then check if error is raised
  762         mock_support.return_value = False
  763         self.assertRaises(exception.InvalidParameterValue,
  764                           ipmi._parse_driver_info, node)
  765         mock_support.assert_called_once_with('single_bridge')
  766 
  767     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  768     def test__parse_driver_info_with_single_bridging_missing_parameters(
  769             self, mock_support):
  770         info = dict(BRIDGE_INFO_DICT)
  771         info['ipmi_bridging'] = 'single'
  772         mock_support.return_value = True
  773         # make sure error is raised when single bridging is selected and the
  774         # required parameters for single bridging are not provided
  775         for param in ['ipmi_target_channel', 'ipmi_target_address']:
  776             del info[param]
  777             node = obj_utils.get_test_node(self.context, driver_info=info)
  778             self.assertRaises(exception.MissingParameterValue,
  779                               ipmi._parse_driver_info,
  780                               node)
  781             self.assertEqual(mock.call('single_bridge'),
  782                              mock_support.call_args)
  783 
  784     def test__parse_driver_info_numeric_password(self):
  785         # ipmi_password must not be converted to int / float
  786         # even if it includes just numbers.
  787         info = dict(INFO_DICT)
  788         info['ipmi_password'] = 12345678
  789         node = obj_utils.get_test_node(self.context, driver_info=info)
  790         ret = ipmi._parse_driver_info(node)
  791         self.assertEqual(u'12345678', ret['password'])
  792         self.assertIsInstance(ret['password'], str)
  793 
  794     def test__parse_driver_info_ipmi_prot_version_1_5(self):
  795         info = dict(INFO_DICT)
  796         info['ipmi_protocol_version'] = '1.5'
  797         node = obj_utils.get_test_node(self.context, driver_info=info)
  798         ret = ipmi._parse_driver_info(node)
  799         self.assertEqual('1.5', ret['protocol_version'])
  800 
  801     def test__parse_driver_info_invalid_ipmi_prot_version(self):
  802         info = dict(INFO_DICT)
  803         info['ipmi_protocol_version'] = '9000'
  804         node = obj_utils.get_test_node(self.context, driver_info=info)
  805         self.assertRaises(exception.InvalidParameterValue,
  806                           ipmi._parse_driver_info, node)
  807 
  808     def test__parse_driver_info_invalid_ipmi_port(self):
  809         info = dict(INFO_DICT)
  810         info['ipmi_port'] = '700000'
  811         node = obj_utils.get_test_node(self.context, driver_info=info)
  812         self.assertRaises(exception.InvalidParameterValue,
  813                           ipmi._parse_driver_info, node)
  814 
  815     def test__parse_driver_info_ipmi_hex_kg_key(self):
  816         info = dict(INFO_DICT)
  817         info['ipmi_hex_kg_key'] = 'A115023E08E23F7F8DC4BB443A1A75F160763A43'
  818         node = obj_utils.get_test_node(self.context, driver_info=info)
  819         ret = ipmi._parse_driver_info(node)
  820         self.assertEqual(info['ipmi_hex_kg_key'], ret['hex_kg_key'])
  821 
  822     def test__parse_driver_info_ipmi_hex_kg_key_odd_chars(self):
  823         info = dict(INFO_DICT)
  824         info['ipmi_hex_kg_key'] = 'A115023E08E23F7F8DC4BB443A1A75F160763A4'
  825         node = obj_utils.get_test_node(self.context, driver_info=info)
  826         self.assertRaises(exception.InvalidParameterValue,
  827                           ipmi._parse_driver_info, node)
  828 
  829     def test__parse_driver_info_ipmi_port_valid(self):
  830         info = dict(INFO_DICT)
  831         info['ipmi_port'] = '623'
  832         node = obj_utils.get_test_node(self.context, driver_info=info)
  833         ret = ipmi._parse_driver_info(node)
  834         self.assertEqual(623, ret['dest_port'])
  835 
  836     def test__parse_driver_info_ipmi_cipher_suite(self):
  837         info = dict(INFO_DICT)
  838         info['ipmi_cipher_suite'] = 0  # absolute power!
  839         node = obj_utils.get_test_node(self.context, driver_info=info)
  840         ret = ipmi._parse_driver_info(node)
  841         self.assertEqual(0, ret['cipher_suite'])
  842 
  843     def test__parse_driver_info_ipmi_cipher_suite_not_a_number(self):
  844         info = dict(INFO_DICT)
  845         info['ipmi_cipher_suite'] = 'I can haz accez'
  846         node = obj_utils.get_test_node(self.context, driver_info=info)
  847         self.assertRaises(exception.InvalidParameterValue,
  848                           ipmi._parse_driver_info, node)
  849 
  850     def test__parse_driver_info_ipmi_cipher_suite_ipmi_1_5(self):
  851         info = dict(INFO_DICT)
  852         info['ipmi_cipher_suite'] = 0
  853         info['ipmi_protocol_version'] = '1.5'
  854         node = obj_utils.get_test_node(self.context, driver_info=info)
  855         self.assertRaises(exception.InvalidParameterValue,
  856                           ipmi._parse_driver_info, node)
  857 
  858     @mock.patch.object(ipmi.LOG, 'warning', spec_set=True, autospec=True)
  859     def test__parse_driver_info_undefined_credentials(self, mock_log):
  860         info = dict(INFO_DICT)
  861         del info['ipmi_username']
  862         del info['ipmi_password']
  863         node = obj_utils.get_test_node(self.context, driver_info=info)
  864         ipmi._parse_driver_info(node)
  865         calls = [
  866             mock.call(u'ipmi_username is not defined or empty for node '
  867                       u'%s: NULL user will be utilized.', self.node.uuid),
  868             mock.call(u'ipmi_password is not defined or empty for node '
  869                       u'%s: NULL password will be utilized.', self.node.uuid),
  870         ]
  871         mock_log.assert_has_calls(calls)
  872 
  873     @mock.patch.object(ipmi.LOG, 'warning', spec_set=True, autospec=True)
  874     def test__parse_driver_info_have_credentials(
  875             self, mock_log):
  876         """Ensure no warnings generated if have credentials"""
  877         info = dict(INFO_DICT)
  878         node = obj_utils.get_test_node(self.context, driver_info=info)
  879         ipmi._parse_driver_info(node)
  880         self.assertFalse(mock_log.called)
  881 
  882     def test__parse_driver_info_terminal_port_specified(self):
  883         info = dict(INFO_DICT)
  884         info['ipmi_terminal_port'] = 10000
  885         node = obj_utils.get_test_node(self.context, driver_info=info)
  886         driver_info = ipmi._parse_driver_info(node)
  887         self.assertEqual(driver_info['port'], 10000)
  888 
  889     def test__parse_driver_info_terminal_port_allocated(self):
  890         info = dict(INFO_DICT)
  891         internal_info = {'allocated_ipmi_terminal_port': 10001}
  892         node = obj_utils.get_test_node(self.context, driver_info=info,
  893                                        driver_internal_info=internal_info)
  894         driver_info = ipmi._parse_driver_info(node)
  895         self.assertEqual(driver_info['port'], 10001)
  896 
  897     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  898     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
  899     @mock.patch.object(utils, 'execute', autospec=True)
  900     def test__exec_ipmitool_first_call_to_address(self, mock_exec,
  901                                                   mock_support):
  902         ipmi.LAST_CMD_TIME = {}
  903         args = [
  904             'ipmitool',
  905             '-I', 'lanplus',
  906             '-H', self.info['address'],
  907             '-L', self.info['priv_level'],
  908             '-U', self.info['username'],
  909             '-v',
  910             '-f', awesome_password_filename,
  911             'A', 'B', 'C',
  912         ]
  913 
  914         mock_support.return_value = False
  915         mock_exec.return_value = (None, None)
  916 
  917         ipmi._exec_ipmitool(self.info, 'A B C')
  918 
  919         mock_support.assert_called_once_with('timing')
  920         mock_exec.assert_called_once_with(*args)
  921         self.assertFalse(self.mock_sleep.called)
  922 
  923     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  924     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
  925     @mock.patch.object(utils, 'execute', autospec=True)
  926     def test__exec_ipmitool_second_call_to_address_sleep(
  927             self, mock_exec, mock_support):
  928         ipmi.LAST_CMD_TIME = {}
  929         args = [[
  930             'ipmitool',
  931             '-I', 'lanplus',
  932             '-H', self.info['address'],
  933             '-L', self.info['priv_level'],
  934             '-U', self.info['username'],
  935             '-v',
  936             '-f', awesome_password_filename,
  937             'A', 'B', 'C',
  938         ], [
  939             'ipmitool',
  940             '-I', 'lanplus',
  941             '-H', self.info['address'],
  942             '-L', self.info['priv_level'],
  943             '-U', self.info['username'],
  944             '-v',
  945             '-f', awesome_password_filename,
  946             'D', 'E', 'F',
  947         ]]
  948 
  949         expected = [mock.call('timing'),
  950                     mock.call('timing')]
  951         mock_support.return_value = False
  952         mock_exec.side_effect = [(None, None), (None, None)]
  953 
  954         ipmi._exec_ipmitool(self.info, 'A B C')
  955         mock_exec.assert_called_with(*args[0])
  956 
  957         ipmi._exec_ipmitool(self.info, 'D E F')
  958         self.assertTrue(self.mock_sleep.called)
  959         self.assertEqual(expected, mock_support.call_args_list)
  960         mock_exec.assert_called_with(*args[1])
  961 
  962     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
  963     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
  964     @mock.patch.object(utils, 'execute', autospec=True)
  965     def test__exec_ipmitool_second_call_to_address_no_sleep(
  966             self, mock_exec, mock_support):
  967         ipmi.LAST_CMD_TIME = {}
  968         args = [[
  969             'ipmitool',
  970             '-I', 'lanplus',
  971             '-H', self.info['address'],
  972             '-L', self.info['priv_level'],
  973             '-U', self.info['username'],
  974             '-v',
  975             '-f', awesome_password_filename,
  976             'A', 'B', 'C',
  977         ], [
  978             'ipmitool',
  979             '-I', 'lanplus',
  980             '-H', self.info['address'],
  981             '-L', self.info['priv_level'],
  982             '-U', self.info['username'],
  983             '-v',
  984             '-f', awesome_password_filename,
  985             'D', 'E', 'F',
  986         ]]
  987 
  988         expected = [mock.call('timing'),
  989                     mock.call('timing')]
  990         mock_support.return_value = False
  991         mock_exec.side_effect = [(None, None), (None, None)]
  992 
  993         ipmi._exec_ipmitool(self.info, 'A B C')
  994         mock_exec.assert_called_with(*args[0])
  995         # act like enough time has passed
  996         ipmi.LAST_CMD_TIME[self.info['address']] = (
  997             time.time() - CONF.ipmi.min_command_interval)
  998         ipmi._exec_ipmitool(self.info, 'D E F')
  999         self.assertFalse(self.mock_sleep.called)
 1000         self.assertEqual(expected, mock_support.call_args_list)
 1001         mock_exec.assert_called_with(*args[1])
 1002 
 1003     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1004     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
 1005     @mock.patch.object(utils, 'execute', autospec=True)
 1006     def test__exec_ipmitool_two_calls_to_diff_address(
 1007             self, mock_exec, mock_support):
 1008         ipmi.LAST_CMD_TIME = {}
 1009         args = [[
 1010             'ipmitool',
 1011             '-I', 'lanplus',
 1012             '-H', self.info['address'],
 1013             '-L', self.info['priv_level'],
 1014             '-U', self.info['username'],
 1015             '-v',
 1016             '-f', awesome_password_filename,
 1017             'A', 'B', 'C',
 1018         ], [
 1019             'ipmitool',
 1020             '-I', 'lanplus',
 1021             '-H', '127.127.127.127',
 1022             '-L', self.info['priv_level'],
 1023             '-U', self.info['username'],
 1024             '-v',
 1025             '-f', awesome_password_filename,
 1026             'D', 'E', 'F',
 1027         ]]
 1028 
 1029         expected = [mock.call('timing'),
 1030                     mock.call('timing')]
 1031         mock_support.return_value = False
 1032         mock_exec.side_effect = [(None, None), (None, None)]
 1033 
 1034         ipmi._exec_ipmitool(self.info, 'A B C')
 1035         mock_exec.assert_called_with(*args[0])
 1036         self.info['address'] = '127.127.127.127'
 1037         ipmi._exec_ipmitool(self.info, 'D E F')
 1038         self.assertFalse(self.mock_sleep.called)
 1039         self.assertEqual(expected, mock_support.call_args_list)
 1040         mock_exec.assert_called_with(*args[1])
 1041 
 1042     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1043     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
 1044     @mock.patch.object(utils, 'execute', autospec=True)
 1045     def test__exec_ipmitool_without_timing(
 1046             self, mock_exec, mock_support):
 1047         args = [
 1048             'ipmitool',
 1049             '-I', 'lanplus',
 1050             '-H', self.info['address'],
 1051             '-L', self.info['priv_level'],
 1052             '-U', self.info['username'],
 1053             '-v',
 1054             '-f', awesome_password_filename,
 1055             'A', 'B', 'C',
 1056         ]
 1057 
 1058         mock_support.return_value = False
 1059         mock_exec.return_value = (None, None)
 1060 
 1061         ipmi._exec_ipmitool(self.info, 'A B C')
 1062 
 1063         mock_support.assert_called_once_with('timing')
 1064         mock_exec.assert_called_once_with(*args)
 1065 
 1066     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1067     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
 1068     @mock.patch.object(utils, 'execute', autospec=True)
 1069     def test__exec_ipmitool_with_timing(
 1070             self, mock_exec, mock_support):
 1071         args = [
 1072             'ipmitool',
 1073             '-I', 'lanplus',
 1074             '-H', self.info['address'],
 1075             '-L', self.info['priv_level'],
 1076             '-U', self.info['username'],
 1077             '-v',
 1078             '-R', '12',
 1079             '-N', '5',
 1080             '-f', awesome_password_filename,
 1081             'A', 'B', 'C',
 1082         ]
 1083 
 1084         mock_support.return_value = True
 1085         mock_exec.return_value = (None, None)
 1086 
 1087         self.config(use_ipmitool_retries=True, group='ipmi')
 1088 
 1089         ipmi._exec_ipmitool(self.info, 'A B C')
 1090 
 1091         mock_support.assert_called_once_with('timing')
 1092         mock_exec.assert_called_once_with(*args)
 1093 
 1094     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1095     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
 1096     @mock.patch.object(utils, 'execute', autospec=True)
 1097     def test__exec_ipmitool_with_ironic_retries(
 1098             self, mock_exec, mock_support):
 1099         args = [
 1100             'ipmitool',
 1101             '-I', 'lanplus',
 1102             '-H', self.info['address'],
 1103             '-L', self.info['priv_level'],
 1104             '-U', self.info['username'],
 1105             '-v',
 1106             '-R', '1',
 1107             '-N', '5',
 1108             '-f', awesome_password_filename,
 1109             'A', 'B', 'C',
 1110         ]
 1111 
 1112         mock_support.return_value = True
 1113         mock_exec.return_value = (None, None)
 1114 
 1115         self.config(use_ipmitool_retries=False, group='ipmi')
 1116 
 1117         ipmi._exec_ipmitool(self.info, 'A B C')
 1118 
 1119         mock_support.assert_called_once_with('timing')
 1120         mock_exec.assert_called_once_with(*args)
 1121 
 1122     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1123     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
 1124     @mock.patch.object(utils, 'execute', autospec=True)
 1125     def test__exec_ipmitool_with_timeout(
 1126             self, mock_exec, mock_support):
 1127         args = [
 1128             'ipmitool',
 1129             '-I', 'lanplus',
 1130             '-H', self.info['address'],
 1131             '-L', self.info['priv_level'],
 1132             '-U', self.info['username'],
 1133             '-v',
 1134             '-R', '12',
 1135             '-N', '5',
 1136             '-f', awesome_password_filename,
 1137             'A', 'B', 'C',
 1138         ]
 1139 
 1140         mock_support.return_value = True
 1141         mock_exec.return_value = (None, None)
 1142 
 1143         self.config(use_ipmitool_retries=True, group='ipmi')
 1144         ipmi._exec_ipmitool(self.info, 'A B C', kill_on_timeout=True)
 1145 
 1146         mock_support.assert_called_once_with('timing')
 1147         mock_exec.assert_called_once_with(*args, timeout=60)
 1148 
 1149     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1150     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
 1151     @mock.patch.object(utils, 'execute', autospec=True)
 1152     def test__exec_ipmitool_with_ironic_retries_multiple(
 1153             self, mock_exec, mock_support):
 1154 
 1155         mock_exec.side_effect = [
 1156             processutils.ProcessExecutionError(
 1157                 stderr="Unknown"
 1158             ),
 1159             processutils.ProcessExecutionError(
 1160                 stderr="Unknown"
 1161             ),
 1162             processutils.ProcessExecutionError(
 1163                 stderr="Unknown"
 1164             ),
 1165         ]
 1166 
 1167         self.config(min_command_interval=1, group='ipmi')
 1168         self.config(command_retry_timeout=3, group='ipmi')
 1169         self.config(use_ipmitool_retries=False, group='ipmi')
 1170 
 1171         self.assertRaises(processutils.ProcessExecutionError,
 1172                           ipmi._exec_ipmitool,
 1173                           self.info, 'A B C')
 1174 
 1175         mock_support.assert_called_once_with('timing')
 1176         self.assertEqual(3, mock_exec.call_count)
 1177 
 1178     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1179     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
 1180     @mock.patch.object(utils, 'execute', autospec=True)
 1181     def test__exec_ipmitool_without_username(
 1182             self, mock_exec, mock_support):
 1183         # An undefined username is treated the same as an empty username and
 1184         # will cause no user (-U) to be specified.
 1185         self.info['username'] = None
 1186         args = [
 1187             'ipmitool',
 1188             '-I', 'lanplus',
 1189             '-H', self.info['address'],
 1190             '-L', self.info['priv_level'],
 1191             '-v',
 1192             '-f', awesome_password_filename,
 1193             'A', 'B', 'C',
 1194         ]
 1195 
 1196         mock_support.return_value = False
 1197         mock_exec.return_value = (None, None)
 1198         ipmi._exec_ipmitool(self.info, 'A B C')
 1199         mock_support.assert_called_once_with('timing')
 1200         mock_exec.assert_called_once_with(*args)
 1201 
 1202     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1203     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
 1204     @mock.patch.object(utils, 'execute', autospec=True)
 1205     def test__exec_ipmitool_with_empty_username(
 1206             self, mock_exec, mock_support):
 1207         # An empty username is treated the same as an undefined username and
 1208         # will cause no user (-U) to be specified.
 1209         self.info['username'] = ""
 1210         args = [
 1211             'ipmitool',
 1212             '-I', 'lanplus',
 1213             '-H', self.info['address'],
 1214             '-L', self.info['priv_level'],
 1215             '-v',
 1216             '-f', awesome_password_filename,
 1217             'A', 'B', 'C',
 1218         ]
 1219 
 1220         mock_support.return_value = False
 1221         mock_exec.return_value = (None, None)
 1222         ipmi._exec_ipmitool(self.info, 'A B C')
 1223         mock_support.assert_called_once_with('timing')
 1224         mock_exec.assert_called_once_with(*args)
 1225 
 1226     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1227     @mock.patch.object(
 1228         ipmi, '_make_password_file', wraps=_make_password_file_stub)
 1229     @mock.patch.object(utils, 'execute', autospec=True)
 1230     def test__exec_ipmitool_without_password(self, mock_exec,
 1231                                              _make_password_file_mock,
 1232                                              mock_support):
 1233         # An undefined password is treated the same as an empty password and
 1234         # will cause a NULL (\0) password to be used"""
 1235         self.info['password'] = None
 1236         args = [
 1237             'ipmitool',
 1238             '-I', 'lanplus',
 1239             '-H', self.info['address'],
 1240             '-L', self.info['priv_level'],
 1241             '-U', self.info['username'],
 1242             '-v',
 1243             '-f', awesome_password_filename,
 1244             'A', 'B', 'C',
 1245         ]
 1246 
 1247         mock_support.return_value = False
 1248         mock_exec.return_value = (None, None)
 1249         ipmi._exec_ipmitool(self.info, 'A B C')
 1250         mock_support.assert_called_once_with('timing')
 1251         mock_exec.assert_called_once_with(*args)
 1252         _make_password_file_mock.assert_called_once_with('\0')
 1253 
 1254     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1255     @mock.patch.object(
 1256         ipmi, '_make_password_file', wraps=_make_password_file_stub)
 1257     @mock.patch.object(utils, 'execute', autospec=True)
 1258     def test__exec_ipmitool_with_empty_password(self, mock_exec,
 1259                                                 _make_password_file_mock,
 1260                                                 mock_support):
 1261         # An empty password is treated the same as an undefined password and
 1262         # will cause a NULL (\0) password to be used"""
 1263         self.info['password'] = ""
 1264         args = [
 1265             'ipmitool',
 1266             '-I', 'lanplus',
 1267             '-H', self.info['address'],
 1268             '-L', self.info['priv_level'],
 1269             '-U', self.info['username'],
 1270             '-v',
 1271             '-f', awesome_password_filename,
 1272             'A', 'B', 'C',
 1273         ]
 1274 
 1275         mock_support.return_value = False
 1276         mock_exec.return_value = (None, None)
 1277         ipmi._exec_ipmitool(self.info, 'A B C')
 1278         mock_support.assert_called_once_with('timing')
 1279         mock_exec.assert_called_once_with(*args)
 1280         _make_password_file_mock.assert_called_once_with('\0')
 1281 
 1282     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1283     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
 1284     @mock.patch.object(utils, 'execute', autospec=True)
 1285     def test__exec_ipmitool_with_dual_bridging(self,
 1286                                                mock_exec,
 1287                                                mock_support):
 1288 
 1289         node = obj_utils.get_test_node(self.context,
 1290                                        driver_info=BRIDGE_INFO_DICT)
 1291         # when support for dual bridge command is called returns True
 1292         mock_support.return_value = True
 1293         info = ipmi._parse_driver_info(node)
 1294         args = [
 1295             'ipmitool',
 1296             '-I', 'lanplus',
 1297             '-H', info['address'],
 1298             '-L', info['priv_level'],
 1299             '-U', info['username'],
 1300             '-m', info['local_address'],
 1301             '-B', info['transit_channel'],
 1302             '-T', info['transit_address'],
 1303             '-b', info['target_channel'],
 1304             '-t', info['target_address'],
 1305             '-v',
 1306             '-f', awesome_password_filename,
 1307             'A', 'B', 'C',
 1308         ]
 1309 
 1310         expected = [mock.call('dual_bridge'),
 1311                     mock.call('timing')]
 1312         # When support for timing command is called returns False
 1313         mock_support.return_value = False
 1314         mock_exec.return_value = (None, None)
 1315         ipmi._exec_ipmitool(info, 'A B C')
 1316         self.assertEqual(expected, mock_support.call_args_list)
 1317         mock_exec.assert_called_once_with(*args)
 1318 
 1319     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
 1320     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1321     @mock.patch.object(utils, 'execute', autospec=True)
 1322     def test__exec_ipmitool_with_single_bridging(self,
 1323                                                  mock_exec,
 1324                                                  mock_support):
 1325         single_bridge_info = dict(BRIDGE_INFO_DICT)
 1326         single_bridge_info['ipmi_bridging'] = 'single'
 1327         node = obj_utils.get_test_node(self.context,
 1328                                        driver_info=single_bridge_info)
 1329         # when support for single bridge command is called returns True
 1330         mock_support.return_value = True
 1331         info = ipmi._parse_driver_info(node)
 1332         info['transit_channel'] = info['transit_address'] = None
 1333 
 1334         args = [
 1335             'ipmitool',
 1336             '-I', 'lanplus',
 1337             '-H', info['address'],
 1338             '-L', info['priv_level'],
 1339             '-U', info['username'],
 1340             '-m', info['local_address'],
 1341             '-b', info['target_channel'],
 1342             '-t', info['target_address'],
 1343             '-v',
 1344             '-f', awesome_password_filename,
 1345             'A', 'B', 'C',
 1346         ]
 1347 
 1348         expected = [mock.call('single_bridge'),
 1349                     mock.call('timing')]
 1350         # When support for timing command is called returns False
 1351         mock_support.return_value = False
 1352         mock_exec.return_value = (None, None)
 1353         ipmi._exec_ipmitool(info, 'A B C')
 1354         self.assertEqual(expected, mock_support.call_args_list)
 1355         mock_exec.assert_called_once_with(*args)
 1356 
 1357     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1358     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
 1359     @mock.patch.object(utils, 'execute', autospec=True)
 1360     def test__exec_ipmitool_exception(self, mock_exec, mock_support):
 1361         args = [
 1362             'ipmitool',
 1363             '-I', 'lanplus',
 1364             '-H', self.info['address'],
 1365             '-L', self.info['priv_level'],
 1366             '-U', self.info['username'],
 1367             '-v',
 1368             '-f', awesome_password_filename,
 1369             'A', 'B', 'C',
 1370         ]
 1371 
 1372         self.config(use_ipmitool_retries=True, group='ipmi')
 1373 
 1374         mock_support.return_value = False
 1375         mock_exec.side_effect = processutils.ProcessExecutionError("x")
 1376         self.assertRaises(processutils.ProcessExecutionError,
 1377                           ipmi._exec_ipmitool,
 1378                           self.info, 'A B C')
 1379         mock_support.assert_called_once_with('timing')
 1380         mock_exec.assert_called_once_with(*args)
 1381         self.assertEqual(1, mock_exec.call_count)
 1382 
 1383     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1384     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
 1385     @mock.patch.object(utils, 'execute', autospec=True)
 1386     def test__exec_ipmitool_IPMI_version_1_5(
 1387             self, mock_exec, mock_support):
 1388         self.info['protocol_version'] = '1.5'
 1389         # Assert it uses "-I lan" (1.5) instead of "-I lanplus" (2.0)
 1390         args = [
 1391             'ipmitool',
 1392             '-I', 'lan',
 1393             '-H', self.info['address'],
 1394             '-L', self.info['priv_level'],
 1395             '-U', self.info['username'],
 1396             '-v',
 1397             '-f', awesome_password_filename,
 1398             'A', 'B', 'C',
 1399         ]
 1400 
 1401         mock_support.return_value = False
 1402         mock_exec.return_value = (None, None)
 1403         ipmi._exec_ipmitool(self.info, 'A B C')
 1404         mock_support.assert_called_once_with('timing')
 1405         mock_exec.assert_called_once_with(*args)
 1406 
 1407     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1408     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
 1409     @mock.patch.object(utils, 'execute', autospec=True)
 1410     def test__exec_ipmitool_with_port(self, mock_exec, mock_support):
 1411         self.info['dest_port'] = '1623'
 1412         ipmi.LAST_CMD_TIME = {}
 1413         args = [
 1414             'ipmitool',
 1415             '-I', 'lanplus',
 1416             '-H', self.info['address'],
 1417             '-L', self.info['priv_level'],
 1418             '-p', '1623',
 1419             '-U', self.info['username'],
 1420             '-v',
 1421             '-f', awesome_password_filename,
 1422             'A', 'B', 'C',
 1423         ]
 1424 
 1425         mock_support.return_value = False
 1426         mock_exec.return_value = (None, None)
 1427 
 1428         ipmi._exec_ipmitool(self.info, 'A B C')
 1429 
 1430         mock_support.assert_called_once_with('timing')
 1431         mock_exec.assert_called_once_with(*args)
 1432         self.assertFalse(self.mock_sleep.called)
 1433 
 1434     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1435     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
 1436     @mock.patch.object(utils, 'execute', autospec=True)
 1437     def test__exec_ipmitool_cipher_suite(self, mock_exec, mock_support):
 1438         self.info['cipher_suite'] = '3'
 1439         ipmi.LAST_CMD_TIME = {}
 1440         args = [
 1441             'ipmitool',
 1442             '-I', 'lanplus',
 1443             '-H', self.info['address'],
 1444             '-L', self.info['priv_level'],
 1445             '-U', self.info['username'],
 1446             '-C', '3',
 1447             '-v',
 1448             '-f', awesome_password_filename,
 1449             'A', 'B', 'C',
 1450         ]
 1451 
 1452         mock_support.return_value = False
 1453         mock_exec.return_value = (None, None)
 1454 
 1455         ipmi._exec_ipmitool(self.info, 'A B C')
 1456 
 1457         mock_support.assert_called_once_with('timing')
 1458         mock_exec.assert_called_once_with(*args)
 1459         self.assertFalse(self.mock_sleep.called)
 1460 
 1461     @mock.patch.object(ipmi, '_is_option_supported', autospec=True)
 1462     @mock.patch.object(ipmi, '_make_password_file', _make_password_file_stub)
 1463     @mock.patch.object(utils, 'execute', autospec=True)
 1464     def test__exec_ipmitool_with_check_exit_code(self, mock_exec,
 1465                                                  mock_support):
 1466         args = [
 1467             'ipmitool',
 1468             '-I', 'lanplus',
 1469             '-H', self.info['address'],
 1470             '-L', self.info['priv_level'],
 1471             '-U', self.info['username'],
 1472             '-v',
 1473             '-f', awesome_password_filename,
 1474             'A', 'B', 'C',
 1475         ]
 1476         mock_support.return_value = False
 1477         mock_exec.return_value = (None, None)
 1478         ipmi._exec_ipmitool(self.info, 'A B C', check_exit_code=[0, 1])
 1479         mock_support.assert_called_once_with('timing')
 1480         mock_exec.assert_called_once_with(*args, check_exit_code=[0, 1])
 1481 
 1482     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 1483     def test__power_status_on(self, mock_exec):
 1484         mock_exec.return_value = ["Chassis Power is on\n", None]
 1485 
 1486         state = ipmi._power_status(self.info)
 1487 
 1488         mock_exec.assert_called_once_with(self.info, "power status",
 1489                                           kill_on_timeout=True)
 1490         self.assertEqual(states.POWER_ON, state)
 1491 
 1492     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 1493     def test__power_status_off(self, mock_exec):
 1494         mock_exec.return_value = ["Chassis Power is off\n", None]
 1495 
 1496         state = ipmi._power_status(self.info)
 1497 
 1498         mock_exec.assert_called_once_with(self.info, "power status",
 1499                                           kill_on_timeout=True)
 1500         self.assertEqual(states.POWER_OFF, state)
 1501 
 1502     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 1503     def test__power_status_error(self, mock_exec):
 1504         mock_exec.return_value = ["Chassis Power is badstate\n", None]
 1505 
 1506         state = ipmi._power_status(self.info)
 1507 
 1508         mock_exec.assert_called_once_with(self.info, "power status",
 1509                                           kill_on_timeout=True)
 1510         self.assertEqual(states.ERROR, state)
 1511 
 1512     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 1513     def test__power_status_exception(self, mock_exec):
 1514         mock_exec.side_effect = processutils.ProcessExecutionError("error")
 1515         self.assertRaises(exception.IPMIFailure,
 1516                           ipmi._power_status,
 1517                           self.info)
 1518         mock_exec.assert_called_once_with(self.info, "power status",
 1519                                           kill_on_timeout=True)
 1520 
 1521     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 1522     @mock.patch('oslo_utils.eventletutils.EventletEvent.wait', autospec=True)
 1523     def test__power_on_max_retries(self, sleep_mock, mock_exec):
 1524         self.config(command_retry_timeout=2, group='ipmi')
 1525 
 1526         def side_effect(driver_info, command, **kwargs):
 1527             resp_dict = {"power status": ["Chassis Power is off\n", None],
 1528                          "power on": [None, None]}
 1529             return resp_dict.get(command, ["Bad\n", None])
 1530 
 1531         mock_exec.side_effect = side_effect
 1532 
 1533         expected = [mock.call(self.info, "power on"),
 1534                     mock.call(self.info, "power status", kill_on_timeout=True),
 1535                     mock.call(self.info, "power status", kill_on_timeout=True)]
 1536 
 1537         with task_manager.acquire(self.context, self.node.uuid) as task:
 1538             task.node.properties['vendor'] = 'lolcat'
 1539             self.assertRaises(exception.PowerStateFailure,
 1540                               ipmi._power_on, task, self.info, timeout=2)
 1541 
 1542         self.assertEqual(expected, mock_exec.call_args_list)
 1543 
 1544     @mock.patch.object(ipmi.IPMIManagement, 'detect_vendor', autospec=True)
 1545     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 1546     @mock.patch('oslo_utils.eventletutils.EventletEvent.wait', autospec=True)
 1547     def test__soft_power_off(self, sleep_mock, mock_exec, mock_vendor):
 1548         def side_effect(driver_info, command, **kwargs):
 1549             resp_dict = {"power status": ["Chassis Power is off\n", None],
 1550                          "power soft": [None, None]}
 1551             return resp_dict.get(command, ["Bad\n", None])
 1552 
 1553         mock_vendor.return_value = None
 1554         mock_exec.side_effect = side_effect
 1555 
 1556         expected = [mock.call(self.info, "power soft"),
 1557                     mock.call(self.info, "power status", kill_on_timeout=True)]
 1558 
 1559         with task_manager.acquire(self.context, self.node.uuid) as task:
 1560             state = ipmi._soft_power_off(task, self.info)
 1561 
 1562         self.assertEqual(expected, mock_exec.call_args_list)
 1563         self.assertEqual(states.POWER_OFF, state)
 1564         self.assertTrue(mock_vendor.called)
 1565 
 1566     @mock.patch.object(ipmi.IPMIManagement, 'detect_vendor', autospec=True)
 1567     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 1568     @mock.patch('oslo_utils.eventletutils.EventletEvent.wait', autospec=True)
 1569     def test__soft_power_off_max_retries(self, sleep_mock, mock_exec,
 1570                                          mock_vendor):
 1571 
 1572         def side_effect(driver_info, command, **kwargs):
 1573             resp_dict = {"power status": ["Chassis Power is on\n", None],
 1574                          "power soft": [None, None]}
 1575             return resp_dict.get(command, ["Bad\n", None])
 1576 
 1577         mock_exec.side_effect = side_effect
 1578 
 1579         expected = [mock.call(self.info, "power soft"),
 1580                     mock.call(self.info, "power status", kill_on_timeout=True),
 1581                     mock.call(self.info, "power status", kill_on_timeout=True)]
 1582 
 1583         with task_manager.acquire(self.context, self.node.uuid) as task:
 1584             task.node.properties['vendor'] = 'meow5000'
 1585             self.assertRaises(exception.PowerStateFailure,
 1586                               ipmi._soft_power_off, task, self.info, timeout=2)
 1587 
 1588         self.assertEqual(expected, mock_exec.call_args_list)
 1589         # Should be removed when detect_vendor automatic invocation is moved.
 1590         self.assertFalse(mock_vendor.called)
 1591 
 1592     @mock.patch.object(ipmi, '_power_status', autospec=True)
 1593     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 1594     @mock.patch('oslo_utils.eventletutils.EventletEvent.wait', autospec=True)
 1595     def test___set_and_wait_no_needless_status_polling(
 1596             self, sleep_mock, mock_exec, mock_status):
 1597         # Check that if the call to power state change fails, it doesn't
 1598         # call power_status().
 1599         self.config(command_retry_timeout=2, group='ipmi')
 1600 
 1601         mock_exec.side_effect = exception.IPMIFailure(cmd='power on')
 1602 
 1603         with task_manager.acquire(self.context, self.node.uuid) as task:
 1604             self.assertRaises(exception.IPMIFailure, ipmi._power_on, task,
 1605                               self.info)
 1606         self.assertFalse(mock_status.called)
 1607 
 1608 
 1609 class IPMIToolDriverTestCase(Base):
 1610 
 1611     @mock.patch.object(ipmi, "_parse_driver_info", autospec=True)
 1612     def test_power_validate(self, mock_parse):
 1613         mock_parse.return_value = {}
 1614 
 1615         with task_manager.acquire(self.context, self.node.uuid) as task:
 1616             task.driver.power.validate(task)
 1617             mock_parse.assert_called_once_with(mock.ANY)
 1618 
 1619     def test_get_properties(self):
 1620         expected = ipmi.COMMON_PROPERTIES
 1621         self.assertEqual(expected, self.power.get_properties())
 1622 
 1623         expected = list(ipmi.COMMON_PROPERTIES) + list(ipmi.CONSOLE_PROPERTIES)
 1624         self.assertEqual(sorted(expected),
 1625                          sorted(self.console.get_properties()))
 1626         with task_manager.acquire(self.context, self.node.uuid) as task:
 1627             self.assertEqual(sorted(expected),
 1628                              sorted(task.driver.get_properties()))
 1629 
 1630     @mock.patch.object(ipmi.IPMIManagement, 'detect_vendor', autospec=True)
 1631     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 1632     def test_get_power_state(self, mock_exec, mock_detect):
 1633         returns = iter([["Chassis Power is off\n", None],
 1634                         ["Chassis Power is on\n", None],
 1635                         ["\n", None]])
 1636         expected = [mock.call(self.info, "power status", kill_on_timeout=True),
 1637                     mock.call(self.info, "power status", kill_on_timeout=True),
 1638                     mock.call(self.info, "power status", kill_on_timeout=True)]
 1639         mock_exec.side_effect = returns
 1640         mock_detect.return_value = None
 1641 
 1642         with task_manager.acquire(self.context, self.node.uuid) as task:
 1643             pstate = self.power.get_power_state(task)
 1644             self.assertEqual(states.POWER_OFF, pstate)
 1645 
 1646             pstate = self.power.get_power_state(task)
 1647             self.assertEqual(states.POWER_ON, pstate)
 1648 
 1649             pstate = self.power.get_power_state(task)
 1650             self.assertEqual(states.ERROR, pstate)
 1651 
 1652         self.assertEqual(mock_exec.call_args_list, expected)
 1653         self.assertEqual(3, mock_detect.call_count)
 1654 
 1655     @mock.patch.object(ipmi.IPMIManagement, 'detect_vendor', autospec=True)
 1656     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 1657     def test_get_power_state_exception(self, mock_exec, mock_vendor):
 1658         mock_exec.side_effect = processutils.ProcessExecutionError("error")
 1659         mock_vendor.return_value = None
 1660         with task_manager.acquire(self.context, self.node.uuid) as task:
 1661             self.assertRaises(exception.IPMIFailure,
 1662                               self.power.get_power_state,
 1663                               task)
 1664         mock_exec.assert_called_once_with(self.info, "power status",
 1665                                           kill_on_timeout=True)
 1666         self.assertEqual(1, mock_vendor.call_count)
 1667 
 1668     @mock.patch.object(ipmi, '_power_on', autospec=True)
 1669     @mock.patch.object(ipmi, '_power_off', autospec=True)
 1670     def test_set_power_on_ok(self, mock_off, mock_on):
 1671         self.config(command_retry_timeout=0, group='ipmi')
 1672 
 1673         mock_on.return_value = states.POWER_ON
 1674         with task_manager.acquire(self.context,
 1675                                   self.node.uuid) as task:
 1676             self.power.set_power_state(task, states.POWER_ON)
 1677 
 1678             mock_on.assert_called_once_with(task, self.info, timeout=None)
 1679         self.assertFalse(mock_off.called)
 1680 
 1681     @mock.patch.object(ipmi, '_power_on', autospec=True)
 1682     @mock.patch.object(ipmi, '_power_off', autospec=True)
 1683     def test_set_power_on_timeout_ok(self, mock_off, mock_on):
 1684         self.config(command_retry_timeout=0, group='ipmi')
 1685 
 1686         mock_on.return_value = states.POWER_ON
 1687         with task_manager.acquire(self.context,
 1688                                   self.node.uuid) as task:
 1689             self.power.set_power_state(task, states.POWER_ON, timeout=2)
 1690 
 1691             mock_on.assert_called_once_with(task, self.info, timeout=2)
 1692         self.assertFalse(mock_off.called)
 1693 
 1694     @mock.patch.object(driver_utils, 'ensure_next_boot_device', autospec=True)
 1695     @mock.patch.object(ipmi, '_power_on', autospec=True)
 1696     @mock.patch.object(ipmi, '_power_off', autospec=True)
 1697     def test_set_power_on_with_next_boot(self, mock_off, mock_on,
 1698                                          mock_next_boot):
 1699         self.config(command_retry_timeout=0, group='ipmi')
 1700 
 1701         mock_on.return_value = states.POWER_ON
 1702         with task_manager.acquire(self.context,
 1703                                   self.node.uuid) as task:
 1704             self.power.set_power_state(task, states.POWER_ON)
 1705             mock_next_boot.assert_called_once_with(task, self.info)
 1706 
 1707             mock_on.assert_called_once_with(task, self.info, timeout=None)
 1708         self.assertFalse(mock_off.called)
 1709 
 1710     @mock.patch.object(driver_utils, 'ensure_next_boot_device', autospec=True)
 1711     @mock.patch.object(ipmi, '_power_on', autospec=True)
 1712     @mock.patch.object(ipmi, '_power_off', autospec=True)
 1713     def test_set_power_on_with_next_boot_timeout(self, mock_off, mock_on,
 1714                                                  mock_next_boot):
 1715         self.config(command_retry_timeout=0, group='ipmi')
 1716 
 1717         mock_on.return_value = states.POWER_ON
 1718         with task_manager.acquire(self.context,
 1719                                   self.node.uuid) as task:
 1720             self.power.set_power_state(task, states.POWER_ON, timeout=2)
 1721             mock_next_boot.assert_called_once_with(task, self.info)
 1722 
 1723             mock_on.assert_called_once_with(task, self.info, timeout=2)
 1724         self.assertFalse(mock_off.called)
 1725 
 1726     @mock.patch.object(ipmi, '_power_on', autospec=True)
 1727     @mock.patch.object(ipmi, '_power_off', autospec=True)
 1728     def test_set_power_off_ok(self, mock_off, mock_on):
 1729         self.config(command_retry_timeout=0, group='ipmi')
 1730 
 1731         mock_off.return_value = states.POWER_OFF
 1732 
 1733         with task_manager.acquire(self.context,
 1734                                   self.node.uuid) as task:
 1735             self.power.set_power_state(task, states.POWER_OFF)
 1736 
 1737             mock_off.assert_called_once_with(task, self.info, timeout=None)
 1738         self.assertFalse(mock_on.called)
 1739 
 1740     @mock.patch.object(ipmi, '_power_on', autospec=True)
 1741     @mock.patch.object(ipmi, '_power_off', autospec=True)
 1742     def test_set_power_off_timeout_ok(self, mock_off, mock_on):
 1743         self.config(command_retry_timeout=0, group='ipmi')
 1744 
 1745         mock_off.return_value = states.POWER_OFF
 1746 
 1747         with task_manager.acquire(self.context,
 1748                                   self.node.uuid) as task:
 1749             self.power.set_power_state(task, states.POWER_OFF, timeout=2)
 1750 
 1751             mock_off.assert_called_once_with(task, self.info, timeout=2)
 1752         self.assertFalse(mock_on.called)
 1753 
 1754     @mock.patch.object(ipmi, '_power_on', autospec=True)
 1755     @mock.patch.object(ipmi, '_soft_power_off', autospec=True)
 1756     def test_set_soft_power_off_ok(self, mock_off, mock_on):
 1757         self.config(command_retry_timeout=0, group='ipmi')
 1758 
 1759         mock_off.return_value = states.POWER_OFF
 1760 
 1761         with task_manager.acquire(self.context,
 1762                                   self.node['uuid']) as task:
 1763             self.power.set_power_state(task, states.SOFT_POWER_OFF)
 1764 
 1765             mock_off.assert_called_once_with(task, self.info, timeout=None)
 1766         self.assertFalse(mock_on.called)
 1767 
 1768     @mock.patch.object(ipmi, '_power_on', autospec=True)
 1769     @mock.patch.object(ipmi, '_soft_power_off', autospec=True)
 1770     def test_set_soft_power_off_timeout_ok(self, mock_off, mock_on):
 1771         self.config(command_retry_timeout=0, group='ipmi')
 1772 
 1773         mock_off.return_value = states.POWER_OFF
 1774 
 1775         with task_manager.acquire(self.context,
 1776                                   self.node['uuid']) as task:
 1777             self.power.set_power_state(task, states.SOFT_POWER_OFF, timeout=2)
 1778 
 1779             mock_off.assert_called_once_with(task, self.info, timeout=2)
 1780         self.assertFalse(mock_on.called)
 1781 
 1782     @mock.patch.object(driver_utils, 'ensure_next_boot_device', autospec=True)
 1783     @mock.patch.object(ipmi, '_power_on', autospec=True)
 1784     @mock.patch.object(ipmi, '_soft_power_off', autospec=True)
 1785     def test_set_soft_reboot_ok(self, mock_off, mock_on, mock_next_boot):
 1786         self.config(command_retry_timeout=0, group='ipmi')
 1787 
 1788         mock_off.return_value = states.POWER_OFF
 1789         mock_on.return_value = states.POWER_ON
 1790 
 1791         with task_manager.acquire(self.context,
 1792                                   self.node['uuid']) as task:
 1793             self.power.set_power_state(task, states.SOFT_REBOOT)
 1794             mock_next_boot.assert_called_once_with(task, self.info)
 1795             mock_off.assert_called_once_with(task, self.info, timeout=None)
 1796             mock_on.assert_called_once_with(task, self.info, timeout=None)
 1797 
 1798     @mock.patch.object(driver_utils, 'ensure_next_boot_device', autospec=True)
 1799     @mock.patch.object(ipmi, '_power_on', autospec=True)
 1800     @mock.patch.object(ipmi, '_soft_power_off', autospec=True)
 1801     def test_set_soft_reboot_timeout_ok(self, mock_off, mock_on,
 1802                                         mock_next_boot):
 1803         self.config(command_retry_timeout=0, group='ipmi')
 1804 
 1805         mock_off.return_value = states.POWER_OFF
 1806         mock_on.return_value = states.POWER_ON
 1807 
 1808         with task_manager.acquire(self.context,
 1809                                   self.node['uuid']) as task:
 1810             self.power.set_power_state(task, states.SOFT_REBOOT, timeout=2)
 1811             mock_next_boot.assert_called_once_with(task, self.info)
 1812             mock_off.assert_called_once_with(task, self.info, timeout=2)
 1813             mock_on.assert_called_once_with(task, self.info, timeout=2)
 1814 
 1815     @mock.patch.object(driver_utils, 'ensure_next_boot_device', autospec=True)
 1816     @mock.patch.object(ipmi, '_power_on', autospec=True)
 1817     @mock.patch.object(ipmi, '_soft_power_off', autospec=True)
 1818     def test_set_soft_reboot_timeout_fail(self, mock_off, mock_on,
 1819                                           mock_next_boot):
 1820         self.config(command_retry_timeout=0, group='ipmi')
 1821 
 1822         mock_off.side_effect = exception.PowerStateFailure(
 1823             pstate=states.POWER_ON)
 1824 
 1825         with task_manager.acquire(self.context,
 1826                                   self.node['uuid']) as task:
 1827             self.assertRaises(exception.PowerStateFailure,
 1828                               self.power.set_power_state,
 1829                               task,
 1830                               states.SOFT_REBOOT,
 1831                               timeout=2)
 1832 
 1833             mock_off.assert_called_once_with(task, self.info, timeout=2)
 1834         self.assertFalse(mock_next_boot.called)
 1835         self.assertFalse(mock_on.called)
 1836 
 1837     @mock.patch.object(ipmi, '_power_on', autospec=True)
 1838     @mock.patch.object(ipmi, '_power_off', autospec=True)
 1839     def test_set_power_on_fail(self, mock_off, mock_on):
 1840         self.config(command_retry_timeout=0, group='ipmi')
 1841 
 1842         mock_on.side_effect = exception.PowerStateFailure(
 1843             pstate=states.POWER_ON)
 1844         with task_manager.acquire(self.context,
 1845                                   self.node.uuid) as task:
 1846             self.assertRaises(exception.PowerStateFailure,
 1847                               self.power.set_power_state,
 1848                               task,
 1849                               states.POWER_ON)
 1850 
 1851             mock_on.assert_called_once_with(task, self.info, timeout=None)
 1852         self.assertFalse(mock_off.called)
 1853 
 1854     @mock.patch.object(ipmi, '_power_on', autospec=True)
 1855     @mock.patch.object(ipmi, '_power_off', autospec=True)
 1856     def test_set_power_on_timeout_fail(self, mock_off, mock_on):
 1857         self.config(command_retry_timeout=0, group='ipmi')
 1858 
 1859         mock_on.side_effect = exception.PowerStateFailure(pstate=states.ERROR)
 1860         with task_manager.acquire(self.context,
 1861                                   self.node.uuid) as task:
 1862             self.assertRaises(exception.PowerStateFailure,
 1863                               self.power.set_power_state,
 1864                               task,
 1865                               states.POWER_ON,
 1866                               timeout=2)
 1867 
 1868             mock_on.assert_called_once_with(task, self.info, timeout=2)
 1869         self.assertFalse(mock_off.called)
 1870 
 1871     def test_set_power_invalid_state(self):
 1872         with task_manager.acquire(self.context, self.node.uuid) as task:
 1873             self.assertRaises(exception.InvalidParameterValue,
 1874                               self.power.set_power_state,
 1875                               task,
 1876                               "fake state")
 1877 
 1878     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 1879     def test_send_raw_bytes_ok(self, mock_exec):
 1880         mock_exec.return_value = [None, None]
 1881 
 1882         with task_manager.acquire(self.context,
 1883                                   self.node.uuid) as task:
 1884             self.vendor.send_raw(task, http_method='POST',
 1885                                  raw_bytes='0x00 0x01')
 1886 
 1887         mock_exec.assert_called_once_with(self.info, 'raw 0x00 0x01')
 1888 
 1889     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 1890     def test_send_raw_bytes_fail(self, mock_exec):
 1891         mock_exec.side_effect = exception.PasswordFileFailedToCreate('error')
 1892 
 1893         with task_manager.acquire(self.context,
 1894                                   self.node.uuid) as task:
 1895             self.assertRaises(exception.IPMIFailure,
 1896                               self.vendor.send_raw,
 1897                               task,
 1898                               http_method='POST',
 1899                               raw_bytes='0x00 0x01')
 1900 
 1901     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 1902     def test__bmc_reset_ok(self, mock_exec):
 1903         mock_exec.return_value = [None, None]
 1904 
 1905         with task_manager.acquire(self.context,
 1906                                   self.node.uuid) as task:
 1907             self.vendor.bmc_reset(task, 'POST')
 1908 
 1909         mock_exec.assert_called_once_with(self.info, 'bmc reset warm')
 1910 
 1911     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 1912     def test__bmc_reset_cold(self, mock_exec):
 1913         mock_exec.return_value = [None, None]
 1914 
 1915         with task_manager.acquire(self.context,
 1916                                   self.node.uuid) as task:
 1917             self.vendor.bmc_reset(task, 'POST', warm=False)
 1918 
 1919         mock_exec.assert_called_once_with(self.info, 'bmc reset cold')
 1920 
 1921     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 1922     def test__bmc_reset_fail(self, mock_exec):
 1923         mock_exec.side_effect = processutils.ProcessExecutionError()
 1924 
 1925         with task_manager.acquire(self.context,
 1926                                   self.node.uuid) as task:
 1927             self.assertRaises(exception.IPMIFailure,
 1928                               self.vendor.bmc_reset,
 1929                               task, 'POST')
 1930 
 1931     @mock.patch.object(driver_utils, 'ensure_next_boot_device', autospec=True)
 1932     @mock.patch.object(ipmi, '_power_off', spec_set=types.FunctionType)
 1933     @mock.patch.object(ipmi, '_power_on', spec_set=types.FunctionType)
 1934     @mock.patch.object(ipmi, '_power_status',
 1935                        lambda driver_info: states.POWER_ON)
 1936     def test_reboot_ok(self, mock_on, mock_off, mock_next_boot):
 1937         manager = mock.MagicMock()
 1938         # NOTE(rloo): if autospec is True, then manager.mock_calls is empty
 1939         mock_off.return_value = states.POWER_OFF
 1940         mock_on.return_value = states.POWER_ON
 1941         manager.attach_mock(mock_off, 'power_off')
 1942         manager.attach_mock(mock_on, 'power_on')
 1943 
 1944         with task_manager.acquire(self.context,
 1945                                   self.node.uuid) as task:
 1946             expected = [mock.call.power_off(task, self.info, timeout=None),
 1947                         mock.call.power_on(task, self.info, timeout=None)]
 1948             self.power.reboot(task)
 1949             mock_next_boot.assert_called_once_with(task, self.info)
 1950 
 1951         self.assertEqual(expected, manager.mock_calls)
 1952 
 1953     @mock.patch.object(driver_utils, 'ensure_next_boot_device', autospec=True)
 1954     @mock.patch.object(ipmi, '_power_off', spec_set=types.FunctionType)
 1955     @mock.patch.object(ipmi, '_power_on', spec_set=types.FunctionType)
 1956     @mock.patch.object(ipmi, '_power_status',
 1957                        lambda driver_info: states.POWER_OFF)
 1958     def test_reboot_already_off(self, mock_on, mock_off, mock_next_boot):
 1959         manager = mock.MagicMock()
 1960         # NOTE(rloo): if autospec is True, then manager.mock_calls is empty
 1961         mock_off.return_value = states.POWER_OFF
 1962         mock_on.return_value = states.POWER_ON
 1963         manager.attach_mock(mock_off, 'power_off')
 1964         manager.attach_mock(mock_on, 'power_on')
 1965 
 1966         with task_manager.acquire(self.context,
 1967                                   self.node.uuid) as task:
 1968             expected = [mock.call.power_on(task, self.info, timeout=None)]
 1969             self.power.reboot(task)
 1970             mock_next_boot.assert_called_once_with(task, self.info)
 1971 
 1972         self.assertEqual(expected, manager.mock_calls)
 1973 
 1974     @mock.patch.object(driver_utils, 'ensure_next_boot_device', autospec=True)
 1975     @mock.patch.object(ipmi, '_power_off', spec_set=types.FunctionType)
 1976     @mock.patch.object(ipmi, '_power_on', spec_set=types.FunctionType)
 1977     @mock.patch.object(ipmi, '_power_status',
 1978                        lambda driver_info: states.POWER_ON)
 1979     def test_reboot_timeout_ok(self, mock_on, mock_off, mock_next_boot):
 1980         manager = mock.MagicMock()
 1981         # NOTE(rloo): if autospec is True, then manager.mock_calls is empty
 1982         manager.attach_mock(mock_off, 'power_off')
 1983         manager.attach_mock(mock_on, 'power_on')
 1984 
 1985         with task_manager.acquire(self.context,
 1986                                   self.node.uuid) as task:
 1987             expected = [mock.call.power_off(task, self.info, timeout=2),
 1988                         mock.call.power_on(task, self.info, timeout=2)]
 1989 
 1990             self.power.reboot(task, timeout=2)
 1991             mock_next_boot.assert_called_once_with(task, self.info)
 1992 
 1993             self.assertEqual(expected, manager.mock_calls)
 1994 
 1995     @mock.patch.object(ipmi, '_power_off', spec_set=types.FunctionType)
 1996     @mock.patch.object(ipmi, '_power_on', spec_set=types.FunctionType)
 1997     @mock.patch.object(ipmi, '_power_status',
 1998                        lambda driver_info: states.POWER_ON)
 1999     def test_reboot_fail_power_off(self, mock_on, mock_off):
 2000         manager = mock.MagicMock()
 2001         # NOTE(rloo): if autospec is True, then manager.mock_calls is empty
 2002         mock_off.side_effect = exception.PowerStateFailure(
 2003             pstate=states.POWER_OFF)
 2004         manager.attach_mock(mock_off, 'power_off')
 2005         manager.attach_mock(mock_on, 'power_on')
 2006 
 2007         with task_manager.acquire(self.context,
 2008                                   self.node.uuid) as task:
 2009             expected = [mock.call.power_off(task, self.info, timeout=None)]
 2010             self.assertRaises(exception.PowerStateFailure,
 2011                               self.power.reboot,
 2012                               task)
 2013 
 2014         self.assertEqual(expected, manager.mock_calls)
 2015 
 2016     @mock.patch.object(ipmi, '_power_off', spec_set=types.FunctionType)
 2017     @mock.patch.object(ipmi, '_power_on', spec_set=types.FunctionType)
 2018     @mock.patch.object(ipmi, '_power_status',
 2019                        lambda driver_info: states.POWER_ON)
 2020     def test_reboot_fail_power_on(self, mock_on, mock_off):
 2021         manager = mock.MagicMock()
 2022         # NOTE(rloo): if autospec is True, then manager.mock_calls is empty
 2023         mock_off.return_value = states.POWER_OFF
 2024         mock_on.side_effect = exception.PowerStateFailure(
 2025             pstate=states.POWER_ON)
 2026         manager.attach_mock(mock_off, 'power_off')
 2027         manager.attach_mock(mock_on, 'power_on')
 2028 
 2029         with task_manager.acquire(self.context,
 2030                                   self.node.uuid) as task:
 2031             expected = [mock.call.power_off(task, self.info, timeout=None),
 2032                         mock.call.power_on(task, self.info, timeout=None)]
 2033             self.assertRaises(exception.PowerStateFailure,
 2034                               self.power.reboot,
 2035                               task)
 2036 
 2037         self.assertEqual(expected, manager.mock_calls)
 2038 
 2039     @mock.patch.object(ipmi, '_power_off', spec_set=types.FunctionType)
 2040     @mock.patch.object(ipmi, '_power_on', spec_set=types.FunctionType)
 2041     @mock.patch.object(ipmi, '_power_status',
 2042                        lambda driver_info: states.POWER_ON)
 2043     def test_reboot_timeout_fail(self, mock_on, mock_off):
 2044         manager = mock.MagicMock()
 2045         # NOTE(rloo): if autospec is True, then manager.mock_calls is empty
 2046         mock_on.side_effect = exception.PowerStateFailure(
 2047             pstate=states.POWER_ON)
 2048         manager.attach_mock(mock_off, 'power_off')
 2049         manager.attach_mock(mock_on, 'power_on')
 2050 
 2051         with task_manager.acquire(self.context,
 2052                                   self.node.uuid) as task:
 2053             expected = [mock.call.power_off(task, self.info, timeout=2),
 2054                         mock.call.power_on(task, self.info, timeout=2)]
 2055             self.assertRaises(exception.PowerStateFailure,
 2056                               self.power.reboot,
 2057                               task, timeout=2)
 2058 
 2059         self.assertEqual(expected, manager.mock_calls)
 2060 
 2061     @mock.patch.object(ipmi, '_parse_driver_info', autospec=True)
 2062     def test_vendor_passthru_validate__parse_driver_info_fail(self, info_mock):
 2063         info_mock.side_effect = exception.InvalidParameterValue("bad")
 2064         with task_manager.acquire(self.context, self.node.uuid) as task:
 2065             self.assertRaises(exception.InvalidParameterValue,
 2066                               self.vendor.validate,
 2067                               task, method='send_raw', raw_bytes='0x00 0x01')
 2068             info_mock.assert_called_once_with(task.node)
 2069 
 2070     def test_vendor_passthru_validate__send_raw_bytes_good(self):
 2071         with task_manager.acquire(self.context, self.node.uuid) as task:
 2072             self.vendor.validate(task,
 2073                                  method='send_raw',
 2074                                  http_method='POST',
 2075                                  raw_bytes='0x00 0x01')
 2076 
 2077     def test_vendor_passthru_validate__send_raw_bytes_fail(self):
 2078         with task_manager.acquire(self.context, self.node.uuid) as task:
 2079             self.assertRaises(exception.MissingParameterValue,
 2080                               self.vendor.validate,
 2081                               task, method='send_raw')
 2082 
 2083     @mock.patch.object(ipmi.VendorPassthru, 'send_raw', autospec=True)
 2084     def test_vendor_passthru_call_send_raw_bytes(self, raw_bytes_mock):
 2085         with task_manager.acquire(self.context, self.node.uuid,
 2086                                   shared=False) as task:
 2087             self.vendor.send_raw(task, http_method='POST',
 2088                                  raw_bytes='0x00 0x01')
 2089             raw_bytes_mock.assert_called_once_with(
 2090                 self.vendor, task, http_method='POST',
 2091                 raw_bytes='0x00 0x01')
 2092 
 2093     def test_vendor_passthru_validate__bmc_reset_good(self):
 2094         with task_manager.acquire(self.context, self.node.uuid) as task:
 2095             self.vendor.validate(task, method='bmc_reset')
 2096 
 2097     def test_vendor_passthru_validate__bmc_reset_warm_good(self):
 2098         with task_manager.acquire(self.context, self.node.uuid) as task:
 2099             self.vendor.validate(task, method='bmc_reset', warm=True)
 2100 
 2101     def test_vendor_passthru_validate__bmc_reset_cold_good(self):
 2102         with task_manager.acquire(self.context, self.node.uuid) as task:
 2103             self.vendor.validate(task, method='bmc_reset', warm=False)
 2104 
 2105     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2106     def _vendor_passthru_call_bmc_reset(self, warm, expected,
 2107                                         mock_exec):
 2108         mock_exec.return_value = [None, None]
 2109         with task_manager.acquire(self.context, self.node.uuid,
 2110                                   shared=False) as task:
 2111             self.vendor.bmc_reset(task, 'POST', warm=warm)
 2112             mock_exec.assert_called_once_with(
 2113                 mock.ANY, 'bmc reset %s' % expected)
 2114 
 2115     def test_vendor_passthru_call_bmc_reset_warm(self):
 2116         for param in (True, 'true', 'on', 'y', 'yes'):
 2117             self._vendor_passthru_call_bmc_reset(param, 'warm')
 2118 
 2119     def test_vendor_passthru_call_bmc_reset_cold(self):
 2120         for param in (False, 'false', 'off', 'n', 'no'):
 2121             self._vendor_passthru_call_bmc_reset(param, 'cold')
 2122 
 2123     def test_vendor_passthru_vendor_routes(self):
 2124         expected = ['send_raw', 'bmc_reset']
 2125         with task_manager.acquire(self.context, self.node.uuid,
 2126                                   shared=True) as task:
 2127             vendor_routes = task.driver.vendor.vendor_routes
 2128             self.assertIsInstance(vendor_routes, dict)
 2129             self.assertEqual(sorted(expected), sorted(vendor_routes))
 2130 
 2131     def test_vendor_passthru_driver_routes(self):
 2132         with task_manager.acquire(self.context, self.node.uuid,
 2133                                   shared=True) as task:
 2134             driver_routes = task.driver.vendor.driver_routes
 2135             self.assertIsInstance(driver_routes, dict)
 2136             self.assertEqual({}, driver_routes)
 2137 
 2138     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2139     def test_management_interface_set_boot_device_ok(self, mock_exec):
 2140         mock_exec.return_value = [None, None]
 2141 
 2142         with task_manager.acquire(self.context, self.node.uuid) as task:
 2143             self.management.set_boot_device(task, boot_devices.PXE)
 2144 
 2145         mock_calls = [mock.call(self.info, "raw 0x00 0x08 0x03 0x08"),
 2146                       mock.call(self.info, "chassis bootdev pxe")]
 2147         mock_exec.assert_has_calls(mock_calls)
 2148 
 2149     @mock.patch.object(driver_utils, 'force_persistent_boot', autospec=True)
 2150     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2151     def test_management_interface_no_force_set_boot_device(self,
 2152                                                            mock_exec,
 2153                                                            mock_force_boot):
 2154         mock_exec.return_value = [None, None]
 2155 
 2156         with task_manager.acquire(self.context, self.node.uuid) as task:
 2157             driver_info = task.node.driver_info
 2158             driver_info['ipmi_force_boot_device'] = 'False'
 2159             task.node.driver_info = driver_info
 2160             self.info['force_boot_device'] = 'False'
 2161             self.management.set_boot_device(task, boot_devices.PXE)
 2162 
 2163         mock_calls = [mock.call(self.info, "raw 0x00 0x08 0x03 0x08"),
 2164                       mock.call(self.info, "chassis bootdev pxe")]
 2165         mock_exec.assert_has_calls(mock_calls)
 2166         self.assertFalse(mock_force_boot.called)
 2167 
 2168     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2169     def test_management_interface_force_set_boot_device_ok(self, mock_exec):
 2170         mock_exec.return_value = [None, None]
 2171 
 2172         with task_manager.acquire(self.context, self.node.uuid) as task:
 2173             driver_info = task.node.driver_info
 2174             driver_info['ipmi_force_boot_device'] = 'True'
 2175             task.node.driver_info = driver_info
 2176             self.info['force_boot_device'] = 'True'
 2177             self.management.set_boot_device(task, boot_devices.PXE)
 2178             task.node.refresh()
 2179             self.assertIs(
 2180                 False,
 2181                 task.node.driver_internal_info['is_next_boot_persistent']
 2182             )
 2183 
 2184         mock_calls = [mock.call(self.info, "raw 0x00 0x08 0x03 0x08"),
 2185                       mock.call(self.info, "chassis bootdev pxe")]
 2186         mock_exec.assert_has_calls(mock_calls)
 2187 
 2188     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2189     def test_management_interface_set_boot_device_persistent(self, mock_exec):
 2190         mock_exec.return_value = [None, None]
 2191 
 2192         with task_manager.acquire(self.context, self.node.uuid) as task:
 2193             driver_info = task.node.driver_info
 2194             driver_info['ipmi_force_boot_device'] = 'True'
 2195             task.node.driver_info = driver_info
 2196             self.info['force_boot_device'] = 'True'
 2197             self.management.set_boot_device(task, boot_devices.PXE, True)
 2198             self.assertEqual(
 2199                 boot_devices.PXE,
 2200                 task.node.driver_internal_info['persistent_boot_device'])
 2201 
 2202         mock_calls = [mock.call(self.info, "raw 0x00 0x08 0x03 0x08"),
 2203                       mock.call(self.info, "chassis bootdev pxe")]
 2204         mock_exec.assert_has_calls(mock_calls)
 2205 
 2206     def test_management_interface_set_boot_device_bad_device(self):
 2207         with task_manager.acquire(self.context, self.node.uuid) as task:
 2208             self.assertRaises(exception.InvalidParameterValue,
 2209                               self.management.set_boot_device,
 2210                               task, 'fake-device')
 2211 
 2212     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2213     def test_management_interface_set_boot_device_without_timeout_1(self,
 2214                                                                     mock_exec):
 2215         mock_exec.return_value = [None, None]
 2216 
 2217         with task_manager.acquire(self.context, self.node.uuid) as task:
 2218             driver_info = task.node.driver_info
 2219             driver_info['ipmi_disable_boot_timeout'] = 'False'
 2220             task.node.driver_info = driver_info
 2221             self.management.set_boot_device(task, boot_devices.PXE)
 2222 
 2223         mock_calls = [mock.call(self.info, "chassis bootdev pxe")]
 2224         mock_exec.assert_has_calls(mock_calls)
 2225 
 2226     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2227     def test_management_interface_set_boot_device_without_timeout_2(self,
 2228                                                                     mock_exec):
 2229         CONF.set_override('disable_boot_timeout', False, 'ipmi')
 2230         mock_exec.return_value = [None, None]
 2231 
 2232         with task_manager.acquire(self.context, self.node.uuid) as task:
 2233             self.management.set_boot_device(task, boot_devices.PXE)
 2234 
 2235         mock_calls = [mock.call(self.info, "chassis bootdev pxe")]
 2236         mock_exec.assert_has_calls(mock_calls)
 2237 
 2238     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2239     def test_management_interface_set_boot_device_exec_failed(self, mock_exec):
 2240         mock_exec.side_effect = processutils.ProcessExecutionError()
 2241         with task_manager.acquire(self.context, self.node.uuid) as task:
 2242             self.assertRaises(exception.IPMIFailure,
 2243                               self.management.set_boot_device,
 2244                               task, boot_devices.PXE)
 2245 
 2246     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2247     def test_management_interface_set_boot_device_unknown_exception(self,
 2248                                                                     mock_exec):
 2249 
 2250         class FakeException(Exception):
 2251             pass
 2252 
 2253         mock_exec.side_effect = FakeException('boom')
 2254         with task_manager.acquire(self.context, self.node.uuid) as task:
 2255             self.assertRaises(FakeException,
 2256                               self.management.set_boot_device,
 2257                               task, boot_devices.PXE)
 2258 
 2259     @mock.patch.object(boot_mode_utils, 'get_boot_mode_for_deploy',
 2260                        autospec=True)
 2261     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2262     def test_management_interface_set_boot_device_uefi(self, mock_exec,
 2263                                                        mock_boot_mode):
 2264         mock_boot_mode.return_value = 'uefi'
 2265         mock_exec.return_value = [None, None]
 2266 
 2267         with task_manager.acquire(self.context, self.node.uuid) as task:
 2268             self.management.set_boot_device(task, boot_devices.PXE)
 2269 
 2270         mock_calls = [
 2271             mock.call(self.info, "raw 0x00 0x08 0x03 0x08"),
 2272             mock.call(self.info, "raw 0x00 0x08 0x05 0xa0 0x04 0x00 0x00 0x00")
 2273         ]
 2274         mock_exec.assert_has_calls(mock_calls)
 2275 
 2276     @mock.patch.object(boot_mode_utils, 'get_boot_mode_for_deploy',
 2277                        autospec=True)
 2278     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2279     def test_management_interface_set_boot_device_uefi_and_persistent(
 2280             self, mock_exec, mock_boot_mode):
 2281         mock_boot_mode.return_value = 'uefi'
 2282         mock_exec.return_value = [None, None]
 2283 
 2284         with task_manager.acquire(self.context, self.node.uuid) as task:
 2285             self.management.set_boot_device(task, boot_devices.PXE,
 2286                                             persistent=True)
 2287         mock_calls = [
 2288             mock.call(self.info, "raw 0x00 0x08 0x03 0x08"),
 2289             mock.call(self.info, "raw 0x00 0x08 0x05 0xe0 0x04 0x00 0x00 0x00")
 2290         ]
 2291         mock_exec.assert_has_calls(mock_calls)
 2292 
 2293     @mock.patch.object(boot_mode_utils, 'get_boot_mode_for_deploy',
 2294                        autospec=True)
 2295     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2296     def test_management_interface_set_boot_device_uefi_and_persistent_smci(
 2297             self, mock_exec, mock_boot_mode):
 2298         mock_boot_mode.return_value = 'uefi'
 2299         mock_exec.return_value = [None, None]
 2300         properties = self.node.properties
 2301         properties['vendor'] = 'SuperMicro'
 2302         self.node.properties = properties
 2303         self.node.save()
 2304         with task_manager.acquire(self.context, self.node.uuid) as task:
 2305             self.management.set_boot_device(task, boot_devices.DISK,
 2306                                             persistent=True)
 2307         mock_calls = [
 2308             mock.call(self.info, "raw 0x00 0x08 0x03 0x08"),
 2309             mock.call(self.info, "raw 0x00 0x08 0x05 0xe0 "
 2310                                  "0x24 0x00 0x00 0x00")
 2311         ]
 2312         mock_exec.assert_has_calls(mock_calls)
 2313 
 2314     @mock.patch.object(boot_mode_utils, 'get_boot_mode_for_deploy',
 2315                        autospec=True)
 2316     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2317     def test_management_interface_set_boot_device_uefi_and_onetime_smci(
 2318             self, mock_exec, mock_boot_mode):
 2319         mock_boot_mode.return_value = 'uefi'
 2320         mock_exec.return_value = [None, None]
 2321         properties = self.node.properties
 2322         properties['vendor'] = 'SuperMicro'
 2323         self.node.properties = properties
 2324         self.node.save()
 2325         with task_manager.acquire(self.context, self.node.uuid) as task:
 2326             self.management.set_boot_device(task, boot_devices.DISK,
 2327                                             persistent=False)
 2328         mock_calls = [
 2329             mock.call(self.info, "raw 0x00 0x08 0x03 0x08"),
 2330             mock.call(self.info, "raw 0x00 0x08 0x05 0xa0 "
 2331                                  "0x24 0x00 0x00 0x00")
 2332         ]
 2333         mock_exec.assert_has_calls(mock_calls)
 2334 
 2335     def test_management_interface_get_supported_boot_devices(self):
 2336         with task_manager.acquire(self.context, self.node.uuid) as task:
 2337             expected = [boot_devices.PXE, boot_devices.DISK,
 2338                         boot_devices.CDROM, boot_devices.BIOS,
 2339                         boot_devices.SAFE]
 2340             self.assertEqual(sorted(expected), sorted(task.driver.management.
 2341                              get_supported_boot_devices(task)))
 2342 
 2343     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2344     def test_management_interface_get_boot_device(self, mock_exec):
 2345         # output, expected boot device
 2346         bootdevs = [('Boot Device Selector : '
 2347                      'Force Boot from default Hard-Drive\n',
 2348                      boot_devices.DISK),
 2349                     ('Boot Device Selector : '
 2350                      'Force Boot from default Hard-Drive, request Safe-Mode\n',
 2351                      boot_devices.SAFE),
 2352                     ('Boot Device Selector : '
 2353                      'Force Boot into BIOS Setup\n',
 2354                      boot_devices.BIOS),
 2355                     ('Boot Device Selector : '
 2356                      'Force PXE\n',
 2357                      boot_devices.PXE),
 2358                     ('Boot Device Selector : '
 2359                      'Force Boot from CD/DVD\n',
 2360                      boot_devices.CDROM)]
 2361         with task_manager.acquire(self.context, self.node.uuid) as task:
 2362             for out, expected_device in bootdevs:
 2363                 mock_exec.return_value = (out, '')
 2364                 expected_response = {'boot_device': expected_device,
 2365                                      'persistent': False}
 2366                 self.assertEqual(expected_response,
 2367                                  task.driver.management.get_boot_device(task))
 2368                 mock_exec.assert_called_with(mock.ANY,
 2369                                              "chassis bootparam get 5")
 2370 
 2371     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2372     def test_management_interface_get_boot_device_unknown_dev(self, mock_exec):
 2373         with task_manager.acquire(self.context, self.node.uuid) as task:
 2374             mock_exec.return_value = ('Boot Device Selector : Fake\n', '')
 2375             response = task.driver.management.get_boot_device(task)
 2376             self.assertIsNone(response['boot_device'])
 2377             mock_exec.assert_called_with(mock.ANY, "chassis bootparam get 5")
 2378 
 2379     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2380     def test_management_interface_get_boot_device_fail(self, mock_exec):
 2381         with task_manager.acquire(self.context, self.node.uuid) as task:
 2382             mock_exec.side_effect = processutils.ProcessExecutionError()
 2383             self.assertRaises(exception.IPMIFailure,
 2384                               task.driver.management.get_boot_device, task)
 2385             mock_exec.assert_called_with(mock.ANY, "chassis bootparam get 5")
 2386 
 2387     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2388     def test_management_interface_get_boot_device_persistent(self, mock_exec):
 2389         outputs = [('Options apply to only next boot\n'
 2390                     'Boot Device Selector : Force PXE\n',
 2391                     False),
 2392                    ('Options apply to all future boots\n'
 2393                     'Boot Device Selector : Force PXE\n',
 2394                     True)]
 2395         with task_manager.acquire(self.context, self.node.uuid) as task:
 2396             for out, expected_persistent in outputs:
 2397                 mock_exec.return_value = (out, '')
 2398                 expected_response = {'boot_device': boot_devices.PXE,
 2399                                      'persistent': expected_persistent}
 2400                 self.assertEqual(expected_response,
 2401                                  task.driver.management.get_boot_device(task))
 2402                 mock_exec.assert_called_with(mock.ANY,
 2403                                              "chassis bootparam get 5")
 2404 
 2405     def test_get_force_boot_device_persistent(self):
 2406         with task_manager.acquire(self.context, self.node.uuid) as task:
 2407             task.node.driver_info['ipmi_force_boot_device'] = 'True'
 2408             task.node.driver_internal_info['persistent_boot_device'] = 'pxe'
 2409             bootdev = self.management.get_boot_device(task)
 2410             self.assertEqual('pxe', bootdev['boot_device'])
 2411             self.assertTrue(bootdev['persistent'])
 2412 
 2413     def test_management_interface_validate_good(self):
 2414         with task_manager.acquire(self.context, self.node.uuid) as task:
 2415             task.driver.management.validate(task)
 2416 
 2417     def test_management_interface_validate_fail(self):
 2418         # Missing IPMI driver_info information
 2419         node = obj_utils.create_test_node(self.context,
 2420                                           uuid=uuidutils.generate_uuid(),
 2421                                           management_interface='ipmitool')
 2422         with task_manager.acquire(self.context, node.uuid) as task:
 2423             self.assertRaises(exception.MissingParameterValue,
 2424                               task.driver.management.validate, task)
 2425 
 2426     @mock.patch.object(ipmi.LOG, 'error', spec_set=True, autospec=True)
 2427     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2428     def test_management_interface_inject_nmi_ok(self, mock_exec, mock_log):
 2429         with task_manager.acquire(self.context, self.node.uuid) as task:
 2430             driver_info = ipmi._parse_driver_info(task.node)
 2431             self.management.inject_nmi(task)
 2432 
 2433             mock_exec.assert_called_once_with(driver_info, "power diag")
 2434             self.assertFalse(mock_log.called)
 2435 
 2436     @mock.patch.object(ipmi.LOG, 'error', spec_set=True, autospec=True)
 2437     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2438     def test_management_interface_inject_nmi_fail(self, mock_exec, mock_log):
 2439         mock_exec.side_effect = exception.PasswordFileFailedToCreate('error')
 2440 
 2441         with task_manager.acquire(self.context, self.node.uuid) as task:
 2442             driver_info = ipmi._parse_driver_info(task.node)
 2443             self.assertRaises(exception.IPMIFailure,
 2444                               self.management.inject_nmi,
 2445                               task)
 2446 
 2447             mock_exec.assert_called_once_with(driver_info, "power diag")
 2448             self.assertTrue(mock_log.called)
 2449 
 2450     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2451     def test_detect_vendor(self, mock_exec):
 2452         mock_exec.return_value = (MC_INFO.format(vendor='LolCatMeow'),
 2453                                   None)
 2454         with task_manager.acquire(self.context, self.node.uuid) as task:
 2455             driver_info = ipmi._parse_driver_info(task.node)
 2456             vendor = self.management.detect_vendor(task)
 2457             mock_exec.assert_called_once_with(driver_info, 'mc info')
 2458             self.assertEqual('lolcatmeow', vendor)
 2459 
 2460     def test__parse_ipmi_sensor_data_ok(self):
 2461         fake_sensors_data = """
 2462                             Sensor ID              : Temp (0x1)
 2463                              Entity ID             : 3.1 (Processor)
 2464                              Sensor Type (Analog)  : Temperature
 2465                              Sensor Reading        : -58 (+/- 1) degrees C
 2466                              Status                : ok
 2467                              Nominal Reading       : 50.000
 2468                              Normal Minimum        : 11.000
 2469                              Normal Maximum        : 69.000
 2470                              Upper critical        : 90.000
 2471                              Upper non-critical    : 85.000
 2472                              Positive Hysteresis   : 1.000
 2473                              Negative Hysteresis   : 1.000
 2474 
 2475                             Sensor ID              : Temp (0x2)
 2476                              Entity ID             : 3.2 (Processor)
 2477                              Sensor Type (Analog)  : Temperature
 2478                              Sensor Reading        : 50 (+/- 1) degrees C
 2479                              Status                : ok
 2480                              Nominal Reading       : 50.000
 2481                              Normal Minimum        : 11.000
 2482                              Normal Maximum        : 69.000
 2483                              Upper critical        : 90.000
 2484                              Upper non-critical    : 85.000
 2485                              Positive Hysteresis   : 1.000
 2486                              Negative Hysteresis   : 1.000
 2487 
 2488                             Sensor ID              : FAN MOD 1A RPM (0x30)
 2489                              Entity ID             : 7.1 (System Board)
 2490                              Sensor Type (Analog)  : Fan
 2491                              Sensor Reading        : 8400 (+/- 75) RPM
 2492                              Status                : ok
 2493                              Nominal Reading       : 5325.000
 2494                              Normal Minimum        : 10425.000
 2495                              Normal Maximum        : 14775.000
 2496                              Lower critical        : 4275.000
 2497                              Positive Hysteresis   : 375.000
 2498                              Negative Hysteresis   : 375.000
 2499 
 2500                             Sensor ID              : FAN MOD 1B RPM (0x31)
 2501                              Entity ID             : 7.1 (System Board)
 2502                              Sensor Type (Analog)  : Fan
 2503                              Sensor Reading        : 8550 (+/- 75) RPM
 2504                              Status                : ok
 2505                              Nominal Reading       : 7800.000
 2506                              Normal Minimum        : 10425.000
 2507                              Normal Maximum        : 14775.000
 2508                              Lower critical        : 4275.000
 2509                              Positive Hysteresis   : 375.000
 2510                              Negative Hysteresis   : 375.000
 2511                              """
 2512         expected_return = {
 2513             'Fan': {
 2514                 'FAN MOD 1A RPM (0x30)': {
 2515                     'Status': 'ok',
 2516                     'Sensor Reading': '8400 (+/- 75) RPM',
 2517                     'Entity ID': '7.1 (System Board)',
 2518                     'Normal Minimum': '10425.000',
 2519                     'Positive Hysteresis': '375.000',
 2520                     'Normal Maximum': '14775.000',
 2521                     'Sensor Type (Analog)': 'Fan',
 2522                     'Lower critical': '4275.000',
 2523                     'Negative Hysteresis': '375.000',
 2524                     'Sensor ID': 'FAN MOD 1A RPM (0x30)',
 2525                     'Nominal Reading': '5325.000'
 2526                 },
 2527                 'FAN MOD 1B RPM (0x31)': {
 2528                     'Status': 'ok',
 2529                     'Sensor Reading': '8550 (+/- 75) RPM',
 2530                     'Entity ID': '7.1 (System Board)',
 2531                     'Normal Minimum': '10425.000',
 2532                     'Positive Hysteresis': '375.000',
 2533                     'Normal Maximum': '14775.000',
 2534                     'Sensor Type (Analog)': 'Fan',
 2535                     'Lower critical': '4275.000',
 2536                     'Negative Hysteresis': '375.000',
 2537                     'Sensor ID': 'FAN MOD 1B RPM (0x31)',
 2538                     'Nominal Reading': '7800.000'
 2539                 }
 2540             },
 2541             'Temperature': {
 2542                 'Temp (0x1)': {
 2543                     'Status': 'ok',
 2544                     'Sensor Reading': '-58 (+/- 1) degrees C',
 2545                     'Entity ID': '3.1 (Processor)',
 2546                     'Normal Minimum': '11.000',
 2547                     'Positive Hysteresis': '1.000',
 2548                     'Upper non-critical': '85.000',
 2549                     'Normal Maximum': '69.000',
 2550                     'Sensor Type (Analog)': 'Temperature',
 2551                     'Negative Hysteresis': '1.000',
 2552                     'Upper critical': '90.000',
 2553                     'Sensor ID': 'Temp (0x1)',
 2554                     'Nominal Reading': '50.000'
 2555                 },
 2556                 'Temp (0x2)': {
 2557                     'Status': 'ok',
 2558                     'Sensor Reading': '50 (+/- 1) degrees C',
 2559                     'Entity ID': '3.2 (Processor)',
 2560                     'Normal Minimum': '11.000',
 2561                     'Positive Hysteresis': '1.000',
 2562                     'Upper non-critical': '85.000',
 2563                     'Normal Maximum': '69.000',
 2564                     'Sensor Type (Analog)': 'Temperature',
 2565                     'Negative Hysteresis': '1.000',
 2566                     'Upper critical': '90.000',
 2567                     'Sensor ID': 'Temp (0x2)',
 2568                     'Nominal Reading': '50.000'
 2569                 }
 2570             }
 2571         }
 2572         ret = ipmi._parse_ipmi_sensors_data(self.node, fake_sensors_data)
 2573 
 2574         self.assertEqual(expected_return, ret)
 2575 
 2576     def test__parse_ipmi_sensor_data_missing_sensor_reading(self):
 2577         fake_sensors_data = """
 2578                             Sensor ID              : Temp (0x1)
 2579                              Entity ID             : 3.1 (Processor)
 2580                              Sensor Type (Analog)  : Temperature
 2581                              Status                : ok
 2582                              Nominal Reading       : 50.000
 2583                              Normal Minimum        : 11.000
 2584                              Normal Maximum        : 69.000
 2585                              Upper critical        : 90.000
 2586                              Upper non-critical    : 85.000
 2587                              Positive Hysteresis   : 1.000
 2588                              Negative Hysteresis   : 1.000
 2589 
 2590                             Sensor ID              : Temp (0x2)
 2591                              Entity ID             : 3.2 (Processor)
 2592                              Sensor Type (Analog)  : Temperature
 2593                              Sensor Reading        : 50 (+/- 1) degrees C
 2594                              Status                : ok
 2595                              Nominal Reading       : 50.000
 2596                              Normal Minimum        : 11.000
 2597                              Normal Maximum        : 69.000
 2598                              Upper critical        : 90.000
 2599                              Upper non-critical    : 85.000
 2600                              Positive Hysteresis   : 1.000
 2601                              Negative Hysteresis   : 1.000
 2602 
 2603                             Sensor ID              : FAN MOD 1A RPM (0x30)
 2604                              Entity ID             : 7.1 (System Board)
 2605                              Sensor Type (Analog)  : Fan
 2606                              Sensor Reading        : 8400 (+/- 75) RPM
 2607                              Status                : ok
 2608                              Nominal Reading       : 5325.000
 2609                              Normal Minimum        : 10425.000
 2610                              Normal Maximum        : 14775.000
 2611                              Lower critical        : 4275.000
 2612                              Positive Hysteresis   : 375.000
 2613                              Negative Hysteresis   : 375.000
 2614                              """
 2615         expected_return = {
 2616             'Fan': {
 2617                 'FAN MOD 1A RPM (0x30)': {
 2618                     'Status': 'ok',
 2619                     'Sensor Reading': '8400 (+/- 75) RPM',
 2620                     'Entity ID': '7.1 (System Board)',
 2621                     'Normal Minimum': '10425.000',
 2622                     'Positive Hysteresis': '375.000',
 2623                     'Normal Maximum': '14775.000',
 2624                     'Sensor Type (Analog)': 'Fan',
 2625                     'Lower critical': '4275.000',
 2626                     'Negative Hysteresis': '375.000',
 2627                     'Sensor ID': 'FAN MOD 1A RPM (0x30)',
 2628                     'Nominal Reading': '5325.000'
 2629                 }
 2630             },
 2631             'Temperature': {
 2632                 'Temp (0x2)': {
 2633                     'Status': 'ok',
 2634                     'Sensor Reading': '50 (+/- 1) degrees C',
 2635                     'Entity ID': '3.2 (Processor)',
 2636                     'Normal Minimum': '11.000',
 2637                     'Positive Hysteresis': '1.000',
 2638                     'Upper non-critical': '85.000',
 2639                     'Normal Maximum': '69.000',
 2640                     'Sensor Type (Analog)': 'Temperature',
 2641                     'Negative Hysteresis': '1.000',
 2642                     'Upper critical': '90.000',
 2643                     'Sensor ID': 'Temp (0x2)',
 2644                     'Nominal Reading': '50.000'
 2645                 }
 2646             }
 2647         }
 2648         ret = ipmi._parse_ipmi_sensors_data(self.node, fake_sensors_data)
 2649 
 2650         self.assertEqual(expected_return, ret)
 2651 
 2652     def test__parse_ipmi_sensor_data_debug(self):
 2653         fake_sensors_data = """
 2654 <<  Message tag                        : 0x00
 2655 <<  RMCP+ status                       : no errors
 2656 <<  Maximum privilege level            : admin
 2657 <<  Console Session ID                 : 0xa0a2a3a4
 2658 <<  BMC Session ID                     : 0x02006a01
 2659 <<  Negotiated authenticatin algorithm : hmac_sha1
 2660 <<  Negotiated integrity algorithm     : hmac_sha1_96
 2661 <<  Negotiated encryption algorithm    : aes_cbc_128
 2662 
 2663 
 2664                             Sensor ID              : Temp (0x2)
 2665                              Entity ID             : 3.2 (Processor)
 2666                              Sensor Type (Analog)  : Temperature
 2667                              Sensor Reading        : 50 (+/- 1) degrees C
 2668                              Status                : ok
 2669                              Nominal Reading       : 50.000
 2670                              Normal Minimum        : 11.000
 2671                              Normal Maximum        : 69.000
 2672                              Upper critical        : 90.000
 2673                              Upper non-critical    : 85.000
 2674                              Positive Hysteresis   : 1.000
 2675                              Negative Hysteresis   : 1.000
 2676 
 2677                             Sensor ID              : FAN MOD 1A RPM (0x30)
 2678                              Entity ID             : 7.1 (System Board)
 2679                              Sensor Type (Analog)  : Fan
 2680                              Sensor Reading        : 8400 (+/- 75) RPM
 2681                              Status                : ok
 2682                              Nominal Reading       : 5325.000
 2683                              Normal Minimum        : 10425.000
 2684                              Normal Maximum        : 14775.000
 2685                              Lower critical        : 4275.000
 2686                              Positive Hysteresis   : 375.000
 2687                              Negative Hysteresis   : 375.000
 2688                              """
 2689         expected_return = {
 2690             'Fan': {
 2691                 'FAN MOD 1A RPM (0x30)': {
 2692                     'Status': 'ok',
 2693                     'Sensor Reading': '8400 (+/- 75) RPM',
 2694                     'Entity ID': '7.1 (System Board)',
 2695                     'Normal Minimum': '10425.000',
 2696                     'Positive Hysteresis': '375.000',
 2697                     'Normal Maximum': '14775.000',
 2698                     'Sensor Type (Analog)': 'Fan',
 2699                     'Lower critical': '4275.000',
 2700                     'Negative Hysteresis': '375.000',
 2701                     'Sensor ID': 'FAN MOD 1A RPM (0x30)',
 2702                     'Nominal Reading': '5325.000'
 2703                 }
 2704             },
 2705             'Temperature': {
 2706                 'Temp (0x2)': {
 2707                     'Status': 'ok',
 2708                     'Sensor Reading': '50 (+/- 1) degrees C',
 2709                     'Entity ID': '3.2 (Processor)',
 2710                     'Normal Minimum': '11.000',
 2711                     'Positive Hysteresis': '1.000',
 2712                     'Upper non-critical': '85.000',
 2713                     'Normal Maximum': '69.000',
 2714                     'Sensor Type (Analog)': 'Temperature',
 2715                     'Negative Hysteresis': '1.000',
 2716                     'Upper critical': '90.000',
 2717                     'Sensor ID': 'Temp (0x2)',
 2718                     'Nominal Reading': '50.000'
 2719                 }
 2720             }
 2721         }
 2722         ret = ipmi._parse_ipmi_sensors_data(self.node, fake_sensors_data)
 2723 
 2724         self.assertEqual(expected_return, ret)
 2725 
 2726     def test__parse_ipmi_sensor_data_failed(self):
 2727         fake_sensors_data = "abcdef"
 2728         self.assertRaises(exception.FailedToParseSensorData,
 2729                           ipmi._parse_ipmi_sensors_data,
 2730                           self.node,
 2731                           fake_sensors_data)
 2732 
 2733         fake_sensors_data = "abc:def:ghi"
 2734         self.assertRaises(exception.FailedToParseSensorData,
 2735                           ipmi._parse_ipmi_sensors_data,
 2736                           self.node,
 2737                           fake_sensors_data)
 2738 
 2739     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2740     def test_dump_sdr_ok(self, mock_exec):
 2741         mock_exec.return_value = (None, None)
 2742 
 2743         with task_manager.acquire(self.context, self.node.uuid) as task:
 2744             ipmi.dump_sdr(task, 'foo_file')
 2745 
 2746         mock_exec.assert_called_once_with(self.info, 'sdr dump foo_file')
 2747 
 2748     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2749     def test_dump_sdr_fail(self, mock_exec):
 2750         with task_manager.acquire(self.context, self.node.uuid) as task:
 2751             mock_exec.side_effect = processutils.ProcessExecutionError()
 2752             self.assertRaises(exception.IPMIFailure, ipmi.dump_sdr, task,
 2753                               'foo_file')
 2754         mock_exec.assert_called_once_with(self.info, 'sdr dump foo_file')
 2755 
 2756     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 2757     def test_send_raw_bytes_returns(self, mock_exec):
 2758         fake_ret = ('foo', 'bar')
 2759         mock_exec.return_value = fake_ret
 2760 
 2761         with task_manager.acquire(self.context, self.node.uuid) as task:
 2762             ret = ipmi.send_raw(task, 'fake raw')
 2763 
 2764         self.assertEqual(fake_ret, ret)
 2765 
 2766     @mock.patch.object(console_utils, 'acquire_port', autospec=True)
 2767     def test__allocate_port(self, mock_acquire):
 2768         mock_acquire.return_value = 1234
 2769         with task_manager.acquire(self.context,
 2770                                   self.node.uuid) as task:
 2771             port = ipmi._allocate_port(task)
 2772             mock_acquire.assert_called_once_with(host=None)
 2773             self.assertEqual(port, 1234)
 2774             info = task.node.driver_internal_info
 2775             self.assertEqual(info['allocated_ipmi_terminal_port'], 1234)
 2776 
 2777     @mock.patch.object(console_utils, 'release_port', autospec=True)
 2778     def test__release_allocated_port(self, mock_release):
 2779         info = self.node.driver_internal_info
 2780         info['allocated_ipmi_terminal_port'] = 1234
 2781         self.node.driver_internal_info = info
 2782         self.node.save()
 2783 
 2784         with task_manager.acquire(self.context,
 2785                                   self.node.uuid) as task:
 2786             ipmi._release_allocated_port(task)
 2787             mock_release.assert_called_once_with(1234)
 2788             info = task.node.driver_internal_info
 2789             self.assertIsNone(info.get('allocated_ipmi_terminal_port'))
 2790 
 2791 
 2792 class IPMIToolShellinaboxTestCase(db_base.DbTestCase):
 2793     console_interface = 'ipmitool-shellinabox'
 2794     console_class = ipmi.IPMIShellinaboxConsole
 2795 
 2796     def setUp(self):
 2797         super(IPMIToolShellinaboxTestCase, self).setUp()
 2798         self.config(enabled_console_interfaces=[self.console_interface,
 2799                                                 'no-console'])
 2800         self.node = obj_utils.create_test_node(
 2801             self.context,
 2802             console_interface=self.console_interface,
 2803             driver_info=INFO_DICT)
 2804         self.info = ipmi._parse_driver_info(self.node)
 2805         self.console = self.console_class()
 2806 
 2807     def test_console_validate(self):
 2808         with task_manager.acquire(
 2809                 self.context, self.node.uuid, shared=True) as task:
 2810             task.node.driver_info['ipmi_terminal_port'] = 123
 2811             task.driver.console.validate(task)
 2812 
 2813     def test_console_validate_missing_port(self):
 2814         with task_manager.acquire(
 2815                 self.context, self.node.uuid, shared=True) as task:
 2816             task.node.driver_info.pop('ipmi_terminal_port', None)
 2817             self.assertRaises(exception.MissingParameterValue,
 2818                               task.driver.console.validate, task)
 2819 
 2820     def test_console_validate_missing_port_auto_allocate(self):
 2821         self.config(port_range='10000:20000', group='console')
 2822         with task_manager.acquire(
 2823                 self.context, self.node.uuid, shared=True) as task:
 2824             task.node.driver_info.pop('ipmi_terminal_port', None)
 2825             task.driver.console.validate(task)
 2826 
 2827     def test_console_validate_invalid_port(self):
 2828         with task_manager.acquire(
 2829                 self.context, self.node.uuid, shared=True) as task:
 2830             task.node.driver_info['ipmi_terminal_port'] = ''
 2831             self.assertRaises(exception.InvalidParameterValue,
 2832                               task.driver.console.validate, task)
 2833 
 2834     def test_console_validate_wrong_ipmi_protocol_version(self):
 2835         with task_manager.acquire(
 2836                 self.context, self.node.uuid, shared=True) as task:
 2837             task.node.driver_info['ipmi_terminal_port'] = 123
 2838             task.node.driver_info['ipmi_protocol_version'] = '1.5'
 2839             self.assertRaises(exception.InvalidParameterValue,
 2840                               task.driver.console.validate, task)
 2841 
 2842     def test__get_ipmi_cmd(self):
 2843         with task_manager.acquire(self.context,
 2844                                   self.node.uuid) as task:
 2845             driver_info = ipmi._parse_driver_info(task.node)
 2846             ipmi_cmd = self.console._get_ipmi_cmd(driver_info, 'pw_file')
 2847             expected_ipmi_cmd = ("/:%(uid)s:%(gid)s:HOME:ipmitool "
 2848                                  "-I lanplus -H %(address)s -L ADMINISTRATOR "
 2849                                  "-U %(user)s -f pw_file" %
 2850                                  {'uid': os.getuid(), 'gid': os.getgid(),
 2851                                   'address': driver_info['address'],
 2852                                   'user': driver_info['username']})
 2853         self.assertEqual(expected_ipmi_cmd, ipmi_cmd)
 2854 
 2855     def test__get_ipmi_cmd_without_user(self):
 2856         with task_manager.acquire(self.context,
 2857                                   self.node.uuid) as task:
 2858             driver_info = ipmi._parse_driver_info(task.node)
 2859             driver_info['username'] = None
 2860             ipmi_cmd = self.console._get_ipmi_cmd(driver_info, 'pw_file')
 2861             expected_ipmi_cmd = ("/:%(uid)s:%(gid)s:HOME:ipmitool "
 2862                                  "-I lanplus -H %(address)s -L ADMINISTRATOR "
 2863                                  "-f pw_file" %
 2864                                  {'uid': os.getuid(), 'gid': os.getgid(),
 2865                                   'address': driver_info['address']})
 2866         self.assertEqual(expected_ipmi_cmd, ipmi_cmd)
 2867 
 2868     @mock.patch.object(ipmi, '_allocate_port', autospec=True)
 2869     @mock.patch.object(ipmi.IPMIConsole, '_start_console', autospec=True)
 2870     def test_start_console(self, mock_start, mock_alloc):
 2871         mock_start.return_value = None
 2872         mock_alloc.return_value = 10000
 2873 
 2874         with task_manager.acquire(self.context,
 2875                                   self.node.uuid) as task:
 2876             self.console.start_console(task)
 2877             driver_info = ipmi._parse_driver_info(task.node)
 2878             driver_info.update(port=10000)
 2879         mock_start.assert_called_once_with(
 2880             self.console, driver_info,
 2881             console_utils.start_shellinabox_console)
 2882 
 2883     @mock.patch.object(ipmi, '_allocate_port', autospec=True)
 2884     @mock.patch.object(ipmi, '_parse_driver_info', autospec=True)
 2885     @mock.patch.object(ipmi.IPMIConsole, '_start_console', autospec=True)
 2886     def test_start_console_with_port(self, mock_start, mock_info, mock_alloc):
 2887         mock_start.return_value = None
 2888         mock_info.return_value = {'port': 10000}
 2889 
 2890         with task_manager.acquire(self.context,
 2891                                   self.node.uuid) as task:
 2892             self.console.start_console(task)
 2893         mock_start.assert_called_once_with(
 2894             self.console, {'port': 10000},
 2895             console_utils.start_shellinabox_console)
 2896         mock_alloc.assert_not_called()
 2897 
 2898     @mock.patch.object(ipmi, '_allocate_port', autospec=True)
 2899     @mock.patch.object(ipmi, '_parse_driver_info', autospec=True)
 2900     @mock.patch.object(ipmi.IPMIConsole, '_start_console', autospec=True)
 2901     def test_start_console_alloc_port(self, mock_start, mock_info, mock_alloc):
 2902         mock_start.return_value = None
 2903         mock_info.return_value = {'port': None}
 2904         mock_alloc.return_value = 1234
 2905 
 2906         with task_manager.acquire(self.context,
 2907                                   self.node.uuid) as task:
 2908             self.console.start_console(task)
 2909         mock_start.assert_called_once_with(
 2910             self.console, {'port': 1234},
 2911             console_utils.start_shellinabox_console)
 2912         mock_alloc.assert_called_once_with(mock.ANY)
 2913 
 2914     @mock.patch.object(ipmi.IPMIConsole, '_get_ipmi_cmd', autospec=True)
 2915     @mock.patch.object(console_utils, 'start_shellinabox_console',
 2916                        autospec=True)
 2917     def test__start_console(self, mock_start, mock_ipmi_cmd):
 2918         mock_start.return_value = None
 2919 
 2920         with task_manager.acquire(self.context,
 2921                                   self.node.uuid) as task:
 2922             driver_info = ipmi._parse_driver_info(task.node)
 2923             self.console._start_console(
 2924                 driver_info, console_utils.start_shellinabox_console)
 2925 
 2926         mock_start.assert_called_once_with(self.info['uuid'],
 2927                                            self.info['port'],
 2928                                            mock.ANY)
 2929         mock_ipmi_cmd.assert_called_once_with(self.console,
 2930                                               driver_info, mock.ANY)
 2931 
 2932     @mock.patch.object(console_utils, 'start_shellinabox_console',
 2933                        autospec=True)
 2934     def test__start_console_fail(self, mock_start):
 2935         mock_start.side_effect = exception.ConsoleSubprocessFailed(
 2936             error='error')
 2937 
 2938         with task_manager.acquire(self.context,
 2939                                   self.node.uuid) as task:
 2940             driver_info = ipmi._parse_driver_info(task.node)
 2941             self.assertRaises(exception.ConsoleSubprocessFailed,
 2942                               self.console._start_console,
 2943                               driver_info,
 2944                               console_utils.start_shellinabox_console)
 2945 
 2946     @mock.patch.object(console_utils, 'start_shellinabox_console',
 2947                        autospec=True)
 2948     def test__start_console_fail_nodir(self, mock_start):
 2949         mock_start.side_effect = exception.ConsoleError()
 2950 
 2951         with task_manager.acquire(self.context,
 2952                                   self.node.uuid) as task:
 2953             driver_info = ipmi._parse_driver_info(task.node)
 2954             self.assertRaises(exception.ConsoleError,
 2955                               self.console._start_console,
 2956                               driver_info,
 2957                               console_utils.start_shellinabox_console)
 2958         mock_start.assert_called_once_with(self.node.uuid, mock.ANY, mock.ANY)
 2959 
 2960     @mock.patch.object(console_utils, 'make_persistent_password_file',
 2961                        autospec=True)
 2962     @mock.patch.object(console_utils, 'start_shellinabox_console',
 2963                        autospec=True)
 2964     def test__start_console_empty_password(self, mock_start, mock_pass):
 2965         driver_info = self.node.driver_info
 2966         del driver_info['ipmi_password']
 2967         self.node.driver_info = driver_info
 2968         self.node.save()
 2969 
 2970         with task_manager.acquire(self.context,
 2971                                   self.node.uuid) as task:
 2972             driver_info = ipmi._parse_driver_info(task.node)
 2973             self.console._start_console(
 2974                 driver_info, console_utils.start_shellinabox_console)
 2975 
 2976         mock_pass.assert_called_once_with(mock.ANY, '\0')
 2977         mock_start.assert_called_once_with(self.info['uuid'],
 2978                                            self.info['port'],
 2979                                            mock.ANY)
 2980 
 2981     @mock.patch.object(ipmi, '_release_allocated_port', autospec=True)
 2982     @mock.patch.object(console_utils, 'stop_shellinabox_console',
 2983                        autospec=True)
 2984     def test_stop_console(self, mock_stop, mock_release):
 2985         mock_stop.return_value = None
 2986 
 2987         with task_manager.acquire(self.context,
 2988                                   self.node.uuid) as task:
 2989             self.console.stop_console(task)
 2990 
 2991         mock_stop.assert_called_once_with(self.info['uuid'])
 2992         mock_release.assert_called_once_with(mock.ANY)
 2993 
 2994     @mock.patch.object(console_utils, 'stop_shellinabox_console',
 2995                        autospec=True)
 2996     def test_stop_console_fail(self, mock_stop):
 2997         mock_stop.side_effect = exception.ConsoleError()
 2998 
 2999         with task_manager.acquire(self.context,
 3000                                   self.node.uuid) as task:
 3001             self.assertRaises(exception.ConsoleError,
 3002                               self.console.stop_console,
 3003                               task)
 3004 
 3005         mock_stop.assert_called_once_with(self.node.uuid)
 3006 
 3007     @mock.patch.object(console_utils, 'get_shellinabox_console_url',
 3008                        autospec=True)
 3009     def test_get_console(self, mock_get):
 3010         url = 'http://localhost:4201'
 3011         mock_get.return_value = url
 3012         expected = {'type': 'shellinabox', 'url': url}
 3013 
 3014         with task_manager.acquire(self.context,
 3015                                   self.node.uuid) as task:
 3016             console_info = self.console.get_console(task)
 3017 
 3018         self.assertEqual(expected, console_info)
 3019         mock_get.assert_called_once_with(self.info['port'])
 3020 
 3021 
 3022 class IPMIToolSocatDriverTestCase(IPMIToolShellinaboxTestCase):
 3023     console_interface = 'ipmitool-shellinabox'
 3024     console_class = ipmi.IPMISocatConsole
 3025 
 3026     def test__get_ipmi_cmd(self):
 3027         with task_manager.acquire(self.context,
 3028                                   self.node.uuid) as task:
 3029             driver_info = ipmi._parse_driver_info(task.node)
 3030             ipmi_cmd = self.console._get_ipmi_cmd(driver_info, 'pw_file')
 3031             expected_ipmi_cmd = ("ipmitool -I lanplus -H %(address)s "
 3032                                  "-L ADMINISTRATOR -U %(user)s "
 3033                                  "-f pw_file" %
 3034                                  {'address': driver_info['address'],
 3035                                   'user': driver_info['username']})
 3036         self.assertEqual(expected_ipmi_cmd, ipmi_cmd)
 3037 
 3038     def test__get_ipmi_cmd_without_user(self):
 3039         with task_manager.acquire(self.context,
 3040                                   self.node.uuid) as task:
 3041             driver_info = ipmi._parse_driver_info(task.node)
 3042             driver_info['username'] = None
 3043             ipmi_cmd = self.console._get_ipmi_cmd(driver_info, 'pw_file')
 3044             expected_ipmi_cmd = ("ipmitool -I lanplus -H %(address)s "
 3045                                  "-L ADMINISTRATOR "
 3046                                  "-f pw_file" %
 3047                                  {'address': driver_info['address']})
 3048         self.assertEqual(expected_ipmi_cmd, ipmi_cmd)
 3049 
 3050     def test_console_validate_missing_port_auto_allocate(self):
 3051         self.config(port_range='10000:20000', group='console')
 3052         with task_manager.acquire(
 3053                 self.context, self.node.uuid, shared=True) as task:
 3054             task.node.driver_info.pop('ipmi_terminal_port', None)
 3055             task.driver.console.validate(task)
 3056 
 3057     @mock.patch.object(ipmi, '_allocate_port', autospec=True)
 3058     @mock.patch.object(ipmi.IPMIConsole, '_start_console', autospec=True)
 3059     @mock.patch.object(ipmi.IPMISocatConsole, '_exec_stop_console',
 3060                        autospec=True)
 3061     def test_start_console(self, mock_stop, mock_start, mock_alloc):
 3062         mock_start.return_value = None
 3063         mock_stop.return_value = None
 3064         mock_alloc.return_value = 10000
 3065 
 3066         with task_manager.acquire(self.context,
 3067                                   self.node.uuid) as task:
 3068             self.console.start_console(task)
 3069             driver_info = ipmi._parse_driver_info(task.node)
 3070             driver_info.update(port=10000)
 3071         mock_stop.assert_called_once_with(self.console, driver_info)
 3072         mock_start.assert_called_once_with(
 3073             self.console, driver_info,
 3074             console_utils.start_socat_console)
 3075 
 3076     @mock.patch.object(ipmi, '_allocate_port', autospec=True)
 3077     @mock.patch.object(ipmi, '_parse_driver_info', autospec=True)
 3078     @mock.patch.object(ipmi.IPMIConsole, '_start_console', autospec=True)
 3079     @mock.patch.object(ipmi.IPMISocatConsole, '_exec_stop_console',
 3080                        autospec=True)
 3081     def test_start_console_with_port(self, mock_stop, mock_start, mock_info,
 3082                                      mock_alloc):
 3083         mock_start.return_value = None
 3084         mock_info.return_value = {'port': 10000}
 3085 
 3086         with task_manager.acquire(self.context,
 3087                                   self.node.uuid) as task:
 3088             self.console.start_console(task)
 3089         mock_stop.assert_called_once_with(self.console, mock.ANY)
 3090         mock_start.assert_called_once_with(
 3091             self.console, {'port': 10000},
 3092             console_utils.start_socat_console)
 3093         mock_alloc.assert_not_called()
 3094 
 3095     @mock.patch.object(ipmi, '_allocate_port', autospec=True)
 3096     @mock.patch.object(ipmi, '_parse_driver_info', autospec=True)
 3097     @mock.patch.object(ipmi.IPMIConsole, '_start_console', autospec=True)
 3098     @mock.patch.object(ipmi.IPMISocatConsole, '_exec_stop_console',
 3099                        autospec=True)
 3100     def test_start_console_alloc_port(self, mock_stop, mock_start, mock_info,
 3101                                       mock_alloc):
 3102         self.config(socat_address='2001:dead:beef::1', group='console')
 3103         mock_start.return_value = None
 3104         mock_info.return_value = {'port': None}
 3105         mock_alloc.return_value = 1234
 3106 
 3107         with task_manager.acquire(self.context,
 3108                                   self.node.uuid) as task:
 3109             self.console.start_console(task)
 3110         mock_stop.assert_called_once_with(self.console, mock.ANY)
 3111         mock_start.assert_called_once_with(
 3112             self.console, {'port': 1234},
 3113             console_utils.start_socat_console)
 3114         mock_alloc.assert_called_once_with(mock.ANY, host='2001:dead:beef::1')
 3115 
 3116     @mock.patch.object(ipmi.IPMISocatConsole, '_get_ipmi_cmd', autospec=True)
 3117     @mock.patch.object(console_utils, 'start_socat_console',
 3118                        autospec=True)
 3119     def test__start_console(self, mock_start, mock_ipmi_cmd):
 3120         mock_start.return_value = None
 3121 
 3122         with task_manager.acquire(self.context,
 3123                                   self.node.uuid) as task:
 3124             driver_info = ipmi._parse_driver_info(task.node)
 3125             self.console._start_console(
 3126                 driver_info, console_utils.start_socat_console)
 3127 
 3128         mock_start.assert_called_once_with(self.info['uuid'],
 3129                                            self.info['port'],
 3130                                            mock.ANY)
 3131         mock_ipmi_cmd.assert_called_once_with(self.console,
 3132                                               driver_info, mock.ANY)
 3133 
 3134     @mock.patch.object(console_utils, 'start_socat_console',
 3135                        autospec=True)
 3136     def test__start_console_fail(self, mock_start):
 3137         mock_start.side_effect = exception.ConsoleSubprocessFailed(
 3138             error='error')
 3139 
 3140         with task_manager.acquire(self.context,
 3141                                   self.node.uuid) as task:
 3142             driver_info = ipmi._parse_driver_info(task.node)
 3143             self.assertRaises(exception.ConsoleSubprocessFailed,
 3144                               self.console._start_console,
 3145                               driver_info,
 3146                               console_utils.start_socat_console)
 3147 
 3148         mock_start.assert_called_once_with(self.info['uuid'],
 3149                                            self.info['port'],
 3150                                            mock.ANY)
 3151 
 3152     @mock.patch.object(console_utils, 'start_socat_console',
 3153                        autospec=True)
 3154     def test__start_console_fail_nodir(self, mock_start):
 3155         mock_start.side_effect = exception.ConsoleError()
 3156 
 3157         with task_manager.acquire(self.context,
 3158                                   self.node.uuid) as task:
 3159             driver_info = ipmi._parse_driver_info(task.node)
 3160             self.assertRaises(exception.ConsoleError,
 3161                               self.console._start_console,
 3162                               driver_info,
 3163                               console_utils.start_socat_console)
 3164         mock_start.assert_called_once_with(self.node.uuid, mock.ANY, mock.ANY)
 3165 
 3166     @mock.patch.object(console_utils, 'make_persistent_password_file',
 3167                        autospec=True)
 3168     @mock.patch.object(console_utils, 'start_socat_console',
 3169                        autospec=True)
 3170     def test__start_console_empty_password(self, mock_start, mock_pass):
 3171         driver_info = self.node.driver_info
 3172         del driver_info['ipmi_password']
 3173         self.node.driver_info = driver_info
 3174         self.node.save()
 3175 
 3176         with task_manager.acquire(self.context,
 3177                                   self.node.uuid) as task:
 3178             driver_info = ipmi._parse_driver_info(task.node)
 3179             self.console._start_console(
 3180                 driver_info, console_utils.start_socat_console)
 3181 
 3182         mock_pass.assert_called_once_with(mock.ANY, '\0')
 3183         mock_start.assert_called_once_with(self.info['uuid'],
 3184                                            self.info['port'],
 3185                                            mock.ANY)
 3186 
 3187     @mock.patch.object(ipmi, '_release_allocated_port', autospec=True)
 3188     @mock.patch.object(ipmi.IPMISocatConsole, '_exec_stop_console',
 3189                        autospec=True)
 3190     @mock.patch.object(console_utils, 'stop_socat_console',
 3191                        autospec=True)
 3192     def test_stop_console(self, mock_stop, mock_exec_stop, mock_release):
 3193         mock_stop.return_value = None
 3194 
 3195         with task_manager.acquire(self.context,
 3196                                   self.node.uuid) as task:
 3197             driver_info = ipmi._parse_driver_info(task.node)
 3198             self.console.stop_console(task)
 3199 
 3200         mock_stop.assert_called_once_with(self.info['uuid'])
 3201         mock_exec_stop.assert_called_once_with(self.console,
 3202                                                driver_info)
 3203         mock_release.assert_called_once_with(mock.ANY)
 3204 
 3205     @mock.patch.object(ipmi.IPMISocatConsole, '_exec_stop_console',
 3206                        autospec=True)
 3207     @mock.patch.object(ironic_utils, 'unlink_without_raise',
 3208                        autospec=True)
 3209     @mock.patch.object(console_utils, 'stop_socat_console',
 3210                        autospec=True)
 3211     def test_stop_console_fail(self, mock_stop, mock_unlink, mock_exec_stop):
 3212         mock_stop.side_effect = exception.ConsoleError()
 3213 
 3214         with task_manager.acquire(self.context,
 3215                                   self.node.uuid) as task:
 3216             self.assertRaises(exception.ConsoleError,
 3217                               self.console.stop_console,
 3218                               task)
 3219 
 3220         mock_stop.assert_called_once_with(self.node.uuid)
 3221         mock_unlink.assert_called_once_with(
 3222             ipmi._console_pwfile_path(self.node.uuid))
 3223         self.assertFalse(mock_exec_stop.called)
 3224 
 3225     @mock.patch.object(ipmi, '_exec_ipmitool', autospec=True)
 3226     def test__exec_stop_console(self, mock_exec):
 3227         with task_manager.acquire(self.context,
 3228                                   self.node.uuid) as task:
 3229 
 3230             driver_info = ipmi._parse_driver_info(task.node)
 3231             self.console._exec_stop_console(driver_info)
 3232 
 3233         mock_exec.assert_called_once_with(
 3234             driver_info, 'sol deactivate', check_exit_code=[0, 1])
 3235 
 3236     @mock.patch.object(console_utils, 'get_socat_console_url',
 3237                        autospec=True)
 3238     def test_get_console(self, mock_get_url):
 3239         url = 'tcp://localhost:4201'
 3240         mock_get_url.return_value = url
 3241         expected = {'type': 'socat', 'url': url}
 3242 
 3243         with task_manager.acquire(self.context,
 3244                                   self.node.uuid) as task:
 3245             console_info = self.console.get_console(task)
 3246 
 3247         self.assertEqual(expected, console_info)
 3248         mock_get_url.assert_called_once_with(self.info['port'])