← Back to branch summary

~ubuntu-branches/ubuntu/trusty/maas/trusty-security

« back to all changes in this revision

Viewing changes to src/provisioningserver/drivers/hardware/tests/test_mscm.py

  • Committer: Package Import Robot
  • Author(s): Greg Lutostanski
  • Date: 2014-08-29 13:27:34 UTC
  • mto: (61.1.4 trusty-proposed)
  • mto: This revision was merged to the branch mainline in revision 62.
  • Revision ID: package-import@ubuntu.com-20140829132734-d47evihju2etdwcy
Tags: upstream-1.5.4+bzr2294
Import upstream version 1.5.4+bzr2294

Show diffs side-by-side

added added

removed removed

Lines of Context:
src/provisioningserver/drivers/hardware/tests/test_mscm.py
 
1
# Copyright 2014 Canonical Ltd.  This software is licensed under the
 
2
# GNU Affero General Public License version 3 (see the file LICENSE).
 
3
 
 
4
"""Tests for ``provisioningserver.drivers.hardware.mscm``."""
 
5
 
 
6
from __future__ import (
 
7
    absolute_import,
 
8
    print_function,
 
9
    unicode_literals,
 
10
    )
 
11
 
 
12
str = None
 
13
 
 
14
__metaclass__ = type
 
15
__all__ = []
 
16
 
 
17
from random import randint
 
18
import re
 
19
from StringIO import StringIO
 
20
 
 
21
from maastesting.factory import factory
 
22
from maastesting.matchers import MockCalledOnceWith
 
23
from maastesting.testcase import MAASTestCase
 
24
from mock import Mock
 
25
from provisioningserver.drivers.hardware.mscm import (
 
26
    cartridge_mapping,
 
27
    MSCM_CLI_API,
 
28
    power_control_mscm,
 
29
    probe_and_enlist_mscm,
 
30
    )
 
31
import provisioningserver.custom_hardware.utils as utils
 
32
 
 
33
 
 
34
def make_mscm_api():
 
35
    """Make a MSCM_CLI_API object with randomized parameters."""
 
36
    host = factory.make_hostname('mscm')
 
37
    username = factory.make_name('user')
 
38
    password = factory.make_name('password')
 
39
    return MSCM_CLI_API(host, username, password)
 
40
 
 
41
 
 
42
def make_node_id():
 
43
    """Make a node_id."""
 
44
    return 'c%sn%s' % (randint(1, 45), randint(1, 8))
 
45
 
 
46
 
 
47
def make_show_node_list(length=10):
 
48
    """Make a fake return value for discover_nodes."""
 
49
    return re.findall(r'c\d+n\d', ''.join(make_node_id()
 
50
                                          for _ in xrange(length)))
 
51
 
 
52
 
 
53
def make_show_node_macaddr(length=10):
 
54
    """Make a fake return value for get_node_macaddr."""
 
55
    return ''.join((factory.getRandomMACAddress() + ' ')
 
56
                   for _ in xrange(length))
 
57
 
 
58
 
 
59
class TestRunCliCommand(MAASTestCase):
 
60
    """Tests for ``MSCM_CLI_API.run_cli_command``."""
 
61
 
 
62
    def test_returns_output(self):
 
63
        api = make_mscm_api()
 
64
        ssh_mock = self.patch(api, '_ssh')
 
65
        expected = factory.make_name('output')
 
66
        stdout = StringIO(expected)
 
67
        streams = factory.make_streams(stdout=stdout)
 
68
        ssh_mock.exec_command = Mock(return_value=streams)
 
69
        output = api._run_cli_command(factory.make_name('command'))
 
70
        self.assertEqual(expected, output)
 
71
 
 
72
    def test_connects_and_closes_ssh_client(self):
 
73
        api = make_mscm_api()
 
74
        ssh_mock = self.patch(api, '_ssh')
 
75
        ssh_mock.exec_command = Mock(return_value=factory.make_streams())
 
76
        api._run_cli_command(factory.make_name('command'))
 
77
        self.assertThat(
 
78
            ssh_mock.connect,
 
79
            MockCalledOnceWith(
 
80
                api.host, username=api.username, password=api.password))
 
81
        self.assertThat(ssh_mock.close, MockCalledOnceWith())
 
82
 
 
83
    def test_closes_when_exception_raised(self):
 
84
        api = make_mscm_api()
 
85
        ssh_mock = self.patch(api, '_ssh')
 
86
 
 
87
        def fail():
 
88
            raise Exception('fail')
 
89
 
 
90
        ssh_mock.exec_command = Mock(side_effect=fail)
 
91
        command = factory.make_name('command')
 
92
        self.assertRaises(Exception, api._run_cli_command, command)
 
93
        self.assertThat(ssh_mock.close, MockCalledOnceWith())
 
94
 
 
95
 
 
96
class TestDiscoverNodes(MAASTestCase):
 
97
    """Tests for ``MSCM_CLI_API.discover_nodes``."""
 
98
 
 
99
    def test_discover_nodes(self):
 
100
        api = make_mscm_api()
 
101
        ssh_mock = self.patch(api, '_ssh')
 
102
        expected = make_show_node_list()
 
103
        stdout = StringIO(expected)
 
104
        streams = factory.make_streams(stdout=stdout)
 
105
        ssh_mock.exec_command = Mock(return_value=streams)
 
106
        output = api.discover_nodes()
 
107
        self.assertEqual(expected, output)
 
108
 
 
109
 
 
110
class TestNodeMACAddress(MAASTestCase):
 
111
    """Tests for ``MSCM_CLI_API.get_node_macaddr``."""
 
112
 
 
113
    def test_get_node_macaddr(self):
 
114
        api = make_mscm_api()
 
115
        expected = make_show_node_macaddr()
 
116
        cli_mock = self.patch(api, '_run_cli_command')
 
117
        cli_mock.return_value = expected
 
118
        node_id = make_node_id()
 
119
        output = api.get_node_macaddr(node_id)
 
120
        self.assertEqual(re.findall(r':'.join(['[0-9a-f]{2}'] * 6),
 
121
                                    expected), output)
 
122
 
 
123
 
 
124
class TestNodeArch(MAASTestCase):
 
125
    """Tests for ``MSCM_CLI_API.get_node_arch``."""
 
126
 
 
127
    def test_get_node_arch(self):
 
128
        api = make_mscm_api()
 
129
        expected = '\r\n    Product Name: ProLiant Moonshot Cartridge\r\n'
 
130
        cli_mock = self.patch(api, '_run_cli_command')
 
131
        cli_mock.return_value = expected
 
132
        node_id = make_node_id()
 
133
        output = api.get_node_arch(node_id)
 
134
        key = expected.split('Product Name: ')[1].splitlines()[0]
 
135
        self.assertEqual(cartridge_mapping[key], output)
 
136
 
 
137
 
 
138
class TestGetNodePowerStatus(MAASTestCase):
 
139
    """Tests for ``MSCM_CLI_API.get_node_power_status``."""
 
140
 
 
141
    def test_get_node_power_status(self):
 
142
        api = make_mscm_api()
 
143
        expected = '\r\n  Node #1\r\n    Power State: On\r\n'
 
144
        cli_mock = self.patch(api, '_run_cli_command')
 
145
        cli_mock.return_value = expected
 
146
        node_id = make_node_id()
 
147
        output = api.get_node_power_status(node_id)
 
148
        self.assertEqual(expected.split('Power State: ')[1].splitlines()[0],
 
149
                         output)
 
150
 
 
151
 
 
152
class TestPowerAndConfigureNode(MAASTestCase):
 
153
    """Tests for ``MSCM_CLI_API.configure_node_bootonce_pxe,
 
154
    MSCM_CLI_API.power_node_on, and MSCM_CLI_API.power_node_off``.
 
155
    """
 
156
 
 
157
    scenarios = [
 
158
        ('power_node_on()',
 
159
            dict(method='power_node_on')),
 
160
        ('power_node_off()',
 
161
            dict(method='power_node_off')),
 
162
        ('configure_node_bootonce_pxe()',
 
163
            dict(method='configure_node_bootonce_pxe')),
 
164
    ]
 
165
 
 
166
    def test_returns_expected_outout(self):
 
167
        api = make_mscm_api()
 
168
        ssh_mock = self.patch(api, '_ssh')
 
169
        expected = factory.make_name('output')
 
170
        stdout = StringIO(expected)
 
171
        streams = factory.make_streams(stdout=stdout)
 
172
        ssh_mock.exec_command = Mock(return_value=streams)
 
173
        output = getattr(api, self.method)(make_node_id())
 
174
        self.assertEqual(expected, output)
 
175
 
 
176
 
 
177
class TestPowerControlMSCM(MAASTestCase):
 
178
    """Tests for ``power_control_ucsm``."""
 
179
 
 
180
    def test_power_control_mscm_on_on(self):
 
181
        # power_change and power_status are both 'on'
 
182
        host = factory.make_hostname('mscm')
 
183
        username = factory.make_name('user')
 
184
        password = factory.make_name('password')
 
185
        node_id = make_node_id()
 
186
        bootonce_mock = self.patch(MSCM_CLI_API, 'configure_node_bootonce_pxe')
 
187
        power_status_mock = self.patch(MSCM_CLI_API, 'get_node_power_status')
 
188
        power_status_mock.return_value = 'On'
 
189
        power_node_on_mock = self.patch(MSCM_CLI_API, 'power_node_on')
 
190
        power_node_off_mock = self.patch(MSCM_CLI_API, 'power_node_off')
 
191
 
 
192
        power_control_mscm(host, username, password, node_id,
 
193
                           power_change='on')
 
194
        self.assertThat(bootonce_mock, MockCalledOnceWith(node_id))
 
195
        self.assertThat(power_node_off_mock, MockCalledOnceWith(node_id))
 
196
        self.assertThat(power_node_on_mock, MockCalledOnceWith(node_id))
 
197
 
 
198
    def test_power_control_mscm_on_off(self):
 
199
        # power_change is 'on' and power_status is 'off'
 
200
        host = factory.make_hostname('mscm')
 
201
        username = factory.make_name('user')
 
202
        password = factory.make_name('password')
 
203
        node_id = make_node_id()
 
204
        bootonce_mock = self.patch(MSCM_CLI_API, 'configure_node_bootonce_pxe')
 
205
        power_status_mock = self.patch(MSCM_CLI_API, 'get_node_power_status')
 
206
        power_status_mock.return_value = 'Off'
 
207
        power_node_on_mock = self.patch(MSCM_CLI_API, 'power_node_on')
 
208
 
 
209
        power_control_mscm(host, username, password, node_id,
 
210
                           power_change='on')
 
211
        self.assertThat(bootonce_mock, MockCalledOnceWith(node_id))
 
212
        self.assertThat(power_node_on_mock, MockCalledOnceWith(node_id))
 
213
 
 
214
    def test_power_control_mscm_off_on(self):
 
215
        # power_change is 'off' and power_status is 'on'
 
216
        host = factory.make_hostname('mscm')
 
217
        username = factory.make_name('user')
 
218
        password = factory.make_name('password')
 
219
        node_id = make_node_id()
 
220
        power_status_mock = self.patch(MSCM_CLI_API, 'get_node_power_status')
 
221
        power_status_mock.return_value = 'On'
 
222
        power_node_off_mock = self.patch(MSCM_CLI_API, 'power_node_off')
 
223
 
 
224
        power_control_mscm(host, username, password, node_id,
 
225
                           power_change='off')
 
226
        self.assertThat(power_node_off_mock, MockCalledOnceWith(node_id))
 
227
 
 
228
 
 
229
class TestProbeAndEnlistMSCM(MAASTestCase):
 
230
    """Tests for ``probe_and_enlist_mscm``."""
 
231
 
 
232
    def test_probe_and_enlist(self):
 
233
        host = factory.make_hostname('mscm')
 
234
        username = factory.make_name('user')
 
235
        password = factory.make_name('password')
 
236
        node_id = make_node_id()
 
237
        macs = make_show_node_macaddr(4)
 
238
        arch = 'arm64/xgene-uboot'
 
239
        discover_nodes_mock = self.patch(MSCM_CLI_API, 'discover_nodes')
 
240
        discover_nodes_mock.return_value = [node_id]
 
241
        boot_m2_mock = self.patch(MSCM_CLI_API, 'configure_node_boot_m2')
 
242
        node_arch_mock = self.patch(MSCM_CLI_API, 'get_node_arch')
 
243
        node_arch_mock.return_value = arch
 
244
        node_macs_mock = self.patch(MSCM_CLI_API, 'get_node_macaddr')
 
245
        node_macs_mock.return_value = macs
 
246
        create_node_mock = self.patch(utils, 'create_node')
 
247
        probe_and_enlist_mscm(host, username, password)
 
248
        self.assertThat(discover_nodes_mock, MockCalledOnceWith())
 
249
        self.assertThat(boot_m2_mock, MockCalledOnceWith(node_id))
 
250
        self.assertThat(node_arch_mock, MockCalledOnceWith(node_id))
 
251
        self.assertThat(node_macs_mock, MockCalledOnceWith(node_id))
 
252
        params = {
 
253
            'power_address': host,
 
254
            'power_user': username,
 
255
            'power_pass': password,
 
256
            'node_id': node_id,
 
257
        }
 
258
        self.assertThat(create_node_mock,
 
259
                        MockCalledOnceWith(macs, arch, 'mscm', params))