← Back to branch summary

~ubuntu-branches/ubuntu/trusty/cloud-init/trusty

« back to all changes in this revision

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

  • Committer: Scott Moser
  • Author(s): Ben Howard
  • Date: 2015-11-17 17:02:24 UTC
  • Revision ID: smoser@ubuntu.com-20151117170224-dd9fexd1lnl2e7d1
Tags: 0.7.5-0ubuntu1.15
https://launchpad.net/bugs/1506244
https://launchpad.net/bugs/1506187
* Microsoft Azure:
  - d/patches/lp-1506244-azure-ssh-key-values.patch: AZURE: Add support
    and preference for fabric provided public SSH public key values over
    fingerprints (LP: #1506244).
  - use stable VM instance ID over SharedConfig.xml (LP: #1506187):
    - d/patches/lp-1506187-azure_use_unique_vm_id.patch: use DMI data for
      the stable VM instance ID
    - d/cloud-init.preinst: migrate existing instances to stable VM instance
      ID on upgrade from prior versions of cloud-init.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/unittests/test_datasource/test_azure.py
58
58
 
59
59
    if pubkeys:
60
60
        content += "<SSH><PublicKeys>\n"
61
 
        for fp, path in pubkeys:
 
61
        for fp, path, value in pubkeys:
62
62
            content += " <PublicKey>"
63
 
            content += ("<Fingerprint>%s</Fingerprint><Path>%s</Path>" %
64
 
                        (fp, path))
 
63
            if fp and path:
 
64
                content += ("<Fingerprint>%s</Fingerprint><Path>%s</Path>" %
 
65
                            (fp, path))
 
66
            if value:
 
67
                content += "<Value>%s</Value>" % value
65
68
            content += "</PublicKey>\n"
66
69
        content += "</PublicKeys></SSH>"
67
70
    content += """
117
120
            data['pubkey_files'] = flist
118
121
            return ["pubkey_from: %s" % f for f in flist]
119
122
 
120
 
        def _iid_from_shared_config(path):
121
 
            data['iid_from_shared_cfg'] = path
 
123
        def _get_instance_id():
122
124
            return 'i-my-azure-id'
123
125
 
124
126
        if data.get('ovfcontent') is not None:
125
127
            populate_dir(os.path.join(self.paths.seed_dir, "azure"),
126
128
                         {'ovf-env.xml': data['ovfcontent']})
127
129
 
128
 
 
129
130
        mod = DataSourceAzure
130
131
        mod.BUILTIN_DS_CONFIG = OVERRIDE_BUILTIN_DS_CONFIG
131
132
        mod.BUILTIN_DS_CONFIG['data_dir'] = self.waagent_d
136
137
                            (mod, 'wait_for_files', _wait_for_files),
137
138
                            (mod, 'pubkeys_from_crt_files',
138
139
                             _pubkeys_from_crt_files),
139
 
                            (mod, 'iid_from_shared_config',
140
 
                             _iid_from_shared_config)])
 
140
                            (mod, 'get_instance_id', _get_instance_id)])
141
141
 
142
142
        dsrc = mod.DataSourceAzureNet(
143
143
            data.get('sys_cfg', {}), distro=None, paths=self.paths)
173
173
    def xml_notequals(self, oxml, nxml):
174
174
        try:
175
175
            self.xml_equals(oxml, nxml)
176
 
        except AssertionError as e:
 
176
        except AssertionError:
177
177
            return
178
178
        raise AssertionError("XML is the same")
179
179
 
206
206
        yaml_cfg = "{agent_command: my_command}\n"
207
207
        cfg = yaml.safe_load(yaml_cfg)
208
208
        odata = {'HostName': "myhost", 'UserName': "myuser",
209
 
                'dscfg': {'text': yaml_cfg, 'encoding': 'plain'}}
 
209
                 'dscfg': {'text': yaml_cfg, 'encoding': 'plain'}}
210
210
        data = {'ovfcontent': construct_valid_ovf_env(data=odata)}
211
211
 
212
212
        dsrc = self._get_ds(data)
218
218
        # set dscfg in via base64 encoded yaml
219
219
        cfg = {'agent_command': "my_command"}
220
220
        odata = {'HostName': "myhost", 'UserName': "myuser",
221
 
                'dscfg': {'text': base64.b64encode(yaml.dump(cfg)),
222
 
                          'encoding': 'base64'}}
 
221
                 'dscfg': {'text': base64.b64encode(yaml.dump(cfg)),
 
222
                           'encoding': 'base64'}}
223
223
        data = {'ovfcontent': construct_valid_ovf_env(data=odata)}
224
224
 
225
225
        dsrc = self._get_ds(data)
266
266
        # should equal that after the '$'
267
267
        pos = defuser['passwd'].rfind("$") + 1
268
268
        self.assertEqual(defuser['passwd'],
269
 
            crypt.crypt(odata['UserPassword'], defuser['passwd'][0:pos]))
 
269
                         crypt.crypt(odata['UserPassword'],
 
270
                                     defuser['passwd'][0:pos]))
270
271
 
271
272
    def test_userdata_found(self):
272
273
        mydata = "FOOBAR"
279
280
        self.assertEqual(dsrc.userdata_raw, mydata)
280
281
 
281
282
    def test_no_datasource_expected(self):
282
 
        #no source should be found if no seed_dir and no devs
 
283
        # no source should be found if no seed_dir and no devs
283
284
        data = {}
284
285
        dsrc = self._get_ds({})
285
286
        ret = dsrc.get_data()
286
287
        self.assertFalse(ret)
287
288
        self.assertFalse('agent_invoked' in data)
288
289
 
289
 
    def test_cfg_has_pubkeys(self):
290
 
        odata = {'HostName': "myhost", 'UserName': "myuser"}
291
 
        mypklist = [{'fingerprint': 'fp1', 'path': 'path1'}]
292
 
        pubkeys = [(x['fingerprint'], x['path']) for x in mypklist]
293
 
        data = {'ovfcontent': construct_valid_ovf_env(data=odata,
294
 
                                                      pubkeys=pubkeys)}
295
 
 
296
 
        dsrc = self._get_ds(data)
297
 
        ret = dsrc.get_data()
298
 
        self.assertTrue(ret)
299
 
        for mypk in mypklist:
300
 
            self.assertIn(mypk, dsrc.cfg['_pubkeys'])
 
290
    def test_cfg_has_pubkeys_fingerprint(self):
 
291
        odata = {'HostName': "myhost", 'UserName': "myuser"}
 
292
        mypklist = [{'fingerprint': 'fp1', 'path': 'path1', 'value': ''}]
 
293
        pubkeys = [(x['fingerprint'], x['path'], x['value']) for x in mypklist]
 
294
        data = {'ovfcontent': construct_valid_ovf_env(data=odata,
 
295
                                                      pubkeys=pubkeys)}
 
296
 
 
297
        dsrc = self._get_ds(data)
 
298
        ret = dsrc.get_data()
 
299
        self.assertTrue(ret)
 
300
        for mypk in mypklist:
 
301
            self.assertIn(mypk, dsrc.cfg['_pubkeys'])
 
302
            self.assertIn('pubkey_from', dsrc.metadata['public-keys'][-1])
 
303
 
 
304
    def test_cfg_has_pubkeys_value(self):
 
305
        # make sure that provided key is used over fingerprint
 
306
        odata = {'HostName': "myhost", 'UserName': "myuser"}
 
307
        mypklist = [{'fingerprint': 'fp1', 'path': 'path1', 'value': 'value1'}]
 
308
        pubkeys = [(x['fingerprint'], x['path'], x['value']) for x in mypklist]
 
309
        data = {'ovfcontent': construct_valid_ovf_env(data=odata,
 
310
                                                      pubkeys=pubkeys)}
 
311
 
 
312
        dsrc = self._get_ds(data)
 
313
        ret = dsrc.get_data()
 
314
        self.assertTrue(ret)
 
315
 
 
316
        for mypk in mypklist:
 
317
            self.assertIn(mypk, dsrc.cfg['_pubkeys'])
 
318
            self.assertIn(mypk['value'], dsrc.metadata['public-keys'])
 
319
 
 
320
    def test_cfg_has_no_fingerprint_has_value(self):
 
321
        # test value is used when fingerprint not provided
 
322
        odata = {'HostName': "myhost", 'UserName': "myuser"}
 
323
        mypklist = [{'fingerprint': None, 'path': 'path1', 'value': 'value1'}]
 
324
        pubkeys = [(x['fingerprint'], x['path'], x['value']) for x in mypklist]
 
325
        data = {'ovfcontent': construct_valid_ovf_env(data=odata,
 
326
                                                      pubkeys=pubkeys)}
 
327
 
 
328
        dsrc = self._get_ds(data)
 
329
        ret = dsrc.get_data()
 
330
        self.assertTrue(ret)
 
331
 
 
332
        for mypk in mypklist:
 
333
            self.assertIn(mypk['value'], dsrc.metadata['public-keys'])
301
334
 
302
335
    def test_disabled_bounce(self):
303
336
        pass
324
357
        # Make sure that user can affect disk aliases
325
358
        dscfg = {'disk_aliases': {'ephemeral0': '/dev/sdc'}}
326
359
        odata = {'HostName': "myhost", 'UserName': "myuser",
327
 
                'dscfg': {'text': base64.b64encode(yaml.dump(dscfg)),
328
 
                          'encoding': 'base64'}}
 
360
                 'dscfg': {'text': base64.b64encode(yaml.dump(dscfg)),
 
361
                           'encoding': 'base64'}}
329
362
        usercfg = {'disk_setup': {'/dev/sdc': {'something': '...'},
330
363
                                  'ephemeral0': False}}
331
364
        userdata = '#cloud-config' + yaml.dump(usercfg) + "\n"
384
417
        self.assertTrue(os.path.exists(ovf_env_path))
385
418
        self.xml_equals(xml, load_file(ovf_env_path))
386
419
 
387
 
    def test_existing_ovf_same(self):
388
 
        # waagent/SharedConfig left alone if found ovf-env.xml same as cached
389
 
        odata = {'UserData': base64.b64encode("SOMEUSERDATA")}
390
 
        data = {'ovfcontent': construct_valid_ovf_env(data=odata)}
391
 
 
392
 
        populate_dir(self.waagent_d,
393
 
            {'ovf-env.xml': data['ovfcontent'],
394
 
             'otherfile': 'otherfile-content',
395
 
             'SharedConfig.xml': 'mysharedconfig'})
396
 
 
397
 
        dsrc = self._get_ds(data)
398
 
        ret = dsrc.get_data()
399
 
        self.assertTrue(ret)
400
 
        self.assertTrue(os.path.exists(
401
 
            os.path.join(self.waagent_d, 'ovf-env.xml')))
402
 
        self.assertTrue(os.path.exists(
403
 
            os.path.join(self.waagent_d, 'otherfile')))
404
 
        self.assertTrue(os.path.exists(
405
 
            os.path.join(self.waagent_d, 'SharedConfig.xml')))
406
 
 
407
 
    def test_existing_ovf_diff(self):
408
 
        # waagent/SharedConfig must be removed if ovfenv is found elsewhere
409
 
 
410
 
        # 'get_data' should remove SharedConfig.xml in /var/lib/waagent
411
 
        # if ovf-env.xml differs.
412
 
        cached_ovfenv = construct_valid_ovf_env(
413
 
            {'userdata': base64.b64encode("FOO_USERDATA")})
414
 
        new_ovfenv = construct_valid_ovf_env(
415
 
            {'userdata': base64.b64encode("NEW_USERDATA")})
416
 
 
417
 
        populate_dir(self.waagent_d,
418
 
            {'ovf-env.xml': cached_ovfenv,
419
 
             'SharedConfig.xml': "mysharedconfigxml",
420
 
             'otherfile': 'otherfilecontent'})
421
 
 
422
 
        dsrc = self._get_ds({'ovfcontent': new_ovfenv})
423
 
        ret = dsrc.get_data()
424
 
        self.assertTrue(ret)
425
 
        self.assertEqual(dsrc.userdata_raw, "NEW_USERDATA")
426
 
        self.assertTrue(os.path.exists(
427
 
            os.path.join(self.waagent_d, 'otherfile')))
428
 
        self.assertFalse(
429
 
            os.path.exists(os.path.join(self.waagent_d, 'SharedConfig.xml')))
430
 
        self.assertTrue(
431
 
            os.path.exists(os.path.join(self.waagent_d, 'ovf-env.xml')))
432
 
        new_xml = load_file(os.path.join(self.waagent_d, 'ovf-env.xml'))
433
 
        self.xml_equals(new_ovfenv, new_xml)
434
420
 
435
421
class TestReadAzureOvf(MockerTestCase):
436
422
    def test_invalid_xml_raises_non_azure_ds(self):
437
423
        invalid_xml = "<foo>" + construct_valid_ovf_env(data={})
438
424
        self.assertRaises(DataSourceAzure.BrokenAzureDataSource,
439
 
            DataSourceAzure.read_azure_ovf, invalid_xml)
 
425
                          DataSourceAzure.read_azure_ovf, invalid_xml)
440
426
 
441
427
    def test_load_with_pubkeys(self):
442
 
        mypklist = [{'fingerprint': 'fp1', 'path': 'path1'}]
443
 
        pubkeys = [(x['fingerprint'], x['path']) for x in mypklist]
 
428
        mypklist = [{'fingerprint': 'fp1', 'path': 'path1', 'value': ''}]
 
429
        pubkeys = [(x['fingerprint'], x['path'], x['value']) for x in mypklist]
444
430
        content = construct_valid_ovf_env(pubkeys=pubkeys)
445
 
        (_md, _ud, cfg) = DataSourceAzure.read_azure_ovf(content)
 
431
        (_, _, cfg) = DataSourceAzure.read_azure_ovf(content)
446
432
        for mypk in mypklist:
447
433
            self.assertIn(mypk, cfg['_pubkeys'])
448
434
 
449
435
 
450
 
class TestReadAzureSharedConfig(MockerTestCase):
451
 
    def test_valid_content(self):
452
 
        xml = """<?xml version="1.0" encoding="utf-8"?>
453
 
            <SharedConfig>
454
 
             <Deployment name="MY_INSTANCE_ID">
455
 
              <Service name="myservice"/>
456
 
              <ServiceInstance name="INSTANCE_ID.0" guid="{abcd-uuid}" />
457
 
             </Deployment>
458
 
            <Incarnation number="1"/>
459
 
            </SharedConfig>"""
460
 
        ret = DataSourceAzure.iid_from_shared_config_content(xml)
461
 
        self.assertEqual("MY_INSTANCE_ID", ret)
462
 
 
463
 
 
464
436
def apply_patches(patches):
465
437
    ret = []
466
438
    for (ref, name, replace) in patches: