~smoser/ubuntu/quantal/cloud-init/sru

« back to all changes in this revision

Viewing changes to .pc/lp-1100545-allow-config-drive-from-cdrom.patch/tests/unittests/test_datasource/test_configdrive.py

  • Committer: Scott Moser
  • Date: 2013-01-18 15:28:08 UTC
  • Revision ID: smoser@ubuntu.com-20130118152808-jy5uq9pc79t82r85
Tags: 0.7.0-0ubuntu2.3~ppa0
debian/patches/lp-1100545-allow-config-drive-from-cdrom.patch:
in config-drive data to be provided from a CD-ROM (LP: #1100545)

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
from copy import copy
 
2
import json
 
3
import os
 
4
import os.path
 
5
import shutil
 
6
import tempfile
 
7
from unittest import TestCase
 
8
 
 
9
from cloudinit import helpers
 
10
from cloudinit import settings
 
11
from cloudinit.sources import DataSourceConfigDrive as ds
 
12
from cloudinit import util
 
13
 
 
14
 
 
15
PUBKEY = u'ssh-rsa AAAAB3NzaC1....sIkJhq8wdX+4I3A4cYbYP ubuntu@server-460\n'
 
16
EC2_META = {
 
17
    'ami-id': 'ami-00000001',
 
18
    'ami-launch-index': 0,
 
19
    'ami-manifest-path': 'FIXME',
 
20
    'block-device-mapping': {
 
21
        'ami': 'sda1',
 
22
        'ephemeral0': 'sda2',
 
23
        'root': '/dev/sda1',
 
24
        'swap': 'sda3'},
 
25
    'hostname': 'sm-foo-test.novalocal',
 
26
    'instance-action': 'none',
 
27
    'instance-id': 'i-00000001',
 
28
    'instance-type': 'm1.tiny',
 
29
    'local-hostname': 'sm-foo-test.novalocal',
 
30
    'local-ipv4': None,
 
31
    'placement': {'availability-zone': 'nova'},
 
32
    'public-hostname': 'sm-foo-test.novalocal',
 
33
    'public-ipv4': '',
 
34
    'public-keys': {'0': {'openssh-key': PUBKEY}},
 
35
    'reservation-id': 'r-iru5qm4m',
 
36
    'security-groups': ['default']
 
37
}
 
38
USER_DATA = '#!/bin/sh\necho This is user data\n'
 
39
OSTACK_META = {
 
40
    'availability_zone': 'nova',
 
41
    'files': [{'content_path': '/content/0000', 'path': '/etc/foo.cfg'},
 
42
              {'content_path': '/content/0001', 'path': '/etc/bar/bar.cfg'}],
 
43
    'hostname': 'sm-foo-test.novalocal',
 
44
    'meta': {'dsmode': 'local', 'my-meta': 'my-value'},
 
45
    'name': 'sm-foo-test',
 
46
    'public_keys': {'mykey': PUBKEY},
 
47
    'uuid': 'b0fa911b-69d4-4476-bbe2-1c92bff6535c'}
 
48
 
 
49
CONTENT_0 = 'This is contents of /etc/foo.cfg\n'
 
50
CONTENT_1 = '# this is /etc/bar/bar.cfg\n'
 
51
 
 
52
CFG_DRIVE_FILES_V2 = {
 
53
  'ec2/2009-04-04/meta-data.json': json.dumps(EC2_META),
 
54
  'ec2/2009-04-04/user-data': USER_DATA,
 
55
  'ec2/latest/meta-data.json': json.dumps(EC2_META),
 
56
  'ec2/latest/user-data': USER_DATA,
 
57
  'openstack/2012-08-10/meta_data.json': json.dumps(OSTACK_META),
 
58
  'openstack/2012-08-10/user_data': USER_DATA,
 
59
  'openstack/content/0000': CONTENT_0,
 
60
  'openstack/content/0001': CONTENT_1,
 
61
  'openstack/latest/meta_data.json': json.dumps(OSTACK_META),
 
62
  'openstack/latest/user_data': USER_DATA}
 
63
 
 
64
 
 
65
class TestConfigDriveDataSource(TestCase):
 
66
 
 
67
    def setUp(self):
 
68
        super(TestConfigDriveDataSource, self).setUp()
 
69
        self.tmp = tempfile.mkdtemp()
 
70
 
 
71
    def tearDown(self):
 
72
        try:
 
73
            shutil.rmtree(self.tmp)
 
74
        except OSError:
 
75
            pass
 
76
 
 
77
    def test_dir_valid(self):
 
78
        """Verify a dir is read as such."""
 
79
 
 
80
        populate_dir(self.tmp, CFG_DRIVE_FILES_V2)
 
81
 
 
82
        found = ds.read_config_drive_dir(self.tmp)
 
83
 
 
84
        expected_md = copy(OSTACK_META)
 
85
        expected_md['instance-id'] = expected_md['uuid']
 
86
 
 
87
        self.assertEqual(USER_DATA, found['userdata'])
 
88
        self.assertEqual(expected_md, found['metadata'])
 
89
        self.assertEqual(found['files']['/etc/foo.cfg'], CONTENT_0)
 
90
        self.assertEqual(found['files']['/etc/bar/bar.cfg'], CONTENT_1)
 
91
 
 
92
    def test_seed_dir_valid_extra(self):
 
93
        """Verify extra files do not affect datasource validity."""
 
94
 
 
95
        data = copy(CFG_DRIVE_FILES_V2)
 
96
        data["myfoofile.txt"] = "myfoocontent"
 
97
        data["openstack/latest/random-file.txt"] = "random-content"
 
98
 
 
99
        populate_dir(self.tmp, data)
 
100
 
 
101
        found = ds.read_config_drive_dir(self.tmp)
 
102
 
 
103
        expected_md = copy(OSTACK_META)
 
104
        expected_md['instance-id'] = expected_md['uuid']
 
105
 
 
106
        self.assertEqual(expected_md, found['metadata'])
 
107
 
 
108
    def test_seed_dir_bad_json_metadata(self):
 
109
        """Verify that bad json in metadata raises BrokenConfigDriveDir."""
 
110
        data = copy(CFG_DRIVE_FILES_V2)
 
111
 
 
112
        data["openstack/2012-08-10/meta_data.json"] = "non-json garbage {}"
 
113
        data["openstack/latest/meta_data.json"] = "non-json garbage {}"
 
114
 
 
115
        populate_dir(self.tmp, data)
 
116
 
 
117
        self.assertRaises(ds.BrokenConfigDriveDir,
 
118
                          ds.read_config_drive_dir, self.tmp)
 
119
 
 
120
    def test_seed_dir_no_configdrive(self):
 
121
        """Verify that no metadata raises NonConfigDriveDir."""
 
122
 
 
123
        my_d = os.path.join(self.tmp, "non-configdrive")
 
124
        data = copy(CFG_DRIVE_FILES_V2)
 
125
        data["myfoofile.txt"] = "myfoocontent"
 
126
        data["openstack/latest/random-file.txt"] = "random-content"
 
127
        data["content/foo"] = "foocontent"
 
128
 
 
129
        self.assertRaises(ds.NonConfigDriveDir,
 
130
                          ds.read_config_drive_dir, my_d)
 
131
 
 
132
    def test_seed_dir_missing(self):
 
133
        """Verify that missing seed_dir raises NonConfigDriveDir."""
 
134
        my_d = os.path.join(self.tmp, "nonexistantdirectory")
 
135
        self.assertRaises(ds.NonConfigDriveDir,
 
136
                          ds.read_config_drive_dir, my_d)
 
137
 
 
138
    def test_find_candidates(self):
 
139
        devs_with_answers = {
 
140
            "TYPE=vfat": [],
 
141
            "TYPE=iso9660": ["/dev/vdb"],
 
142
            "LABEL=config-2": ["/dev/vdb"],
 
143
        }
 
144
 
 
145
        def my_devs_with(criteria):
 
146
            return devs_with_answers[criteria]
 
147
 
 
148
        try:
 
149
            orig_find_devs_with = util.find_devs_with
 
150
            util.find_devs_with = my_devs_with
 
151
 
 
152
            self.assertEqual(["/dev/vdb"], ds.find_candidate_devs())
 
153
 
 
154
            # add a vfat item
 
155
            # zdd reverse sorts after vdb, but config-2 label is preferred
 
156
            devs_with_answers['TYPE=vfat'] = ["/dev/zdd"]
 
157
            self.assertEqual(["/dev/vdb", "/dev/zdd"],
 
158
                             ds.find_candidate_devs())
 
159
 
 
160
            # verify that partitions are not considered
 
161
            devs_with_answers = {"TYPE=vfat": ["/dev/sda1"],
 
162
                "TYPE=iso9660": [], "LABEL=config-2": ["/dev/vdb3"]}
 
163
            self.assertEqual([], ds.find_candidate_devs())
 
164
 
 
165
        finally:
 
166
            util.find_devs_with = orig_find_devs_with
 
167
 
 
168
    def test_pubkeys_v2(self):
 
169
        """Verify that public-keys work in config-drive-v2."""
 
170
        populate_dir(self.tmp, CFG_DRIVE_FILES_V2)
 
171
        myds = cfg_ds_from_dir(self.tmp)
 
172
        self.assertEqual(myds.get_public_ssh_keys(),
 
173
           [OSTACK_META['public_keys']['mykey']])
 
174
 
 
175
 
 
176
def cfg_ds_from_dir(seed_d):
 
177
    found = ds.read_config_drive_dir(seed_d)
 
178
    cfg_ds = ds.DataSourceConfigDrive(settings.CFG_BUILTIN, None,
 
179
                                      helpers.Paths({}))
 
180
    populate_ds_from_read_config(cfg_ds, seed_d, found)
 
181
    return cfg_ds
 
182
 
 
183
 
 
184
def populate_ds_from_read_config(cfg_ds, source, results):
 
185
    """Patch the DataSourceConfigDrive from the results of
 
186
    read_config_drive_dir hopefully in line with what it would have
 
187
    if cfg_ds.get_data had been successfully called"""
 
188
    cfg_ds.source = source
 
189
    cfg_ds.metadata = results.get('metadata')
 
190
    cfg_ds.ec2_metadata = results.get('ec2-metadata')
 
191
    cfg_ds.userdata_raw = results.get('userdata')
 
192
    cfg_ds.version = results.get('cfgdrive_ver')
 
193
 
 
194
 
 
195
def populate_dir(seed_dir, files):
 
196
    for (name, content) in files.iteritems():
 
197
        path = os.path.join(seed_dir, name)
 
198
        dirname = os.path.dirname(path)
 
199
        if not os.path.isdir(dirname):
 
200
            os.makedirs(dirname)
 
201
        with open(path, "w") as fp:
 
202
            fp.write(content)
 
203
            fp.close()
 
204
 
 
205
# vi: ts=4 expandtab