← Back to branch summary

~smoser/ubuntu/vivid/cloud-init/snappy

« back to all changes in this revision

Viewing changes to tests/unittests/test_datasource/test_azure.py

  • Committer: Scott Moser
  • Date: 2015-02-27 20:55:58 UTC
  • mfrom: (355.2.8 vivid)
  • Revision ID: smoser@ubuntu.com-20150227205558-glrwdgxqkaz6zyxa
https://launchpad.net/bugs/1247132
https://launchpad.net/bugs/1355343
https://launchpad.net/bugs/1383794
https://launchpad.net/bugs/1387340
https://launchpad.net/bugs/1404311
https://launchpad.net/bugs/1422388
https://launchpad.net/bugs/1422993
https://launchpad.net/bugs/1423972
https://launchpad.net/bugs/1424900
* Merge with vivid at 0.7.7~bzr1067-0ubuntu1
* New upstream snapshot.
  * fix broken consumption of gzipped user-data (LP: #1424900)
  * functional user-data on Azure again (LP: #1423972)
  * CloudStack: support fetching password from virtual router (LP: #1422388)
* New upstream snapshot.
  * Fix for ascii decode in DataSourceAzure (LP: #1422993).
* New upstream snapshot.
  * support for gpt partitioning, utilized in Azure [Daniel Watkins]
  * fix bug in exception handling in mount_cb.
* New upstream snapshot.
  * move to python3 (LP: #1247132)
  * systemd: run cloud-init before systemd-user-sessions.service
  * Use the GCE short hostname. (LP: #1383794)
  * Enable user-data encoding support for GCE. (LP: #1404311)
  * Update to use a newer and better OMNIBUS_URL
  * Be more tolerant of 'ssh_authorized_keys' types
  * Fix parse_ssh_config failing in ssh_util.py
  * Increase the robustness/configurability of the chef module.
  * retain trailing newline from template files when using
    jinja2 (LP: #1355343)
  * fix broken output handling (LP: #1387340)
  * digital ocean datasource
  * update url in config drive documentation
  * freebsd: enable correct behavior on Ec2.
  * freebsd: Use the proper virtio FreeBSD network interface name.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/unittests/test_datasource/test_azure.py
1
1
from cloudinit import helpers
2
 
from cloudinit.util import load_file
 
2
from cloudinit.util import b64e, decode_binary, load_file
3
3
from cloudinit.sources import DataSourceAzure
4
 
from ..helpers import populate_dir
5
 
 
6
 
import base64
 
4
from ..helpers import TestCase, populate_dir
 
5
 
 
6
try:
 
7
    from unittest import mock
 
8
except ImportError:
 
9
    import mock
 
10
try:
 
11
    from contextlib import ExitStack
 
12
except ImportError:
 
13
    from contextlib2 import ExitStack
 
14
 
7
15
import crypt
8
 
from mocker import MockerTestCase
9
16
import os
10
17
import stat
11
18
import yaml
 
19
import shutil
 
20
import tempfile
 
21
import unittest
12
22
 
13
23
 
14
24
def construct_valid_ovf_env(data=None, pubkeys=None, userdata=None):
40
50
        content += "<%s%s>%s</%s>\n" % (key, attrs, val, key)
41
51
 
42
52
    if userdata:
43
 
        content += "<UserData>%s</UserData>\n" % (base64.b64encode(userdata))
 
53
        content += "<UserData>%s</UserData>\n" % (b64e(userdata))
44
54
 
45
55
    if pubkeys:
46
56
        content += "<SSH><PublicKeys>\n"
66
76
    return content
67
77
 
68
78
 
69
 
class TestAzureDataSource(MockerTestCase):
 
79
class TestAzureDataSource(TestCase):
70
80
 
71
81
    def setUp(self):
72
 
        # makeDir comes from MockerTestCase
73
 
        self.tmp = self.makeDir()
 
82
        super(TestAzureDataSource, self).setUp()
 
83
        self.tmp = tempfile.mkdtemp()
 
84
        self.addCleanup(shutil.rmtree, self.tmp)
74
85
 
75
86
        # patch cloud_dir, so our 'seed_dir' is guaranteed empty
76
87
        self.paths = helpers.Paths({'cloud_dir': self.tmp})
77
88
        self.waagent_d = os.path.join(self.tmp, 'var', 'lib', 'waagent')
78
89
 
79
 
        self.unapply = []
 
90
        self.patches = ExitStack()
 
91
        self.addCleanup(self.patches.close)
 
92
 
80
93
        super(TestAzureDataSource, self).setUp()
81
94
 
82
 
    def tearDown(self):
83
 
        apply_patches([i for i in reversed(self.unapply)])
84
 
        super(TestAzureDataSource, self).tearDown()
85
 
 
86
95
    def apply_patches(self, patches):
87
 
        ret = apply_patches(patches)
88
 
        self.unapply += ret
 
96
        for module, name, new in patches:
 
97
            self.patches.enter_context(mock.patch.object(module, name, new))
89
98
 
90
99
    def _get_ds(self, data):
91
100
 
117
126
        mod = DataSourceAzure
118
127
        mod.BUILTIN_DS_CONFIG['data_dir'] = self.waagent_d
119
128
 
120
 
        self.apply_patches([(mod, 'list_possible_azure_ds_devs', dsdevs)])
121
 
 
122
 
        self.apply_patches([(mod, 'invoke_agent', _invoke_agent),
123
 
                            (mod, 'wait_for_files', _wait_for_files),
124
 
                            (mod, 'pubkeys_from_crt_files',
125
 
                             _pubkeys_from_crt_files),
126
 
                            (mod, 'iid_from_shared_config',
127
 
                             _iid_from_shared_config),
128
 
                            (mod, 'apply_hostname_bounce',
129
 
                             _apply_hostname_bounce), ])
 
129
        self.apply_patches([
 
130
            (mod, 'list_possible_azure_ds_devs', dsdevs),
 
131
            (mod, 'invoke_agent', _invoke_agent),
 
132
            (mod, 'wait_for_files', _wait_for_files),
 
133
            (mod, 'pubkeys_from_crt_files', _pubkeys_from_crt_files),
 
134
            (mod, 'iid_from_shared_config', _iid_from_shared_config),
 
135
            (mod, 'apply_hostname_bounce', _apply_hostname_bounce),
 
136
            ])
130
137
 
131
138
        dsrc = mod.DataSourceAzureNet(
132
139
            data.get('sys_cfg', {}), distro=None, paths=self.paths)
153
160
        ret = dsrc.get_data()
154
161
        self.assertTrue(ret)
155
162
        self.assertTrue(os.path.isdir(self.waagent_d))
156
 
        self.assertEqual(stat.S_IMODE(os.stat(self.waagent_d).st_mode), 0700)
 
163
        self.assertEqual(stat.S_IMODE(os.stat(self.waagent_d).st_mode), 0o700)
157
164
 
158
165
    def test_user_cfg_set_agent_command_plain(self):
159
166
        # set dscfg in via plaintext
174
181
        # set dscfg in via base64 encoded yaml
175
182
        cfg = {'agent_command': "my_command"}
176
183
        odata = {'HostName': "myhost", 'UserName': "myuser",
177
 
                'dscfg': {'text': base64.b64encode(yaml.dump(cfg)),
 
184
                'dscfg': {'text': b64e(yaml.dump(cfg)),
178
185
                          'encoding': 'base64'}}
179
186
        data = {'ovfcontent': construct_valid_ovf_env(data=odata)}
180
187
 
224
231
        self.assertEqual(defuser['passwd'],
225
232
            crypt.crypt(odata['UserPassword'], defuser['passwd'][0:pos]))
226
233
 
 
234
    def test_userdata_plain(self):
 
235
        mydata = "FOOBAR"
 
236
        odata = {'UserData': {'text': mydata, 'encoding': 'plain'}}
 
237
        data = {'ovfcontent': construct_valid_ovf_env(data=odata)}
 
238
 
 
239
        dsrc = self._get_ds(data)
 
240
        ret = dsrc.get_data()
 
241
        self.assertTrue(ret)
 
242
        self.assertEqual(decode_binary(dsrc.userdata_raw), mydata)
 
243
 
227
244
    def test_userdata_found(self):
228
245
        mydata = "FOOBAR"
229
 
        odata = {'UserData': base64.b64encode(mydata)}
 
246
        odata = {'UserData': {'text': b64e(mydata), 'encoding': 'base64'}}
230
247
        data = {'ovfcontent': construct_valid_ovf_env(data=odata)}
231
248
 
232
249
        dsrc = self._get_ds(data)
233
250
        ret = dsrc.get_data()
234
251
        self.assertTrue(ret)
235
 
        self.assertEqual(dsrc.userdata_raw, mydata)
 
252
        self.assertEqual(dsrc.userdata_raw, mydata.encode('utf-8'))
236
253
 
237
254
    def test_no_datasource_expected(self):
238
255
        # no source should be found if no seed_dir and no devs
274
291
                                   'command': 'my-bounce-command',
275
292
                                   'hostname_command': 'my-hostname-command'}}
276
293
        odata = {'HostName': "xhost",
277
 
                'dscfg': {'text': base64.b64encode(yaml.dump(cfg)),
 
294
                'dscfg': {'text': b64e(yaml.dump(cfg)),
278
295
                          'encoding': 'base64'}}
279
296
        data = {'ovfcontent': construct_valid_ovf_env(data=odata)}
280
297
        self._get_ds(data).get_data()
289
306
        # config specifying set_hostname off should not bounce
290
307
        cfg = {'set_hostname': False}
291
308
        odata = {'HostName': "xhost",
292
 
                'dscfg': {'text': base64.b64encode(yaml.dump(cfg)),
 
309
                'dscfg': {'text': b64e(yaml.dump(cfg)),
293
310
                          'encoding': 'base64'}}
294
311
        data = {'ovfcontent': construct_valid_ovf_env(data=odata)}
295
312
        self._get_ds(data).get_data()
318
335
        # Make sure that user can affect disk aliases
319
336
        dscfg = {'disk_aliases': {'ephemeral0': '/dev/sdc'}}
320
337
        odata = {'HostName': "myhost", 'UserName': "myuser",
321
 
                'dscfg': {'text': base64.b64encode(yaml.dump(dscfg)),
 
338
                'dscfg': {'text': b64e(yaml.dump(dscfg)),
322
339
                          'encoding': 'base64'}}
323
340
        usercfg = {'disk_setup': {'/dev/sdc': {'something': '...'},
324
341
                                  'ephemeral0': False}}
340
357
        dsrc = self._get_ds(data)
341
358
        dsrc.get_data()
342
359
 
343
 
        self.assertEqual(userdata, dsrc.userdata_raw)
 
360
        self.assertEqual(userdata.encode('us-ascii'), dsrc.userdata_raw)
344
361
 
345
362
    def test_ovf_env_arrives_in_waagent_dir(self):
346
363
        xml = construct_valid_ovf_env(data={}, userdata="FOODATA")
353
370
        self.assertTrue(os.path.exists(ovf_env_path))
354
371
        self.assertEqual(xml, load_file(ovf_env_path))
355
372
 
 
373
    def test_ovf_can_include_unicode(self):
 
374
        xml = construct_valid_ovf_env(data={})
 
375
        xml = u'\ufeff{0}'.format(xml)
 
376
        dsrc = self._get_ds({'ovfcontent': xml})
 
377
        dsrc.get_data()
 
378
 
356
379
    def test_existing_ovf_same(self):
357
380
        # waagent/SharedConfig left alone if found ovf-env.xml same as cached
358
 
        odata = {'UserData': base64.b64encode("SOMEUSERDATA")}
 
381
        odata = {'UserData': b64e("SOMEUSERDATA")}
359
382
        data = {'ovfcontent': construct_valid_ovf_env(data=odata)}
360
383
 
361
384
        populate_dir(self.waagent_d,
379
402
        # 'get_data' should remove SharedConfig.xml in /var/lib/waagent
380
403
        # if ovf-env.xml differs.
381
404
        cached_ovfenv = construct_valid_ovf_env(
382
 
            {'userdata': base64.b64encode("FOO_USERDATA")})
 
405
            {'userdata': b64e("FOO_USERDATA")})
383
406
        new_ovfenv = construct_valid_ovf_env(
384
 
            {'userdata': base64.b64encode("NEW_USERDATA")})
 
407
            {'userdata': b64e("NEW_USERDATA")})
385
408
 
386
409
        populate_dir(self.waagent_d,
387
410
            {'ovf-env.xml': cached_ovfenv,
391
414
        dsrc = self._get_ds({'ovfcontent': new_ovfenv})
392
415
        ret = dsrc.get_data()
393
416
        self.assertTrue(ret)
394
 
        self.assertEqual(dsrc.userdata_raw, "NEW_USERDATA")
 
417
        self.assertEqual(dsrc.userdata_raw, b"NEW_USERDATA")
395
418
        self.assertTrue(os.path.exists(
396
419
            os.path.join(self.waagent_d, 'otherfile')))
397
420
        self.assertFalse(
402
425
            load_file(os.path.join(self.waagent_d, 'ovf-env.xml')))
403
426
 
404
427
 
405
 
class TestReadAzureOvf(MockerTestCase):
 
428
class TestReadAzureOvf(TestCase):
406
429
    def test_invalid_xml_raises_non_azure_ds(self):
407
430
        invalid_xml = "<foo>" + construct_valid_ovf_env(data={})
408
431
        self.assertRaises(DataSourceAzure.BrokenAzureDataSource,
417
440
            self.assertIn(mypk, cfg['_pubkeys'])
418
441
 
419
442
 
420
 
class TestReadAzureSharedConfig(MockerTestCase):
 
443
class TestReadAzureSharedConfig(unittest.TestCase):
421
444
    def test_valid_content(self):
422
445
        xml = """<?xml version="1.0" encoding="utf-8"?>
423
446
            <SharedConfig>
429
452
            </SharedConfig>"""
430
453
        ret = DataSourceAzure.iid_from_shared_config_content(xml)
431
454
        self.assertEqual("MY_INSTANCE_ID", ret)
432
 
 
433
 
 
434
 
def apply_patches(patches):
435
 
    ret = []
436
 
    for (ref, name, replace) in patches:
437
 
        if replace is None:
438
 
            continue
439
 
        orig = getattr(ref, name)
440
 
        setattr(ref, name, replace)
441
 
        ret.append((ref, name, orig))
442
 
    return ret