text: import os
  import re
  import json
  import unittest
  import mock
  import time
  import datetime
  from localstack.constants import LAMBDA_TEST_ROLE
  from localstack.utils.common import isoformat_milliseconds, save_file, new_tmp_dir, mkdir
  from localstack.services.awslambda import lambda_api, lambda_executors
  from localstack.utils.aws.aws_models import LambdaFunction
  
  TEST_EVENT_SOURCE_ARN = 'arn:aws:sqs:eu-west-1:000000000000:testq'
  TEST_SECRETSMANANAGER_EVENT_SOURCE_ARN = 'arn:aws:secretsmanager:us-east-1:000000000000:secret:mysecret-kUBhE'
  
  
  class TestLambdaAPI(unittest.TestCase):
      CODE_SIZE = 50
      CODE_SHA_256 = '/u60ZpAA9bzZPVwb8d4390i5oqP1YAObUwV03CZvsWA='
      UPDATED_CODE_SHA_256 = '/u6A='
      MEMORY_SIZE = 128
      ROLE = 'arn:aws:iam::123456:role/role-name'
      LAST_MODIFIED = datetime.datetime.utcnow()
      TRACING_CONFIG = {'Mode': 'PassThrough'}
      REVISION_ID = 'e54dbcf8-e3ef-44ab-9af7-8dbef510608a'
      HANDLER = 'index.handler'
      RUNTIME = 'node.js4.3'
      TIMEOUT = 60  # Default value, hardcoded
      FUNCTION_NAME = 'test1'
      ALIAS_NAME = 'alias1'
      ALIAS2_NAME = 'alias2'
      RESOURCENOTFOUND_EXCEPTION = 'ResourceNotFoundException'
      RESOURCENOTFOUND_MESSAGE = 'Function not found: %s'
      ALIASEXISTS_EXCEPTION = 'ResourceConflictException'
      ALIASEXISTS_MESSAGE = 'Alias already exists: %s'
      ALIASNOTFOUND_EXCEPTION = 'ResourceNotFoundException'
      ALIASNOTFOUND_MESSAGE = 'Alias not found: %s'
      TEST_UUID = 'Test'
      TAGS = {'hello': 'world', 'env': 'prod'}
  
      def setUp(self):
          lambda_api.cleanup()
          self.maxDiff = None
          self.app = lambda_api.app
          self.app.testing = True
          self.client = self.app.test_client()
  
      def test_get_non_existent_function_returns_error(self):
          with self.app.test_request_context():
              result = json.loads(lambda_api.get_function('non_existent_function_name').get_data())
              self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])
              self.assertEqual(
                  self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn('non_existent_function_name'),
                  result['message'])
  
      def test_get_event_source_mapping(self):
          region = lambda_api.LambdaRegion.get()
          with self.app.test_request_context():
              region.event_source_mappings.append({'UUID': self.TEST_UUID})
              result = lambda_api.get_event_source_mapping(self.TEST_UUID)
              self.assertEqual(json.loads(result.get_data()).get('UUID'), self.TEST_UUID)
  
      def test_get_event_sources(self):
          region = lambda_api.LambdaRegion.get()
          with self.app.test_request_context():
              region.event_source_mappings.append(
                  {
                      'UUID': self.TEST_UUID,
                      'EventSourceArn': 'the_arn'
                  })
  
              # Match source ARN
              result = lambda_api.get_event_sources(source_arn='the_arn')
              self.assertEqual(len(result), 1)
              self.assertEqual(result[0].get('UUID'), self.TEST_UUID)
  
              # No partial match on source ARN
              result = lambda_api.get_event_sources(source_arn='the_')
              self.assertEqual(len(result), 0)
  
      def test_get_event_sources_with_paths(self):
          region = lambda_api.LambdaRegion.get()
          with self.app.test_request_context():
              region.event_source_mappings.append(
                  {
                      'UUID': self.TEST_UUID,
                      'EventSourceArn': 'the_arn/path/subpath'
                  })
  
              # Do partial match on paths
              result = lambda_api.get_event_sources(source_arn='the_arn')
              self.assertEqual(len(result), 1)
              result = lambda_api.get_event_sources(source_arn='the_arn/path')
              self.assertEqual(len(result), 1)
  
      def test_delete_event_source_mapping(self):
          region = lambda_api.LambdaRegion.get()
          with self.app.test_request_context():
              region.event_source_mappings.append({'UUID': self.TEST_UUID})
              result = lambda_api.delete_event_source_mapping(self.TEST_UUID)
              self.assertEqual(json.loads(result.get_data()).get('UUID'), self.TEST_UUID)
              self.assertEqual(0, len(region.event_source_mappings))
  
      def test_invoke_RETURNS_415_WHEN_not_json_input(self):
          with self.app.test_request_context() as context:
              context.request._cached_data = '~notjsonrequest~'
              response = lambda_api.invoke_function(self.FUNCTION_NAME)
              self.assertEqual('415 UNSUPPORTED MEDIA TYPE', response.status)
  
      def _request_response(self, context):
          context.request._cached_data = '{}'
          context.request.args = {'Qualifier': '$LATEST'}
          context.request.environ['HTTP_X_AMZ_INVOCATION_TYPE'] = 'RequestResponse'
          self._create_function(self.FUNCTION_NAME)
  
      @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')
      def test_invoke_plain_text_response(self, mock_run_lambda):
          with self.app.test_request_context() as context:
              self._request_response(context)
              mock_run_lambda.return_value = '~notjsonresponse~'
              response = lambda_api.invoke_function(self.FUNCTION_NAME)
              self.assertEqual('~notjsonresponse~', response[0])
              self.assertEqual(200, response[1])
              self._assert_contained({'Content-Type': 'text/plain'}, response[2])
  
      @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')
      def test_invoke_empty_plain_text_response(self, mock_run_lambda):
          with self.app.test_request_context() as context:
              self._request_response(context)
              mock_run_lambda.return_value = ''
              response = lambda_api.invoke_function(self.FUNCTION_NAME)
              self.assertEqual('', response[0])
              self.assertEqual(200, response[1])
              self._assert_contained({'Content-Type': 'text/plain'}, response[2])
  
      @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')
      def test_invoke_empty_map_json_response(self, mock_run_lambda):
          with self.app.test_request_context() as context:
              self._request_response(context)
              mock_run_lambda.return_value = '{}'
              response = lambda_api.invoke_function(self.FUNCTION_NAME)
              self.assertEqual(b'{}\n', response[0].response[0])
              self.assertEqual(200, response[1])
              self._assert_contained({'Content-Type': 'application/json'}, response[2])
  
      @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')
      def test_invoke_populated_map_json_response(self, mock_run_lambda):
          with self.app.test_request_context() as context:
              self._request_response(context)
              mock_run_lambda.return_value = '{"bool":true,"int":1}'
              response = lambda_api.invoke_function(self.FUNCTION_NAME)
              self.assertEqual(b'{"bool":true,"int":1}\n', response[0].response[0])
              self.assertEqual(200, response[1])
              self._assert_contained({'Content-Type': 'application/json'}, response[2])
  
      @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')
      def test_invoke_empty_list_json_response(self, mock_run_lambda):
          with self.app.test_request_context() as context:
              self._request_response(context)
              mock_run_lambda.return_value = '[]'
              response = lambda_api.invoke_function(self.FUNCTION_NAME)
              self.assertEqual(b'[]\n', response[0].response[0])
              self.assertEqual(200, response[1])
              self._assert_contained({'Content-Type': 'application/json'}, response[2])
  
      @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')
      def test_invoke_populated_list_json_response(self, mock_run_lambda):
          with self.app.test_request_context() as context:
              self._request_response(context)
              mock_run_lambda.return_value = '[true,1,"thing"]'
              response = lambda_api.invoke_function(self.FUNCTION_NAME)
              self.assertEqual(b'[true,1,"thing"]\n', response[0].response[0])
              self.assertEqual(200, response[1])
              self._assert_contained({'Content-Type': 'application/json'}, response[2])
  
      @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')
      def test_invoke_string_json_response(self, mock_run_lambda):
          with self.app.test_request_context() as context:
              self._request_response(context)
              mock_run_lambda.return_value = '"thing"'
              response = lambda_api.invoke_function(self.FUNCTION_NAME)
              self.assertEqual(b'"thing"\n', response[0].response[0])
              self.assertEqual(200, response[1])
              self._assert_contained({'Content-Type': 'application/json'}, response[2])
  
      @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')
      def test_invoke_integer_json_response(self, mock_run_lambda):
          with self.app.test_request_context() as context:
              self._request_response(context)
              mock_run_lambda.return_value = '1234'
              response = lambda_api.invoke_function(self.FUNCTION_NAME)
              self.assertEqual(b'1234\n', response[0].response[0])
              self.assertEqual(200, response[1])
              self._assert_contained({'Content-Type': 'application/json'}, response[2])
  
      @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')
      def test_invoke_float_json_response(self, mock_run_lambda):
          with self.app.test_request_context() as context:
              self._request_response(context)
              mock_run_lambda.return_value = '1.3'
              response = lambda_api.invoke_function(self.FUNCTION_NAME)
              self.assertEqual(b'1.3\n', response[0].response[0])
              self.assertEqual(200, response[1])
              self._assert_contained({'Content-Type': 'application/json'}, response[2])
  
      @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')
      def test_invoke_boolean_json_response(self, mock_run_lambda):
          with self.app.test_request_context() as context:
              self._request_response(context)
              mock_run_lambda.return_value = 'true'
              response = lambda_api.invoke_function(self.FUNCTION_NAME)
              self.assertEqual(b'true\n', response[0].response[0])
              self.assertEqual(200, response[1])
              self._assert_contained({'Content-Type': 'application/json'}, response[2])
  
      @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')
      def test_invoke_null_json_response(self, mock_run_lambda):
          with self.app.test_request_context() as context:
              self._request_response(context)
              mock_run_lambda.return_value = 'null'
              response = lambda_api.invoke_function(self.FUNCTION_NAME)
              self.assertEqual(b'null\n', response[0].response[0])
              self.assertEqual(200, response[1])
              self._assert_contained({'Content-Type': 'application/json'}, response[2])
  
      def test_create_event_source_mapping(self):
          self.client.post('{0}/event-source-mappings/'.format(lambda_api.PATH_ROOT),
              data=json.dumps({'FunctionName': 'test-lambda-function', 'EventSourceArn': TEST_EVENT_SOURCE_ARN}))
  
          listResponse = self.client.get('{0}/event-source-mappings/'.format(lambda_api.PATH_ROOT))
          listResult = json.loads(listResponse.get_data())
  
          eventSourceMappings = listResult.get('EventSourceMappings')
  
          self.assertEqual(1, len(eventSourceMappings))
          self.assertEqual('Enabled', eventSourceMappings[0]['State'])
  
      def test_create_event_source_mapping_self_managed_event_source(self):
          self.client.post('{0}/event-source-mappings/'.format(lambda_api.PATH_ROOT),
              data=json.dumps({'FunctionName': 'test-lambda-function',
                  'Topics': ['test'],
                  'SourceAccessConfigurations': [
                      {'Type': 'SASL_SCRAM_512_AUTH', 'URI': TEST_SECRETSMANANAGER_EVENT_SOURCE_ARN}
                  ],
                  'SelfManagedEventSource': {'Endpoints': {'KAFKA_BOOTSTRAP_SERVERS': ['127.0.0.1:9092']}}
              }))
          listResponse = self.client.get('{0}/event-source-mappings/'.format(lambda_api.PATH_ROOT))
          listResult = json.loads(listResponse.get_data())
  
          eventSourceMappings = listResult.get('EventSourceMappings')
  
          self.assertEqual(1, len(eventSourceMappings))
          self.assertEqual('Enabled', eventSourceMappings[0]['State'])
  
      def test_create_disabled_event_source_mapping(self):
          createResponse = self.client.post('{0}/event-source-mappings/'.format(lambda_api.PATH_ROOT),
                              data=json.dumps({'FunctionName': 'test-lambda-function',
                                               'EventSourceArn': TEST_EVENT_SOURCE_ARN,
                                               'Enabled': 'false'}))
          createResult = json.loads(createResponse.get_data())
  
          self.assertEqual('Disabled', createResult['State'])
  
          getResponse = self.client.get('{0}/event-source-mappings/{1}'.format(lambda_api.PATH_ROOT,
                          createResult.get('UUID')))
          getResult = json.loads(getResponse.get_data())
  
          self.assertEqual('Disabled', getResult['State'])
  
      def test_update_event_source_mapping(self):
          createResponse = self.client.post('{0}/event-source-mappings/'.format(lambda_api.PATH_ROOT),
                              data=json.dumps({'FunctionName': 'test-lambda-function',
                                               'EventSourceArn': TEST_EVENT_SOURCE_ARN,
                                               'Enabled': 'true'}))
          createResult = json.loads(createResponse.get_data())
  
          putResponse = self.client.put('{0}/event-source-mappings/{1}'.format(lambda_api.PATH_ROOT,
                          createResult.get('UUID')), data=json.dumps({'Enabled': 'false'}))
          putResult = json.loads(putResponse.get_data())
  
          self.assertEqual('Disabled', putResult['State'])
  
          getResponse = self.client.get('{0}/event-source-mappings/{1}'.format(lambda_api.PATH_ROOT,
                          createResult.get('UUID')))
          getResult = json.loads(getResponse.get_data())
  
          self.assertEqual('Disabled', getResult['State'])
  
      def test_update_event_source_mapping_self_managed_event_source(self):
          createResponse = self.client.post('{0}/event-source-mappings/'.format(lambda_api.PATH_ROOT),
              data=json.dumps({'FunctionName': 'test-lambda-function',
                  'Topics': ['test'],
                  'SourceAccessConfigurations': [
                      {'Type': 'SASL_SCRAM_512_AUTH', 'URI': TEST_SECRETSMANANAGER_EVENT_SOURCE_ARN}
                  ],
                  'SelfManagedEventSource': {'Endpoints': {'KAFKA_BOOTSTRAP_SERVERS': ['127.0.0.1:9092']}},
                  'Enabled': 'true'
              }))
          createResult = json.loads(createResponse.get_data())
  
          putResponse = self.client.put('{0}/event-source-mappings/{1}'.format(lambda_api.PATH_ROOT,
                          createResult.get('UUID')), data=json.dumps({'Enabled': 'false'}))
          putResult = json.loads(putResponse.get_data())
  
          self.assertEqual('Disabled', putResult['State'])
  
          getResponse = self.client.get('{0}/event-source-mappings/{1}'.format(lambda_api.PATH_ROOT,
                          createResult.get('UUID')))
          getResult = json.loads(getResponse.get_data())
  
          self.assertEqual('Disabled', getResult['State'])
  
      def test_publish_function_version(self):
          with self.app.test_request_context():
              self._create_function(self.FUNCTION_NAME)
  
              result = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())
              result.pop('RevisionId', None)  # we need to remove this, since this is random, so we cannot know its value
  
              expected_result = dict()
              expected_result['CodeSize'] = self.CODE_SIZE
              expected_result['CodeSha256'] = self.CODE_SHA_256
              expected_result['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':1'
              expected_result['FunctionName'] = str(self.FUNCTION_NAME)
              expected_result['Handler'] = str(self.HANDLER)
              expected_result['Runtime'] = str(self.RUNTIME)
              expected_result['Timeout'] = self.TIMEOUT
              expected_result['Description'] = ''
              expected_result['MemorySize'] = self.MEMORY_SIZE
              expected_result['Role'] = self.ROLE
              expected_result['KMSKeyArn'] = None
              expected_result['VpcConfig'] = None
              expected_result['LastModified'] = isoformat_milliseconds(self.LAST_MODIFIED) + '+0000'
              expected_result['TracingConfig'] = self.TRACING_CONFIG
              expected_result['Version'] = '1'
              expected_result['State'] = 'Active'
              expected_result['LastUpdateStatus'] = 'Successful'
              expected_result['PackageType'] = None
              expected_result['ImageConfig'] = {}
              self.assertDictEqual(expected_result, result)
  
      def test_publish_update_version_increment(self):
          with self.app.test_request_context():
              self._create_function(self.FUNCTION_NAME)
              lambda_api.publish_version(self.FUNCTION_NAME)
  
              self._update_function_code(self.FUNCTION_NAME)
              result = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())
              result.pop('RevisionId', None)  # we need to remove this, since this is random, so we cannot know its value
  
              expected_result = dict()
              expected_result['CodeSize'] = self.CODE_SIZE
              expected_result['CodeSha256'] = self.UPDATED_CODE_SHA_256
              expected_result['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':2'
              expected_result['FunctionName'] = str(self.FUNCTION_NAME)
              expected_result['Handler'] = str(self.HANDLER)
              expected_result['Runtime'] = str(self.RUNTIME)
              expected_result['Timeout'] = self.TIMEOUT
              expected_result['Description'] = ''
              expected_result['MemorySize'] = self.MEMORY_SIZE
              expected_result['Role'] = self.ROLE
              expected_result['KMSKeyArn'] = None
              expected_result['VpcConfig'] = None
              expected_result['LastModified'] = isoformat_milliseconds(self.LAST_MODIFIED) + '+0000'
              expected_result['TracingConfig'] = self.TRACING_CONFIG
              expected_result['Version'] = '2'
              expected_result['State'] = 'Active'
              expected_result['LastUpdateStatus'] = 'Successful'
              expected_result['PackageType'] = None
              expected_result['ImageConfig'] = {}
              self.assertDictEqual(expected_result, result)
  
      def test_publish_non_existant_function_version_returns_error(self):
          with self.app.test_request_context():
              result = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())
              self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])
              self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),
                               result['message'])
  
      def test_list_function_versions(self):
          with self.app.test_request_context():
              self._create_function(self.FUNCTION_NAME)
              lambda_api.publish_version(self.FUNCTION_NAME)
              lambda_api.publish_version(self.FUNCTION_NAME)
  
              result = json.loads(lambda_api.list_versions(self.FUNCTION_NAME).get_data())
              for version in result['Versions']:
                  # we need to remove this, since this is random, so we cannot know its value
                  version.pop('RevisionId', None)
  
              latest_version = dict()
              latest_version['CodeSize'] = self.CODE_SIZE
              latest_version['CodeSha256'] = self.CODE_SHA_256
              latest_version['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':$LATEST'
              latest_version['FunctionName'] = str(self.FUNCTION_NAME)
              latest_version['Handler'] = str(self.HANDLER)
              latest_version['Runtime'] = str(self.RUNTIME)
              latest_version['Timeout'] = self.TIMEOUT
              latest_version['Description'] = ''
              latest_version['MemorySize'] = self.MEMORY_SIZE
              latest_version['Role'] = self.ROLE
              latest_version['KMSKeyArn'] = None
              latest_version['VpcConfig'] = None
              latest_version['LastModified'] = isoformat_milliseconds(self.LAST_MODIFIED) + '+0000'
              latest_version['TracingConfig'] = self.TRACING_CONFIG
              latest_version['Version'] = '$LATEST'
              latest_version['State'] = 'Active'
              latest_version['LastUpdateStatus'] = 'Successful'
              latest_version['PackageType'] = None
              latest_version['ImageConfig'] = {}
              version1 = dict(latest_version)
              version1['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':1'
              version1['Version'] = '1'
              expected_result = {'Versions': sorted([latest_version, version],
                                                    key=lambda k: str(k.get('Version')))}
              self.assertDictEqual(expected_result, result)
  
      def test_list_non_existant_function_versions_returns_error(self):
          with self.app.test_request_context():
              result = json.loads(lambda_api.list_versions(self.FUNCTION_NAME).get_data())
              self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])
              self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),
                               result['message'])
  
      def test_create_alias(self):
          self._create_function(self.FUNCTION_NAME)
          self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))
  
          response = self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),
                           data=json.dumps({'Name': self.ALIAS_NAME, 'FunctionVersion': '1',
                               'Description': ''}))
          result = json.loads(response.get_data())
          result.pop('RevisionId', None)  # we need to remove this, since this is random, so we cannot know its value
  
          expected_result = {'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME,
                             'FunctionVersion': '1', 'Description': '', 'Name': self.ALIAS_NAME}
          self.assertDictEqual(expected_result, result)
  
      def test_create_alias_on_non_existant_function_returns_error(self):
          with self.app.test_request_context():
              result = json.loads(lambda_api.create_alias(self.FUNCTION_NAME).get_data())
              self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])
              self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),
                               result['message'])
  
      def test_create_alias_returns_error_if_already_exists(self):
          self._create_function(self.FUNCTION_NAME)
          self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))
          data = json.dumps({'Name': self.ALIAS_NAME, 'FunctionVersion': '1', 'Description': ''})
          self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME), data=data)
  
          response = self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),
                                      data=data)
          result = json.loads(response.get_data())
  
          alias_arn = lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME
          self.assertEqual(self.ALIASEXISTS_EXCEPTION, result['__type'])
          self.assertEqual(self.ALIASEXISTS_MESSAGE % alias_arn,
                           result['message'])
  
      def test_update_alias(self):
          self._create_function(self.FUNCTION_NAME)
          self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))
          self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),
                           data=json.dumps({
                               'Name': self.ALIAS_NAME, 'FunctionVersion': '1', 'Description': ''}))
  
          response = self.client.put('{0}/functions/{1}/aliases/{2}'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME,
                                                                            self.ALIAS_NAME),
                                     data=json.dumps({'FunctionVersion': '$LATEST', 'Description': 'Test-Description'}))
          result = json.loads(response.get_data())
          result.pop('RevisionId', None)  # we need to remove this, since this is random, so we cannot know its value
  
          expected_result = {'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME,
                             'FunctionVersion': '$LATEST', 'Description': 'Test-Description',
                             'Name': self.ALIAS_NAME}
          self.assertDictEqual(expected_result, result)
  
      def test_update_alias_on_non_existant_function_returns_error(self):
          with self.app.test_request_context():
              result = json.loads(lambda_api.update_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())
              self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])
              self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),
                               result['message'])
  
      def test_update_alias_on_non_existant_alias_returns_error(self):
          with self.app.test_request_context():
              self._create_function(self.FUNCTION_NAME)
              result = json.loads(lambda_api.update_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())
              alias_arn = lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME
              self.assertEqual(self.ALIASNOTFOUND_EXCEPTION, result['__type'])
              self.assertEqual(self.ALIASNOTFOUND_MESSAGE % alias_arn, result['message'])
  
      def test_get_alias(self):
          self._create_function(self.FUNCTION_NAME)
          self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))
          self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),
                           data=json.dumps({
                               'Name': self.ALIAS_NAME, 'FunctionVersion': '1', 'Description': ''}))
  
          response = self.client.get('{0}/functions/{1}/aliases/{2}'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME,
                                                                            self.ALIAS_NAME))
          result = json.loads(response.get_data())
          result.pop('RevisionId', None)  # we need to remove this, since this is random, so we cannot know its value
  
          expected_result = {'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME,
                             'FunctionVersion': '1', 'Description': '',
                             'Name': self.ALIAS_NAME}
          self.assertDictEqual(expected_result, result)
  
      def test_get_alias_on_non_existant_function_returns_error(self):
          with self.app.test_request_context():
              result = json.loads(lambda_api.get_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())
              self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])
              self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),
                               result['message'])
  
      def test_get_alias_on_non_existant_alias_returns_error(self):
          with self.app.test_request_context():
              self._create_function(self.FUNCTION_NAME)
              result = json.loads(lambda_api.get_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())
              alias_arn = lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME
              self.assertEqual(self.ALIASNOTFOUND_EXCEPTION, result['__type'])
              self.assertEqual(self.ALIASNOTFOUND_MESSAGE % alias_arn, result['message'])
  
      def test_list_aliases(self):
          self._create_function(self.FUNCTION_NAME)
          self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))
  
          self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),
                           data=json.dumps({'Name': self.ALIAS2_NAME, 'FunctionVersion': '$LATEST'}))
          self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),
                           data=json.dumps({'Name': self.ALIAS_NAME, 'FunctionVersion': '1',
                                            'Description': self.ALIAS_NAME}))
  
          response = self.client.get('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))
          result = json.loads(response.get_data())
          for alias in result['Aliases']:
              alias.pop('RevisionId', None)  # we need to remove this, since this is random, so we cannot know its value
          expected_result = {'Aliases': [
              {
                  'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME,
                  'FunctionVersion': '1',
                  'Name': self.ALIAS_NAME,
                  'Description': self.ALIAS_NAME
              },
              {
                  'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS2_NAME,
                  'FunctionVersion': '$LATEST',
                  'Name': self.ALIAS2_NAME,
                  'Description': ''
              }
          ]}
          self.assertDictEqual(expected_result, result)
  
      def test_list_non_existant_function_aliases_returns_error(self):
          with self.app.test_request_context():
              result = json.loads(lambda_api.list_aliases(self.FUNCTION_NAME).get_data())
              self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])
              self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),
                               result['message'])
  
      def test_get_container_name(self):
          executor = lambda_executors.EXECUTOR_CONTAINERS_REUSE
          name = executor.get_container_name('arn:aws:lambda:us-east-1:00000000:function:my_function_name')
          self.assertEqual(name, 'localstack_lambda_arn_aws_lambda_us-east-1_00000000_function_my_function_name')
  
      def test_concurrency(self):
          with self.app.test_request_context():
              self._create_function(self.FUNCTION_NAME)
              # note: PutFunctionConcurrency is mounted at: /2017-10-31
              # NOT lambda_api.PATH_ROOT
              # https://docs.aws.amazon.com/lambda/latest/dg/API_PutFunctionConcurrency.html
              concurrency_data = {'ReservedConcurrentExecutions': 10}
              response = self.client.put('/2017-10-31/functions/{0}/concurrency'.format(self.FUNCTION_NAME),
                                         data=json.dumps(concurrency_data))
  
              result = json.loads(response.get_data())
              self.assertDictEqual(concurrency_data, result)
  
              response = self.client.get('/2019-09-30/functions/{0}/concurrency'.format(self.FUNCTION_NAME))
              self.assertDictEqual(concurrency_data, result)
  
              response = self.client.delete('/2017-10-31/functions/{0}/concurrency'.format(self.FUNCTION_NAME))
              self.assertIsNotNone('ReservedConcurrentExecutions', result)
  
      def test_concurrency_get_function(self):
          with self.app.test_request_context():
              self._create_function(self.FUNCTION_NAME)
              # note: PutFunctionConcurrency is mounted at: /2017-10-31
              # NOT lambda_api.PATH_ROOT
              # https://docs.aws.amazon.com/lambda/latest/dg/API_PutFunctionConcurrency.html
              concurrency_data = {'ReservedConcurrentExecutions': 10}
              self.client.put('/2017-10-31/functions/{0}/concurrency'.format(self.FUNCTION_NAME),
                              data=json.dumps(concurrency_data))
  
              response = self.client.get('{0}/functions/{1}'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))
  
              result = json.loads(response.get_data())
              self.assertTrue('Concurrency' in result)
              self.assertDictEqual(concurrency_data, result['Concurrency'])
  
      def test_list_tags(self):
          with self.app.test_request_context():
              self._create_function(self.FUNCTION_NAME, self.TAGS)
              arn = lambda_api.func_arn(self.FUNCTION_NAME)
              response = self.client.get('{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn))
              result = json.loads(response.get_data())
              self.assertTrue('Tags' in result)
              self.assertDictEqual(self.TAGS, result['Tags'])
  
      def test_tag_resource(self):
          with self.app.test_request_context():
              self._create_function(self.FUNCTION_NAME)
              arn = lambda_api.func_arn(self.FUNCTION_NAME)
              response = self.client.get('{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn))
              result = json.loads(response.get_data())
              self.assertTrue('Tags' in result)
              self.assertDictEqual({}, result['Tags'])
  
              self.client.post('{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn), data=json.dumps({'Tags': self.TAGS}))
              response = self.client.get('{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn))
              result = json.loads(response.get_data())
              self.assertTrue('Tags' in result)
              self.assertDictEqual(self.TAGS, result['Tags'])
  
      def test_tag_non_existent_function_returns_error(self):
          with self.app.test_request_context():
              arn = lambda_api.func_arn('non-existent-function')
              response = self.client.post(
                  '{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn),
                  data=json.dumps({'Tags': self.TAGS}))
              result = json.loads(response.get_data())
              self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])
              self.assertEqual(
                  self.RESOURCENOTFOUND_MESSAGE % arn,
                  result['message'])
  
      def test_untag_resource(self):
          with self.app.test_request_context():
              self._create_function(self.FUNCTION_NAME, tags=self.TAGS)
              arn = lambda_api.func_arn(self.FUNCTION_NAME)
              response = self.client.get('{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn))
              result = json.loads(response.get_data())
              self.assertTrue('Tags' in result)
              self.assertDictEqual(self.TAGS, result['Tags'])
  
              self.client.delete('{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn), query_string={'tagKeys': 'env'})
              response = self.client.get('{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn))
              result = json.loads(response.get_data())
              self.assertTrue('Tags' in result)
              self.assertDictEqual({'hello': 'world'}, result['Tags'])
  
      def test_java_options_empty_return_empty_value(self):
          lambda_executors.config.LAMBDA_JAVA_OPTS = ''
          result = lambda_executors.Util.get_java_opts()
          self.assertFalse(result)
  
      def test_java_options_with_only_memory_options(self):
          expected = '-Xmx512M'
          result = self.prepare_java_opts(expected)
          self.assertEqual(expected, result)
  
      def test_java_options_with_memory_options_and_agentlib_option(self):
          expected = '.*transport=dt_socket,server=y,suspend=y,address=[0-9]+'
          result = self.prepare_java_opts('-Xmx512M -agentlib:jdwp=transport=dt_socket,server=y'
                                        ',suspend=y,address=_debug_port_')
          self.assertTrue(re.match(expected, result))
          self.assertTrue(lambda_executors.Util.debug_java_port is not False)
  
      def test_java_options_with_unset_debug_port(self):
          options = [
              '-agentlib:jdwp=transport=dt_socket,server=y,address=_debug_port_,suspend=y',
              '-agentlib:jdwp=transport=dt_socket,server=y,address=localhost:_debug_port_,suspend=y',
              '-agentlib:jdwp=transport=dt_socket,server=y,address=127.0.0.1:_debug_port_,suspend=y',
              '-agentlib:jdwp=transport=dt_socket,server=y,address=*:_debug_port_,suspend=y',
              '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=_debug_port_',
              '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=localhost:_debug_port_',
              '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=127.0.0.1:_debug_port_',
              '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=*:_debug_port_'
          ]
  
          expected_results = [
              '-agentlib:jdwp=transport=dt_socket,server=y,address=([0-9]+),suspend=y',
              '-agentlib:jdwp=transport=dt_socket,server=y,address=localhost:([0-9]+),suspend=y',
              '-agentlib:jdwp=transport=dt_socket,server=y,address=127.0.0.1:([0-9]+),suspend=y',
              '-agentlib:jdwp=transport=dt_socket,server=y,address=\\*:([0-9]+),suspend=y',
              '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=([0-9]+)',
              '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=localhost:([0-9]+)',
              '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=127.0.0.1:([0-9]+)',
              '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=\\*:([0-9]+)'
          ]
  
          for i in range(len(options)):
              result = self.prepare_java_opts(options[i])
              m = re.match(expected_results[i], result)
              self.assertTrue(m)
              self.assertEqual(m.groups()[0], lambda_executors.Util.debug_java_port)
  
      def test_java_options_with_configured_debug_port(self):
          options = [
              '-agentlib:jdwp=transport=dt_socket,server=y,address=1234,suspend=y',
              '-agentlib:jdwp=transport=dt_socket,server=y,address=localhost:1234,suspend=y',
              '-agentlib:jdwp=transport=dt_socket,server=y,address=127.0.0.1:1234,suspend=y',
              '-agentlib:jdwp=transport=dt_socket,server=y,address=*:1234,suspend=y',
              '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=1234',
              '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=localhost:1234',
              '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=127.0.0.1:1234',
              '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=*:1234'
          ]
  
          for item in options:
              result = self.prepare_java_opts(item)
              self.assertEqual('1234', lambda_executors.Util.debug_java_port)
              self.assertEqual(item, result)
  
      def prepare_java_opts(self, java_opts):
          lambda_executors.config.LAMBDA_JAVA_OPTS = java_opts
          result = lambda_executors.Util.get_java_opts()
          return result
  
      def test_get_java_lib_folder_classpath(self):
          jar_file = os.path.join(new_tmp_dir(), 'foo.jar')
          save_file(jar_file, '')
          classpath = lambda_executors.Util.get_java_classpath(jar_file)
          self.assertIn('.:foo.jar', classpath)
          self.assertIn('*.jar', classpath)
  
      def test_get_java_lib_folder_classpath_no_directories(self):
          base_dir = new_tmp_dir()
          jar_file = os.path.join(base_dir, 'foo.jar')
          save_file(jar_file, '')
          lib_file = os.path.join(base_dir, 'lib', 'lib.jar')
          mkdir(os.path.dirname(lib_file))
          save_file(lib_file, '')
          classpath = lambda_executors.Util.get_java_classpath(jar_file)
          self.assertIn(':foo.jar', classpath)
          self.assertIn('lib/lib.jar:', classpath)
          self.assertIn(':*.jar', classpath)
  
      def test_get_java_lib_folder_classpath_archive_is_None(self):
          self.assertRaises(TypeError, lambda_executors.Util.get_java_classpath, None)
  
      @mock.patch('localstack.services.awslambda.lambda_executors.store_cloudwatch_logs')
      def test_executor_store_logs_can_handle_milliseconds(self, mock_store_cloudwatch_logs):
          mock_details = mock.Mock()
          t_sec = time.time()  # plain old epoch secs
          t_ms = time.time() * 1000  # epoch ms as a long-int like AWS
  
          # pass t_ms millisecs to _store_logs
          lambda_executors._store_logs(mock_details, 'mock log output', t_ms)
  
          # expect the computed log-stream-name to having a prefix matching the date derived from t_sec
          today = datetime.datetime.utcfromtimestamp(t_sec).strftime('%Y/%m/%d')
          log_stream_name = mock_store_cloudwatch_logs.call_args_list[0].args[1]
          parts = log_stream_name.split('/')
          date_part = '/'.join(parts[:3])
          self.assertEqual(date_part, today)
  
      def _create_function(self, function_name, tags={}):
          region = lambda_api.LambdaRegion.get()
          arn = lambda_api.func_arn(function_name)
          region.lambdas[arn] = LambdaFunction(arn)
          region.lambdas[arn].versions = {
              '$LATEST': {'CodeSize': self.CODE_SIZE, 'CodeSha256': self.CODE_SHA_256, 'RevisionId': self.REVISION_ID}
          }
          region.lambdas[arn].handler = self.HANDLER
          region.lambdas[arn].runtime = self.RUNTIME
          region.lambdas[arn].timeout = self.TIMEOUT
          region.lambdas[arn].tags = tags
          region.lambdas[arn].envvars = {}
          region.lambdas[arn].last_modified = self.LAST_MODIFIED
          region.lambdas[arn].role = self.ROLE
          region.lambdas[arn].memory_size = self.MEMORY_SIZE
  
      def _update_function_code(self, function_name, tags={}):
          region = lambda_api.LambdaRegion.get()
          arn = lambda_api.func_arn(function_name)
          region.lambdas[arn].versions.update({
              '$LATEST': {'CodeSize': self.CODE_SIZE,
              'CodeSha256': self.UPDATED_CODE_SHA_256,
              'RevisionId': self.REVISION_ID}
          })
  
      def _assert_contained(self, child, parent):
          self.assertTrue(set(child.items()).issubset(set(parent.items())))
  
  
  class TestLambdaEventInvokeConfig(unittest.TestCase):
      CODE_SIZE = 50
      CODE_SHA_256 = '/u60ZpAA9bzZPVwb8d4390i5oqP1YAObUwV03CZvsWA='
      MEMORY_SIZE = 128
      ROLE = LAMBDA_TEST_ROLE
      LAST_MODIFIED = datetime.datetime.utcnow()
      REVISION_ID = 'e54dbcf8-e3ef-44ab-9af7-8dbef510608a'
      HANDLER = 'index.handler'
      RUNTIME = 'node.js4.3'
      TIMEOUT = 60
      FUNCTION_NAME = 'test1'
      RETRY_ATTEMPTS = 5
      EVENT_AGE = 360
      DL_QUEUE = 'arn:aws:sqs:us-east-1:000000000000:dlQueue'
      LAMBDA_OBJ = LambdaFunction(lambda_api.func_arn('test1'))
  
      def _create_function(self, function_name, tags={}):
          self.LAMBDA_OBJ.versions = {
              '$LATEST': {'CodeSize': self.CODE_SIZE, 'CodeSha256': self.CODE_SHA_256, 'RevisionId': self.REVISION_ID}
          }
          self.LAMBDA_OBJ.handler = self.HANDLER
          self.LAMBDA_OBJ.runtime = self.RUNTIME
          self.LAMBDA_OBJ.timeout = self.TIMEOUT
          self.LAMBDA_OBJ.tags = tags
          self.LAMBDA_OBJ.envvars = {}
          self.LAMBDA_OBJ.last_modified = self.LAST_MODIFIED
          self.LAMBDA_OBJ.role = self.ROLE
          self.LAMBDA_OBJ.memory_size = self.MEMORY_SIZE
  
      # TODO: remove this test case. Already added it in integration test case
      def test_put_function_event_invoke_config(self):
          # creating a lambda function
          self._create_function(self.FUNCTION_NAME)
  
          # calling put_function_event_invoke_config
          payload = {
              'DestinationConfig': {
                  'OnFailure': {
                      'Destination': self.DL_QUEUE
                  }
              },
              'MaximumEventAgeInSeconds': self.EVENT_AGE,
              'MaximumRetryAttempts': self.RETRY_ATTEMPTS
          }
          response = self.LAMBDA_OBJ.put_function_event_invoke_config(
              payload
          )
          # checking if response is not None
          self.assertIsNotNone(response)
  
          # calling get_function_event_invoke_config
          response = self.LAMBDA_OBJ.get_function_event_invoke_config()
  
          # verifying set values
          self.assertEqual(response['FunctionArn'], self.LAMBDA_OBJ.id)
          self.assertEqual(response['MaximumRetryAttempts'], self.RETRY_ATTEMPTS)
          self.assertEqual(response['MaximumEventAgeInSeconds'], self.EVENT_AGE)
          self.assertEqual(response['DestinationConfig']['OnFailure']['Destination'], self.DL_QUEUE)
  