birkeh/o3de
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'main' into LYN-3969

Browse Source
This commit is contained in:
clujames
2021-06-01 22:09:16 -07:00
parent 17024d6cf4 262c1c1132
commit 60da667e8b
1337 changed files with 27079 additions and 20397 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AutomatedTesting/Gem/PythonTests/AWS/Windows/aws_metrics/__init__.py
+10
View File
@@ -0,0 +1,10 @@
"""
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.
"""
AutomatedTesting/Gem/PythonTests/AWS/Windows/aws_metrics/aws_metrics_automation_test.py
+237
View File
@@ -0,0 +1,237 @@
"""
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 logging
import os
import pytest
import time
import typing
from datetime import datetime
import ly_test_tools.log.log_monitor
from assetpipeline.ap_fixtures.asset_processor_fixture import asset_processor as asset_processor
from AWS.common.aws_utils import aws_utils
from AWS.common.aws_credentials import aws_credentials
from AWS.Windows.resource_mappings.resource_mappings import resource_mappings
from AWS.Windows.cdk.cdk import cdk
from .aws_metrics_utils import aws_metrics_utils
AWS_METRICS_FEATURE_NAME = 'AWSMetrics'
GAME_LOG_NAME = 'Game.log'
logger = logging.getLogger(__name__)
def setup(launcher: ly_test_tools.launchers.Launcher,
cdk: cdk,
asset_processor: asset_processor,
resource_mappings: resource_mappings,
context_variable: str = '') -> typing.Tuple[ly_test_tools.log.log_monitor.LogMonitor, str, str]:
"""
Set up the CDK application and start the log monitor.
:param launcher: Client launcher for running the test level.
:param cdk: CDK application for deploying the AWS resources.
:param asset_processor: asset_processor fixture.
:param resource_mappings: resource_mappings fixture.
:param context_variable: context_variable for enable optional CDK feature.
:return log monitor object, metrics file path and the metrics stack name.
"""
logger.info(f'Cdk stack names:\n{cdk.list()}')
stacks = cdk.deploy(context_variable=context_variable)
resource_mappings.populate_output_keys(stacks)
asset_processor.start()
asset_processor.wait_for_idle()
metrics_file_path = os.path.join(launcher.workspace.paths.project(), 'user',
AWS_METRICS_FEATURE_NAME, 'metrics.json')
remove_file(metrics_file_path)
file_to_monitor = os.path.join(launcher.workspace.paths.project_log(), GAME_LOG_NAME)
remove_file(file_to_monitor)
# Initialize the log monitor.
log_monitor = ly_test_tools.log.log_monitor.LogMonitor(launcher=launcher, log_file_path=file_to_monitor)
return log_monitor, metrics_file_path, stacks[0]
def monitor_metrics_submission(log_monitor: ly_test_tools.log.log_monitor.LogMonitor) -> None:
"""
Monitor the messages and notifications for submitting metrics.
:param log_monitor: Log monitor to check the log messages.
"""
expected_lines = [
'(Script) - Submitted metrics without buffer.',
'(Script) - Submitted metrics with buffer.',
'(Script) - Metrics is sent successfully.'
]
unexpected_lines = [
'(Script) - Failed to submit metrics without buffer.',
'(Script) - Failed to submit metrics with buffer.',
'(Script) - Failed to send metrics.'
]
result = log_monitor.monitor_log_for_lines(
expected_lines=expected_lines,
unexpected_lines=unexpected_lines,
halt_on_unexpected=True)
# Assert the log monitor detected expected lines and did not detect any unexpected lines.
assert result, (
f'Log monitoring failed. Used expected_lines values: {expected_lines} & '
f'unexpected_lines values: {unexpected_lines}')
def remove_file(file_path: str) -> None:
"""
Remove a local file and its directory.
:param file_path: Path to the local file.
"""
if os.path.exists(file_path):
os.remove(file_path)
file_dir = os.path.dirname(file_path)
if os.path.exists(file_dir) and len(os.listdir(file_dir)) == 0:
os.rmdir(file_dir)
@pytest.mark.SUITE_periodic
@pytest.mark.usefixtures('automatic_process_killer')
@pytest.mark.parametrize('project', ['AutomatedTesting'])
@pytest.mark.parametrize('level', ['AWS/Metrics'])
@pytest.mark.parametrize('feature_name', [AWS_METRICS_FEATURE_NAME])
@pytest.mark.parametrize('resource_mappings_filename', ['aws_resource_mappings.json'])
@pytest.mark.parametrize('profile_name', ['AWSAutomationTest'])
@pytest.mark.parametrize('region_name', ['us-west-2'])
@pytest.mark.parametrize('assume_role_arn', ['arn:aws:iam::645075835648:role/o3de-automation-tests'])
@pytest.mark.parametrize('session_name', ['o3de-Automation-session'])
class TestAWSMetrics_Windows(object):
def test_AWSMetrics_RealTimeAnalytics_MetricsSentToCloudWatch(self,
level: str,
launcher: ly_test_tools.launchers.Launcher,
asset_processor: pytest.fixture,
workspace: pytest.fixture,
aws_utils: aws_utils,
aws_credentials: aws_credentials,
resource_mappings: resource_mappings,
cdk: cdk,
aws_metrics_utils: aws_metrics_utils,
):
"""
Tests that the submitted metrics are sent to CloudWatch for real-time analytics.
"""
log_monitor, metrics_file_path, stack_name = setup(launcher, cdk, asset_processor, resource_mappings)
# Start the Kinesis Data Analytics application for real-time analytics.
analytics_application_name = f'{stack_name}-AnalyticsApplication'
aws_metrics_utils.start_kinesis_data_analytics_application(analytics_application_name)
launcher.args = ['+LoadLevel', level]
launcher.args.extend(['-rhi=null'])
with launcher.start(launch_ap=False):
start_time = datetime.utcnow()
monitor_metrics_submission(log_monitor)
# Verify that operational health metrics are delivered to CloudWatch.
aws_metrics_utils.verify_cloud_watch_delivery(
'AWS/Lambda',
'Invocations',
[{'Name': 'FunctionName',
'Value': f'{stack_name}-AnalyticsProcessingLambda'}],
start_time)
logger.info('Operational health metrics sent to CloudWatch.')
aws_metrics_utils.verify_cloud_watch_delivery(
AWS_METRICS_FEATURE_NAME,
'TotalLogins',
[],
start_time)
logger.info('Real-time metrics sent to CloudWatch.')
# Stop the Kinesis Data Analytics application.
aws_metrics_utils.stop_kinesis_data_analytics_application(analytics_application_name)
def test_AWSMetrics_UnauthorizedUser_RequestRejected(self,
level: str,
launcher: ly_test_tools.launchers.Launcher,
cdk: cdk,
aws_credentials: aws_credentials,
asset_processor: pytest.fixture,
resource_mappings: resource_mappings,
workspace: pytest.fixture):
"""
Tests that unauthorized users cannot send metrics events to the AWS backed backend.
"""
log_monitor, metrics_file_path, stack_name = setup(launcher, cdk, asset_processor, resource_mappings)
# Set invalid AWS credentials.
launcher.args = ['+LoadLevel', level, '+cl_awsAccessKey', 'AKIAIOSFODNN7EXAMPLE',
'+cl_awsSecretKey', 'wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY']
launcher.args.extend(['-rhi=null'])
with launcher.start(launch_ap=False):
result = log_monitor.monitor_log_for_lines(
expected_lines=['(Script) - Failed to send metrics.'],
unexpected_lines=['(Script) - Metrics is sent successfully.'],
halt_on_unexpected=True)
assert result, 'Metrics events are sent successfully by unauthorized user'
logger.info('Unauthorized user is rejected to send metrics.')
def test_AWSMetrics_BatchAnalytics_MetricsDeliveredToS3(self,
level: str,
launcher: ly_test_tools.launchers.Launcher,
cdk: cdk,
aws_credentials: aws_credentials,
asset_processor: pytest.fixture,
resource_mappings: resource_mappings,
aws_utils: aws_utils,
aws_metrics_utils: aws_metrics_utils,
workspace: pytest.fixture):
"""
Tests that the submitted metrics are sent to the data lake for batch analytics.
"""
log_monitor, metrics_file_path, stack_name = setup(launcher, cdk, asset_processor, resource_mappings,
context_variable='batch_processing=true')
analytics_bucket_name = aws_metrics_utils.get_analytics_bucket_name(stack_name)
launcher.args = ['+LoadLevel', level]
launcher.args.extend(['-rhi=null'])
with launcher.start(launch_ap=False):
start_time = datetime.utcnow()
monitor_metrics_submission(log_monitor)
# Verify that operational health metrics are delivered to CloudWatch.
aws_metrics_utils.verify_cloud_watch_delivery(
'AWS/Lambda',
'Invocations',
[{'Name': 'FunctionName',
'Value': f'{stack_name}-EventsProcessingLambda'}],
start_time)
logger.info('Operational health metrics sent to CloudWatch.')
aws_metrics_utils.verify_s3_delivery(analytics_bucket_name)
logger.info('Metrics sent to S3.')
# Run the glue crawler to populate the AWS Glue Data Catalog with tables.
aws_metrics_utils.run_glue_crawler(f'{stack_name}-EventsCrawler')
# Run named queries on the table to verify the batch analytics.
aws_metrics_utils.run_named_queries(f'{stack_name}-AthenaWorkGroup')
logger.info('Query metrics from S3 successfully.')
# Kinesis Data Firehose buffers incoming data before it delivers it to Amazon S3. Sleep for the
# default interval (60s) to make sure that all the metrics are sent to the bucket before cleanup.
time.sleep(60)
# Empty the S3 bucket. S3 buckets can only be deleted successfully when it doesn't contain any object.
aws_metrics_utils.empty_s3_bucket(analytics_bucket_name)
AutomatedTesting/Gem/PythonTests/AWS/Windows/aws_metrics/aws_metrics_utils.py
+252
View File
@@ -0,0 +1,252 @@
"""
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 logging
import pathlib
import pytest
import typing
from datetime import datetime
from botocore.exceptions import WaiterError
from AWS.common.aws_utils import AwsUtils
from .aws_metrics_waiters import KinesisAnalyticsApplicationUpdatedWaiter, \
CloudWatchMetricsDeliveredWaiter, DataLakeMetricsDeliveredWaiter, GlueCrawlerReadyWaiter
logging.getLogger('boto').setLevel(logging.CRITICAL)
# Expected directory and file extension for the S3 objects.
EXPECTED_S3_DIRECTORY = 'firehose_events/'
EXPECTED_S3_OBJECT_EXTENSION = '.parquet'
class AWSMetricsUtils:
"""
Provide utils functions for the AWSMetrics gem to interact with the deployed resources.
"""
def __init__(self, aws_utils: AwsUtils):
self._aws_util = aws_utils
def start_kinesis_data_analytics_application(self, application_name: str) -> None:
"""
Start the Kenisis Data Analytics application for real-time analytics.
:param application_name: Name of the Kenisis Data Analytics application.
"""
input_id = self.get_kinesis_analytics_application_input_id(application_name)
assert input_id, 'invalid Kinesis Data Analytics application input.'
client = self._aws_util.client('kinesisanalytics')
try:
client.start_application(
ApplicationName=application_name,
InputConfigurations=[
{
'Id': input_id,
'InputStartingPositionConfiguration': {
'InputStartingPosition': 'NOW'
}
},
]
)
except client.exceptions.ResourceInUseException:
# The application has been started.
return
try:
KinesisAnalyticsApplicationUpdatedWaiter(client, 'RUNNING').wait(application_name=application_name)
except WaiterError as e:
assert False, f'Failed to start the Kinesis Data Analytics application: {str(e)}.'
def get_kinesis_analytics_application_input_id(self, application_name: str) -> str:
"""
Get the input ID for the Kenisis Data Analytics application.
:param application_name: Name of the Kenisis Data Analytics application.
:return: Input ID for the Kenisis Data Analytics application.
"""
client = self._aws_util.client('kinesisanalytics')
response = client.describe_application(
ApplicationName=application_name
)
if not response:
return ''
input_descriptions = response.get('ApplicationDetail', {}).get('InputDescriptions', [])
if len(input_descriptions) != 1:
return ''
return input_descriptions[0].get('InputId', '')
def stop_kinesis_data_analytics_application(self, application_name: str) -> None:
"""
Stop the Kenisis Data Analytics application.
:param application_name: Name of the Kenisis Data Analytics application.
"""
client = self._aws_util.client('kinesisanalytics')
client.stop_application(
ApplicationName=application_name
)
try:
KinesisAnalyticsApplicationUpdatedWaiter(client, 'READY').wait(application_name=application_name)
except WaiterError as e:
assert False, f'Failed to stop the Kinesis Data Analytics application: {str(e)}.'
def verify_cloud_watch_delivery(self, namespace: str, metrics_name: str,
dimensions: typing.List[dict], start_time: datetime) -> None:
"""
Verify that the expected metrics is delivered to CloudWatch.
:param namespace: Namespace of the metrics.
:param metrics_name: Name of the metrics.
:param dimensions: Dimensions of the metrics.
:param start_time: Start time for generating the metrics.
"""
client = self._aws_util.client('cloudwatch')
try:
CloudWatchMetricsDeliveredWaiter(client).wait(
namespace=namespace,
metrics_name=metrics_name,
dimensions=dimensions,
start_time=start_time
)
except WaiterError as e:
assert False, f'Failed to deliver metrics to CloudWatch: {str(e)}.'
def verify_s3_delivery(self, analytics_bucket_name: str) -> None:
"""
Verify that metrics are delivered to S3 for batch analytics successfully.
:param analytics_bucket_name: Name of the deployed S3 bucket.
"""
client = self._aws_util.client('s3')
bucket_name = analytics_bucket_name
try:
DataLakeMetricsDeliveredWaiter(client).wait(bucket_name=bucket_name, prefix=EXPECTED_S3_DIRECTORY)
except WaiterError as e:
assert False, f'Failed to find the S3 directory for storing metrics data: {str(e)}.'
# Check whether the data is converted to the expected data format.
response = client.list_objects_v2(
Bucket=bucket_name,
Prefix=EXPECTED_S3_DIRECTORY
)
assert response.get('KeyCount', 0) != 0, f'Failed to deliver metrics to the S3 bucket {bucket_name}.'
s3_objects = response.get('Contents', [])
for s3_object in s3_objects:
key = s3_object.get('Key', '')
assert pathlib.Path(key).suffix == EXPECTED_S3_OBJECT_EXTENSION, \
f'Invalid data format is found in the S3 bucket {bucket_name}'
def run_glue_crawler(self, crawler_name: str) -> None:
"""
Run the Glue crawler and wait for it to finish.
:param crawler_name: Name of the Glue crawler
"""
client = self._aws_util.client('glue')
try:
client.start_crawler(
Name=crawler_name
)
except client.exceptions.CrawlerRunningException:
# The crawler has already been started.
return
try:
GlueCrawlerReadyWaiter(client).wait(crawler_name=crawler_name)
except WaiterError as e:
assert False, f'Failed to run the Glue crawler: {str(e)}.'
def run_named_queries(self, work_group: str) -> None:
"""
Run the named queries under the specific Athena work group.
:param work_group: Name of the Athena work group.
"""
client = self._aws_util.client('athena')
# List all the named queries.
response = client.list_named_queries(
WorkGroup=work_group
)
named_query_ids = response.get('NamedQueryIds', [])
# Run each of the queries.
for named_query_id in named_query_ids:
get_named_query_response = client.get_named_query(
NamedQueryId=named_query_id
)
named_query = get_named_query_response.get('NamedQuery', {})
start_query_execution_response = client.start_query_execution(
QueryString=named_query.get('QueryString', ''),
QueryExecutionContext={
'Database': named_query.get('Database', '')
},
WorkGroup=work_group
)
# Wait for the query to finish.
state = 'RUNNING'
while state == 'QUEUED' or state == 'RUNNING':
get_query_execution_response = client.get_query_execution(
QueryExecutionId=start_query_execution_response.get('QueryExecutionId', '')
)
state = get_query_execution_response.get('QueryExecution', {}).get('Status', {}).get('State', '')
assert state == 'SUCCEEDED', f'Failed to run the named query {named_query.get("Name", {})}'
def empty_s3_bucket(self, bucket_name: str) -> None:
"""
Empty the S3 bucket following:
https://boto3.amazonaws.com/v1/documentation/api/latest/guide/migrations3.html
:param bucket_name: Name of the S3 bucket.
"""
s3 = self._aws_util.resource('s3')
bucket = s3.Bucket(bucket_name)
for key in bucket.objects.all():
key.delete()
def get_analytics_bucket_name(self, stack_name: str) -> str:
"""
Get the name of the deployed S3 bucket.
:param stack_name: Name of the CloudFormation stack.
:return: Name of the deployed S3 bucket.
"""
client = self._aws_util.client('cloudformation')
response = client.describe_stack_resources(
StackName=stack_name
)
resources = response.get('StackResources', [])
for resource in resources:
if resource.get('ResourceType') == 'AWS::S3::Bucket':
return resource.get('PhysicalResourceId', '')
return ''
@pytest.fixture(scope='function')
def aws_metrics_utils(
request: pytest.fixture,
aws_utils: pytest.fixture):
"""
Fixture for the AWS metrics util functions.
:param request: _pytest.fixtures.SubRequest class that handles getting
a pytest fixture from a pytest function/fixture.
:param aws_utils: aws_utils fixture.
"""
aws_utils_obj = AWSMetricsUtils(aws_utils)
return aws_utils_obj
AutomatedTesting/Gem/PythonTests/AWS/Windows/aws_metrics/aws_metrics_waiters.py
+142
View File
@@ -0,0 +1,142 @@
"""
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 botocore.client
import logging
from datetime import timedelta
from AWS.common.custom_waiter import CustomWaiter, WaitState
logging.getLogger('boto').setLevel(logging.CRITICAL)
class KinesisAnalyticsApplicationUpdatedWaiter(CustomWaiter):
"""
Subclass of the base custom waiter class.
Wait for the Kinesis analytics application being updated to a specific status.
"""
def __init__(self, client: botocore.client, status: str):
"""
Initialize the waiter.
:param client: Boto3 client to use.
:param status: Expected status.
"""
super().__init__(
'KinesisAnalyticsApplicationUpdated',
'DescribeApplication',
'ApplicationDetail.ApplicationStatus',
{status: WaitState.SUCCESS},
client)
def wait(self, application_name: str):
"""
Wait for the expected status.
:param application_name: Name of the Kinesis analytics application.
"""
self._wait(ApplicationName=application_name)
class GlueCrawlerReadyWaiter(CustomWaiter):
"""
Subclass of the base custom waiter class.
Wait for the Glue crawler to finish its processing.
"""
def __init__(self, client: botocore.client):
"""
Initialize the waiter.
:param client: Boto3 client to use.
"""
super().__init__(
'GlueCrawlerReady',
'GetCrawler',
'Crawler.State',
{'READY': WaitState.SUCCESS},
client)
def wait(self, crawler_name):
"""
Wait for the expected status.
:param crawler_name: Name of the Glue crawler.
"""
self._wait(Name=crawler_name)
class DataLakeMetricsDeliveredWaiter(CustomWaiter):
"""
Subclass of the base custom waiter class.
Wait for the expected directory being created in the S3 bucket.
"""
def __init__(self, client: botocore.client):
"""
Initialize the waiter.
:param client: Boto3 client to use.
"""
super().__init__(
'DataLakeMetricsDelivered',
'ListObjectsV2',
'KeyCount > `0`',
{True: WaitState.SUCCESS},
client)
def wait(self, bucket_name, prefix):
"""
Wait for the expected directory being created.
:param bucket_name: Name of the S3 bucket.
:param prefix: Name of the expected directory prefix.
"""
self._wait(Bucket=bucket_name, Prefix=prefix)
class CloudWatchMetricsDeliveredWaiter(CustomWaiter):
"""
Subclass of the base custom waiter class.
Wait for the expected metrics being delivered to CloudWatch.
"""
def __init__(self, client: botocore.client):
"""
Initialize the waiter.
:param client: Boto3 client to use.
"""
super().__init__(
'CloudWatchMetricsDelivered',
'GetMetricStatistics',
'length(Datapoints) > `0`',
{True: WaitState.SUCCESS},
client)
def wait(self, namespace, metrics_name, dimensions, start_time):
"""
Wait for the expected metrics being delivered.
:param namespace: Namespace of the metrics.
:param metrics_name: Name of the metrics.
:param dimensions: Dimensions of the metrics.
:param start_time: Start time for generating the metrics.
"""
self._wait(
Namespace=namespace,
MetricName=metrics_name,
Dimensions=dimensions,
StartTime=start_time,
EndTime=start_time + timedelta(0, self.timeout),
Period=60,
Statistics=[
'SampleCount'
],
Unit='Count'
)
AutomatedTesting/Gem/PythonTests/AWS/Windows/cdk/cdk.py
+56 -1
View File
@@ -16,12 +16,15 @@ import boto3
import ly_test_tools.environment.process_utils as process_utils
from typing import List
BOOTSTRAP_STACK_NAME = 'CDKToolkit'
BOOTSTRAP_STAGING_BUCKET_LOGIC_ID = 'StagingBucket'
class Cdk:
"""
Cdk class that provides methods to run cdk application commands.
Expects system to have NodeJS, AWS CLI and CDK installed globally and have their paths setup as env variables.
"""
def __init__(self, cdk_path: str, project: str, account_id: str,
workspace: pytest.fixture, session: boto3.session.Session):
"""
@@ -49,12 +52,24 @@ class Cdk:
env=self._cdk_env,
shell=True)
def bootstrap(self) -> None:
"""
Deploy the bootstrap stack.
"""
bootstrap_cmd = ['cdk', 'bootstrap',
f'aws://{self._cdk_env["O3DE_AWS_DEPLOY_ACCOUNT"]}/{self._cdk_env["O3DE_AWS_DEPLOY_REGION"]}']
process_utils.check_call(
bootstrap_cmd,
cwd=self._cdk_path,
env=self._cdk_env,
shell=True)
def list(self) -> List[str]:
"""
lists cdk stack names
:return List of cdk stack names
"""
if not self._cdk_path:
return []
@@ -126,6 +141,38 @@ class Cdk:
self._stacks = []
self._cdk_path = ''
@staticmethod
def remove_bootstrap_stack(aws_utils: pytest.fixture) -> None:
"""
Remove the CDK bootstrap stack.
:param aws_utils: aws_utils fixture.
"""
# Check if the bootstrap stack exists.
response = aws_utils.client('cloudformation').describe_stacks(
StackName=BOOTSTRAP_STACK_NAME
)
stacks = response.get('Stacks', [])
if not stacks:
return
# Clear the bootstrap staging bucket before deleting the bootstrap stack.
response = aws_utils.client('cloudformation').describe_stack_resource(
StackName=BOOTSTRAP_STACK_NAME,
LogicalResourceId=BOOTSTRAP_STAGING_BUCKET_LOGIC_ID
)
staging_bucket_name = response.get('StackResourceDetail', {}).get('PhysicalResourceId', '')
if staging_bucket_name:
s3 = aws_utils.resource('s3')
bucket = s3.Bucket(staging_bucket_name)
for key in bucket.objects.all():
key.delete()
# Delete the bootstrap stack.
aws_utils.client('cloudformation').delete_stack(
StackName=BOOTSTRAP_STACK_NAME
)
@pytest.fixture(scope='function')
def cdk(
@@ -134,6 +181,7 @@ def cdk(
feature_name: str,
workspace: pytest.fixture,
aws_utils: pytest.fixture,
bootstrap_required: bool = True,
destroy_stacks_on_teardown: bool = True) -> Cdk:
"""
Fixture for setting up a Cdk
@@ -143,6 +191,8 @@ def cdk(
:param feature_name: Feature gem name to expect cdk folder in.
:param workspace: ly_test_tools workspace fixture.
:param aws_utils: aws_utils fixture.
:param bootstrap_required: Whether the bootstrap stack needs to be created to
provision resources the AWS CDK needs to perform the deployment.
:param destroy_stacks_on_teardown: option to control calling destroy ot the end of test.
:return Cdk class object.
"""
@@ -150,9 +200,14 @@ def cdk(
cdk_path = f'{workspace.paths.engine_root()}/Gems/{feature_name}/cdk'
cdk_obj = Cdk(cdk_path, project, aws_utils.assume_account_id(), workspace, aws_utils.assume_session())
if bootstrap_required:
cdk_obj.bootstrap()
def teardown():
if destroy_stacks_on_teardown:
cdk_obj.destroy()
cdk_obj.remove_bootstrap_stack(aws_utils)
request.addfinalizer(teardown)
return cdk_obj