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
|
from tests.unittests import helpers
from cloudinit import ec2_utils as eu
import httpretty as hp
class TestEc2Util(helpers.TestCase):
VERSION = 'latest'
@hp.activate
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.assertEquals('stuff', userdata)
@hp.activate
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.assertEquals('', userdata)
@hp.activate
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.assertEquals('', userdata)
@hp.activate
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, eu.combine_url(base_url, 'hostname'),
status=200, body='ec2.fake.host.name.com')
hp.register_uri(hp.GET, eu.combine_url(base_url, 'instance-id'),
status=200, body='123')
hp.register_uri(hp.GET, eu.combine_url(base_url, 'ami-launch-index'),
status=200, body='1')
md = eu.get_instance_metadata(self.VERSION, retries=0)
self.assertEquals(md['hostname'], 'ec2.fake.host.name.com')
self.assertEquals(md['instance-id'], '123')
self.assertEquals(md['ami-launch-index'], '1')
@hp.activate
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, eu.combine_url(base_url, 'hostname'),
status=200, body='ec2.fake.host.name.com')
hp.register_uri(hp.GET, eu.combine_url(base_url, 'instance-id'),
status=200, body='123')
hp.register_uri(hp.GET, eu.combine_url(base_url, 'public-keys/'),
status=200, body='0=my-public-key')
hp.register_uri(hp.GET,
eu.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.assertEquals(md['hostname'], 'ec2.fake.host.name.com')
self.assertEquals(md['instance-id'], '123')
self.assertEquals(1, len(md['public-keys']))
@hp.activate
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, eu.combine_url(base_url, 'hostname'),
status=200, body='ec2.fake.host.name.com')
hp.register_uri(hp.GET, eu.combine_url(base_url, 'instance-id'),
status=200, body='123')
hp.register_uri(hp.GET, eu.combine_url(base_url, 'public-keys/'),
status=200,
body="\n".join(['0=my-public-key', '1=my-other-key']))
hp.register_uri(hp.GET,
eu.combine_url(base_url, 'public-keys/0/openssh-key'),
status=200, body='ssh-rsa AAAA.....wZEf my-public-key')
hp.register_uri(hp.GET,
eu.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.assertEquals(md['hostname'], 'ec2.fake.host.name.com')
self.assertEquals(md['instance-id'], '123')
self.assertEquals(2, len(md['public-keys']))
@hp.activate
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, eu.combine_url(base_url, 'hostname'),
status=200, body='ec2.fake.host.name.com')
hp.register_uri(hp.GET, eu.combine_url(base_url, 'instance-id'),
status=200, body='123')
hp.register_uri(hp.GET,
eu.combine_url(base_url, 'block-device-mapping/'),
status=200,
body="\n".join(['ami', 'ephemeral0']))
hp.register_uri(hp.GET,
eu.combine_url(base_url, 'block-device-mapping/ami'),
status=200,
body="sdb")
hp.register_uri(hp.GET,
eu.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.assertEquals(md['hostname'], 'ec2.fake.host.name.com')
self.assertEquals(md['instance-id'], '123')
bdm = md['block-device-mapping']
self.assertEquals(2, len(bdm))
self.assertEquals(bdm['ami'], 'sdb')
self.assertEquals(bdm['ephemeral0'], 'sdc')
|
