~oddbloke/cloud-init/fix-gce-az

« back to all changes in this revision

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

  • Committer: Daniel Watkins
  • Date: 2015-07-20 14:33:25 UTC
  • mfrom: (1118.2.2 fix-gce-az)
  • Revision ID: daniel.watkins@canonical.com-20150720143325-566rbxalnspf33vt
Return a sensible value for DataSourceGCE.availability_zone.

Show diffs side-by-side

added added

removed removed

Lines of Context:
55
55
    r'http://metadata.google.internal./computeMetadata/v1/.*')
56
56
 
57
57
 
58
 
def _new_request_callback(gce_meta=None):
59
 
    if not gce_meta:
 
58
def _set_mock_metadata(gce_meta=None):
 
59
    if gce_meta is None:
60
60
        gce_meta = GCE_META
61
61
 
62
62
    def _request_callback(method, uri, headers):
70
70
        else:
71
71
            return (404, headers, '')
72
72
 
73
 
    return _request_callback
74
 
 
75
 
 
 
73
    httpretty.register_uri(httpretty.GET, MD_URL_RE, body=_request_callback)
 
74
 
 
75
 
 
76
@httpretty.activate
76
77
class TestDataSourceGCE(test_helpers.HttprettyTestCase):
77
78
 
78
79
    def setUp(self):
81
82
            helpers.Paths({}))
82
83
        super(TestDataSourceGCE, self).setUp()
83
84
 
84
 
    @httpretty.activate
85
85
    def test_connection(self):
86
 
        httpretty.register_uri(
87
 
            httpretty.GET, MD_URL_RE,
88
 
            body=_new_request_callback())
89
 
 
 
86
        _set_mock_metadata()
90
87
        success = self.ds.get_data()
91
88
        self.assertTrue(success)
92
89
 
93
90
        req_header = httpretty.last_request().headers
94
91
        self.assertDictContainsSubset(HEADERS, req_header)
95
92
 
96
 
    @httpretty.activate
97
93
    def test_metadata(self):
98
 
        httpretty.register_uri(
99
 
            httpretty.GET, MD_URL_RE,
100
 
            body=_new_request_callback())
 
94
        _set_mock_metadata()
101
95
        self.ds.get_data()
102
96
 
103
97
        shostname = GCE_META.get('instance/hostname').split('.')[0]
107
101
        self.assertEqual(GCE_META.get('instance/id'),
108
102
                         self.ds.get_instance_id())
109
103
 
110
 
        self.assertEqual(GCE_META.get('instance/zone'),
111
 
                         self.ds.availability_zone)
112
 
 
113
104
        self.assertEqual(GCE_META.get('instance/attributes/user-data'),
114
105
                         self.ds.get_userdata_raw())
115
106
 
116
107
    # test partial metadata (missing user-data in particular)
117
 
    @httpretty.activate
118
108
    def test_metadata_partial(self):
119
 
        httpretty.register_uri(
120
 
            httpretty.GET, MD_URL_RE,
121
 
            body=_new_request_callback(GCE_META_PARTIAL))
 
109
        _set_mock_metadata(GCE_META_PARTIAL)
122
110
        self.ds.get_data()
123
111
 
124
112
        self.assertEqual(GCE_META_PARTIAL.get('instance/id'),
127
115
        shostname = GCE_META_PARTIAL.get('instance/hostname').split('.')[0]
128
116
        self.assertEqual(shostname, self.ds.get_hostname())
129
117
 
130
 
    @httpretty.activate
131
118
    def test_metadata_encoding(self):
132
 
        httpretty.register_uri(
133
 
            httpretty.GET, MD_URL_RE,
134
 
            body=_new_request_callback(GCE_META_ENCODING))
 
119
        _set_mock_metadata(GCE_META_ENCODING)
135
120
        self.ds.get_data()
136
121
 
137
122
        decoded = b64decode(
138
123
            GCE_META_ENCODING.get('instance/attributes/user-data'))
139
124
        self.assertEqual(decoded, self.ds.get_userdata_raw())
140
125
 
141
 
    @httpretty.activate
142
126
    def test_missing_required_keys_return_false(self):
143
127
        for required_key in ['instance/id', 'instance/zone',
144
128
                             'instance/hostname']:
145
129
            meta = GCE_META_PARTIAL.copy()
146
130
            del meta[required_key]
147
 
            httpretty.register_uri(httpretty.GET, MD_URL_RE,
148
 
                                   body=_new_request_callback(meta))
 
131
            _set_mock_metadata(meta)
149
132
            self.assertEqual(False, self.ds.get_data())
150
133
            httpretty.reset()
151
134
 
152
 
    @httpretty.activate
153
135
    def test_project_level_ssh_keys_are_used(self):
154
 
        httpretty.register_uri(httpretty.GET, MD_URL_RE,
155
 
                               body=_new_request_callback())
 
136
        _set_mock_metadata()
156
137
        self.ds.get_data()
157
138
 
158
139
        # we expect a list of public ssh keys with user names stripped
159
140
        self.assertEqual(['ssh-rsa AA2..+aRD0fyVw== root@server'],
160
141
                         self.ds.get_public_ssh_keys())
161
142
 
162
 
    @httpretty.activate
163
143
    def test_instance_level_ssh_keys_are_used(self):
164
144
        key_content = 'ssh-rsa JustAUser root@server'
165
145
        meta = GCE_META.copy()
166
146
        meta['instance/attributes/sshKeys'] = 'user:{0}'.format(key_content)
167
147
 
168
 
        httpretty.register_uri(httpretty.GET, MD_URL_RE,
169
 
                               body=_new_request_callback(meta))
 
148
        _set_mock_metadata(meta)
170
149
        self.ds.get_data()
171
150
 
172
151
        self.assertIn(key_content, self.ds.get_public_ssh_keys())
173
152
 
174
 
    @httpretty.activate
175
153
    def test_instance_level_keys_replace_project_level_keys(self):
176
154
        key_content = 'ssh-rsa JustAUser root@server'
177
155
        meta = GCE_META.copy()
178
156
        meta['instance/attributes/sshKeys'] = 'user:{0}'.format(key_content)
179
157
 
180
 
        httpretty.register_uri(httpretty.GET, MD_URL_RE,
181
 
                               body=_new_request_callback(meta))
 
158
        _set_mock_metadata(meta)
182
159
        self.ds.get_data()
183
160
 
184
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)