[LYN-2243] Create end-to-end automation tests for the metrics gem (#23)
[LYN-2243] Create end-to-end automation tests for the metrics gem
This commit is contained in:
Junbo Liang
GitHub
@@ -0,0 +1,134 @@
|
||||
"""
|
||||
All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
|
||||
its licensors.
|
||||
For complete copyright and license terms please see the LICENSE at the root of this
|
||||
distribution (the "License"). All use of this software is governed by the License,
|
||||
or, if provided, by the license below or the license accompanying this file. Do not
|
||||
remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
"""
|
||||
|
||||
import boto3
|
||||
import configparser
|
||||
import logging
|
||||
import os
|
||||
import pytest
|
||||
import typing
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
logging.getLogger('boto').setLevel(logging.CRITICAL)
|
||||
|
||||
|
||||
class AwsCredentials:
|
||||
def __init__(self, profile_name: str):
|
||||
self._profile_name = profile_name
|
||||
|
||||
self._credentials_path = os.environ.get('AWS_SHARED_CREDENTIALS_FILE')
|
||||
if not self._credentials_path:
|
||||
# Home directory location varies based on the operating system, but is referred to using the environment
|
||||
# variables %UserProfile% in Windows and $HOME or ~ (tilde) in Unix-based systems.
|
||||
self._credentials_path = os.path.join(os.environ.get('UserProfile', os.path.expanduser('~')),
|
||||
'.aws', 'credentials')
|
||||
self._credentials_file_exists = os.path.exists(self._credentials_path)
|
||||
|
||||
self._credentials = configparser.ConfigParser()
|
||||
self._credentials.read(self._credentials_path)
|
||||
|
||||
def get_aws_credentials(self) -> typing.Tuple[str, str, str]:
|
||||
"""
|
||||
Get aws credentials stored in the specific named profile.
|
||||
|
||||
:return AWS credentials.
|
||||
"""
|
||||
access_key_id = self._get_aws_credential_attribute_value('aws_access_key_id')
|
||||
secret_access_key = self._get_aws_credential_attribute_value('aws_secret_access_key')
|
||||
session_token = self._get_aws_credential_attribute_value('aws_session_token')
|
||||
|
||||
return access_key_id, secret_access_key, session_token
|
||||
|
||||
def set_aws_credentials_by_session(self, session: boto3.Session) -> None:
|
||||
"""
|
||||
Set AWS credentials stored in the specific named profile using an assumed role session.
|
||||
|
||||
:param session: assumed role session.
|
||||
"""
|
||||
credentials = session.get_credentials().get_frozen_credentials()
|
||||
self.set_aws_credentials(credentials.access_key, credentials.secret_key, credentials.token)
|
||||
|
||||
def set_aws_credentials(self, aws_access_key_id: str, aws_secret_access_key: str,
|
||||
aws_session_token: str) -> None:
|
||||
"""
|
||||
Set AWS credentials stored in the specific named profile.
|
||||
|
||||
:param aws_access_key_id: AWS access key id.
|
||||
:param aws_secret_access_key: AWS secrete access key.
|
||||
:param aws_session_token: AWS assumed role session.
|
||||
"""
|
||||
self._set_aws_credential_attribute_value('aws_access_key_id', aws_access_key_id)
|
||||
self._set_aws_credential_attribute_value('aws_secret_access_key', aws_secret_access_key)
|
||||
self._set_aws_credential_attribute_value('aws_session_token', aws_session_token)
|
||||
|
||||
if (len(self._credentials.sections()) == 0) and (not self._credentials_file_exists):
|
||||
os.remove(self._credentials_path)
|
||||
return
|
||||
|
||||
with open(self._credentials_path, 'w+') as credential_file:
|
||||
self._credentials.write(credential_file)
|
||||
|
||||
def _get_aws_credential_attribute_value(self, attribute_name: str) -> str:
|
||||
"""
|
||||
Get the value of an AWS credential attribute stored in the specific named profile.
|
||||
|
||||
:param attribute_name: Name of the AWS credential attribute.
|
||||
:return Value of the AWS credential attribute.
|
||||
"""
|
||||
try:
|
||||
value = self._credentials.get(self._profile_name, attribute_name)
|
||||
except configparser.NoSectionError:
|
||||
# Named profile or key doesn't exist
|
||||
value = None
|
||||
except configparser.NoOptionError:
|
||||
# Named profile doesn't have the specified attribute
|
||||
value = None
|
||||
|
||||
return value
|
||||
|
||||
def _set_aws_credential_attribute_value(self, attribute_name: str, attribute_value: str) -> None:
|
||||
"""
|
||||
Set the value of an AWS credential attribute stored in the specific named profile.
|
||||
|
||||
:param attribute_name: Name of the AWS credential attribute.
|
||||
:param attribute_value: Value of the AWS credential attribute.
|
||||
"""
|
||||
if self._profile_name not in self._credentials:
|
||||
self._credentials[self._profile_name] = {}
|
||||
|
||||
if attribute_value is None:
|
||||
self._credentials.remove_option(self._profile_name, attribute_name)
|
||||
# Remove the named profile if it doesn't have any AWS credential attribute.
|
||||
if len(self._credentials[self._profile_name]) == 0:
|
||||
self._credentials.remove_section(self._profile_name)
|
||||
else:
|
||||
self._credentials[self._profile_name][attribute_name] = attribute_value
|
||||
|
||||
|
||||
@pytest.fixture(scope='function')
|
||||
def aws_credentials(request: pytest.fixture, aws_utils: pytest.fixture, profile_name: str):
|
||||
"""
|
||||
Fixture for setting up temporary AWS credentials from assume role.
|
||||
|
||||
:param request: _pytest.fixtures.SubRequest class that handles getting
|
||||
a pytest fixture from a pytest function/fixture.
|
||||
:param aws_utils: aws_utils fixture.
|
||||
:param profile_name: Named AWS profile to store temporary credentials.
|
||||
"""
|
||||
aws_credentials_obj = AwsCredentials(profile_name)
|
||||
original_access_key, original_secret_access_key, original_token = aws_credentials_obj.get_aws_credentials()
|
||||
aws_credentials_obj.set_aws_credentials_by_session(aws_utils.assume_session())
|
||||
|
||||
def teardown():
|
||||
# Reset to the named profile using the original AWS credentials
|
||||
aws_credentials_obj.set_aws_credentials(original_access_key, original_secret_access_key, original_token)
|
||||
request.addfinalizer(teardown)
|
||||
|
||||
return aws_credentials_obj
|
||||
@@ -1,82 +1,90 @@
|
||||
"""
|
||||
All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
|
||||
its licensors.
|
||||
For complete copyright and license terms please see the LICENSE at the root of this
|
||||
distribution (the "License"). All use of this software is governed by the License,
|
||||
or, if provided, by the license below or the license accompanying this file. Do not
|
||||
remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
"""
|
||||
import boto3
|
||||
import pytest
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class AwsUtils:
|
||||
|
||||
def __init__(self, arn: str, session_name: str, region_name: str):
|
||||
local_session = boto3.Session(profile_name='default')
|
||||
local_sts_client = local_session.client('sts')
|
||||
self._local_account_id = local_sts_client.get_caller_identity()["Account"]
|
||||
logger.info(f'Local Account Id: {self._local_account_id}')
|
||||
|
||||
response = local_sts_client.assume_role(RoleArn=arn, RoleSessionName=session_name)
|
||||
|
||||
self._assume_session = boto3.Session(aws_access_key_id=response['Credentials']['AccessKeyId'],
|
||||
aws_secret_access_key=response['Credentials']['SecretAccessKey'],
|
||||
aws_session_token=response['Credentials']['SessionToken'],
|
||||
region_name=region_name)
|
||||
|
||||
assume_sts_client = self._assume_session.client('sts')
|
||||
assume_account_id = assume_sts_client.get_caller_identity()["Account"]
|
||||
logger.info(f'Assume Account Id: {assume_account_id}')
|
||||
self._assume_account_id = assume_account_id
|
||||
|
||||
def client(self, service: str):
|
||||
"""
|
||||
Get the client for a specific AWS service from configured session
|
||||
:return: Client for the AWS service.
|
||||
"""
|
||||
return self._assume_session.client(service)
|
||||
|
||||
def assume_session(self):
|
||||
return self._assume_session
|
||||
|
||||
def local_account_id(self):
|
||||
return self._local_account_id
|
||||
|
||||
def assume_account_id(self):
|
||||
return self._assume_account_id
|
||||
|
||||
def destroy(self) -> None:
|
||||
"""
|
||||
clears stored session
|
||||
"""
|
||||
self._assume_session = None
|
||||
|
||||
|
||||
@pytest.fixture(scope='function')
|
||||
def aws_utils(
|
||||
request: pytest.fixture,
|
||||
assume_role_arn: str,
|
||||
session_name: str,
|
||||
region_name: str):
|
||||
"""
|
||||
Fixture for setting up a Cdk
|
||||
:param request: _pytest.fixtures.SubRequest class that handles getting
|
||||
a pytest fixture from a pytest function/fixture.
|
||||
:param assume_role_arn: Role used to fetch temporary aws credentials, configure service clients with obtained credentials.
|
||||
:param session_name: Session name to set.
|
||||
:param region_name: AWS account region to set for session.
|
||||
:return AWSUtils class object.
|
||||
"""
|
||||
aws_utils_obj = AwsUtils(assume_role_arn, session_name, region_name)
|
||||
|
||||
def teardown():
|
||||
aws_utils_obj.destroy()
|
||||
|
||||
request.addfinalizer(teardown)
|
||||
|
||||
return aws_utils_obj
|
||||
"""
|
||||
All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
|
||||
its licensors.
|
||||
For complete copyright and license terms please see the LICENSE at the root of this
|
||||
distribution (the "License"). All use of this software is governed by the License,
|
||||
or, if provided, by the license below or the license accompanying this file. Do not
|
||||
remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
"""
|
||||
import boto3
|
||||
import pytest
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
logging.getLogger('boto').setLevel(logging.CRITICAL)
|
||||
|
||||
|
||||
class AwsUtils:
|
||||
|
||||
def __init__(self, arn: str, session_name: str, region_name: str):
|
||||
local_session = boto3.Session(profile_name='default')
|
||||
local_sts_client = local_session.client('sts')
|
||||
self._local_account_id = local_sts_client.get_caller_identity()["Account"]
|
||||
logger.info(f'Local Account Id: {self._local_account_id}')
|
||||
|
||||
response = local_sts_client.assume_role(RoleArn=arn, RoleSessionName=session_name)
|
||||
|
||||
self._assume_session = boto3.Session(aws_access_key_id=response['Credentials']['AccessKeyId'],
|
||||
aws_secret_access_key=response['Credentials']['SecretAccessKey'],
|
||||
aws_session_token=response['Credentials']['SessionToken'],
|
||||
region_name=region_name)
|
||||
|
||||
assume_sts_client = self._assume_session.client('sts')
|
||||
assume_account_id = assume_sts_client.get_caller_identity()["Account"]
|
||||
logger.info(f'Assume Account Id: {assume_account_id}')
|
||||
self._assume_account_id = assume_account_id
|
||||
|
||||
def client(self, service: str):
|
||||
"""
|
||||
Get the client for a specific AWS service from configured session
|
||||
:return: Client for the AWS service.
|
||||
"""
|
||||
return self._assume_session.client(service)
|
||||
|
||||
def resource(self, service: str):
|
||||
"""
|
||||
Get the resource for a specific AWS service from configured session
|
||||
:return: Client for the AWS service.
|
||||
"""
|
||||
return self._assume_session.resource(service)
|
||||
|
||||
def assume_session(self):
|
||||
return self._assume_session
|
||||
|
||||
def local_account_id(self):
|
||||
return self._local_account_id
|
||||
|
||||
def assume_account_id(self):
|
||||
return self._assume_account_id
|
||||
|
||||
def destroy(self) -> None:
|
||||
"""
|
||||
clears stored session
|
||||
"""
|
||||
self._assume_session = None
|
||||
|
||||
|
||||
@pytest.fixture(scope='function')
|
||||
def aws_utils(
|
||||
request: pytest.fixture,
|
||||
assume_role_arn: str,
|
||||
session_name: str,
|
||||
region_name: str):
|
||||
"""
|
||||
Fixture for AWS util functions
|
||||
:param request: _pytest.fixtures.SubRequest class that handles getting
|
||||
a pytest fixture from a pytest function/fixture.
|
||||
:param assume_role_arn: Role used to fetch temporary aws credentials, configure service clients with obtained credentials.
|
||||
:param session_name: Session name to set.
|
||||
:param region_name: AWS account region to set for session.
|
||||
:return AWSUtils class object.
|
||||
"""
|
||||
aws_utils_obj = AwsUtils(assume_role_arn, session_name, region_name)
|
||||
|
||||
def teardown():
|
||||
aws_utils_obj.destroy()
|
||||
|
||||
request.addfinalizer(teardown)
|
||||
|
||||
return aws_utils_obj
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
"""
|
||||
All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
|
||||
its licensors.
|
||||
|
||||
For complete copyright and license terms please see the LICENSE at the root of this
|
||||
distribution (the "License"). All use of this software is governed by the License,
|
||||
or, if provided, by the license below or the license accompanying this file. Do not
|
||||
remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
"""
|
||||
|
||||
from enum import Enum
|
||||
import botocore.client
|
||||
import botocore.waiter
|
||||
import logging
|
||||
|
||||
logging.getLogger('boto').setLevel(logging.CRITICAL)
|
||||
|
||||
|
||||
class WaitState(Enum):
|
||||
SUCCESS = 'success'
|
||||
FAILURE = 'failure'
|
||||
|
||||
|
||||
class CustomWaiter:
|
||||
"""
|
||||
Base class for a custom waiter.
|
||||
|
||||
Modified from:
|
||||
https://docs.aws.amazon.com/code-samples/latest/catalog/python-demo_tools-custom_waiter.py.html
|
||||
"""
|
||||
def __init__(
|
||||
self, name: str, operation: str, argument: str,
|
||||
acceptors: dict, client: botocore.client, delay: int = 30, max_tries: int = 10,
|
||||
matcher='path'):
|
||||
"""
|
||||
Subclasses should pass specific operations, arguments, and acceptors to
|
||||
their superclass.
|
||||
|
||||
:param name: The name of the waiter. This can be any descriptive string.
|
||||
:param operation: The operation to wait for. This must match the casing of
|
||||
the underlying operation model, which is typically in
|
||||
CamelCase.
|
||||
:param argument: The dict keys used to access the result of the operation, in
|
||||
dot notation. For example, 'Job.Status' will access
|
||||
result['Job']['Status'].
|
||||
:param acceptors: The list of acceptors that indicate the wait is over. These
|
||||
can indicate either success or failure. The acceptor values
|
||||
are compared to the result of the operation after the
|
||||
argument keys are applied.
|
||||
:param client: The Boto3 client.
|
||||
:param delay: The number of seconds to wait between each call to the operation. Default to 30 seconds.
|
||||
:param max_tries: The maximum number of tries before exiting. Default to 10.
|
||||
:param matcher: The kind of matcher to use. Default to 'path'.
|
||||
"""
|
||||
self.name = name
|
||||
self.operation = operation
|
||||
self.argument = argument
|
||||
self.client = client
|
||||
self.waiter_model = botocore.waiter.WaiterModel({
|
||||
'version': 2,
|
||||
'waiters': {
|
||||
name: {
|
||||
"delay": delay,
|
||||
"operation": operation,
|
||||
"maxAttempts": max_tries,
|
||||
"acceptors": [{
|
||||
"state": state.value,
|
||||
"matcher": matcher,
|
||||
"argument": argument,
|
||||
"expected": expected
|
||||
} for expected, state in acceptors.items()]
|
||||
}}})
|
||||
self.waiter = botocore.waiter.create_waiter_with_client(
|
||||
self.name, self.waiter_model, self.client)
|
||||
|
||||
self._timeout = delay * max_tries
|
||||
|
||||
def _wait(self, **kwargs):
|
||||
"""
|
||||
Starts the botocore wait loop.
|
||||
|
||||
:param kwargs: Keyword arguments that are passed to the operation being polled.
|
||||
"""
|
||||
self.waiter.wait(**kwargs)
|
||||
|
||||
@property
|
||||
def timeout(self):
|
||||
return self._timeout
|
||||
|
||||
|
||||
Reference in New Issue
Block a user