diff options
Diffstat (limited to 'tests/unittests/test_ec2_util.py')
| -rw-r--r-- | tests/unittests/test_ec2_util.py | 376 |
1 files changed, 237 insertions, 139 deletions
diff --git a/tests/unittests/test_ec2_util.py b/tests/unittests/test_ec2_util.py index e8e0b5b1..f447d295 100644 --- a/tests/unittests/test_ec2_util.py +++ b/tests/unittests/test_ec2_util.py @@ -2,178 +2,276 @@ import httpretty as hp -from tests.unittests import helpers - from cloudinit import ec2_utils as eu from cloudinit import url_helper as uh +from tests.unittests import helpers class TestEc2Util(helpers.HttprettyTestCase): - VERSION = 'latest' + 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) + 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')) + 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) + 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) + 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) + 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) + 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) + 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) + 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') + 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') + 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') + 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'])) + 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') + 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'])) + 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") + 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(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') + 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') + 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(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) + self.assertEqual(iam["info"]["LastUpdated"], "2016-10-27T17:29:39Z") + self.assertNotIn("security-credentials", iam) + # vi: ts=4 expandtab |