diff --git a/tests/domain/cloud/test_aws_csp.py b/tests/domain/cloud/test_aws_csp.py index d6b37186..c71e3fa0 100644 --- a/tests/domain/cloud/test_aws_csp.py +++ b/tests/domain/cloud/test_aws_csp.py @@ -20,3 +20,31 @@ def test_create_environment_raises_x_when_account_creation_fails(mock_aws): environment = EnvironmentFactory.create() with pytest.raises(EnvironmentCreationException): mock_aws.create_environment(AUTH_CREDENTIALS, environment.creator, environment) + + +def test_create_atat_admin_user_succeeds(mock_aws): + root_user_info = mock_aws.create_atat_admin_user( + AUTH_CREDENTIALS, "csp_environment_id" + ) + assert { + "id": "user-id", + "username": "user-name", + "resource_id": "user-arn", + "credentials": {"key": "access-key-id", "secret_key": "secret-access-key"}, + } == root_user_info + + +@pytest.mark.mock_boto3({"iam.create_user.already_exists": True}) +def test_create_atat_admin_when_user_already_exists(mock_aws): + root_user_info = mock_aws.create_atat_admin_user( + AUTH_CREDENTIALS, "csp_environment_id" + ) + assert { + "id": "user-id", + "username": "user-name", + "resource_id": "user-arn", + "credentials": {"key": "access-key-id", "secret_key": "secret-access-key"}, + } == root_user_info + + iam_client = mock_aws.boto3.client("iam") + iam_client.get_user.assert_any_call(UserName="atat") diff --git a/tests/mock_boto3.py b/tests/mock_boto3.py index 5a646936..4926a100 100644 --- a/tests/mock_boto3.py +++ b/tests/mock_boto3.py @@ -52,14 +52,103 @@ def mock_boto_organizations(_config=None, **kwargs): return mock +def mock_boto_iam(_config=None, **kwargs): + user_already_exists = _config.get("iam.create_user.already_exists", False) + + import boto3 + + real_iam_client = boto3.client("iam", **kwargs) + mock = Mock(wraps=real_iam_client) + mock.exceptions.EntityAlreadyExistsException = ( + real_iam_client.exceptions.EntityAlreadyExistsException + ) + + mock.put_user_policy = Mock(return_value={"ResponseMetadata": {}}) + + if user_already_exists: + + def _raise(**kwargs): + raise real_iam_client.exceptions.EntityAlreadyExistsException( + {"Error": {}}, "operation-name" + ) + + mock.create_user = Mock(side_effect=_raise) + else: + mock.create_user = Mock( + return_value={ + "User": { + "UserId": "user-id", + "Arn": "user-arn", + "UserName": "user-name", + } + } + ) + + mock.get_user = Mock( + return_value={ + "User": {"UserId": "user-id", "Arn": "user-arn", "UserName": "user-name"} + } + ) + + mock.create_access_key = Mock( + return_value={ + "AccessKey": { + "AccessKeyId": "access-key-id", + "SecretAccessKey": "secret-access-key", + } + } + ) + + return mock + + +def mock_boto_sts(_config=None, **kwargs): + import boto3 + + mock = Mock(wraps=boto3.client("sts", **kwargs)) + mock.assume_role = Mock( + return_value={ + "Credentials": { + "AccessKeyId": "access-key-id", + "SecretAccessKey": "secret-access-key", + "SessionToken": "session-token", + } + } + ) + + return mock + + class MockBoto3: + CLIENTS = { + "organizations": mock_boto_organizations, + "iam": mock_boto_iam, + "sts": mock_boto_sts, + } + def __init__(self, config=None): self.config = config or {} + self.client_instances = {} def client(self, client_name, **kwargs): - return {"organizations": mock_boto_organizations}[client_name]( - **kwargs, _config=self.config - ) + """ + Return a new mock client for the given `client_name`, either by + retrieving it from the `client_instances` cache or by instantiating + it for the first time. + + Params should be the same ones you'd pass to `boto3.client`. + """ + + if client_name in self.client_instances: + return self.client_instances[client_name] + + try: + client_fn = self.CLIENTS[client_name] + client_instance = client_fn(**kwargs, _config=self.config) + self.client_instances[client_name] = client_instance + return client_instance + except KeyError: + raise ValueError(f"MockBoto3: {client_name} client is not yet implemented.") @pytest.fixture(scope="function")