234195908

Build # 234195908

Build Type

github-actions

Committed by GitHub

Commit Message

Merge 4258ac828 into eeddff26c

Pull Request Pull Request #160: Implement s3 Hashnaming and artifacts option

Run Details

360 of 381 branches covered (94.49%)

Branch coverage included in aggregate %.

105 of 107 new or added lines in 6 files covered. (98.13%)

3 existing lines in 1 file now uncovered.

1176 of 1205 relevant lines covered (97.59%)

1.95 hits per line

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

92.22

/formica/helper.py

from .s3 import temporary_bucket


def name(*names):
    name = "".join(map(lambda name: name.title(), names))
    name = "".join(e for e in name if e.isalnum())
    return name


def collect_stack_set_vars(args):
    variables = args.vars or {}
    if args.organization_variables:
        variables.update(accounts_and_regions())
    else:
        if args.organization_region_variables:
            variables.update(aws_regions())
        if args.organization_account_variables:
            variables.update(aws_accounts())

    return variables


def collect_vars(args):
    variables = collect_stack_set_vars(args)
    if args.artifacts:
        variables.update(artifact_variables(args.artifacts))

    return variables


def accounts_and_regions():
    acc = aws_accounts()
    acc.update(aws_regions())
    return acc


def aws_regions():
    import boto3

    ec2 = boto3.client("ec2")
    regions = ec2.describe_regions()
    regions = [r["RegionName"] for r in regions["Regions"]]
    return {"AWSRegions": regions}


def aws_accounts():
    import boto3

    organizations = boto3.client("organizations")
    sts = boto3.client("sts")

    paginator = organizations.get_paginator("list_accounts")

    pages = paginator.paginate()
    accounts = [
        {"Id": a["Id"], "Name": a["Name"], "Email": a["Email"]}
        for page in pages
        for a in page["Accounts"]
        if a["Status"] == "ACTIVE"
    ]
    account_id = sts.get_caller_identity()["Account"]
    return {
        "AWSMainAccount": [a for a in accounts if a["Id"] == account_id][0],
        "AWSAccounts": accounts,
        "AWSSubAccounts": [a for a in accounts if a["Id"] != account_id],
    }


def main_account_id():
    import boto3

    sts = boto3.client("sts")
    identity = sts.get_caller_identity()
    return identity["Account"]


def artifact_variables(artifacts):
    class Artifact:
        def __init__(self, key, bucket):
            self.key = key
            self.bucket = bucket

    artifact_keys = {}
    with temporary_bucket() as t:
        for a in artifacts:
            artifact_keys[a] = t.add_file(a)
        finished_vars = {key: Artifact(value, t.name) for key, value in artifact_keys.items()}
    return {"artifacts": finished_vars}


def with_artifacts(function):
    def handle_artifacts(args):
        if args.artifacts:
            with temporary_bucket() as t:
                for a in args.artifacts:
                    t.add_file(a)
                t.upload()
                function(args)
        else:
            function(args)

    return handle_artifacts

1	from .s3 import temporary_bucket	2✔
2
3
4	def name(*names):	2✔
UNCOV 5	name = "".join(map(lambda name: name.title(), names))	×
UNCOV 6	name = "".join(e for e in name if e.isalnum())	×
UNCOV 7	return name	×
8
9
10	def collect_stack_set_vars(args):	2✔
11	variables = args.vars or {}	2✔
12	if args.organization_variables:	2✔
13	variables.update(accounts_and_regions())	2✔
14	else:
15	if args.organization_region_variables:	2✔
16	variables.update(aws_regions())	2✔
17	if args.organization_account_variables:	2✔
18	variables.update(aws_accounts())	2✔
19
20	return variables	2✔
21
22
23	def collect_vars(args):	2✔
24	variables = collect_stack_set_vars(args)	2✔
25	if args.artifacts:	2✔
26	variables.update(artifact_variables(args.artifacts))	2✔
27
28	return variables	2✔
29
30
31	def accounts_and_regions():	2✔
32	acc = aws_accounts()	2✔
33	acc.update(aws_regions())	2✔
34	return acc	2✔
35
36
37	def aws_regions():	2✔
38	import boto3	2✔
39
40	ec2 = boto3.client("ec2")	2✔
41	regions = ec2.describe_regions()	2✔
42	regions = [r["RegionName"] for r in regions["Regions"]]	2✔
43	return {"AWSRegions": regions}	2✔
44
45
46	def aws_accounts():	2✔
47	import boto3	2✔
48
49	organizations = boto3.client("organizations")	2✔
50	sts = boto3.client("sts")	2✔
51
52	paginator = organizations.get_paginator("list_accounts")	2✔
53
54	pages = paginator.paginate()	2✔
55	accounts = [	2✔
56	{"Id": a["Id"], "Name": a["Name"], "Email": a["Email"]}
57	for page in pages
58	for a in page["Accounts"]
59	if a["Status"] == "ACTIVE"
60	]
61	account_id = sts.get_caller_identity()["Account"]	2✔
62	return {	2✔
63	"AWSMainAccount": [a for a in accounts if a["Id"] == account_id][0],
64	"AWSAccounts": accounts,
65	"AWSSubAccounts": [a for a in accounts if a["Id"] != account_id],
66	}
67
68
69	def main_account_id():	2✔
70	import boto3	2✔
71
72	sts = boto3.client("sts")	2✔
73	identity = sts.get_caller_identity()	2✔
74	return identity["Account"]	2✔
75
76
77	def artifact_variables(artifacts):	2✔
78	class Artifact:	2✔
79	def __init__(self, key, bucket):	2✔
80	self.key = key	2✔
81	self.bucket = bucket	2✔
82
83	artifact_keys = {}	2✔
84	with temporary_bucket() as t:	2✔
85	for a in artifacts:	2✔
86	artifact_keys[a] = t.add_file(a)	2✔
87	finished_vars = {key: Artifact(value, t.name) for key, value in artifact_keys.items()}	2✔
88	return {"artifacts": finished_vars}	2✔
89
90
91	def with_artifacts(function):	2✔
92	def handle_artifacts(args):	2✔
93	if args.artifacts:	2✔
94	with temporary_bucket() as t:	2✔
95	for a in args.artifacts:	2✔
96	t.add_file(a)	2✔
97	t.upload()	2✔
98	function(args)	2✔
99	else:
100	function(args)	2✔
101
102	return handle_artifacts	2✔

theserverlessway / formica / 234195908

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