← Back to branch summary

~ubuntu-branches/ubuntu/vivid/ironic/vivid-updates

« back to all changes in this revision

Viewing changes to ironic/tests/drivers/amt/test_power.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2015-03-30 11:14:57 UTC
  • mfrom: (1.2.6)
  • Revision ID: package-import@ubuntu.com-20150330111457-kr4ju3guf22m4vbz
Tags: 2015.1~b3-0ubuntu1
https://launchpad.net/bugs/1358820
https://launchpad.net/bugs/1429962
* New upstream release.
  + d/control: 
    - Align with upstream dependencies.
    - Add dh-python to build-dependencies.
    - Add psmisc as a dependency. (LP: #1358820)
  + d/p/fix-requirements.patch: Rediffed.
  + d/ironic-conductor.init.in: Fixed typos in LSB headers,
    thanks to JJ Asghar. (LP: #1429962)

Show diffs side-by-side

added added

removed removed

Lines of Context:
ironic/tests/drivers/amt/test_power.py
 
1
#
 
2
# Licensed under the Apache License, Version 2.0 (the "License"); you may
 
3
# not use this file except in compliance with the License. You may obtain
 
4
# a copy of the License at
 
5
#
 
6
#      http://www.apache.org/licenses/LICENSE-2.0
 
7
#
 
8
# Unless required by applicable law or agreed to in writing, software
 
9
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
10
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
11
# License for the specific language governing permissions and limitations
 
12
# under the License.
 
13
 
 
14
"""
 
15
Test class for AMT ManagementInterface
 
16
"""
 
17
 
 
18
import mock
 
19
from oslo_config import cfg
 
20
 
 
21
from ironic.common import boot_devices
 
22
from ironic.common import exception
 
23
from ironic.common import states
 
24
from ironic.conductor import task_manager
 
25
from ironic.drivers.modules.amt import common as amt_common
 
26
from ironic.drivers.modules.amt import management as amt_mgmt
 
27
from ironic.drivers.modules.amt import power as amt_power
 
28
from ironic.drivers.modules.amt import resource_uris
 
29
from ironic.tests.conductor import utils as mgr_utils
 
30
from ironic.tests.db import base as db_base
 
31
from ironic.tests.db import utils as db_utils
 
32
from ironic.tests.drivers.drac import utils as test_utils
 
33
from ironic.tests.objects import utils as obj_utils
 
34
 
 
35
INFO_DICT = db_utils.get_test_amt_info()
 
36
CONF = cfg.CONF
 
37
 
 
38
 
 
39
class AMTPowerInteralMethodsTestCase(db_base.DbTestCase):
 
40
 
 
41
    def setUp(self):
 
42
        super(AMTPowerInteralMethodsTestCase, self).setUp()
 
43
        mgr_utils.mock_the_extension_manager(driver='fake_amt')
 
44
        self.info = INFO_DICT
 
45
        self.node = obj_utils.create_test_node(self.context,
 
46
                                               driver='fake_amt',
 
47
                                               driver_info=self.info)
 
48
        CONF.set_override('max_attempts', 2, 'amt')
 
49
        CONF.set_override('action_wait', 0, 'amt')
 
50
 
 
51
    @mock.patch.object(amt_common, 'get_wsman_client')
 
52
    def test__set_power_state(self, mock_client_pywsman):
 
53
        namespace = resource_uris.CIM_PowerManagementService
 
54
        mock_client = mock_client_pywsman.return_value
 
55
        amt_power._set_power_state(self.node, states.POWER_ON)
 
56
        mock_client.wsman_invoke.assert_called_once_with(mock.ANY,
 
57
            namespace, 'RequestPowerStateChange', mock.ANY)
 
58
 
 
59
    @mock.patch.object(amt_common, 'get_wsman_client')
 
60
    def test__set_power_state_fail(self, mock_client_pywsman):
 
61
        mock_client = mock_client_pywsman.return_value
 
62
        mock_client.wsman_invoke.side_effect = exception.AMTFailure('x')
 
63
        self.assertRaises(exception.AMTFailure,
 
64
                          amt_power._set_power_state,
 
65
                          self.node, states.POWER_ON)
 
66
 
 
67
    @mock.patch.object(amt_common, 'get_wsman_client')
 
68
    def test__power_status(self, mock_gwc):
 
69
        namespace = resource_uris.CIM_AssociatedPowerManagementService
 
70
        result_xml = test_utils.build_soap_xml([{'PowerState':
 
71
                                                 '2'}],
 
72
                                               namespace)
 
73
        mock_doc = test_utils.mock_wsman_root(result_xml)
 
74
        mock_client = mock_gwc.return_value
 
75
        mock_client.wsman_get.return_value = mock_doc
 
76
        self.assertEqual(
 
77
            states.POWER_ON, amt_power._power_status(self.node))
 
78
 
 
79
        result_xml = test_utils.build_soap_xml([{'PowerState':
 
80
                                                 '8'}],
 
81
                                               namespace)
 
82
        mock_doc = test_utils.mock_wsman_root(result_xml)
 
83
        mock_client = mock_gwc.return_value
 
84
        mock_client.wsman_get.return_value = mock_doc
 
85
        self.assertEqual(
 
86
            states.POWER_OFF, amt_power._power_status(self.node))
 
87
 
 
88
        result_xml = test_utils.build_soap_xml([{'PowerState':
 
89
                                                 '4'}],
 
90
                                               namespace)
 
91
        mock_doc = test_utils.mock_wsman_root(result_xml)
 
92
        mock_client = mock_gwc.return_value
 
93
        mock_client.wsman_get.return_value = mock_doc
 
94
        self.assertEqual(
 
95
            states.ERROR, amt_power._power_status(self.node))
 
96
 
 
97
    @mock.patch.object(amt_common, 'get_wsman_client')
 
98
    def test__power_status_fail(self, mock_gwc):
 
99
        mock_client = mock_gwc.return_value
 
100
        mock_client.wsman_get.side_effect = exception.AMTFailure('x')
 
101
        self.assertRaises(exception.AMTFailure,
 
102
                          amt_power._power_status,
 
103
                          self.node)
 
104
 
 
105
    @mock.patch.object(amt_mgmt.AMTManagement, 'ensure_next_boot_device')
 
106
    @mock.patch.object(amt_power, '_power_status')
 
107
    @mock.patch.object(amt_power, '_set_power_state')
 
108
    def test__set_and_wait_power_on_with_boot_device(self, mock_sps,
 
109
                                                     mock_ps, mock_snbd):
 
110
        target_state = states.POWER_ON
 
111
        boot_device = boot_devices.PXE
 
112
        mock_ps.side_effect = [states.POWER_OFF, states.POWER_ON]
 
113
        mock_snbd.return_value = None
 
114
        with task_manager.acquire(self.context, self.node.uuid,
 
115
                                  shared=True) as task:
 
116
            task.node.driver_internal_info['amt_boot_device'] = boot_device
 
117
            self.assertEqual(states.POWER_ON,
 
118
                             amt_power._set_and_wait(task, target_state))
 
119
            mock_snbd.assert_called_with(task.node, boot_devices.PXE)
 
120
            mock_sps.assert_called_once_with(task.node, states.POWER_ON)
 
121
            mock_ps.assert_called_with(task.node)
 
122
 
 
123
    @mock.patch.object(amt_power, '_power_status')
 
124
    @mock.patch.object(amt_power, '_set_power_state')
 
125
    def test__set_and_wait_power_on_without_boot_device(self, mock_sps,
 
126
                                                        mock_ps):
 
127
        target_state = states.POWER_ON
 
128
        mock_ps.side_effect = [states.POWER_OFF, states.POWER_ON]
 
129
        with task_manager.acquire(self.context, self.node.uuid,
 
130
                                  shared=True) as task:
 
131
            self.assertEqual(states.POWER_ON,
 
132
                             amt_power._set_and_wait(task, target_state))
 
133
            mock_sps.assert_called_once_with(task.node, states.POWER_ON)
 
134
            mock_ps.assert_called_with(task.node)
 
135
 
 
136
        boot_device = boot_devices.DISK
 
137
        self.node.driver_internal_info['amt_boot_device'] = boot_device
 
138
        mock_ps.side_effect = [states.POWER_OFF, states.POWER_ON]
 
139
        with task_manager.acquire(self.context, self.node.uuid,
 
140
                                  shared=True) as task:
 
141
            self.assertEqual(states.POWER_ON,
 
142
                             amt_power._set_and_wait(task, target_state))
 
143
            mock_sps.assert_called_with(task.node, states.POWER_ON)
 
144
            mock_ps.assert_called_with(task.node)
 
145
 
 
146
    def test__set_and_wait_wrong_target_state(self):
 
147
        target_state = 'fake-state'
 
148
        with task_manager.acquire(self.context, self.node.uuid,
 
149
                                  shared=True) as task:
 
150
            self.assertRaises(exception.InvalidParameterValue,
 
151
                              amt_power._set_and_wait, task, target_state)
 
152
 
 
153
    @mock.patch.object(amt_power, '_power_status')
 
154
    @mock.patch.object(amt_power, '_set_power_state')
 
155
    def test__set_and_wait_exceed_iterations(self, mock_sps,
 
156
                                             mock_ps):
 
157
        target_state = states.POWER_ON
 
158
        mock_ps.side_effect = [states.POWER_OFF, states.POWER_OFF,
 
159
                               states.POWER_OFF]
 
160
        mock_sps.return_value = exception.AMTFailure('x')
 
161
        with task_manager.acquire(self.context, self.node.uuid,
 
162
                                  shared=True) as task:
 
163
            self.assertRaises(exception.PowerStateFailure,
 
164
                              amt_power._set_and_wait, task, target_state)
 
165
            mock_sps.assert_called_with(task.node, states.POWER_ON)
 
166
            mock_ps.assert_called_with(task.node)
 
167
            self.assertEqual(3, mock_ps.call_count)
 
168
 
 
169
    @mock.patch.object(amt_power, '_power_status')
 
170
    def test__set_and_wait_already_target_state(self, mock_ps):
 
171
        target_state = states.POWER_ON
 
172
        mock_ps.side_effect = [states.POWER_ON]
 
173
        with task_manager.acquire(self.context, self.node.uuid,
 
174
                                  shared=True) as task:
 
175
            self.assertEqual(states.POWER_ON,
 
176
                         amt_power._set_and_wait(task, target_state))
 
177
            mock_ps.assert_called_with(task.node)
 
178
 
 
179
    @mock.patch.object(amt_power, '_power_status')
 
180
    @mock.patch.object(amt_power, '_set_power_state')
 
181
    def test__set_and_wait_power_off(self, mock_sps, mock_ps):
 
182
        target_state = states.POWER_OFF
 
183
        mock_ps.side_effect = [states.POWER_ON, states.POWER_OFF]
 
184
        with task_manager.acquire(self.context, self.node.uuid,
 
185
                                  shared=True) as task:
 
186
            self.assertEqual(states.POWER_OFF,
 
187
                         amt_power._set_and_wait(task, target_state))
 
188
            mock_sps.assert_called_once_with(task.node, states.POWER_OFF)
 
189
            mock_ps.assert_called_with(task.node)
 
190
 
 
191
 
 
192
class AMTPowerTestCase(db_base.DbTestCase):
 
193
 
 
194
    def setUp(self):
 
195
        super(AMTPowerTestCase, self).setUp()
 
196
        mgr_utils.mock_the_extension_manager(driver='fake_amt')
 
197
        self.info = INFO_DICT
 
198
        self.node = obj_utils.create_test_node(self.context,
 
199
                                               driver='fake_amt',
 
200
                                               driver_info=self.info)
 
201
 
 
202
    def test_get_properties(self):
 
203
        expected = amt_common.COMMON_PROPERTIES
 
204
        with task_manager.acquire(self.context, self.node.uuid,
 
205
                                  shared=True) as task:
 
206
            self.assertEqual(expected, task.driver.get_properties())
 
207
 
 
208
    @mock.patch.object(amt_common, 'parse_driver_info')
 
209
    def test_validate(self, mock_drvinfo):
 
210
        with task_manager.acquire(self.context, self.node.uuid,
 
211
                                  shared=True) as task:
 
212
            task.driver.power.validate(task)
 
213
            mock_drvinfo.assert_called_once_with(task.node)
 
214
 
 
215
    @mock.patch.object(amt_common, 'parse_driver_info')
 
216
    def test_validate_fail(self, mock_drvinfo):
 
217
        with task_manager.acquire(self.context, self.node.uuid,
 
218
                                  shared=True) as task:
 
219
            mock_drvinfo.side_effect = exception.InvalidParameterValue('x')
 
220
            self.assertRaises(exception.InvalidParameterValue,
 
221
                              task.driver.power.validate,
 
222
                              task)
 
223
 
 
224
    @mock.patch.object(amt_power, '_power_status')
 
225
    def test_get_power_state(self, mock_ps):
 
226
        with task_manager.acquire(self.context, self.node.uuid,
 
227
                                  shared=True) as task:
 
228
            mock_ps.return_value = states.POWER_ON
 
229
            self.assertEqual(states.POWER_ON,
 
230
                             task.driver.power.get_power_state(task))
 
231
            mock_ps.assert_called_once_with(task.node)
 
232
 
 
233
    @mock.patch.object(amt_power, '_set_and_wait')
 
234
    def test_set_power_state(self, mock_saw):
 
235
        with task_manager.acquire(self.context, self.node.uuid,
 
236
                                  shared=False) as task:
 
237
            pstate = states.POWER_ON
 
238
            mock_saw.return_value = states.POWER_ON
 
239
            task.driver.power.set_power_state(task, pstate)
 
240
            mock_saw.assert_called_once_with(task, pstate)
 
241
 
 
242
    @mock.patch.object(amt_power, '_set_and_wait')
 
243
    def test_set_power_state_fail(self, mock_saw):
 
244
        with task_manager.acquire(self.context, self.node.uuid,
 
245
                                  shared=False) as task:
 
246
            pstate = states.POWER_ON
 
247
            mock_saw.side_effect = exception.PowerStateFailure('x')
 
248
            self.assertRaises(exception.PowerStateFailure,
 
249
                              task.driver.power.set_power_state,
 
250
                              task, pstate)
 
251
            mock_saw.assert_called_once_with(task, pstate)
 
252
 
 
253
    @mock.patch.object(amt_power, '_set_and_wait')
 
254
    def test_reboot(self, mock_saw):
 
255
        with task_manager.acquire(self.context, self.node.uuid,
 
256
                                  shared=False) as task:
 
257
            task.driver.power.reboot(task)
 
258
            calls = [mock.call(task, states.POWER_OFF),
 
259
                     mock.call(task, states.POWER_ON)]
 
260
            mock_saw.has_calls(calls)