# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import eventlet
import ironic_discoverd
from ironic_discoverd import client
import mock
from ironic.common import driver_factory
from ironic.common import exception
from ironic.common import keystone
from ironic.common import states
from ironic.conductor import task_manager
from ironic.drivers.modules import discoverd
from ironic.tests.conductor import utils as mgr_utils
from ironic.tests.db import base as db_base
from ironic.tests.objects import utils as obj_utils
[docs]class DisabledTestCase(db_base.DbTestCase):
[docs] def setUp(self):
super(DisabledTestCase, self).setUp()
def _do_mock(self):
# NOTE(dtantsur): fake driver always has inspection, using another one
mgr_utils.mock_the_extension_manager("pxe_ssh")
self.driver = driver_factory.get_driver("pxe_ssh")
[docs] def test_disabled(self):
self.config(enabled=False, group='discoverd')
self._do_mock()
self.assertIsNone(self.driver.inspect)
# NOTE(dtantsur): it's expected that fake_discoverd fails to load
# in this case
self.assertRaises(exception.DriverLoadError,
mgr_utils.mock_the_extension_manager,
"fake_discoverd")
[docs] def test_enabled(self):
self.config(enabled=True, group='discoverd')
self._do_mock()
self.assertIsNotNone(self.driver.inspect)
@mock.patch.object(discoverd, 'ironic_discoverd', None)
[docs] def test_init_discoverd_not_imported(self):
self.assertRaises(exception.DriverLoadError,
discoverd.DiscoverdInspect)
@mock.patch.object(ironic_discoverd, '__version_info__', (1, 0, 0))
[docs] def test_init_ok(self):
self.config(enabled=True, group='discoverd')
discoverd.DiscoverdInspect()
@mock.patch.object(ironic_discoverd, '__version_info__', (0, 2, 2))
[docs] def test_init_old_version(self):
self.config(enabled=True, group='discoverd')
self.assertRaises(exception.DriverLoadError,
discoverd.DiscoverdInspect)
[docs]class BaseTestCase(db_base.DbTestCase):
[docs] def setUp(self):
super(BaseTestCase, self).setUp()
self.config(enabled=True, group='discoverd')
mgr_utils.mock_the_extension_manager("fake_discoverd")
self.driver = driver_factory.get_driver("fake_discoverd")
self.node = obj_utils.get_test_node(self.context)
self.task = mock.Mock(spec=task_manager.TaskManager)
self.task.context = mock.Mock()
self.task.shared = False
self.task.node = self.node
self.task.driver = self.driver
[docs]class CommonFunctionsTestCase(BaseTestCase):
[docs] def test_validate_ok(self):
self.driver.inspect.validate(self.task)
[docs] def test_get_properties(self):
res = self.driver.inspect.get_properties()
self.assertEqual({}, res)
[docs] def test_create_if_enabled(self):
res = discoverd.DiscoverdInspect.create_if_enabled('driver')
self.assertIsInstance(res, discoverd.DiscoverdInspect)
@mock.patch.object(discoverd.LOG, 'info')
[docs] def test_create_if_enabled_disabled(self, warn_mock):
self.config(enabled=False, group='discoverd')
res = discoverd.DiscoverdInspect.create_if_enabled('driver')
self.assertIsNone(res)
self.assertTrue(warn_mock.called)
@mock.patch.object(eventlet, 'spawn_n', lambda f, *a, **kw: f(*a, **kw))
@mock.patch.object(client, 'introspect')
[docs]class InspectHardwareTestCase(BaseTestCase):
[docs] def test_ok(self, mock_introspect):
self.assertEqual(states.INSPECTING,
self.driver.inspect.inspect_hardware(self.task))
mock_introspect.assert_called_once_with(
self.node.uuid,
auth_token=self.task.context.auth_token)
[docs] def test_url(self, mock_introspect):
self.config(service_url='meow', group='discoverd')
self.assertEqual(states.INSPECTING,
self.driver.inspect.inspect_hardware(self.task))
mock_introspect.assert_called_once_with(
self.node.uuid,
auth_token=self.task.context.auth_token,
base_url='meow')
@mock.patch.object(task_manager, 'acquire')
[docs] def test_error(self, mock_acquire, mock_introspect):
mock_introspect.side_effect = RuntimeError('boom')
self.driver.inspect.inspect_hardware(self.task)
mock_introspect.assert_called_once_with(
self.node.uuid,
auth_token=self.task.context.auth_token)
task = mock_acquire.return_value.__enter__.return_value
self.assertIn('boom', task.node.last_error)
task.process_event.assert_called_once_with('fail')
@mock.patch.object(keystone, 'get_admin_auth_token', lambda: 'the token')
@mock.patch.object(client, 'get_status')
[docs]class CheckStatusTestCase(BaseTestCase):
[docs] def setUp(self):
super(CheckStatusTestCase, self).setUp()
self.node.provision_state = states.INSPECTING
[docs] def test_not_inspecting(self, mock_get):
self.node.provision_state = states.MANAGEABLE
discoverd._check_status(self.task)
self.assertFalse(mock_get.called)
[docs] def test_not_discoverd(self, mock_get):
self.task.driver.inspect = object()
discoverd._check_status(self.task)
self.assertFalse(mock_get.called)
[docs] def test_not_finished(self, mock_get):
mock_get.return_value = {}
discoverd._check_status(self.task)
mock_get.assert_called_once_with(self.node.uuid,
auth_token='the token')
self.assertFalse(self.task.process_event.called)
[docs] def test_exception_ignored(self, mock_get):
mock_get.side_effect = RuntimeError('boom')
discoverd._check_status(self.task)
mock_get.assert_called_once_with(self.node.uuid,
auth_token='the token')
self.assertFalse(self.task.process_event.called)
[docs] def test_status_ok(self, mock_get):
mock_get.return_value = {'finished': True}
discoverd._check_status(self.task)
mock_get.assert_called_once_with(self.node.uuid,
auth_token='the token')
self.task.process_event.assert_called_once_with('done')
[docs] def test_status_error(self, mock_get):
mock_get.return_value = {'error': 'boom'}
discoverd._check_status(self.task)
mock_get.assert_called_once_with(self.node.uuid,
auth_token='the token')
self.task.process_event.assert_called_once_with('fail')
self.assertIn('boom', self.node.last_error)
[docs] def test_service_url(self, mock_get):
self.config(service_url='meow', group='discoverd')
mock_get.return_value = {'finished': True}
discoverd._check_status(self.task)
mock_get.assert_called_once_with(self.node.uuid,
auth_token='the token',
base_url='meow')
self.task.process_event.assert_called_once_with('done')
@mock.patch.object(eventlet.greenthread, 'spawn_n',
lambda f, *a, **kw: f(*a, **kw))
@mock.patch.object(ironic_discoverd, '__version_info__', (1, 0, 0))
@mock.patch.object(task_manager, 'acquire', autospec=True)
@mock.patch.object(discoverd, '_check_status', autospec=True)
[docs]class PeriodicTaskTestCase(BaseTestCase):
[docs] def test_ok(self, mock_check, mock_acquire):
mgr = mock.Mock(spec=['iter_nodes'])
mgr.iter_nodes.return_value = [('1', 'd1'), ('2', 'd2')]
tasks = [mock.sentinel.task1, mock.sentinel.task2]
mock_acquire.side_effect = (
mock.MagicMock(__enter__=mock.Mock(return_value=task))
for task in tasks
)
discoverd.DiscoverdInspect()._periodic_check_result(
mgr, mock.sentinel.context)
mock_check.assert_any_call(tasks[0])
mock_check.assert_any_call(tasks[1])
self.assertEqual(2, mock_acquire.call_count)
[docs] def test_node_locked(self, mock_check, mock_acquire):
mgr = mock.Mock(spec=['iter_nodes'])
mgr.iter_nodes.return_value = [('1', 'd1'), ('2', 'd2')]
mock_acquire.side_effect = exception.NodeLocked("boom")
discoverd.DiscoverdInspect()._periodic_check_result(
mgr, mock.sentinel.context)
self.assertFalse(mock_check.called)
self.assertEqual(2, mock_acquire.call_count)