Add mechanism to delete CF resources of deleted stacks
diff --git a/localstack/services/cloudformation/cloudformation_listener.py b/localstack/services/cloudformation/cloudformation_listener.py
index d935472d0b975..e29602169a672 100644
--- a/localstack/services/cloudformation/cloudformation_listener.py
+++ b/localstack/services/cloudformation/cloudformation_listener.py
@@ -126,15 +126,19 @@ def forward_request(self, method, path, data, headers):
             req_data = urlparse.parse_qs(to_str(data))
             req_data = dict([(k, v[0]) for k, v in req_data.items()])
             action = req_data.get('Action')
+            stack_name = req_data.get('StackName')
 
             if action == 'CreateStack':
-                stack_name = req_data.get('StackName')
                 event_publisher.fire_event(event_publisher.EVENT_CLOUDFORMATION_CREATE_STACK,
                     payload={'n': event_publisher.get_hash(stack_name)})
 
+            if action == 'DeleteStack':
+                client = aws_stack.connect_to_service('cloudformation')
+                stack_resources = client.list_stack_resources(StackName=stack_name)['StackResourceSummaries']
+                template_deployer.delete_stack(stack_name, stack_resources)
+
             if action == 'DescribeStackEvents':
                 # fix an issue where moto cannot handle ARNs as stack names (or missing names)
-                stack_name = req_data.get('StackName')
                 run_fix = not stack_name
                 if stack_name:
                     if stack_name.startswith('arn:aws:cloudformation'):
@@ -169,12 +173,19 @@ def return_response(self, method, path, data, headers, response):
         if response.status_code >= 400:
             LOG.debug('Error response from CloudFormation (%s) %s %s: %s' %
                       (response.status_code, method, path, response.content))
+
         if response._content:
             aws_stack.fix_account_id_in_arns(response)
 
     def _list_stack_names(self):
         client = aws_stack.connect_to_service('cloudformation')
-        stack_names = [s['StackName'] for s in client.list_stacks()['StackSummaries']]
+        stacks = client.list_stacks()['StackSummaries']
+        stack_names = []
+        for stack in stacks:
+            status = stack['StackStatus']
+            if 'FAILED' in status or 'DELETE' in status:
+                continue
+            stack_names.append(stack['StackName'])
         return stack_names
 
 
diff --git a/localstack/services/cloudformation/cloudformation_starter.py b/localstack/services/cloudformation/cloudformation_starter.py
index caab48c8e43a4..c3a8e696c82d5 100644
--- a/localstack/services/cloudformation/cloudformation_starter.py
+++ b/localstack/services/cloudformation/cloudformation_starter.py
@@ -8,6 +8,7 @@
 from moto.sqs import models as sqs_models
 from moto.core import BaseModel
 from moto.server import main as moto_main
+from moto.kinesis import models as kinesis_models
 from moto.dynamodb import models as dynamodb_models
 from moto.dynamodb2 import models as dynamodb2_models
 from moto.awslambda import models as lambda_models
@@ -128,6 +129,8 @@ def update_physical_resource_id(resource):
         elif isinstance(resource, service_models.StepFunctionsActivity):
             act_arn = aws_stack.stepfunctions_activity_arn(resource.params.get('Name'))
             resource.physical_resource_id = act_arn
+        elif isinstance(resource, kinesis_models.Stream):
+            resource.physical_resource_id = resource.stream_name
         else:
             LOG.warning('Unable to determine physical_resource_id for resource %s' % type(resource))
 
diff --git a/localstack/utils/cloudformation/template_deployer.py b/localstack/utils/cloudformation/template_deployer.py
index b3edc33f5312e..6d99b1580bb5b 100644
--- a/localstack/utils/cloudformation/template_deployer.py
+++ b/localstack/utils/cloudformation/template_deployer.py
@@ -14,6 +14,7 @@
 from localstack.services.awslambda.lambda_api import get_handler_file_from_name
 
 ACTION_CREATE = 'create'
+ACTION_DELETE = 'delete'
 PLACEHOLDER_RESOURCE_NAME = '__resource_name__'
 
 LOG = logging.getLogger(__name__)
@@ -74,6 +75,12 @@ def get_lambda_code_param(params, **kwargs):
                 'ACL': lambda params, **kwargs: convert_acl_cf_to_s3(params.get('AccessControl', 'PublicRead')),
                 'CreateBucketConfiguration': lambda params, **kwargs: get_bucket_location_config()
             }
+        },
+        'delete': {
+            'function': 'delete_bucket',
+            'parameters': {
+                'Bucket': 'PhysicalResourceId'
+            }
         }
     },
     'SQS::Queue': {
@@ -87,6 +94,12 @@ def get_lambda_code_param(params, **kwargs):
                 ),
                 'tags': 'Tags'
             }
+        },
+        'delete': {
+            'function': 'delete_queue',
+            'parameters': {
+                'QueueUrl': 'PhysicalResourceId'
+            }
         }
     },
     'SNS::Topic': {
@@ -96,6 +109,12 @@ def get_lambda_code_param(params, **kwargs):
                 'Name': 'TopicName',
                 'Tags': 'Tags'
             }
+        },
+        'delete': {
+            'function': 'delete_topic',
+            'parameters': {
+                'TopicArn': 'PhysicalResourceId'
+            }
         }
     },
     'Logs::LogGroup': {
@@ -254,6 +273,12 @@ def get_lambda_code_param(params, **kwargs):
             'defaults': {
                 'ShardCount': 1
             }
+        },
+        'delete': {
+            'function': 'delete_stream',
+            'parameters': {
+                'StreamName': 'PhysicalResourceId'
+            }
         }
     },
     'StepFunctions::StateMachine': {
@@ -337,7 +362,7 @@ def get_resource_type(resource):
 
 
 def get_service_name(resource):
-    res_type = resource.get('Type', '')
+    res_type = resource.get('Type', resource.get('ResourceType', ''))
     parts = res_type.split('::')
     if len(parts) == 1:
         return None
@@ -376,9 +401,6 @@ def get_client(resource, func_config):
     resource_config = RESOURCE_TO_FUNCTION.get(resource_type)
     if resource_config is None:
         raise Exception('CloudFormation deployment for resource type %s not yet implemented' % resource_type)
-    if ACTION_CREATE not in resource_config:
-        # nothing to do for this resource
-        return
     try:
         if func_config.get('boto_client') == 'resource':
             return aws_stack.connect_to_resource(service)
@@ -698,15 +720,23 @@ def remove_nones(o, **kwargs):
 
 
 def deploy_resource(resource_id, resources, stack_name):
+    return execute_resource_action(resource_id, resources, stack_name, ACTION_CREATE)
+
+
+def delete_resource(resource_id, resources, stack_name):
+    return execute_resource_action(resource_id, resources, stack_name, ACTION_DELETE)
+
+
+def execute_resource_action(resource_id, resources, stack_name, action_name):
     resource = resources[resource_id]
     resource_type = get_resource_type(resource)
     func_details = RESOURCE_TO_FUNCTION.get(resource_type)
-    if not func_details:
-        LOG.warning('Resource type not yet implemented: %s' % resource_type)
+    if not func_details or action_name not in func_details:
+        LOG.warning('Action "%s" for resource type %s not yet implemented' % (action_name, resource_type))
         return
 
-    LOG.debug('Deploying resource type "%s" id "%s"' % (resource_type, resource_id))
-    func_details = func_details[ACTION_CREATE]
+    LOG.debug('Running action "%s" for resource type "%s" id "%s"' % (action_name, resource_type, resource_id))
+    func_details = func_details[action_name]
     func_details = func_details if isinstance(func_details, list) else [func_details]
     results = []
     for func in func_details:
@@ -716,12 +746,12 @@ def deploy_resource(resource_id, resources, stack_name):
             continue
         client = get_client(resource, func)
         if client:
-            result = deploy_resource_via_sdk_function(resource_id, resources, resource_type, func, stack_name)
+            result = configure_resource_via_sdk(resource_id, resources, resource_type, func, stack_name)
             results.append(result)
     return (results or [None])[0]
 
 
-def deploy_resource_via_sdk_function(resource_id, resources, resource_type, func_details, stack_name):
+def configure_resource_via_sdk(resource_id, resources, resource_type, func_details, stack_name):
     resource = resources[resource_id]
     client = get_client(resource, func_details)
     function = getattr(client, func_details['function'])
@@ -781,7 +811,7 @@ def deploy_resource_via_sdk_function(resource_id, resources, resource_type, func
 
     # invoke function
     try:
-        LOG.debug('Request for creating resource type "%s" in region %s: %s %s' % (
+        LOG.debug('Request for resource type "%s" in region %s: %s %s' % (
             resource_type, aws_stack.get_region(), func_details['function'], params))
         result = function(**params)
     except Exception as e:
@@ -847,6 +877,14 @@ def deploy_template(template, stack_name):
         'after %s iterations. Remaining (%s): %s' % (iters, len(next), next))
 
 
+def delete_stack(stack_name, stack_resources):
+    resources = dict([(r['LogicalResourceId'], common.clone_safe(r)) for r in stack_resources])
+    for key, resource in resources.items():
+        resources[key]['Properties'] = common.clone_safe(resource)
+    for resource_id, resource in resources.items():
+        delete_resource(resource_id, resources, stack_name)
+
+
 # --------
 # Util methods for analyzing resource dependencies
 # --------
diff --git a/localstack/utils/common.py b/localstack/utils/common.py
index 0fb3281a28e64..8b0de6563c80e 100644
--- a/localstack/utils/common.py
+++ b/localstack/utils/common.py
@@ -970,6 +970,10 @@ def clone(item):
     return json.loads(json.dumps(item))
 
 
+def clone_safe(item):
+    return clone(json_safe(item))
+
+
 def remove_non_ascii(text):
     # text = unicode(text, "utf-8")
     text = text.decode('utf-8', CODEC_HANDLER_UNDERSCORE)
diff --git a/tests/integration/test_cloudformation.py b/tests/integration/test_cloudformation.py
index 9af346facdd81..bb44f7ec74dac 100644
--- a/tests/integration/test_cloudformation.py
+++ b/tests/integration/test_cloudformation.py
@@ -51,7 +51,10 @@ def bucket_exists(name):
 def queue_exists(name):
     sqs_client = aws_stack.connect_to_service('sqs')
     queues = sqs_client.list_queues()
-    url = name if '://' in name else aws_stack.get_sqs_queue_url(name)
+    try:
+        url = name if '://' in name else aws_stack.get_sqs_queue_url(name)
+    except Exception:
+        return False
     for queue_url in queues['QueueUrls']:
         if queue_url == url:
             return True
@@ -114,8 +117,9 @@ def get_topic_arns():
 
 class CloudFormationTest(unittest.TestCase):
 
-    def test_apply_template(self):
+    def test_create_delete_stack(self):
         cloudformation = aws_stack.connect_to_resource('cloudformation')
+        cf_client = aws_stack.connect_to_service('cloudformation')
         s3 = aws_stack.connect_to_service('s3')
         sns = aws_stack.connect_to_service('sns')
         apigateway = aws_stack.connect_to_service('apigateway')
@@ -132,16 +136,12 @@ def check_stack():
 
         retry(check_stack, retries=3, sleep=2)
 
-        # assert that bucket has been created
+        # assert that resources have been created
         assert bucket_exists('cf-test-bucket-1')
-        # assert that queue has been created
         assert queue_exists('cf-test-queue-1')
-        # assert that topic has been created
         topic_arn = topic_exists('%s-test-topic-1-1' % stack_name)
         assert topic_arn
-        # assert that stream has been created
         assert stream_exists('cf-test-stream-1')
-        # assert that queue has been created
         resource = describe_stack_resource(stack_name, 'SQSQueueNoNameProperty')
         assert queue_exists(resource['PhysicalResourceId'])
 
@@ -153,6 +153,7 @@ def check_stack():
             {'Key': 'foo', 'Value': 'cf-test-bucket-1'},
             {'Key': 'bar', 'Value': aws_stack.s3_bucket_arn('cf-test-bucket-1')}
         ])
+
         # assert that subscriptions have been created
         subs = sns.list_subscriptions()['Subscriptions']
         subs = [s for s in subs if (':%s:cf-test-queue-1' % TEST_AWS_ACCOUNT_ID) in s['Endpoint']]
@@ -167,6 +168,15 @@ def check_stack():
         types = [r['responseType'] for r in responses]
         self.assertEqual(set(types), set(['UNAUTHORIZED', 'DEFAULT_5XX']))
 
+        # delete the stack
+        cf_client.delete_stack(StackName=stack_name)
+
+        # assert that resources have been deleted
+        assert not bucket_exists('cf-test-bucket-1')
+        assert not queue_exists('cf-test-queue-1')
+        assert not topic_exists('%s-test-topic-1-1' % stack_name)
+        retry(lambda: self.assertFalse(stream_exists('cf-test-stream-1')))
+
     def test_list_stack_events(self):
         cloudformation = aws_stack.connect_to_service('cloudformation')
         response = cloudformation.describe_stack_events()

Add mechanism to delete CF resources of deleted stacks - addresses #999