← Back to branch summary

~smoser/ubuntu/vivid/cloud-init/snappy

« back to all changes in this revision

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

  • Committer: Scott Moser
  • Date: 2015-02-27 20:55:58 UTC
  • mfrom: (355.2.8 vivid)
  • Revision ID: smoser@ubuntu.com-20150227205558-glrwdgxqkaz6zyxa
https://launchpad.net/bugs/1247132
https://launchpad.net/bugs/1355343
https://launchpad.net/bugs/1383794
https://launchpad.net/bugs/1387340
https://launchpad.net/bugs/1404311
https://launchpad.net/bugs/1422388
https://launchpad.net/bugs/1422993
https://launchpad.net/bugs/1423972
https://launchpad.net/bugs/1424900
* Merge with vivid at 0.7.7~bzr1067-0ubuntu1
* New upstream snapshot.
  * fix broken consumption of gzipped user-data (LP: #1424900)
  * functional user-data on Azure again (LP: #1423972)
  * CloudStack: support fetching password from virtual router (LP: #1422388)
* New upstream snapshot.
  * Fix for ascii decode in DataSourceAzure (LP: #1422993).
* New upstream snapshot.
  * support for gpt partitioning, utilized in Azure [Daniel Watkins]
  * fix bug in exception handling in mount_cb.
* New upstream snapshot.
  * move to python3 (LP: #1247132)
  * systemd: run cloud-init before systemd-user-sessions.service
  * Use the GCE short hostname. (LP: #1383794)
  * Enable user-data encoding support for GCE. (LP: #1404311)
  * Update to use a newer and better OMNIBUS_URL
  * Be more tolerant of 'ssh_authorized_keys' types
  * Fix parse_ssh_config failing in ssh_util.py
  * Increase the robustness/configurability of the chef module.
  * retain trailing newline from template files when using
    jinja2 (LP: #1355343)
  * fix broken output handling (LP: #1387340)
  * digital ocean datasource
  * update url in config drive documentation
  * freebsd: enable correct behavior on Ec2.
  * freebsd: Use the proper virtio FreeBSD network interface name.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/unittests/test_datasource/test_gce.py
18
18
import httpretty
19
19
import re
20
20
 
21
 
from urlparse import urlparse
 
21
from base64 import b64encode, b64decode
 
22
from six.moves.urllib_parse import urlparse
22
23
 
23
24
from cloudinit import settings
24
25
from cloudinit import helpers
30
31
    'instance/id': '123',
31
32
    'instance/zone': 'foo/bar',
32
33
    'project/attributes/sshKeys': 'user:ssh-rsa AA2..+aRD0fyVw== root@server',
33
 
    'instance/hostname': 'server.project-name.local',
34
 
    'instance/attributes/user-data': '/bin/echo foo\n',
 
34
    'instance/hostname': 'server.project-foo.local',
 
35
    'instance/attributes/user-data': b'/bin/echo foo\n',
35
36
}
36
37
 
37
38
GCE_META_PARTIAL = {
38
 
    'instance/id': '123',
39
 
    'instance/hostname': 'server.project-name.local',
 
39
    'instance/id': '1234',
 
40
    'instance/hostname': 'server.project-bar.local',
 
41
    'instance/zone': 'bar/baz',
 
42
}
 
43
 
 
44
GCE_META_ENCODING = {
 
45
    'instance/id': '12345',
 
46
    'instance/hostname': 'server.project-baz.local',
 
47
    'instance/zone': 'baz/bang',
 
48
    'instance/attributes/user-data': b64encode(b'/bin/echo baz\n'),
 
49
    'instance/attributes/user-data-encoding': 'base64',
40
50
}
41
51
 
42
52
HEADERS = {'X-Google-Metadata-Request': 'True'}
43
 
MD_URL_RE = re.compile(r'http://metadata.google.internal./computeMetadata/v1/.*')
44
 
 
45
 
 
46
 
def _request_callback(method, uri, headers):
47
 
    url_path = urlparse(uri).path
48
 
    if url_path.startswith('/computeMetadata/v1/'):
49
 
        path = url_path.split('/computeMetadata/v1/')[1:][0]
50
 
    else:
51
 
        path = None
52
 
    if path in GCE_META:
53
 
        return (200, headers, GCE_META.get(path))
54
 
    else:
55
 
        return (404, headers, '')
 
53
MD_URL_RE = re.compile(
 
54
    r'http://metadata.google.internal./computeMetadata/v1/.*')
 
55
 
 
56
 
 
57
def _new_request_callback(gce_meta=None):
 
58
    if not gce_meta:
 
59
        gce_meta = GCE_META
 
60
 
 
61
    def _request_callback(method, uri, headers):
 
62
        url_path = urlparse(uri).path
 
63
        if url_path.startswith('/computeMetadata/v1/'):
 
64
            path = url_path.split('/computeMetadata/v1/')[1:][0]
 
65
        else:
 
66
            path = None
 
67
        if path in gce_meta:
 
68
            return (200, headers, gce_meta.get(path))
 
69
        else:
 
70
            return (404, headers, '')
 
71
 
 
72
    return _request_callback
56
73
 
57
74
 
58
75
class TestDataSourceGCE(test_helpers.HttprettyTestCase):
67
84
    def test_connection(self):
68
85
        httpretty.register_uri(
69
86
            httpretty.GET, MD_URL_RE,
70
 
            body=_request_callback)
 
87
            body=_new_request_callback())
71
88
 
72
89
        success = self.ds.get_data()
73
90
        self.assertTrue(success)
79
96
    def test_metadata(self):
80
97
        httpretty.register_uri(
81
98
            httpretty.GET, MD_URL_RE,
82
 
            body=_request_callback)
 
99
            body=_new_request_callback())
83
100
        self.ds.get_data()
84
101
 
85
 
        self.assertEqual(GCE_META.get('instance/hostname'),
 
102
        shostname = GCE_META.get('instance/hostname').split('.')[0]
 
103
        self.assertEqual(shostname,
86
104
                         self.ds.get_hostname())
87
105
 
88
106
        self.assertEqual(GCE_META.get('instance/id'),
103
121
    def test_metadata_partial(self):
104
122
        httpretty.register_uri(
105
123
            httpretty.GET, MD_URL_RE,
106
 
            body=_request_callback)
 
124
            body=_new_request_callback(GCE_META_PARTIAL))
107
125
        self.ds.get_data()
108
126
 
109
127
        self.assertEqual(GCE_META_PARTIAL.get('instance/id'),
110
128
                         self.ds.get_instance_id())
111
129
 
112
 
        self.assertEqual(GCE_META_PARTIAL.get('instance/hostname'),
113
 
                         self.ds.get_hostname())
 
130
        shostname = GCE_META_PARTIAL.get('instance/hostname').split('.')[0]
 
131
        self.assertEqual(shostname, self.ds.get_hostname())
 
132
 
 
133
    @httpretty.activate
 
134
    def test_metadata_encoding(self):
 
135
        httpretty.register_uri(
 
136
            httpretty.GET, MD_URL_RE,
 
137
            body=_new_request_callback(GCE_META_ENCODING))
 
138
        self.ds.get_data()
 
139
 
 
140
        decoded = b64decode(
 
141
            GCE_META_ENCODING.get('instance/attributes/user-data'))
 
142
        self.assertEqual(decoded, self.ds.get_userdata_raw())