'Pytest-cov with Moto in list_stacks() from Boto3
I'm trying to create a test with pytest and moto that check if the StackStatus from the dictionary returned from the function list_stacks() (https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/cloudformation.html#CloudFormation.Client.list_stacks) is DELETE_COMPLETE.
I create a pytest.fixture for the client connection:
@pytest.fixture(scope='function')
def cf(aws_credentials):
with mock_cloudformation():
yield boto3.client('cloudformation')
And I created a dummie template:
@pytest.fixture(scope='function')
def template_body_data():
'The Cloud Formation template'
template_data = {
'Resources': {'MyS3Bucket': {'Type': 'AWS::S3::Bucket', 'Properties': {}}}
}
return template_data
And in my test function I create the stack and after I deleted:
@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
The function that is being testes is this one:
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
I'm facing a issue:
It complains that my template doesn't have a 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
But when I add one, it giver another error:
================================= 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
Solution 1:[1]
Very late answer, but since you specified an S3 bucket type in the template, the cloudformation client will likely try making calls to S3 to check if the bucket resource exists. So we would also need to mock those s3 calls. You need to put a @mock_s3 annotation in that test method:
@mock_s3
@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
Sources
This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.
Source: Stack Overflow
| Solution | Source |
|---|---|
| Solution 1 | Laurent P |