Skip to content

Commit

Permalink
Fix pylint tests in cloudformation.py (#31)
Browse files Browse the repository at this point in the history
  • Loading branch information
@rmc3
Jeffrey Dugas authored and rmc3 committed May 29, 2018
1 parent 579873c commit 881364f
Showing 1 changed file with 69 additions and 85 deletions.
154 changes: 69 additions & 85 deletions cumulogenesis/services/cloudformation.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@
from botocore.client import ClientError

from cumulogenesis.log_handling import LOGGER as logger
# raise NotImplementedError()


class CloudformationService(object):
Expand All @@ -22,8 +21,8 @@ def __init__(self, session_builder):
#
# Remove the below once implemented.
#
#self.session = session
#self.client = boto3.client('cloudformation', session)
self.session = session_builder.get_base_session()
self.client = boto3.client('cloudformation', self.session)

def upsert_stackset(self, organization, stackset_name):
'''
Expand All @@ -35,13 +34,12 @@ def upsert_stackset(self, organization, stackset_name):
all required parameter stacks have been upserted, necessary account,
orgunit resources are in place).
'''
aws_stackset = self._retrieve_stackset(stackset_name=stackset.name)
logger.info('Using organization rooted in account %s', organization.root_account_id)
logger.info('Discovering stackset %s', stackset_name)
aws_stackset = self._retrieve_stackset(stackset_name=stackset_name)
if aws_stackset:
logger.debug(aws_stackset)
# involves diff of a stack-set's innate attributes
return self._update_stackset(stackset=stackset)

return self._create_stackset(stackset=stackset)
return self._update_stackset(stackset=aws_stackset)
return self._create_stackset(stackset=aws_stackset)

def upsert_account_parameter_stacks(self, organization):
'''
Expand Down Expand Up @@ -76,47 +74,71 @@ def upsert_account_parameter_stacks(self, organization):
The caller is responsible for ensuring that the account exists
before calling this method.
'''
pass
raise NotImplementedError()

def load_stacksets(self, organization):
def load_stacksets(self, organization, status='ACTIVE'):
'''
Not implemented
Loads existing policies into organization.policies.
This should update organization.stacksets with existing stacksets.
The caller is responsible for ensuring that accounts and orgunits
have already been loaded into the Organization model so that policy
associations can be loaded into them.
'''
pass
stack_sets = []
paginator = self.client.get_paginator('list_stack_sets')

def _retrieve_stackset(self, stackset_name):
'''
{
'StackSet': {
'StackSetName': 'string',
'StackSetId': 'string',
'Description': 'string',
'Status': 'ACTIVE'|'DELETED',
'TemplateBody': 'string',
'Parameters': [
for page in paginator.paginate(Status=status):
if page['Summaries']:
stack_sets = stack_sets + page['Summaries']

for stack_set in stack_sets:
name = stack_set['StackSetName']
stack_set_detail = self._retrieve_stackset(stackset_name=name)
stack_instances = self._retrieve_stackset_instances(stackset_name=name)

stack_set_model = {
'name': name,
'id': stack_set['StackSetId'],
'description': stack_set['Description'],
'status': stack_set['Status'],
'arn': stack_set_detail['StackSetARN'],
'administration_role_arn': stack_set_detail['AdministrationRoleARN'],
'capabilities': stack_set_detail['Capabilities'],
'parameters': [
{
'ParameterKey': 'string',
'ParameterValue': 'string',
'UsePreviousValue': True|False,
'ResolvedValue': 'string'
},
],
'Capabilities': [
'CAPABILITY_IAM'|'CAPABILITY_NAMED_IAM',
'key': parameter['ParameterKey'],
'value': parameter['ParameterValue'],
'use_previous': parameter['UsePreviousValue']
}
for parameter in stack_set_detail['Parameters']
],
'Tags': [
'tags': [
{
'Key': 'string',
'Value': 'string'
},
'key': tag['Key'],
'value': tag['Value']
}
for tag in stack_set_detail['Tags']
],
'StackSetARN': 'string',
'AdministrationRoleARN': 'string'
'template_body': stack_set_detail['TemplateBody'],
'instances': [
{
'stack_set_id': instance['StackSetId'],
'region': instance['Region'],
'account': instance['Account'],
'stack_id': instance['StackId'],
'status': instance['Status'],
'reason': instance.get('StatusReason', None)
}
for instance in stack_instances
]
}
}
'''

organization.stack_sets[name] = stack_set_model
# policy_targets_res = self.client.list_targets_for_policy(PolicyId=policy["Id"])
# for target in policy_targets_res["Targets"]:
# self._add_policy_to_target(org_model=organization, target=target, policy_name=policy['Name'])

def _retrieve_stackset(self, stackset_name):
try:
response = self.client.describe_stack_set(
StackSetName=stackset_name
Expand All @@ -126,40 +148,13 @@ def _retrieve_stackset(self, stackset_name):
if err.response['Error']['Code'] != "ResourceNotFoundException":
print('Error: {}.'.format(err.response['Error']['Message']))
raise err
return None
else:
return None

def _create_stackset(self, stackset):
'''
{
'StackSetId': 'string'
}
'''
response = self.client.create_stack_set(
StackSetName=stackset.name,
Description='string',
TemplateBody='string',
TemplateURL='string',
# Parameters=[
# {
# 'ParameterKey': 'string',
# 'ParameterValue': 'string',
# 'UsePreviousValue': False,
# 'ResolvedValue': 'string'
# },
# ],
Capabilities=[
'CAPABILITY_IAM',
'CAPABILITY_NAMED_IAM'
],
# Tags=[
# {
# 'Key': 'string',
# 'Value': 'string'
# }
# ]
AdministrationRoleARN='string'
)
return response
raise NotImplementedError()

def _update_stackset(self, stackset):
'''
Expand Down Expand Up @@ -237,32 +232,21 @@ def upsert_stack_instance(self, stackset, accounts):
)
return response

def _retrieve_stackset_instances(self, stackset):
def _retrieve_stackset_instances(self, stackset_name):
'''
Sadly, boto3 1.7.12 lacks a paginator for `{cloudformation client}.list_stack_instances`,
so we create our own.
[
{
'StackSetId': 'string',
'Region': 'string',
'Account': 'string',
'StackId': 'string',
'Status': 'CURRENT'|'OUTDATED'|'INOPERABLE',
'StatusReason': 'string'
}
]
'''
results = []
next_token = None
while True:
response = self._page_stackset_instances(
stackset_name=stackset.name,
stackset_name=stackset_name,
next_token=next_token
)
results.extend(response.get('Summaries'))
next_token = response.get('NextToken')
next_token = response.get('NextToken', None)
if next_token is None:
break
return results

def _page_stackset_instances(self, stackset_name, next_token=None):
'''
Expand Down

0 comments on commit 881364f

Please sign in to comment.