2

我正在尝试使用 pytest 和 moto 创建一个测试,检查StackStatus字典中的是否从函数返回list_stacks()https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/cloudformation.html# CloudFormation.Client.list_stacks)是DELETE_COMPLETE.

我为客户端连接创建了一个 pytest.fixture:

@pytest.fixture(scope='function')
def cf(aws_credentials):
    with mock_cloudformation():
        yield boto3.client('cloudformation')

我创建了一个虚拟模板:

@pytest.fixture(scope='function')
def template_body_data():
    'The Cloud Formation template'
    template_data = {
        'Resources': {'MyS3Bucket': {'Type': 'AWS::S3::Bucket', 'Properties': {}}}
    }
    return template_data

在我的测试函数中,我创建了堆栈,在我删除之后:

@mock_cloudformation
def test_deleted_stack(cf, template_body_data):
    params = {'StackName': 'teste', 'TemplateBody': yaml.dump(template_body_data)}
    cf.create_stack(**params)
    cf.delete_stack(StackName='teste')
    assert check_stack_exists('teste') is False

睾丸的功能是这个:

def check_stack_exists(stack_name):
    cf = get_client()
    list_stack = cf.list_stacks()['StackSummaries']
    for stack in list_stack:
        if stack['StackStatus'] == 'DELETE_COMPLETE':
            continue
        if stack['StackName'] == stack_name:
            return True
    return False

我面临一个问题:

它抱怨我的模板没有 BucketName:


___________________ test_check_stack_exists_deleted_stack ___________________

cf = <botocore.client.CloudFormation object at 0x7f59ca68f580>
template_body_data = {'Resources': {'MyS3Bucket': {'Properties': {}, 'Type': 'AWS::S3::Bucket'}}}

    @mock_cloudformation
    def test_check_stack_exists_deleted_stack(cf, template_body_data):
        'Test the stack search if the stack exists'
        params = {'StackName': 'teste', 'TemplateBody': yaml.dump(template_body_data)}
        cf.create_stack(**params)
>       cf.delete_stack(StackName='teste')

tests/test_cl_uploader.py:87: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/client.py:316: in _api_call
    return self._make_api_call(operation_name, kwargs)
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/client.py:621: in _make_api_call
    http, parsed_response = self._make_request(
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/client.py:641: in _make_request
    return self._endpoint.make_request(operation_model, request_dict)
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/endpoint.py:102: in make_request
    return self._send_request(request_dict, operation_model)
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/endpoint.py:136: in _send_request
    while self._needs_retry(attempts, operation_model, request_dict,
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/endpoint.py:253: in _needs_retry
    responses = self._event_emitter.emit(
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/hooks.py:356: in emit
    return self._emitter.emit(aliased_event_name, **kwargs)
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/hooks.py:228: in emit
    return self._emit(event_name, kwargs)
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/hooks.py:211: in _emit
    response = handler(**kwargs)
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/retryhandler.py:183: in __call__
    if self._checker(attempts, response, caught_exception):
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/retryhandler.py:250: in __call__
    should_retry = self._should_retry(attempt_number, response,
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/retryhandler.py:269: in _should_retry
    return self._checker(attempt_number, response, caught_exception)
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/retryhandler.py:316: in __call__
    checker_response = checker(attempt_number, response,
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/retryhandler.py:222: in __call__
    return self._check_caught_exception(
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/retryhandler.py:359: in _check_caught_exception
    raise caught_exception
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/endpoint.py:197: in _do_get_response
    responses = self._event_emitter.emit(event_name, request=request)
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/hooks.py:356: in emit
    return self._emitter.emit(aliased_event_name, **kwargs)
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/hooks.py:228: in emit
    return self._emit(event_name, kwargs)
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/hooks.py:211: in _emit
    response = handler(**kwargs)
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/moto/core/models.py:322: in __call__
    status, headers, body = response_callback(
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/moto/core/responses.py:202: in dispatch
    return cls()._dispatch(*args, **kwargs)
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/moto/core/responses.py:312: in _dispatch
    return self.call_action()
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/moto/core/responses.py:397: in call_action
    response = method()
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/moto/cloudformation/responses.py:380: in delete_stack
    self.cloudformation_backend.delete_stack(name_or_stack_id)
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/moto/cloudformation/models.py:734: in delete_stack
    self.delete_stack(stack.stack_id)
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/moto/cloudformation/models.py:726: in delete_stack
    stack.delete()
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/moto/cloudformation/models.py:363: in delete
    self.resource_map.delete()
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/moto/cloudformation/parsing.py:677: in delete
    raise last_exception
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <moto.cloudformation.parsing.ResourceMap object at 0x7f59cabd3940>

    def delete(self):
        remaining_resources = set(self.resources)
        tries = 1
        while remaining_resources and tries < 5:
            for resource in remaining_resources.copy():
                parsed_resource = self._parsed_resources.get(resource)
                try:
                    if parsed_resource and hasattr(parsed_resource, "delete"):
                        parsed_resource.delete(self._region_name)
                    else:
                        resource_name_attribute = (
                            parsed_resource.cloudformation_name_type()
                            if hasattr(parsed_resource, "cloudformation_name_type")
                            else resource_name_property_from_type(parsed_resource.type)
                        )
                        if resource_name_attribute:
                            resource_json = self._resource_json_map[
                                parsed_resource.logical_resource_id
                            ]
>                           resource_name = resource_json["Properties"][
                                resource_name_attribute
                            ]
E                           KeyError: 'BucketName'

../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/moto/cloudformation/parsing.py:662: KeyError

但是当我添加一个时,它会给出另一个错误:

================================= FAILURES ==================================
___________________ test_check_stack_exists_deleted_stack ___________________

cf = <botocore.client.CloudFormation object at 0x7f7a02ca4ee0>
template_body_data = {'Resources': {'MyS3Bucket': {'Properties': {'BucketName': 'teste'}, 'Type': 'AWS::S3::Bucket'}}}

    @mock_cloudformation
    def test_check_stack_exists_deleted_stack(cf, template_body_data):
        'Test the stack search if the stack exists'
        params = {'StackName': 'teste', 'TemplateBody': yaml.dump(template_body_data)}
>       cf.create_stack(**params)

tests/test_cl_uploader.py:86: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/client.py:316: in _api_call
    return self._make_api_call(operation_name, kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <botocore.client.CloudFormation object at 0x7f7a02ca4ee0>
operation_name = 'CreateStack'
api_params = {'StackName': 'teste', 'TemplateBody': 'Resources:\n  MyS3Bucket:\n    Properties:\n      BucketName: teste\n    Type: AWS::S3::Bucket\n'}

    def _make_api_call(self, operation_name, api_params):
        operation_model = self._service_model.operation_model(operation_name)
        service_name = self._service_model.service_name
        history_recorder.record('API_CALL', {
            'service': service_name,
            'operation': operation_name,
            'params': api_params,
        })
        if operation_model.deprecated:
            logger.debug('Warning: %s.%s() is deprecated',
                         service_name, operation_name)
        request_context = {
            'client_region': self.meta.region_name,
            'client_config': self.meta.config,
            'has_streaming_input': operation_model.has_streaming_input,
            'auth_type': operation_model.auth_type,
        }
        request_dict = self._convert_to_request_dict(
            api_params, operation_model, context=request_context)
    
        service_id = self._service_model.service_id.hyphenize()
        handler, event_response = self.meta.events.emit_until_response(
            'before-call.{service_id}.{operation_name}'.format(
                service_id=service_id,
                operation_name=operation_name),
            model=operation_model, params=request_dict,
            request_signer=self._request_signer, context=request_context)
    
        if event_response is not None:
            http, parsed_response = event_response
        else:
            http, parsed_response = self._make_request(
                operation_model, request_dict, request_context)
    
        self.meta.events.emit(
            'after-call.{service_id}.{operation_name}'.format(
                service_id=service_id,
                operation_name=operation_name),
            http_response=http, parsed=parsed_response,
            model=operation_model, context=request_context
        )
    
        if http.status_code >= 300:
            error_code = parsed_response.get("Error", {}).get("Code")
            error_class = self.exceptions.from_code(error_code)
>           raise error_class(parsed_response, operation_name)
E           botocore.exceptions.ClientError: An error occurred (Unknown) when calling the CreateStack operation: Unknown

../../.cache/pypoetry/virtualenvs/cl-uploader-12nYBdPj-py3.8/lib/python3.8/site-packages/botocore/client.py:635: ClientError

4

0 回答 0