← Back to branch summary

~cloud-init-dev/cloud-init/trunk

« back to all changes in this revision

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

  • Committer: Scott Moser
  • Date: 2016-08-10 15:06:15 UTC
  • Revision ID: smoser@ubuntu.com-20160810150615-ma2fv107w3suy1ma
README: Mention move of revision control to git.

cloud-init development has moved its revision control to git.
It is available at 
  https://code.launchpad.net/cloud-init

Clone with 
  git clone https://git.launchpad.net/cloud-init
or
  git clone git+ssh://git.launchpad.net/cloud-init

For more information see
  https://git.launchpad.net/cloud-init/tree/HACKING.rst

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/unittests/test_datasource/test_gce.py
1
 
#
2
 
#    Copyright (C) 2014 Vaidas Jablonskis
3
 
#
4
 
#    Author: Vaidas Jablonskis <jablonskis@gmail.com>
5
 
#
6
 
#    This program is free software: you can redistribute it and/or modify
7
 
#    it under the terms of the GNU General Public License version 3, as
8
 
#    published by the Free Software Foundation.
9
 
#
10
 
#    This program is distributed in the hope that it will be useful,
11
 
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
12
 
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13
 
#    GNU General Public License for more details.
14
 
#
15
 
#    You should have received a copy of the GNU General Public License
16
 
#    along with this program.  If not, see <http://www.gnu.org/licenses/>.
17
 
 
18
 
import re
19
 
 
20
 
from base64 import b64encode, b64decode
21
 
from six.moves.urllib_parse import urlparse
22
 
 
23
 
from cloudinit import helpers
24
 
from cloudinit import settings
25
 
from cloudinit.sources import DataSourceGCE
26
 
 
27
 
from .. import helpers as test_helpers
28
 
 
29
 
httpretty = test_helpers.import_httpretty()
30
 
 
31
 
GCE_META = {
32
 
    'instance/id': '123',
33
 
    'instance/zone': 'foo/bar',
34
 
    'project/attributes/sshKeys': 'user:ssh-rsa AA2..+aRD0fyVw== root@server',
35
 
    'instance/hostname': 'server.project-foo.local',
36
 
    'instance/attributes/user-data': b'/bin/echo foo\n',
37
 
}
38
 
 
39
 
GCE_META_PARTIAL = {
40
 
    'instance/id': '1234',
41
 
    'instance/hostname': 'server.project-bar.local',
42
 
    'instance/zone': 'bar/baz',
43
 
}
44
 
 
45
 
GCE_META_ENCODING = {
46
 
    'instance/id': '12345',
47
 
    'instance/hostname': 'server.project-baz.local',
48
 
    'instance/zone': 'baz/bang',
49
 
    'instance/attributes/user-data': b64encode(b'/bin/echo baz\n'),
50
 
    'instance/attributes/user-data-encoding': 'base64',
51
 
}
52
 
 
53
 
HEADERS = {'X-Google-Metadata-Request': 'True'}
54
 
MD_URL_RE = re.compile(
55
 
    r'http://metadata.google.internal/computeMetadata/v1/.*')
56
 
 
57
 
 
58
 
def _set_mock_metadata(gce_meta=None):
59
 
    if gce_meta is None:
60
 
        gce_meta = GCE_META
61
 
 
62
 
    def _request_callback(method, uri, headers):
63
 
        url_path = urlparse(uri).path
64
 
        if url_path.startswith('/computeMetadata/v1/'):
65
 
            path = url_path.split('/computeMetadata/v1/')[1:][0]
66
 
        else:
67
 
            path = None
68
 
        if path in gce_meta:
69
 
            return (200, headers, gce_meta.get(path))
70
 
        else:
71
 
            return (404, headers, '')
72
 
 
73
 
    httpretty.register_uri(httpretty.GET, MD_URL_RE, body=_request_callback)
74
 
 
75
 
 
76
 
@httpretty.activate
77
 
class TestDataSourceGCE(test_helpers.HttprettyTestCase):
78
 
 
79
 
    def setUp(self):
80
 
        self.ds = DataSourceGCE.DataSourceGCE(
81
 
            settings.CFG_BUILTIN, None,
82
 
            helpers.Paths({}))
83
 
        super(TestDataSourceGCE, self).setUp()
84
 
 
85
 
    def test_connection(self):
86
 
        _set_mock_metadata()
87
 
        success = self.ds.get_data()
88
 
        self.assertTrue(success)
89
 
 
90
 
        req_header = httpretty.last_request().headers
91
 
        self.assertDictContainsSubset(HEADERS, req_header)
92
 
 
93
 
    def test_metadata(self):
94
 
        _set_mock_metadata()
95
 
        self.ds.get_data()
96
 
 
97
 
        shostname = GCE_META.get('instance/hostname').split('.')[0]
98
 
        self.assertEqual(shostname,
99
 
                         self.ds.get_hostname())
100
 
 
101
 
        self.assertEqual(GCE_META.get('instance/id'),
102
 
                         self.ds.get_instance_id())
103
 
 
104
 
        self.assertEqual(GCE_META.get('instance/attributes/user-data'),
105
 
                         self.ds.get_userdata_raw())
106
 
 
107
 
    # test partial metadata (missing user-data in particular)
108
 
    def test_metadata_partial(self):
109
 
        _set_mock_metadata(GCE_META_PARTIAL)
110
 
        self.ds.get_data()
111
 
 
112
 
        self.assertEqual(GCE_META_PARTIAL.get('instance/id'),
113
 
                         self.ds.get_instance_id())
114
 
 
115
 
        shostname = GCE_META_PARTIAL.get('instance/hostname').split('.')[0]
116
 
        self.assertEqual(shostname, self.ds.get_hostname())
117
 
 
118
 
    def test_metadata_encoding(self):
119
 
        _set_mock_metadata(GCE_META_ENCODING)
120
 
        self.ds.get_data()
121
 
 
122
 
        decoded = b64decode(
123
 
            GCE_META_ENCODING.get('instance/attributes/user-data'))
124
 
        self.assertEqual(decoded, self.ds.get_userdata_raw())
125
 
 
126
 
    def test_missing_required_keys_return_false(self):
127
 
        for required_key in ['instance/id', 'instance/zone',
128
 
                             'instance/hostname']:
129
 
            meta = GCE_META_PARTIAL.copy()
130
 
            del meta[required_key]
131
 
            _set_mock_metadata(meta)
132
 
            self.assertEqual(False, self.ds.get_data())
133
 
            httpretty.reset()
134
 
 
135
 
    def test_project_level_ssh_keys_are_used(self):
136
 
        _set_mock_metadata()
137
 
        self.ds.get_data()
138
 
 
139
 
        # we expect a list of public ssh keys with user names stripped
140
 
        self.assertEqual(['ssh-rsa AA2..+aRD0fyVw== root@server'],
141
 
                         self.ds.get_public_ssh_keys())
142
 
 
143
 
    def test_instance_level_ssh_keys_are_used(self):
144
 
        key_content = 'ssh-rsa JustAUser root@server'
145
 
        meta = GCE_META.copy()
146
 
        meta['instance/attributes/sshKeys'] = 'user:{0}'.format(key_content)
147
 
 
148
 
        _set_mock_metadata(meta)
149
 
        self.ds.get_data()
150
 
 
151
 
        self.assertIn(key_content, self.ds.get_public_ssh_keys())
152
 
 
153
 
    def test_instance_level_keys_replace_project_level_keys(self):
154
 
        key_content = 'ssh-rsa JustAUser root@server'
155
 
        meta = GCE_META.copy()
156
 
        meta['instance/attributes/sshKeys'] = 'user:{0}'.format(key_content)
157
 
 
158
 
        _set_mock_metadata(meta)
159
 
        self.ds.get_data()
160
 
 
161
 
        self.assertEqual([key_content], self.ds.get_public_ssh_keys())
162
 
 
163
 
    def test_only_last_part_of_zone_used_for_availability_zone(self):
164
 
        _set_mock_metadata()
165
 
        self.ds.get_data()
166
 
        self.assertEqual('bar', self.ds.availability_zone)