summaryrefslogtreecommitdiff
path: root/tests/unittests/test_ec2_util.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/unittests/test_ec2_util.py')
-rw-r--r--tests/unittests/test_ec2_util.py376
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
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-10 07:12:55 +0000