x/cloud-init
Code Issues Proposed changes

Add ec2 utils tests and httpretty requirement for http mocking

Browse Source
This commit is contained in:
Joshua Harlow 2014-01-10 17:17:06 -08:00
parent 1aa01899bb
commit 9f49f6b039
3 changed files with 140 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Requires
View File

@ -29,8 +29,8 @@ argparse
# Requests handles ssl correctly! # Requests handles ssl correctly!
requests requests
# Boto for ec2
boto
# For patching pieces of cloud-config together # For patching pieces of cloud-config together
jsonpatch jsonpatch
# For http testing (only needed for testing)
httpretty>=0.7.1

8
cloudinit/ec2_utils.py
View File

@ -81,9 +81,9 @@ class MetadataMaterializer(object):
if len(contents) > 1: if len(contents) > 1:
# What a PITA... # What a PITA...
(ident, sub_contents) = contents (ident, sub_contents) = contents
checked_ident = util.safe_int(ident) ident = util.safe_int(ident)
if checked_ident is not None: if ident is not None:
resource = "%s/openssh-key" % (checked_ident) resource = "%s/openssh-key" % (ident)
field_name = sub_contents field_name = sub_contents
leaves[field_name] = resource leaves[field_name] = resource
return (leaves, children) return (leaves, children)
@ -145,7 +145,7 @@ def get_instance_userdata(api_version='latest',
return str(response) return str(response)
except Exception: except Exception:
util.logexc(LOG, "Failed fetching userdata from url %s", ud_url) util.logexc(LOG, "Failed fetching userdata from url %s", ud_url)
return None return ''
def get_instance_metadata(api_version='latest', def get_instance_metadata(api_version='latest',

133
tests/unittests/test_ec2_util.py Normal file
View File

@ -0,0 +1,133 @@
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/'),
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_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="\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')