Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
function.memory_size = 256
plan = self.determine_plan(function)
existing_params = {
'function_name': 'appname-dev-function_name',
'role_arn': 'role:arn',
'zip_contents': mock.ANY,
'runtime': 'python2.7',
'environment_variables': {},
'tags': {},
'timeout': 60,
'security_group_ids': [],
'subnet_ids': [],
'layers': None
}
expected_params = dict(memory_size=256, **existing_params)
expected = [models.APICall(
method_name='update_function',
params=expected_params,
),
models.APICall(
method_name='delete_function_concurrency',
params={
'function_name': 'appname-dev-function_name',
},
output_var='reserved_concurrency_result',
)]
# update_function
self.assert_apicall_equals(plan[0], expected[0])
# delete_function_concurrency
self.assert_apicall_equals(plan[3], expected[1])
variable_name='websocket_api_url',
),
models.RecordResourceVariable(
resource_type='websocket_api',
resource_name='websocket_api',
name='websocket_api_id',
variable_name='websocket_api_id',
),
models.APICall(
method_name='add_permission_for_apigateway_v2',
params={'function_name': 'appname-dev-function_name_connect',
'region_name': Variable('region_name'),
'account_id': Variable('account_id'),
'api_id': Variable('websocket_api_id')},
),
models.APICall(
method_name='add_permission_for_apigateway_v2',
params={'function_name': 'appname-dev-function_name_message',
'region_name': Variable('region_name'),
'account_id': Variable('account_id'),
'api_id': Variable('websocket_api_id')},
),
models.APICall(
method_name='add_permission_for_apigateway_v2',
params={
'function_name': 'appname-dev-function_name_disconnect',
'region_name': Variable('region_name'),
'account_id': Variable('account_id'),
'api_id': Variable('websocket_api_id'),
},
deployed = {
'resources': [{
'name': 'test-s3-event',
'resource_type': 's3_event',
'bucket': 'OLDBUCKET',
'lambda_arn': 'lambda_arn',
}]
}
config = FakeConfig(deployed)
self.execute(plan, config)
assert plan[-2:] == [
models.APICall(
method_name='disconnect_s3_bucket_from_lambda',
params={'bucket': 'OLDBUCKET', 'function_arn': 'lambda_arn'},
),
models.APICall(
method_name='remove_permission_for_s3_event',
params={'bucket': 'OLDBUCKET', 'function_arn': 'lambda_arn'},
),
def test_can_return_created_resources(self):
params = {}
call = APICall('create_function', params,
output_var='myfunction_arn')
self.mock_client.create_function.return_value = 'function:arn'
record_instruction = RecordResourceVariable(
resource_type='lambda_function',
resource_name='myfunction',
name='myfunction_arn',
variable_name='myfunction_arn',
)
self.execute([call, record_instruction])
assert self.executor.resource_values == [{
'name': 'myfunction',
'myfunction_arn': 'function:arn',
'resource_type': 'lambda_function',
}]
def test_can_print_ui_messages(self):
params = {'name': 'foo', 'trust_policy': {'trust': 'policy'},
'policy': {'iam': 'policy'}}
call = APICall('create_role', params)
messages = {id(call): 'Creating role'}
self.execute([call], messages)
self.mock_client.create_role.assert_called_with(**params)
self.ui.write.assert_called_with('Creating role')
params={
'function_name': 'appname-dev-function_name',
'role_arn': 'role:arn',
'zip_contents': mock.ANY,
'runtime': 'python2.7',
'handler': 'app.app',
'environment_variables': {},
'tags': {},
'timeout': 60,
'memory_size': 128,
'security_group_ids': [],
'subnet_ids': [],
'layers': None
},
),
models.APICall(
method_name='delete_function_concurrency',
params={
'function_name': 'appname-dev-function_name',
},
output_var='reserved_concurrency_result',
)]
# create_function
self.assert_apicall_equals(plan[0], expected[0])
# delete_function_concurrency
self.assert_apicall_equals(plan[2], expected[1])
assert list(self.last_plan.messages.values()) == [
'Creating lambda function: appname-dev-function_name\n',
]
params={'swagger_document': resource.swagger_doc,
'endpoint_type': resource.endpoint_type},
output_var='rest_api_id',
), "Creating Rest API\n"),
models.RecordResourceVariable(
resource_type='rest_api',
resource_name=resource.resource_name,
name='rest_api_id',
variable_name='rest_api_id',
),
]
else:
deployed = self._remote_state.resource_deployed_values(resource)
shared_plan_epilogue.insert(
0,
models.APICall(
method_name='get_rest_api',
params={'rest_api_id': Variable('rest_api_id')},
output_var='rest_api')
)
shared_plan_patch_ops.append({
'op': 'replace',
'path': StringFormat(
'/endpointConfiguration/types/%s' % (
'{rest_api[endpointConfiguration][types][0]}'),
['rest_api']),
'value': resource.endpoint_type}
)
plan = shared_plan_preamble + [
models.StoreValue(
name='rest_api_id',
value=deployed['rest_api_id']),
if isinstance(resource, models.ScheduledEvent):
resource = cast(models.ScheduledEvent, resource)
params['schedule_expression'] = resource.schedule_expression
if resource.rule_description is not None:
params['rule_description'] = resource.rule_description
else:
resource = cast(models.CloudWatchEvent, resource)
params['event_pattern'] = resource.event_pattern
plan = [
models.APICall(
method_name='get_or_create_rule_arn',
params=params,
output_var='rule-arn',
),
models.APICall(
method_name='connect_rule_to_lambda',
params={'rule_name': resource.rule_name,
'function_arn': function_arn}
),
models.APICall(
method_name='add_permission_for_cloudwatch_event',
params={'rule_arn': Variable('rule-arn'),
'function_arn': function_arn},
),
# You need to remove targets (which have IDs)
# before you can delete a rule.
models.RecordResourceValue(
resource_type='cloudwatch_event',
resource_name=resource.resource_name,
name='rule_name',
value=resource.rule_name,
def _create_route_for_key(self, route_key):
# type: (str) -> models.APICall
integration_id = {
'$connect': 'connect-integration-id',
'$disconnect': 'disconnect-integration-id',
}.get(route_key, 'message-integration-id')
return models.APICall(
method_name='create_websocket_route',
params={
'api_id': Variable('websocket_api_id'),
'route_key': route_key,
'integration_id': Variable(integration_id),
},
function_arn = Variable(
'%s_lambda_arn' % resource.lambda_function.resource_name
)
params = {'rule_name': resource.rule_name}
if isinstance(resource, models.ScheduledEvent):
resource = cast(models.ScheduledEvent, resource)
params['schedule_expression'] = resource.schedule_expression
if resource.rule_description is not None:
params['rule_description'] = resource.rule_description
else:
resource = cast(models.CloudWatchEvent, resource)
params['event_pattern'] = resource.event_pattern
plan = [
models.APICall(
method_name='get_or_create_rule_arn',
params=params,
output_var='rule-arn',
),
models.APICall(
method_name='connect_rule_to_lambda',
params={'rule_name': resource.rule_name,
'function_arn': function_arn}
),
models.APICall(
method_name='add_permission_for_cloudwatch_event',
params={'rule_arn': Variable('rule-arn'),
'function_arn': function_arn},
),
# You need to remove targets (which have IDs)
# before you can delete a rule.