298 lines
12 KiB
Python
298 lines
12 KiB
Python
import boto3
|
|
import json
|
|
import pytest
|
|
from botocore.exceptions import ClientError
|
|
from moto import mock_cloudformation, mock_organizations, mock_sqs, settings
|
|
from moto.core import DEFAULT_ACCOUNT_ID
|
|
from moto.cloudformation import cloudformation_backends as cf_backends
|
|
from moto.sqs import sqs_backends
|
|
from unittest import SkipTest, TestCase
|
|
from uuid import uuid4
|
|
|
|
|
|
TEMPLATE_DCT = {
|
|
"AWSTemplateFormatVersion": "2010-09-09",
|
|
"Description": "Template creating one queue",
|
|
"Resources": {
|
|
"MyQueue": {"Type": "AWS::SQS::Queue", "Properties": {"QueueName": "testqueue"}}
|
|
},
|
|
}
|
|
|
|
|
|
class TestStackSetMultipleAccounts(TestCase):
|
|
def setUp(self) -> None:
|
|
# Set up multiple organizations/accounts
|
|
# Our tests will create resources within these orgs/accounts
|
|
#
|
|
# | ROOT |
|
|
# / \
|
|
# / \
|
|
# [ou01] [ou02]
|
|
# | / | \
|
|
# | / [acct02] \
|
|
# [acct01] / \
|
|
# [ou21] [ou22]
|
|
# | |
|
|
# [acct21] [acct22]
|
|
#
|
|
if settings.TEST_SERVER_MODE:
|
|
raise SkipTest(
|
|
"These tests directly access the backend, which is not possible in ServerMode"
|
|
)
|
|
mockname = "mock-account"
|
|
mockdomain = "moto-example.org"
|
|
mockemail = "@".join([mockname, mockdomain])
|
|
|
|
self.org = boto3.client("organizations", region_name="us-east-1")
|
|
self.org.create_organization(FeatureSet="ALL")
|
|
self.acct01 = self.org.create_account(AccountName=mockname, Email=mockemail)[
|
|
"CreateAccountStatus"
|
|
]["AccountId"]
|
|
self.acct02 = self.org.create_account(AccountName=mockname, Email=mockemail)[
|
|
"CreateAccountStatus"
|
|
]["AccountId"]
|
|
self.acct21 = self.org.create_account(AccountName=mockname, Email=mockemail)[
|
|
"CreateAccountStatus"
|
|
]["AccountId"]
|
|
self.acct22 = self.org.create_account(AccountName=mockname, Email=mockemail)[
|
|
"CreateAccountStatus"
|
|
]["AccountId"]
|
|
|
|
root_id = self.org.list_roots()["Roots"][0]["Id"]
|
|
|
|
self.ou01 = self.org.create_organizational_unit(ParentId=root_id, Name="ou1")[
|
|
"OrganizationalUnit"
|
|
]["Id"]
|
|
self.ou02 = self.org.create_organizational_unit(ParentId=root_id, Name="ou2")[
|
|
"OrganizationalUnit"
|
|
]["Id"]
|
|
self.ou21 = self.org.create_organizational_unit(
|
|
ParentId=self.ou02, Name="ou021"
|
|
)["OrganizationalUnit"]["Id"]
|
|
self.ou22 = self.org.create_organizational_unit(
|
|
ParentId=self.ou02, Name="ou022"
|
|
)["OrganizationalUnit"]["Id"]
|
|
self.org.move_account(
|
|
AccountId=self.acct01, SourceParentId=root_id, DestinationParentId=self.ou01
|
|
)
|
|
self.org.move_account(
|
|
AccountId=self.acct02, SourceParentId=root_id, DestinationParentId=self.ou02
|
|
)
|
|
self.org.move_account(
|
|
AccountId=self.acct21, SourceParentId=root_id, DestinationParentId=self.ou21
|
|
)
|
|
self.org.move_account(
|
|
AccountId=self.acct22, SourceParentId=root_id, DestinationParentId=self.ou22
|
|
)
|
|
|
|
self.name = "a" + str(uuid4())[0:6]
|
|
self.client = boto3.client("cloudformation", region_name="us-east-1")
|
|
|
|
def _verify_stack_instance(self, accnt, region=None):
|
|
resp = self.client.describe_stack_instance(
|
|
StackSetName=self.name,
|
|
StackInstanceAccount=accnt,
|
|
StackInstanceRegion=region or "us-east-1",
|
|
)["StackInstance"]
|
|
assert resp["Account"] == accnt
|
|
|
|
def _verify_queues(self, accnt, expected, region=None):
|
|
assert (
|
|
list(sqs_backends[accnt][region or "us-east-1"].queues.keys()) == expected
|
|
)
|
|
|
|
|
|
@mock_cloudformation
|
|
@mock_organizations
|
|
@mock_sqs
|
|
class TestServiceManagedStacks(TestStackSetMultipleAccounts):
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
self.client.create_stack_set(
|
|
StackSetName=self.name,
|
|
Description="test creating stack set in multiple accounts",
|
|
TemplateBody=json.dumps(TEMPLATE_DCT),
|
|
PermissionModel="SERVICE_MANAGED",
|
|
AutoDeployment={"Enabled": False},
|
|
)
|
|
|
|
def test_create_instances__specifying_accounts(self):
|
|
with pytest.raises(ClientError) as exc:
|
|
self.client.create_stack_instances(
|
|
StackSetName=self.name, Accounts=["888781156701"], Regions=["us-east-1"]
|
|
)
|
|
err = exc.value.response["Error"]
|
|
assert err["Code"] == "ValidationError"
|
|
assert (
|
|
err["Message"]
|
|
== "StackSets with SERVICE_MANAGED permission model can only have OrganizationalUnit as target"
|
|
)
|
|
|
|
def test_create_instances__specifying_only_accounts_in_deployment_targets(self):
|
|
with pytest.raises(ClientError) as exc:
|
|
self.client.create_stack_instances(
|
|
StackSetName=self.name,
|
|
DeploymentTargets={
|
|
"Accounts": ["888781156701"],
|
|
},
|
|
Regions=["us-east-1"],
|
|
)
|
|
err = exc.value.response["Error"]
|
|
assert err["Code"] == "ValidationError"
|
|
assert err["Message"] == "OrganizationalUnitIds are required"
|
|
|
|
def test_create_instances___with_invalid_ou(self):
|
|
with pytest.raises(ClientError) as exc:
|
|
self.client.create_stack_instances(
|
|
StackSetName=self.name,
|
|
DeploymentTargets={"OrganizationalUnitIds": ["unknown"]},
|
|
Regions=["us-east-1"],
|
|
)
|
|
err = exc.value.response["Error"]
|
|
assert err["Code"] == "ValidationError"
|
|
assert (
|
|
err["Message"]
|
|
== "1 validation error detected: Value '[unknown]' at 'deploymentTargets.organizationalUnitIds' failed to satisfy constraint: Member must satisfy constraint: [Member must have length less than or equal to 68, Member must have length greater than or equal to 6, Member must satisfy regular expression pattern: ^(ou-[a-z0-9]{4,32}-[a-z0-9]{8,32}|r-[a-z0-9]{4,32})$]"
|
|
)
|
|
|
|
def test_create_instances__single_ou(self):
|
|
self.client.create_stack_instances(
|
|
StackSetName=self.name,
|
|
DeploymentTargets={"OrganizationalUnitIds": [self.ou01]},
|
|
Regions=["us-east-1"],
|
|
)
|
|
self._verify_stack_instance(self.acct01)
|
|
|
|
# We have a queue in our account connected to OU-01
|
|
self._verify_queues(self.acct01, ["testqueue"])
|
|
# No queue has been created in our main account, or unrelated accounts
|
|
self._verify_queues(DEFAULT_ACCOUNT_ID, [])
|
|
self._verify_queues(self.acct02, [])
|
|
self._verify_queues(self.acct21, [])
|
|
self._verify_queues(self.acct22, [])
|
|
|
|
def test_create_instances__ou_with_kiddos(self):
|
|
self.client.create_stack_instances(
|
|
StackSetName=self.name,
|
|
DeploymentTargets={"OrganizationalUnitIds": [self.ou02]},
|
|
Regions=["us-east-1"],
|
|
)
|
|
self._verify_stack_instance(self.acct02)
|
|
self._verify_stack_instance(self.acct21)
|
|
self._verify_stack_instance(self.acct22)
|
|
|
|
# No queue has been created in our main account
|
|
self._verify_queues(DEFAULT_ACCOUNT_ID, expected=[])
|
|
self._verify_queues(self.acct01, expected=[])
|
|
# But we do have a queue in our (child)-accounts of OU-02
|
|
self._verify_queues(self.acct02, ["testqueue"])
|
|
self._verify_queues(self.acct21, ["testqueue"])
|
|
self._verify_queues(self.acct22, ["testqueue"])
|
|
|
|
|
|
@mock_cloudformation
|
|
@mock_organizations
|
|
@mock_sqs
|
|
class TestSelfManagedStacks(TestStackSetMultipleAccounts):
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
# Assumptions: All the necessary IAM roles are created
|
|
# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/stacksets-prereqs-self-managed.html
|
|
self.client.create_stack_set(
|
|
StackSetName=self.name,
|
|
Description="test creating stack set in multiple accounts",
|
|
TemplateBody=json.dumps(TEMPLATE_DCT),
|
|
)
|
|
|
|
def test_create_instances__specifying_accounts(self):
|
|
self.client.create_stack_instances(
|
|
StackSetName=self.name, Accounts=[self.acct01], Regions=["us-east-2"]
|
|
)
|
|
|
|
self._verify_stack_instance(self.acct01, region="us-east-2")
|
|
|
|
# acct01 has a Stack
|
|
assert len(cf_backends[self.acct01]["us-east-2"].stacks) == 1
|
|
# Other acounts do not
|
|
assert len(cf_backends[self.acct02]["us-east-2"].stacks) == 0
|
|
assert len(cf_backends[self.acct21]["us-east-2"].stacks) == 0
|
|
assert len(cf_backends[self.acct22]["us-east-2"].stacks) == 0
|
|
assert len(cf_backends[DEFAULT_ACCOUNT_ID]["us-east-2"].stacks) == 0
|
|
|
|
# acct01 has a queue
|
|
self._verify_queues(self.acct01, ["testqueue"], region="us-east-2")
|
|
# other accounts are empty
|
|
self._verify_queues(self.acct02, [], region="us-east-2")
|
|
self._verify_queues(self.acct21, [], region="us-east-2")
|
|
self._verify_queues(self.acct22, [], region="us-east-2")
|
|
self._verify_queues(DEFAULT_ACCOUNT_ID, [], region="us-east-2")
|
|
|
|
def test_create_instances__multiple_accounts(self):
|
|
self.client.create_stack_instances(
|
|
StackSetName=self.name,
|
|
Accounts=[self.acct01, self.acct02],
|
|
Regions=["us-east-2"],
|
|
)
|
|
|
|
# acct01 and 02 have a Stack
|
|
assert len(cf_backends[self.acct01]["us-east-2"].stacks) == 1
|
|
assert len(cf_backends[self.acct02]["us-east-2"].stacks) == 1
|
|
# Other acounts do not
|
|
assert len(cf_backends[self.acct21]["us-east-2"].stacks) == 0
|
|
assert len(cf_backends[self.acct22]["us-east-2"].stacks) == 0
|
|
assert len(cf_backends[DEFAULT_ACCOUNT_ID]["us-east-2"].stacks) == 0
|
|
|
|
# acct01 and 02 have the queue
|
|
self._verify_queues(self.acct01, ["testqueue"], region="us-east-2")
|
|
self._verify_queues(self.acct02, ["testqueue"], region="us-east-2")
|
|
# other accounts are empty
|
|
self._verify_queues(self.acct21, [], region="us-east-2")
|
|
self._verify_queues(self.acct22, [], region="us-east-2")
|
|
self._verify_queues(DEFAULT_ACCOUNT_ID, [], region="us-east-2")
|
|
|
|
def test_delete_instances(self):
|
|
# Create two stack instances
|
|
self.client.create_stack_instances(
|
|
StackSetName=self.name,
|
|
Accounts=[self.acct01, self.acct02],
|
|
Regions=["us-east-2"],
|
|
)
|
|
|
|
# Delete one
|
|
self.client.delete_stack_instances(
|
|
StackSetName=self.name,
|
|
Accounts=[self.acct01],
|
|
Regions=["us-east-2"],
|
|
RetainStacks=False,
|
|
)
|
|
|
|
# Act02 still has it's stack
|
|
assert len(cf_backends[self.acct02]["us-east-2"].stacks) == 1
|
|
# Other Stacks are removed
|
|
assert len(cf_backends[self.acct01]["us-east-2"].stacks) == 0
|
|
assert len(cf_backends[self.acct21]["us-east-2"].stacks) == 0
|
|
assert len(cf_backends[self.acct22]["us-east-2"].stacks) == 0
|
|
assert len(cf_backends[DEFAULT_ACCOUNT_ID]["us-east-2"].stacks) == 0
|
|
|
|
# Acct02 still has it's queue as well
|
|
self._verify_queues(self.acct02, ["testqueue"], region="us-east-2")
|
|
# Other queues are removed
|
|
self._verify_queues(self.acct01, [], region="us-east-2")
|
|
self._verify_queues(self.acct21, [], region="us-east-2")
|
|
self._verify_queues(self.acct22, [], region="us-east-2")
|
|
self._verify_queues(DEFAULT_ACCOUNT_ID, [], region="us-east-2")
|
|
|
|
def test_create_instances__single_ou(self):
|
|
with pytest.raises(ClientError) as exc:
|
|
self.client.create_stack_instances(
|
|
StackSetName=self.name,
|
|
DeploymentTargets={"OrganizationalUnitIds": [self.ou01]},
|
|
Regions=["us-east-1"],
|
|
)
|
|
err = exc.value.response["Error"]
|
|
assert err["Code"] == "ValidationError"
|
|
assert (
|
|
err["Message"]
|
|
== "StackSets with SELF_MANAGED permission model can only have accounts as target"
|
|
)
|