Added testcase for Instance Profile Information Presented in Instance Metadata Service #228
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,166 @@ | ||
| #!/usr/bin/env python | ||
| # | ||
| # | ||
| # Description: This script encompasses test cases/modules concerning instance specific behavior and | ||
| # features regarding the metadata service with regards to IAM roles and instance profiles in Eucalyptus. | ||
| # The test cases/modules that are executed can be found in the script under the "tests" list. | ||
|
|
||
| import time | ||
| from concurrent.futures import ThreadPoolExecutor | ||
| import threading | ||
| from eucaops import Eucaops | ||
| from eutester.euinstance import EuInstance | ||
| from eutester.eutestcase import EutesterTestCase | ||
| from eucaops import EC2ops | ||
| import os | ||
| import re | ||
| import random | ||
| import json | ||
|
|
||
| class InstanceBasics(EutesterTestCase): | ||
|
There was a problem hiding this comment. Please rename the class to something that hasnt been used yet. |
||
| def __init__( self, name="InstanceBasics", credpath=None, region=None, config_file=None, password=None, emi=None, zone=None, | ||
| user_data=None, instance_user=None, instance_profile_name=None, instance_profile_arn=None, **kwargs): | ||
| """ | ||
| EC2 API tests focused on instance store instances | ||
|
|
||
| :param credpath: Path to directory containing eucarc file | ||
| :param region: EC2 Region to run testcase in | ||
| :param config_file: Configuration file path | ||
| :param password: SSH password for bare metal machines if config is passed and keys arent synced | ||
| :param emi: Image id to use for test | ||
| :param zone: Availability Zone to run test in | ||
| :param user_data: User Data to pass to instance | ||
| :param instance_profile_name: Instance Profile name to pass to instance | ||
| :param instance_profile_arn: Instance Profile ARN to pass to instance | ||
| :param instance_user: User to login to instance as | ||
| :param kwargs: Additional arguments | ||
| """ | ||
| super(InstanceBasics, self).__init__(name=name) | ||
| if region: | ||
| self.tester = EC2ops(credpath=credpath, region=region) | ||
| else: | ||
| self.tester = Eucaops(config_file=config_file, password=password, credpath=credpath) | ||
| self.instance_timeout = 480 | ||
|
|
||
| ### Add and authorize a group for the instance | ||
| self.group = self.tester.add_group(group_name="group-" + str(time.time())) | ||
| self.tester.authorize_group_by_name(group_name=self.group.name) | ||
| self.tester.authorize_group_by_name(group_name=self.group.name, port=-1, protocol="icmp" ) | ||
| ### Generate a keypair for the instance | ||
| self.keypair = self.tester.add_keypair( "keypair-" + str(time.time())) | ||
| self.keypath = '%s/%s.pem' % (os.curdir, self.keypair.name) | ||
| if emi: | ||
| self.image = emi | ||
| else: | ||
| self.image = self.tester.get_emi(root_device_type="instance-store",not_location="loadbalancer") | ||
| self.address = None | ||
| self.volume = None | ||
| self.private_addressing = False | ||
| if instance_profile_name: | ||
| self.instance_profile_name = instance_profile_name | ||
| self.instance_profile_arn = None | ||
| elif instance_profile_arn: | ||
| self.instance_profile_arn = instance_profile_arn | ||
| self.instance_profile_name = None | ||
| if not zone: | ||
| zones = self.tester.ec2.get_all_zones() | ||
| self.zone = random.choice(zones).name | ||
| else: | ||
| self.zone = zone | ||
| self.reservation = None | ||
| self.reservation_lock = threading.Lock() | ||
| self.run_instance_params = {'image': self.image, 'user_data': user_data, 'username': instance_user, | ||
| 'keypair': self.keypair.name, 'group': self.group.name,'zone': self.zone, | ||
| 'instance_profile_name': self.instance_profile_name, | ||
| 'instance_profile_arn': self.instance_profile_arn, 'timeout': self.instance_timeout} | ||
| self.managed_network = True | ||
|
|
||
| ### If I have access to the underlying infrastructure I can look | ||
| ### at the network mode and only run certain tests where it makes sense | ||
| if hasattr(self.tester,"service_manager"): | ||
| cc = self.tester.get_component_machines("cc")[0] | ||
| network_mode = cc.sys("cat " + self.tester.eucapath + "/etc/eucalyptus/eucalyptus.conf | grep MODE")[0] | ||
| if re.search("(SYSTEM|STATIC)", network_mode): | ||
| self.managed_network = False | ||
|
|
||
| def set_reservation(self, reservation): | ||
| self.reservation_lock.acquire() | ||
| self.reservation = reservation | ||
| self.reservation_lock.release() | ||
|
|
||
| def clean_method(self): | ||
| self.tester.cleanup_artifacts() | ||
|
|
||
| def InstanceProfileChecks(self): | ||
|
There was a problem hiding this comment. Since most of this code is the same as instancetest, can we just add this method there? |
||
| """ | ||
| This case was developed to test the metadata service regarding instance profile of an instance for consistency. | ||
| The following meta-data attributes are tested: | ||
| - iam/security-credentials/<role-name> | ||
| - iam/info/instance-profile-arn | ||
| - iam/info/instance-profile-id | ||
| - iam/info/last-updated-date | ||
| If any of these tests fail, the test case will error out; logging the results. | ||
| """ | ||
| if not self.reservation: | ||
| reservation = self.tester.run_instance(**self.run_instance_params) | ||
| else: | ||
| reservation = self.reservation | ||
| for instance in reservation.instances: | ||
| # Check to see if instance profile ARN/Name matches metadata instance-profile-arn | ||
| if self.instance_profile_name: | ||
| self.assertTrue(re.search(self.instance_profile_name, | ||
| instance.get_metadata("iam/info/instance-profile-arn")[0]), 'Incorrect Instance Profile Name') | ||
| else: | ||
| self.assertTrue(re.match(instance.get_metadata("iam/info/instance-profile-arn")[0], | ||
| self.instance_profile_arn), 'Incorrect Instance Profile ARN') | ||
| # Check to see if instance profile ARN is at least 20 characters and at a maximum 2048 characters | ||
| # based on AWS IAM InstanceProfile data type definition | ||
| self.assertTrue(len(instance.get_metadata("iam/info/instance-profile-arn")[0]) >= 20, 'Instance Profile ARN is less than 20 characters') | ||
| self.assertTrue(len(instance.get_metadata("iam/info/instance-profile-arn")[0]) <= 2048, 'Instance Profile ARN is greater than 2048 characters') | ||
| # Check to see if instance profile ID is at least 16 characters and at a maximum 32 characters | ||
| # based on AWS IAM InstanceProfile data type definition | ||
| self.assertTrue(len(instance.get_metadata("iam/info/instance-profile-id")[0]) >= 16, 'Instance Profile ID is less than 16 characters') | ||
| self.assertTrue(len(instance.get_metadata("iam/info/instance-profile-id")[0]) <= 32, 'Instance Profile ID is greater than 32 characters') | ||
| # Check date format of instance profile LastUpdated | ||
| self.assertTrue(re.match('(\d{4})[/.-](\d{2})[/.-](\d{2})T(\d{2})[/.:](\d{2})[/.:](\d{2})Z',instance.get_metadata("iam/info/last-updated-date")[0]), | ||
| 'Incorrect date format for Instance Profile LastUpdated') | ||
| # Check to see if iam/security-credentials/<role-name> is available, then check contents | ||
| self.assertTrue(instance.get_metadata("iam/security-credentials/")[0], 'IAM Role Not Available in Metadata') | ||
| try: | ||
| role_name = instance.get_metadata("iam/security-credentials/")[0] | ||
| temp_creds = json.loads(''.join(instance.get_metadata("iam/security-credentials/%s"%role_name))) | ||
| except Exception, e: | ||
| self.tester.debug("Unable to access IAM Role temporary credentials:\n"+ str(e)) | ||
| raise e | ||
| # Check date format of temporary credentials LastUpdated variable | ||
| self.assertTrue(re.match('(\d{4})[/.-](\d{2})[/.-](\d{2})T(\d{2})[/.:](\d{2})[/.:](\d{2})Z',temp_creds['LastUpdated'].encode('ascii')), | ||
| 'Incorrect date format for temporary credentials LastUpdated variable') | ||
| # Check to see if AccessKeyId is at least 16 characters and at a maximum 32 characters | ||
| # based on AWS STS Credentials data type definition | ||
| self.assertTrue(len(temp_creds['AccessKeyId'].encode('ascii')) >= 16, 'AccessKeyId is less than 16 characters') | ||
| self.assertTrue(len(temp_creds['AccessKeyId'].encode('ascii')) <= 32, 'AccessKeyId is greater than 32 characters') | ||
| self.assertTrue(temp_creds['SecretAccessKey'].encode('ascii'), "SecretAccessKey does not exist in " + role_name + " temporary credentials") | ||
| self.assertTrue(temp_creds['Token'].encode('ascii'), "Token does not exist in " + role_name + " temporary credentials") | ||
| # Check date format of temporary credentials Expiration variable | ||
| self.assertTrue(re.match('(\d{4})[/.-](\d{2})[/.-](\d{2})T(\d{2})[/.:](\d{2})[/.:](\d{2})Z',temp_creds['Expiration'].encode('ascii')), | ||
| 'Incorrect date format for temporary credentials Expiration variable') | ||
| self.set_reservation(reservation) | ||
| return reservation | ||
|
|
||
| if __name__ == "__main__": | ||
| testcase= EutesterTestCase(name='instanceprofiletest') | ||
| testcase.setup_parser(description="Test the Eucalyptus EC2 instance profile metadata functionality.") | ||
| testcase.get_args() | ||
| instancetestsuite= testcase.do_with_args(InstanceBasics) | ||
|
|
||
| ### Either use the list of tests passed from config/command line to determine what subset of tests to run | ||
| list = testcase.args.tests or [ "InstanceProfileChecks"] | ||
| ### Convert test suite methods to EutesterUnitTest objects | ||
| unit_list = [] | ||
| for test in list: | ||
| test = getattr(instancetestsuite,test) | ||
| unit_list.append(testcase.create_testunit_from_method(test)) | ||
| testcase.clean_method = instancetestsuite.clean_method | ||
| result = testcase.run_test_case_list(unit_list) | ||
| exit(result) | ||
|
|
||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can we just add 1 param which is the name, and then translate to the arn if necessary?