483114725

Build Type

github-actions

Committed by GitHub

Commit Message

Merge 4a412a000 into e55ddd656

Pull Request Pull Request #171: Add option to create nested Stack Sets

Run Details

376 of 401 branches covered (93.77%)

Branch coverage included in aggregate %.

8 of 8 new or added lines in 2 files covered. (100.0%)

3 existing lines in 1 file now uncovered.

1215 of 1248 relevant lines covered (97.36%)

1.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.18

/formica/change_set.py

import json
import sys

import logging
from formica.s3 import temporary_bucket
from botocore.exceptions import ClientError, WaiterError
from texttable import Texttable
import boto3

from formica import CHANGE_SET_FORMAT
import time

CHANGE_SET_HEADER = ["Action", "LogicalId", "PhysicalId", "Type", "Replacement", "Changed"]

logger = logging.getLogger(__name__)

cf = boto3.client("cloudformation")


class ChangeSet:
    def create(
        self,
        template="",
        change_set_type="",
        parameters=None,
        tags=None,
        capabilities=None,
        role_arn=None,
        s3=False,
        resource_types=False,
        use_previous_template=False,
        use_previous_parameters=False,
    ):
        optional_arguments = {}
        parameters_set = []
        if use_previous_parameters:
            stacks = cf.describe_stacks(StackName=self.stack)
            parameters_set = [
                {"ParameterKey": p["ParameterKey"], "UsePreviousValue": True}
                for p in stacks["Stacks"][0]["Parameters"]
            ]
        if parameters:
            for (key, value) in parameters.items():
                item = next((x for x in parameters_set if x["ParameterKey"] == key), None)
                values = {"ParameterKey": key, "ParameterValue": str(value), "UsePreviousValue": False}
                if item:
                    item.update(values)
                else:
                    parameters_set.append(values)

        if parameters_set:
            optional_arguments["Parameters"] = parameters_set

        if tags:
            optional_arguments["Tags"] = [{"Key": key, "Value": str(value)} for (key, value) in tags.items()]
        if role_arn:
            optional_arguments["RoleARN"] = role_arn
        if capabilities:
            optional_arguments["Capabilities"] = capabilities
        if change_set_type == "UPDATE":
            self.remove_existing_changeset()
        if resource_types:
            optional_arguments["ResourceTypes"] = list(
                set([resource["Type"] for key, resource in json.loads(template)["Resources"].items()])
            )

        if use_previous_template:
            optional_arguments["UsePreviousTemplate"] = True
            self.__change_and_wait(change_set_type, optional_arguments)
        else:
            if s3:
                with temporary_bucket(self.stack) as t:
                    file_name = t.add(template)
                    t.upload()
                    bucket_name = t.name
                    template_url = "https://{}.s3.amazonaws.com/{}".format(bucket_name, file_name)
                    self.__change_and_wait(change_set_type, {"TemplateURL": template_url, **optional_arguments})
            else:
                self.__change_and_wait(change_set_type, {"TemplateBody": template, **optional_arguments})

    def __change_and_wait(self, change_set_type, optional_arguments):
        try:
            cf.create_change_set(
                StackName=self.stack,
                ChangeSetName=self.name,
                ChangeSetType=change_set_type,
                **optional_arguments,
                IncludeNestedStacks=self.nested_change_sets,
            )
            logger.info("Change set submitted, waiting for CloudFormation to calculate changes ...")
            waiter = cf.get_waiter("change_set_create_complete")
            waiter.wait(ChangeSetName=self.name, StackName=self.stack, WaiterConfig=dict(Delay=10, MaxAttempts=120))
            logger.info("Change set created successfully")
        except WaiterError as e:
            status_reason = e.last_response.get("StatusReason", "")
            logger.info(status_reason)
            if "didn't contain changes" not in status_reason:
                sys.exit(1)

    def __init__(self, stack, arn="", nested_change_sets=False):
        self.name = CHANGE_SET_FORMAT.format(stack=stack)
        self.stack = stack
        self.change_set_arn = arn
        self.nested_change_sets = nested_change_sets

    def describe(self, print_metadata=True):
        if self.change_set_arn:
            cs_options = dict(ChangeSetName=self.change_set_arn)
        else:
            cs_options = dict(StackName=self.stack, ChangeSetName=self.name)
        change_set = cf.describe_change_set(**cs_options)
        table = Texttable(max_width=150)

        if print_metadata:
            logger.info("Deployment metadata:")
            parameters = ", ".join(
                [
                    parameter["ParameterKey"] + "=" + parameter["ParameterValue"]
                    for parameter in change_set.get("Parameters", [])
                ]
            )
            table.add_row(["Parameters", parameters])
            tags = [tag["Key"] + "=" + tag["Value"] for tag in change_set.get("Tags", [])]
            table.add_row(["Tags ", ", ".join(tags)])
            table.add_row(["Capabilities ", ", ".join(change_set.get("Capabilities", []))])
            logger.info(table.draw() + "\n")
            logger.info("Resource Changes:")

        table.reset()
        table = Texttable(max_width=150)
        table.add_rows([CHANGE_SET_HEADER])

        def __change_detail(change):
            target_ = change["Target"]
            attribute = target_["Attribute"]
            if attribute == "Properties":
                return target_.get("Name", "")
            else:
                return attribute

        for change in change_set["Changes"]:
            resource_change = change["ResourceChange"]
            table.add_row(
                [
                    resource_change["Action"],
                    resource_change["LogicalResourceId"],
                    resource_change.get("PhysicalResourceId", ""),
                    resource_change["ResourceType"],
                    resource_change.get("Replacement", ""),
                    ", ".join(sorted(set([__change_detail(c) for c in resource_change["Details"]]))),
                ]
            )

        logger.info(table.draw())

        nested_change_sets = [
            (change["ResourceChange"]["LogicalResourceId"], change["ResourceChange"]["ChangeSetId"])
            for change in change_set["Changes"]
            if change["ResourceChange"]["ResourceType"] == "AWS::CloudFormation::Stack"
            and change["ResourceChange"].get("ChangeSetId")
        ]
        for nested_change_set in nested_change_sets:
            logger.info(f"\nChanges for nested Stack: {nested_change_set[0]}")
            ChangeSet(stack=self.stack, arn=nested_change_set[1]).describe(print_metadata=False)

    def remove_existing_changeset(self):
        try:
            id = cf.describe_change_set(StackName=self.stack, ChangeSetName=self.name)["ChangeSetId"]
            logger.info("Removing existing change set")
            cf.delete_change_set(ChangeSetName=id)

            # Sleep to let Cloudformation remove
            for _ in range(100):
                cf.describe_change_set(StackName=self.stack, ChangeSetName=self.name)
                time.sleep(5)
            raise Exception("Old Change Set could not be removed, please retry")
        except ClientError as e:
            if e.response["Error"]["Code"] != "ChangeSetNotFound":
                raise e

1	import json	2✔
2	import sys	2✔
3
4	import logging	2✔
5	from formica.s3 import temporary_bucket	2✔
6	from botocore.exceptions import ClientError, WaiterError	2✔
7	from texttable import Texttable	2✔
8	import boto3	2✔
9
10	from formica import CHANGE_SET_FORMAT	2✔
11	import time	2✔
12
13	CHANGE_SET_HEADER = ["Action", "LogicalId", "PhysicalId", "Type", "Replacement", "Changed"]	2✔
14
15	logger = logging.getLogger(__name__)	2✔
16
17	cf = boto3.client("cloudformation")	2✔
18
19
20	class ChangeSet:	2✔
21	def create(	2✔
22	self,
23	template="",
24	change_set_type="",
25	parameters=None,
26	tags=None,
27	capabilities=None,
28	role_arn=None,
29	s3=False,
30	resource_types=False,
31	use_previous_template=False,
32	use_previous_parameters=False,
33	):
34	optional_arguments = {}	2✔
35	parameters_set = []	2✔
36	if use_previous_parameters:	2✔
37	stacks = cf.describe_stacks(StackName=self.stack)	2✔
38	parameters_set = [	2✔
39	{"ParameterKey": p["ParameterKey"], "UsePreviousValue": True}
40	for p in stacks["Stacks"][0]["Parameters"]
41	]
42	if parameters:	2✔
43	for (key, value) in parameters.items():	2✔
44	item = next((x for x in parameters_set if x["ParameterKey"] == key), None)	2✔
45	values = {"ParameterKey": key, "ParameterValue": str(value), "UsePreviousValue": False}	2✔
46	if item:	2!
47	item.update(values)	×
48	else:
49	parameters_set.append(values)	2✔
50
51	if parameters_set:	2✔
52	optional_arguments["Parameters"] = parameters_set	2✔
53
54	if tags:	2✔
55	optional_arguments["Tags"] = [{"Key": key, "Value": str(value)} for (key, value) in tags.items()]	2✔
56	if role_arn:	2✔
57	optional_arguments["RoleARN"] = role_arn	2✔
58	if capabilities:	2✔
59	optional_arguments["Capabilities"] = capabilities	2✔
60	if change_set_type == "UPDATE":	2✔
61	self.remove_existing_changeset()	2✔
62	if resource_types:	2✔
63	optional_arguments["ResourceTypes"] = list(	2✔
64	set([resource["Type"] for key, resource in json.loads(template)["Resources"].items()])
65	)
66
67	if use_previous_template:	2✔
68	optional_arguments["UsePreviousTemplate"] = True	2✔
69	self.__change_and_wait(change_set_type, optional_arguments)	2✔
70	else:
71	if s3:	2✔
72	with temporary_bucket(self.stack) as t:	2✔
73	file_name = t.add(template)	2✔
74	t.upload()	2✔
75	bucket_name = t.name	2✔
76	template_url = "https://{}.s3.amazonaws.com/{}".format(bucket_name, file_name)	2✔
77	self.__change_and_wait(change_set_type, {"TemplateURL": template_url, **optional_arguments})	2✔
78	else:
79	self.__change_and_wait(change_set_type, {"TemplateBody": template, **optional_arguments})	2✔
80
81	def __change_and_wait(self, change_set_type, optional_arguments):	2✔
82	try:	2✔
83	cf.create_change_set(	2✔
84	StackName=self.stack,
85	ChangeSetName=self.name,
86	ChangeSetType=change_set_type,
87	**optional_arguments,
88	IncludeNestedStacks=self.nested_change_sets,
89	)
90	logger.info("Change set submitted, waiting for CloudFormation to calculate changes ...")	2✔
91	waiter = cf.get_waiter("change_set_create_complete")	2✔
92	waiter.wait(ChangeSetName=self.name, StackName=self.stack, WaiterConfig=dict(Delay=10, MaxAttempts=120))	2✔
93	logger.info("Change set created successfully")	2✔
94	except WaiterError as e:	2✔
95	status_reason = e.last_response.get("StatusReason", "")	2✔
96	logger.info(status_reason)	2✔
97	if "didn't contain changes" not in status_reason:	2✔
98	sys.exit(1)	2✔
99
100	def __init__(self, stack, arn="", nested_change_sets=False):	2✔
101	self.name = CHANGE_SET_FORMAT.format(stack=stack)	2✔
102	self.stack = stack	2✔
103	self.change_set_arn = arn	2✔
104	self.nested_change_sets = nested_change_sets	2✔
105
106	def describe(self, print_metadata=True):	2✔
107	if self.change_set_arn:	2!
UNCOV 108	cs_options = dict(ChangeSetName=self.change_set_arn)	×
109	else:
110	cs_options = dict(StackName=self.stack, ChangeSetName=self.name)	2✔
111	change_set = cf.describe_change_set(**cs_options)	2✔
112	table = Texttable(max_width=150)	2✔
113
114	if print_metadata:	2!
115	logger.info("Deployment metadata:")	2✔
116	parameters = ", ".join(	2✔
117	[
118	parameter["ParameterKey"] + "=" + parameter["ParameterValue"]
119	for parameter in change_set.get("Parameters", [])
120	]
121	)
122	table.add_row(["Parameters", parameters])	2✔
123	tags = [tag["Key"] + "=" + tag["Value"] for tag in change_set.get("Tags", [])]	2✔
124	table.add_row(["Tags ", ", ".join(tags)])	2✔
125	table.add_row(["Capabilities ", ", ".join(change_set.get("Capabilities", []))])	2✔
126	logger.info(table.draw() + "\n")	2✔
127	logger.info("Resource Changes:")	2✔
128
129	table.reset()	2✔
130	table = Texttable(max_width=150)	2✔
131	table.add_rows([CHANGE_SET_HEADER])	2✔
132
133	def __change_detail(change):	2✔
134	target_ = change["Target"]	2✔
135	attribute = target_["Attribute"]	2✔
136	if attribute == "Properties":	2✔
137	return target_.get("Name", "")	2✔
138	else:
139	return attribute	2✔
140
141	for change in change_set["Changes"]:	2✔
142	resource_change = change["ResourceChange"]	2✔
143	table.add_row(	2✔
144	[
145	resource_change["Action"],
146	resource_change["LogicalResourceId"],
147	resource_change.get("PhysicalResourceId", ""),
148	resource_change["ResourceType"],
149	resource_change.get("Replacement", ""),
150	", ".join(sorted(set([__change_detail(c) for c in resource_change["Details"]]))),
151	]
152	)
153
154	logger.info(table.draw())	2✔
155
156	nested_change_sets = [	2✔
157	(change["ResourceChange"]["LogicalResourceId"], change["ResourceChange"]["ChangeSetId"])
158	for change in change_set["Changes"]
159	if change["ResourceChange"]["ResourceType"] == "AWS::CloudFormation::Stack"
160	and change["ResourceChange"].get("ChangeSetId")
161	]
162	for nested_change_set in nested_change_sets:	2!
UNCOV 163	logger.info(f"\nChanges for nested Stack: {nested_change_set[0]}")	×
UNCOV 164	ChangeSet(stack=self.stack, arn=nested_change_set[1]).describe(print_metadata=False)	×
165
166	def remove_existing_changeset(self):	2✔
167	try:	2✔
168	id = cf.describe_change_set(StackName=self.stack, ChangeSetName=self.name)["ChangeSetId"]	2✔
169	logger.info("Removing existing change set")	2✔
170	cf.delete_change_set(ChangeSetName=id)	2✔
171
172	# Sleep to let Cloudformation remove
173	for _ in range(100):	2✔
174	cf.describe_change_set(StackName=self.stack, ChangeSetName=self.name)	2✔
175	time.sleep(5)	2✔
176	raise Exception("Old Change Set could not be removed, please retry")	2✔
177	except ClientError as e:	2✔
178	if e.response["Error"]["Code"] != "ChangeSetNotFound":	2✔
179	raise e	2✔

theserverlessway / formica / 483114725

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous