summaryrefslogtreecommitdiff
path: root/tests/unittests/test_ec2_util.py
blob: 3f50f57d97683d86a3f7fa688c16fe3dcd2c06f7 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
# This file is part of cloud-init. See LICENSE file for license information.

import httpretty as hp

from cloudinit.tests import helpers

from cloudinit import ec2_utils as eu
from cloudinit import url_helper as uh


class TestEc2Util(helpers.HttprettyTestCase):
    VERSION = 'latest'

    def test_userdata_fetch(self):
        hp.register_uri(hp.GET,
                        'http://169.254.169.254/%s/user-data' % (self.VERSION),
                        body='stuff',
                        status=200)
        userdata = eu.get_instance_userdata(self.VERSION)
        self.assertEqual('stuff', userdata.decode('utf-8'))

    def test_userdata_fetch_fail_not_found(self):
        hp.register_uri(hp.GET,
                        'http://169.254.169.254/%s/user-data' % (self.VERSION),
                        status=404)
        userdata = eu.get_instance_userdata(self.VERSION, retries=0)
        self.assertEqual('', userdata)

    def test_userdata_fetch_fail_server_dead(self):
        hp.register_uri(hp.GET,
                        'http://169.254.169.254/%s/user-data' % (self.VERSION),
                        status=500)
        userdata = eu.get_instance_userdata(self.VERSION, retries=0)
        self.assertEqual('', userdata)

    def test_userdata_fetch_fail_server_not_found(self):
        hp.register_uri(hp.GET,
                        'http://169.254.169.254/%s/user-data' % (self.VERSION),
                        status=404)
        userdata = eu.get_instance_userdata(self.VERSION)
        self.assertEqual('', userdata)

    def test_metadata_fetch_no_keys(self):
        base_url = 'http://169.254.169.254/%s/meta-data/' % (self.VERSION)
        hp.register_uri(hp.GET, base_url, status=200,
                        body="\n".join(['hostname',
                                        'instance-id',
                                        'ami-launch-index']))
        hp.register_uri(hp.GET, uh.combine_url(base_url, 'hostname'),
                        status=200, body='ec2.fake.host.name.com')
        hp.register_uri(hp.GET, uh.combine_url(base_url, 'instance-id'),
                        status=200, body='123')
        hp.register_uri(hp.GET, uh.combine_url(base_url, 'ami-launch-index'),
                        status=200, body='1')
        md = eu.get_instance_metadata(self.VERSION, retries=0)
        self.assertEqual(md['hostname'], 'ec2.fake.host.name.com')
        self.assertEqual(md['instance-id'], '123')
        self.assertEqual(md['ami-launch-index'], '1')

    def test_metadata_fetch_key(self):
        base_url = 'http://169.254.169.254/%s/meta-data/' % (self.VERSION)
        hp.register_uri(hp.GET, base_url, status=200,
                        body="\n".join(['hostname',
                                        'instance-id',
                                        'public-keys/']))
        hp.register_uri(hp.GET, uh.combine_url(base_url, 'hostname'),
                        status=200, body='ec2.fake.host.name.com')
        hp.register_uri(hp.GET, uh.combine_url(base_url, 'instance-id'),
                        status=200, body='123')
        hp.register_uri(hp.GET, uh.combine_url(base_url, 'public-keys/'),
                        status=200, body='0=my-public-key')
        hp.register_uri(hp.GET,
                        uh.combine_url(base_url, 'public-keys/0/openssh-key'),
                        status=200, body='ssh-rsa AAAA.....wZEf my-public-key')
        md = eu.get_instance_metadata(self.VERSION, retries=0, timeout=0.1)
        self.assertEqual(md['hostname'], 'ec2.fake.host.name.com')
        self.assertEqual(md['instance-id'], '123')
        self.assertEqual(1, len(md['public-keys']))

    def test_metadata_fetch_with_2_keys(self):
        base_url = 'http://169.254.169.254/%s/meta-data/' % (self.VERSION)
        hp.register_uri(hp.GET, base_url, status=200,
                        body="\n".join(['hostname',
                                        'instance-id',
                                        'public-keys/']))
        hp.register_uri(hp.GET, uh.combine_url(base_url, 'hostname'),
                        status=200, body='ec2.fake.host.name.com')
        hp.register_uri(hp.GET, uh.combine_url(base_url, 'instance-id'),
                        status=200, body='123')
        hp.register_uri(hp.GET, uh.combine_url(base_url, 'public-keys/'),
                        status=200,
                        body="\n".join(['0=my-public-key', '1=my-other-key']))
        hp.register_uri(hp.GET,
                        uh.combine_url(base_url, 'public-keys/0/openssh-key'),
                        status=200, body='ssh-rsa AAAA.....wZEf my-public-key')
        hp.register_uri(hp.GET,
                        uh.combine_url(base_url, 'public-keys/1/openssh-key'),
                        status=200, body='ssh-rsa AAAA.....wZEf my-other-key')
        md = eu.get_instance_metadata(self.VERSION, retries=0, timeout=0.1)
        self.assertEqual(md['hostname'], 'ec2.fake.host.name.com')
        self.assertEqual(md['instance-id'], '123')
        self.assertEqual(2, len(md['public-keys']))

    def test_metadata_fetch_bdm(self):
        base_url = 'http://169.254.169.254/%s/meta-data/' % (self.VERSION)
        hp.register_uri(hp.GET, base_url, status=200,
                        body="\n".join(['hostname',
                                        'instance-id',
                                        'block-device-mapping/']))
        hp.register_uri(hp.GET, uh.combine_url(base_url, 'hostname'),
                        status=200, body='ec2.fake.host.name.com')
        hp.register_uri(hp.GET, uh.combine_url(base_url, 'instance-id'),
                        status=200, body='123')
        hp.register_uri(hp.GET,
                        uh.combine_url(base_url, 'block-device-mapping/'),
                        status=200,
                        body="\n".join(['ami', 'ephemeral0']))
        hp.register_uri(hp.GET,
                        uh.combine_url(base_url, 'block-device-mapping/ami'),
                        status=200,
                        body="sdb")
        hp.register_uri(hp.GET,
                        uh.combine_url(base_url,
                                       'block-device-mapping/ephemeral0'),
                        status=200,
                        body="sdc")
        md = eu.get_instance_metadata(self.VERSION, retries=0, timeout=0.1)
        self.assertEqual(md['hostname'], 'ec2.fake.host.name.com')
        self.assertEqual(md['instance-id'], '123')
        bdm = md['block-device-mapping']
        self.assertEqual(2, len(bdm))
        self.assertEqual(bdm['ami'], 'sdb')
        self.assertEqual(bdm['ephemeral0'], 'sdc')

    def test_metadata_no_security_credentials(self):
        base_url = 'http://169.254.169.254/%s/meta-data/' % (self.VERSION)
        hp.register_uri(hp.GET, base_url, status=200,
                        body="\n".join(['instance-id',
                                        'iam/']))
        hp.register_uri(hp.GET, uh.combine_url(base_url, 'instance-id'),
                        status=200, body='i-0123451689abcdef0')
        hp.register_uri(hp.GET,
                        uh.combine_url(base_url, 'iam/'),
                        status=200,
                        body="\n".join(['info/', 'security-credentials/']))
        hp.register_uri(hp.GET,
                        uh.combine_url(base_url, 'iam/info/'),
                        status=200,
                        body='LastUpdated')
        hp.register_uri(hp.GET,
                        uh.combine_url(base_url, 'iam/info/LastUpdated'),
                        status=200, body='2016-10-27T17:29:39Z')
        hp.register_uri(hp.GET,
                        uh.combine_url(base_url, 'iam/security-credentials/'),
                        status=200,
                        body='ReadOnly/')
        hp.register_uri(hp.GET,
                        uh.combine_url(base_url,
                                       'iam/security-credentials/ReadOnly/'),
                        status=200,
                        body="\n".join(['LastUpdated', 'Expiration']))
        hp.register_uri(hp.GET,
                        uh.combine_url(
                            base_url,
                            'iam/security-credentials/ReadOnly/LastUpdated'),
                        status=200, body='2016-10-27T17:28:17Z')
        hp.register_uri(hp.GET,
                        uh.combine_url(
                            base_url,
                            'iam/security-credentials/ReadOnly/Expiration'),
                        status=200, body='2016-10-28T00:00:34Z')
        md = eu.get_instance_metadata(self.VERSION, retries=0, timeout=0.1)
        self.assertEqual(md['instance-id'], 'i-0123451689abcdef0')
        iam = md['iam']
        self.assertEqual(1, len(iam))
        self.assertEqual(iam['info']['LastUpdated'], '2016-10-27T17:29:39Z')
        self.assertNotIn('security-credentials', iam)

# vi: ts=4 expandtab