~smoser/ubuntu/raring/cloud-init/lp-123315-1231490

« back to all changes in this revision

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

  • Committer: Package Import Robot
  • Author(s): Scott Moser
  • Date: 2013-03-07 17:33:59 UTC
  • mfrom: (247.1.5 raring)
  • Revision ID: package-import@ubuntu.com-20130307173359-oljs0x66uqxb35l8
Tags: 0.7.2~bzr795-0ubuntu1
* New upstream snapshot.
  * documentation on write-files module (LP: #1111205)
  * support for specifying package versions in package installs
  * DataSourceNoCloud: allow specifyin user-data and meta-data in
    the datasource config (LP: #1115833)
  * work around bug in upstart for now (1124384)
  * support resizing btrfs fileystems
  * parse ssh keys more correctly (LP: #1136343)
  * upstart/cloud-init-nonet.conf: handle sigterm gracefully (LP: #1015223)
  * support growing partitions (LP: #1136936)
  * use --force-unsafe-io for dpkg installations to improve speed
    This is sane as it happens on instance initialization.
  * more powerful and user-suppliable cloud-config merge mechanisms
    (LP: #1023179)

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
from cloudinit import helpers
 
2
from cloudinit.sources import DataSourceNoCloud
 
3
from cloudinit import util
 
4
from tests.unittests.helpers import populate_dir
 
5
 
 
6
from mocker import MockerTestCase
 
7
import os
 
8
import yaml
 
9
 
 
10
 
 
11
class TestNoCloudDataSource(MockerTestCase):
 
12
 
 
13
    def setUp(self):
 
14
        self.tmp = self.makeDir()
 
15
        self.paths = helpers.Paths({'cloud_dir': self.tmp})
 
16
 
 
17
        self.cmdline = "root=TESTCMDLINE"
 
18
 
 
19
        self.unapply = []
 
20
        self.apply_patches([(util, 'get_cmdline', self._getcmdline)])
 
21
        super(TestNoCloudDataSource, self).setUp()
 
22
 
 
23
    def tearDown(self):
 
24
        apply_patches([i for i in reversed(self.unapply)])
 
25
        super(TestNoCloudDataSource, self).setUp()
 
26
 
 
27
    def apply_patches(self, patches):
 
28
        ret = apply_patches(patches)
 
29
        self.unapply += ret
 
30
 
 
31
    def _getcmdline(self):
 
32
        return self.cmdline
 
33
 
 
34
    def test_nocloud_seed_dir(self):
 
35
        md = {'instance-id': 'IID', 'dsmode': 'local'}
 
36
        ud = "USER_DATA_HERE"
 
37
        populate_dir(os.path.join(self.paths.seed_dir, "nocloud"),
 
38
                     {'user-data': ud, 'meta-data': yaml.safe_dump(md)})
 
39
 
 
40
        sys_cfg = {
 
41
            'datasource': {'NoCloud': {'fs_label': None}}
 
42
        }
 
43
 
 
44
        ds = DataSourceNoCloud.DataSourceNoCloud
 
45
 
 
46
        dsrc = ds(sys_cfg=sys_cfg, distro=None, paths=self.paths)
 
47
        ret = dsrc.get_data()
 
48
        self.assertEqual(dsrc.userdata_raw, ud)
 
49
        self.assertEqual(dsrc.metadata, md)
 
50
        self.assertTrue(ret)
 
51
 
 
52
    def test_fs_label(self):
 
53
        #find_devs_with should not be called ff fs_label is None
 
54
        ds = DataSourceNoCloud.DataSourceNoCloud
 
55
 
 
56
        class PsuedoException(Exception):
 
57
            pass
 
58
 
 
59
        def my_find_devs_with(*args, **kwargs):
 
60
            _f = (args, kwargs)
 
61
            raise PsuedoException
 
62
 
 
63
        self.apply_patches([(util, 'find_devs_with', my_find_devs_with)])
 
64
 
 
65
        # by default, NoCloud should search for filesystems by label
 
66
        sys_cfg = {'datasource': {'NoCloud': {}}}
 
67
        dsrc = ds(sys_cfg=sys_cfg, distro=None, paths=self.paths)
 
68
        self.assertRaises(PsuedoException, dsrc.get_data)
 
69
 
 
70
        # but disabling searching should just end up with None found
 
71
        sys_cfg = {'datasource': {'NoCloud': {'fs_label': None}}}
 
72
        dsrc = ds(sys_cfg=sys_cfg, distro=None, paths=self.paths)
 
73
        ret = dsrc.get_data()
 
74
        self.assertFalse(ret)
 
75
 
 
76
    def test_no_datasource_expected(self):
 
77
        #no source should be found if no cmdline, config, and fs_label=None
 
78
        sys_cfg = {'datasource': {'NoCloud': {'fs_label': None}}}
 
79
 
 
80
        ds = DataSourceNoCloud.DataSourceNoCloud
 
81
        dsrc = ds(sys_cfg=sys_cfg, distro=None, paths=self.paths)
 
82
        self.assertFalse(dsrc.get_data())
 
83
 
 
84
    def test_seed_in_config(self):
 
85
        ds = DataSourceNoCloud.DataSourceNoCloud
 
86
 
 
87
        data = {
 
88
            'fs_label': None,
 
89
            'meta-data': {'instance-id': 'IID'},
 
90
            'user-data': "USER_DATA_RAW",
 
91
        }
 
92
 
 
93
        sys_cfg = {'datasource': {'NoCloud': data}}
 
94
        dsrc = ds(sys_cfg=sys_cfg, distro=None, paths=self.paths)
 
95
        ret = dsrc.get_data()
 
96
        self.assertEqual(dsrc.userdata_raw, "USER_DATA_RAW")
 
97
        self.assertEqual(dsrc.metadata.get('instance-id'), 'IID')
 
98
        self.assertTrue(ret)
 
99
 
 
100
 
 
101
class TestParseCommandLineData(MockerTestCase):
 
102
 
 
103
    def test_parse_cmdline_data_valid(self):
 
104
        ds_id = "ds=nocloud"
 
105
        pairs = (
 
106
            ("root=/dev/sda1 %(ds_id)s", {}),
 
107
            ("%(ds_id)s; root=/dev/foo", {}),
 
108
            ("%(ds_id)s", {}),
 
109
            ("%(ds_id)s;", {}),
 
110
            ("%(ds_id)s;s=SEED", {'seedfrom': 'SEED'}),
 
111
            ("%(ds_id)s;seedfrom=SEED;local-hostname=xhost",
 
112
             {'seedfrom': 'SEED', 'local-hostname': 'xhost'}),
 
113
            ("%(ds_id)s;h=xhost",
 
114
             {'local-hostname': 'xhost'}),
 
115
            ("%(ds_id)s;h=xhost;i=IID",
 
116
             {'local-hostname': 'xhost', 'instance-id': 'IID'}),
 
117
        )
 
118
 
 
119
        for (fmt, expected) in pairs:
 
120
            fill = {}
 
121
            cmdline = fmt % {'ds_id': ds_id}
 
122
            ret = DataSourceNoCloud.parse_cmdline_data(ds_id=ds_id, fill=fill,
 
123
                                                       cmdline=cmdline)
 
124
            self.assertEqual(expected, fill)
 
125
            self.assertTrue(ret)
 
126
 
 
127
    def test_parse_cmdline_data_none(self):
 
128
        ds_id = "ds=foo"
 
129
        cmdlines = (
 
130
            "root=/dev/sda1 ro",
 
131
            "console=/dev/ttyS0 root=/dev/foo",
 
132
            "",
 
133
            "ds=foocloud",
 
134
            "ds=foo-net",
 
135
            "ds=nocloud;s=SEED",
 
136
        )
 
137
 
 
138
        for cmdline in cmdlines:
 
139
            fill = {}
 
140
            ret = DataSourceNoCloud.parse_cmdline_data(ds_id=ds_id, fill=fill,
 
141
                                                       cmdline=cmdline)
 
142
            self.assertEqual(fill, {})
 
143
            self.assertFalse(ret)
 
144
 
 
145
 
 
146
def apply_patches(patches):
 
147
    ret = []
 
148
    for (ref, name, replace) in patches:
 
149
        if replace is None:
 
150
            continue
 
151
        orig = getattr(ref, name)
 
152
        setattr(ref, name, replace)
 
153
        ret.append((ref, name, orig))
 
154
    return ret
 
155
 
 
156
 
 
157
# vi: ts=4 expandtab