summaryrefslogtreecommitdiff
path: root/tests/unittests/test_datasource/test_altcloud.py
blob: 3119bfac0c01a2766f9553eba431c4e7608a3400 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
# Copyright (C) 2009-2010 Canonical Ltd.
# Copyright (C) 2012 Hewlett-Packard Development Company, L.P.
# Copyright (C) 2012 Yahoo! Inc.
#
# Author: Joe VLcek <JVLcek@RedHat.com>
#
# This file is part of cloud-init. See LICENSE file for license information.

'''
This test file exercises the code in sources DataSourceAltCloud.py
'''

import os
import shutil
import tempfile

from cloudinit import helpers
from cloudinit import util

from cloudinit.tests.helpers import CiTestCase, mock

import cloudinit.sources.DataSourceAltCloud as dsac

OS_UNAME_ORIG = getattr(os, 'uname')


def _write_user_data_files(mount_dir, value):
    '''
    Populate the deltacloud_user_data_file the user_data_file
    which would be populated with user data.
    '''
    deltacloud_user_data_file = mount_dir + '/deltacloud-user-data.txt'
    user_data_file = mount_dir + '/user-data.txt'

    udfile = open(deltacloud_user_data_file, 'w')
    udfile.write(value)
    udfile.close()
    os.chmod(deltacloud_user_data_file, 0o664)

    udfile = open(user_data_file, 'w')
    udfile.write(value)
    udfile.close()
    os.chmod(user_data_file, 0o664)


def _remove_user_data_files(mount_dir,
                            dc_file=True,
                            non_dc_file=True):
    '''
    Remove the test files: deltacloud_user_data_file and
    user_data_file
    '''
    deltacloud_user_data_file = mount_dir + '/deltacloud-user-data.txt'
    user_data_file = mount_dir + '/user-data.txt'

    # Ignore any failures removeing files that are already gone.
    if dc_file:
        try:
            os.remove(deltacloud_user_data_file)
        except OSError:
            pass

    if non_dc_file:
        try:
            os.remove(user_data_file)
        except OSError:
            pass


def _dmi_data(expected):
    '''
    Spoof the data received over DMI
    '''
    def _data(key):
        return expected

    return _data


class TestGetCloudType(CiTestCase):
    '''Test to exercise method: DataSourceAltCloud.get_cloud_type()'''

    with_logs = True

    def setUp(self):
        '''Set up.'''
        super(TestGetCloudType, self).setUp()
        self.tmp = self.tmp_dir()
        self.paths = helpers.Paths({'cloud_dir': self.tmp})
        self.dmi_data = util.read_dmi_data
        # We have a different code path for arm to deal with LP1243287
        # We have to switch arch to x86_64 to avoid test failure
        force_arch('x86_64')

    def tearDown(self):
        # Reset
        util.read_dmi_data = self.dmi_data
        force_arch()

    def test_cloud_info_file_ioerror(self):
        """Return UNKNOWN when /etc/sysconfig/cloud-info exists but errors."""
        self.assertEqual('/etc/sysconfig/cloud-info', dsac.CLOUD_INFO_FILE)
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        # Attempting to read the directory generates IOError
        with mock.patch.object(dsac, 'CLOUD_INFO_FILE', self.tmp):
            self.assertEqual('UNKNOWN', dsrc.get_cloud_type())
        self.assertIn(
            "[Errno 21] Is a directory: '%s'" % self.tmp,
            self.logs.getvalue())

    def test_cloud_info_file(self):
        """Return uppercase stripped content from /etc/sysconfig/cloud-info."""
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        cloud_info = self.tmp_path('cloud-info', dir=self.tmp)
        util.write_file(cloud_info, ' OverRiDdeN CloudType ')
        # Attempting to read the directory generates IOError
        with mock.patch.object(dsac, 'CLOUD_INFO_FILE', cloud_info):
            self.assertEqual('OVERRIDDEN CLOUDTYPE', dsrc.get_cloud_type())

    def test_rhev(self):
        '''
        Test method get_cloud_type() for RHEVm systems.
        Forcing read_dmi_data return to match a RHEVm system: RHEV Hypervisor
        '''
        util.read_dmi_data = _dmi_data('RHEV')
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        self.assertEqual('RHEV', dsrc.get_cloud_type())

    def test_vsphere(self):
        '''
        Test method get_cloud_type() for vSphere systems.
        Forcing read_dmi_data return to match a vSphere system: RHEV Hypervisor
        '''
        util.read_dmi_data = _dmi_data('VMware Virtual Platform')
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        self.assertEqual('VSPHERE', dsrc.get_cloud_type())

    def test_unknown(self):
        '''
        Test method get_cloud_type() for unknown systems.
        Forcing read_dmi_data return to match an unrecognized return.
        '''
        util.read_dmi_data = _dmi_data('Unrecognized Platform')
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        self.assertEqual('UNKNOWN', dsrc.get_cloud_type())


class TestGetDataCloudInfoFile(CiTestCase):
    '''
    Test to exercise method: DataSourceAltCloud.get_data()
    With a contrived CLOUD_INFO_FILE
    '''
    def setUp(self):
        '''Set up.'''
        self.tmp = self.tmp_dir()
        self.paths = helpers.Paths(
            {'cloud_dir': self.tmp, 'run_dir': self.tmp})
        self.cloud_info_file = self.tmp_path('cloud-info', dir=self.tmp)

    def test_rhev(self):
        '''Success Test module get_data() forcing RHEV.'''

        util.write_file(self.cloud_info_file, 'RHEV')
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        dsrc.user_data_rhevm = lambda: True
        with mock.patch.object(dsac, 'CLOUD_INFO_FILE', self.cloud_info_file):
            self.assertEqual(True, dsrc.get_data())
        self.assertEqual('altcloud', dsrc.cloud_name)
        self.assertEqual('altcloud', dsrc.platform_type)
        self.assertEqual('rhev (/dev/fd0)', dsrc.subplatform)

    def test_vsphere(self):
        '''Success Test module get_data() forcing VSPHERE.'''

        util.write_file(self.cloud_info_file, 'VSPHERE')
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        dsrc.user_data_vsphere = lambda: True
        with mock.patch.object(dsac, 'CLOUD_INFO_FILE', self.cloud_info_file):
            self.assertEqual(True, dsrc.get_data())
        self.assertEqual('altcloud', dsrc.cloud_name)
        self.assertEqual('altcloud', dsrc.platform_type)
        self.assertEqual('vsphere (unknown)', dsrc.subplatform)

    def test_fail_rhev(self):
        '''Failure Test module get_data() forcing RHEV.'''

        util.write_file(self.cloud_info_file, 'RHEV')
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        dsrc.user_data_rhevm = lambda: False
        with mock.patch.object(dsac, 'CLOUD_INFO_FILE', self.cloud_info_file):
            self.assertEqual(False, dsrc.get_data())

    def test_fail_vsphere(self):
        '''Failure Test module get_data() forcing VSPHERE.'''

        util.write_file(self.cloud_info_file, 'VSPHERE')
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        dsrc.user_data_vsphere = lambda: False
        with mock.patch.object(dsac, 'CLOUD_INFO_FILE', self.cloud_info_file):
            self.assertEqual(False, dsrc.get_data())

    def test_unrecognized(self):
        '''Failure Test module get_data() forcing unrecognized.'''

        util.write_file(self.cloud_info_file, 'unrecognized')
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        with mock.patch.object(dsac, 'CLOUD_INFO_FILE', self.cloud_info_file):
            self.assertEqual(False, dsrc.get_data())


class TestGetDataNoCloudInfoFile(CiTestCase):
    '''
    Test to exercise method: DataSourceAltCloud.get_data()
    Without a CLOUD_INFO_FILE
    '''
    def setUp(self):
        '''Set up.'''
        self.tmp = self.tmp_dir()
        self.paths = helpers.Paths(
            {'cloud_dir': self.tmp, 'run_dir': self.tmp})
        self.dmi_data = util.read_dmi_data
        dsac.CLOUD_INFO_FILE = \
            'no such file'
        # We have a different code path for arm to deal with LP1243287
        # We have to switch arch to x86_64 to avoid test failure
        force_arch('x86_64')

    def tearDown(self):
        # Reset
        dsac.CLOUD_INFO_FILE = \
            '/etc/sysconfig/cloud-info'
        util.read_dmi_data = self.dmi_data
        # Return back to original arch
        force_arch()

    def test_rhev_no_cloud_file(self):
        '''Test No cloud info file module get_data() forcing RHEV.'''

        util.read_dmi_data = _dmi_data('RHEV Hypervisor')
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        dsrc.user_data_rhevm = lambda: True
        self.assertEqual(True, dsrc.get_data())

    def test_vsphere_no_cloud_file(self):
        '''Test No cloud info file module get_data() forcing VSPHERE.'''

        util.read_dmi_data = _dmi_data('VMware Virtual Platform')
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        dsrc.user_data_vsphere = lambda: True
        self.assertEqual(True, dsrc.get_data())

    def test_failure_no_cloud_file(self):
        '''Test No cloud info file module get_data() forcing unrecognized.'''

        util.read_dmi_data = _dmi_data('Unrecognized Platform')
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        self.assertEqual(False, dsrc.get_data())


class TestUserDataRhevm(CiTestCase):
    '''
    Test to exercise method: DataSourceAltCloud.user_data_rhevm()
    '''
    def setUp(self):
        '''Set up.'''
        self.paths = helpers.Paths({'cloud_dir': '/tmp'})
        self.mount_dir = self.tmp_dir()
        _write_user_data_files(self.mount_dir, 'test user data')
        self.add_patch(
            'cloudinit.sources.DataSourceAltCloud.modprobe_floppy',
            'm_modprobe_floppy', return_value=None)
        self.add_patch(
            'cloudinit.sources.DataSourceAltCloud.util.udevadm_settle',
            'm_udevadm_settle', return_value=('', ''))
        self.add_patch(
            'cloudinit.sources.DataSourceAltCloud.util.mount_cb',
            'm_mount_cb')

    def test_mount_cb_fails(self):
        '''Test user_data_rhevm() where mount_cb fails.'''

        self.m_mount_cb.side_effect = util.MountFailedError("Failed Mount")
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        self.assertEqual(False, dsrc.user_data_rhevm())

    def test_modprobe_fails(self):
        '''Test user_data_rhevm() where modprobe fails.'''

        self.m_modprobe_floppy.side_effect = util.ProcessExecutionError(
            "Failed modprobe")
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        self.assertEqual(False, dsrc.user_data_rhevm())

    def test_no_modprobe_cmd(self):
        '''Test user_data_rhevm() with no modprobe command.'''

        self.m_modprobe_floppy.side_effect = util.ProcessExecutionError(
            "No such file or dir")
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        self.assertEqual(False, dsrc.user_data_rhevm())

    def test_udevadm_fails(self):
        '''Test user_data_rhevm() where udevadm fails.'''

        self.m_udevadm_settle.side_effect = util.ProcessExecutionError(
            "Failed settle.")
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        self.assertEqual(False, dsrc.user_data_rhevm())

    def test_no_udevadm_cmd(self):
        '''Test user_data_rhevm() with no udevadm command.'''

        self.m_udevadm_settle.side_effect = OSError("No such file or dir")
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        self.assertEqual(False, dsrc.user_data_rhevm())


class TestUserDataVsphere(CiTestCase):
    '''
    Test to exercise method: DataSourceAltCloud.user_data_vsphere()
    '''
    def setUp(self):
        '''Set up.'''
        self.tmp = self.tmp_dir()
        self.paths = helpers.Paths({'cloud_dir': self.tmp})
        self.mount_dir = tempfile.mkdtemp()

        _write_user_data_files(self.mount_dir, 'test user data')

    def tearDown(self):
        # Reset

        _remove_user_data_files(self.mount_dir)

        # Attempt to remove the temp dir ignoring errors
        try:
            shutil.rmtree(self.mount_dir)
        except OSError:
            pass

        dsac.CLOUD_INFO_FILE = \
            '/etc/sysconfig/cloud-info'

    @mock.patch("cloudinit.sources.DataSourceAltCloud.util.find_devs_with")
    @mock.patch("cloudinit.sources.DataSourceAltCloud.util.mount_cb")
    def test_user_data_vsphere_no_cdrom(self, m_mount_cb, m_find_devs_with):
        '''Test user_data_vsphere() where mount_cb fails.'''

        m_mount_cb.return_value = []
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        self.assertEqual(False, dsrc.user_data_vsphere())
        self.assertEqual(0, m_mount_cb.call_count)

    @mock.patch("cloudinit.sources.DataSourceAltCloud.util.find_devs_with")
    @mock.patch("cloudinit.sources.DataSourceAltCloud.util.mount_cb")
    def test_user_data_vsphere_mcb_fail(self, m_mount_cb, m_find_devs_with):
        '''Test user_data_vsphere() where mount_cb fails.'''

        m_find_devs_with.return_value = ["/dev/mock/cdrom"]
        m_mount_cb.side_effect = util.MountFailedError("Unable To mount")
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        self.assertEqual(False, dsrc.user_data_vsphere())
        self.assertEqual(1, m_find_devs_with.call_count)
        self.assertEqual(1, m_mount_cb.call_count)

    @mock.patch("cloudinit.sources.DataSourceAltCloud.util.find_devs_with")
    @mock.patch("cloudinit.sources.DataSourceAltCloud.util.mount_cb")
    def test_user_data_vsphere_success(self, m_mount_cb, m_find_devs_with):
        """Test user_data_vsphere() where successful."""
        m_find_devs_with.return_value = ["/dev/mock/cdrom"]
        m_mount_cb.return_value = 'raw userdata from cdrom'
        dsrc = dsac.DataSourceAltCloud({}, None, self.paths)
        cloud_info = self.tmp_path('cloud-info', dir=self.tmp)
        util.write_file(cloud_info, 'VSPHERE')
        self.assertEqual(True, dsrc.user_data_vsphere())
        m_find_devs_with.assert_called_once_with('LABEL=CDROM')
        m_mount_cb.assert_called_once_with(
            '/dev/mock/cdrom', dsac.read_user_data_callback)
        with mock.patch.object(dsrc, 'get_cloud_type', return_value='VSPHERE'):
            self.assertEqual('vsphere (/dev/mock/cdrom)', dsrc.subplatform)


class TestReadUserDataCallback(CiTestCase):
    '''
    Test to exercise method: DataSourceAltCloud.read_user_data_callback()
    '''
    def setUp(self):
        '''Set up.'''
        self.paths = helpers.Paths({'cloud_dir': '/tmp'})
        self.mount_dir = tempfile.mkdtemp()

        _write_user_data_files(self.mount_dir, 'test user data')

    def tearDown(self):
        # Reset

        _remove_user_data_files(self.mount_dir)

        # Attempt to remove the temp dir ignoring errors
        try:
            shutil.rmtree(self.mount_dir)
        except OSError:
            pass

    def test_callback_both(self):
        '''Test read_user_data_callback() with both files.'''

        self.assertEqual('test user data',
                         dsac.read_user_data_callback(self.mount_dir))

    def test_callback_dc(self):
        '''Test read_user_data_callback() with only DC file.'''

        _remove_user_data_files(self.mount_dir,
                                dc_file=False,
                                non_dc_file=True)

        self.assertEqual('test user data',
                         dsac.read_user_data_callback(self.mount_dir))

    def test_callback_non_dc(self):
        '''Test read_user_data_callback() with only non-DC file.'''

        _remove_user_data_files(self.mount_dir,
                                dc_file=True,
                                non_dc_file=False)

        self.assertEqual('test user data',
                         dsac.read_user_data_callback(self.mount_dir))

    def test_callback_none(self):
        '''Test read_user_data_callback() no files are found.'''

        _remove_user_data_files(self.mount_dir)
        self.assertIsNone(dsac.read_user_data_callback(self.mount_dir))


def force_arch(arch=None):

    def _os_uname():
        return ('LINUX', 'NODENAME', 'RELEASE', 'VERSION', arch)

    if arch:
        setattr(os, 'uname', _os_uname)
    elif arch is None:
        setattr(os, 'uname', OS_UNAME_ORIG)

# vi: ts=4 expandtab