Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def assert_loads_needed_variables(self, plan):
# Parse arn and store region/account id for future
# API calls.
assert plan[0:4] == [
models.BuiltinFunction(
'parse_arn', [Variable('function_name_lambda_arn')],
output_var='parsed_lambda_arn',
),
models.JPSearch('account_id',
input_var='parsed_lambda_arn',
output_var='account_id'),
models.JPSearch('region',
input_var='parsed_lambda_arn',
output_var='region_name'),
# Verify we copy the function arn as needed.
models.CopyVariable(
from_var='function_name_lambda_arn',
to_var='api_handler_lambda_arn'),
]
method_name='update_rest_api',
params={
'rest_api_id': Variable('rest_api_id'),
'patch_operations': [{
'op': 'replace',
'path': '/minimumCompressionSize',
'value': '100',
}],
}
),
models.APICall(
method_name='add_permission_for_apigateway',
params={
'function_name': 'appname-dev-function_name',
'region_name': Variable('region_name'),
'account_id': Variable('account_id'),
'rest_api_id': Variable('rest_api_id'),
}
),
models.APICall(method_name='deploy_rest_api',
params={'rest_api_id': Variable('rest_api_id'),
'api_gateway_stage': 'api'}),
models.StoreValue(
name='rest_api_url',
value=StringFormat(
'https://{rest_api_id}.execute-api.{region_name}'
'.amazonaws.com/api/',
['rest_api_id', 'region_name'],
),
),
models.RecordResourceVariable(
resource_type='rest_api',
def test_can_reference_stored_results_in_api_calls(self):
params = {
'name': Variable('role_name'),
'trust_policy': {'trust': 'policy'},
'policy': {'iam': 'policy'}
}
call = APICall('create_role', params)
self.mock_client.create_role.return_value = 'myrole:arn'
self.executor.variables['role_name'] = 'myrole-name'
self.execute([call])
self.mock_client.create_role.assert_called_with(
name='myrole-name',
trust_policy={'trust': 'policy'},
policy={'iam': 'policy'},
)
plan = self.determine_plan(websocket_api)
self.assert_loads_needed_variables(plan)
assert plan[3:] == [
models.StoreValue(
name='websocket_api_id',
value='my_websocket_api_id',
),
models.APICall(
method_name='get_websocket_routes',
params={'api_id': Variable('websocket_api_id')},
output_var='routes',
),
models.APICall(
method_name='delete_websocket_routes',
params={'api_id': Variable('websocket_api_id'),
'routes': Variable('routes')},
),
models.APICall(
method_name='get_websocket_integrations',
params={'api_id': Variable('websocket_api_id')},
output_var='integrations',
),
models.APICall(
method_name='delete_websocket_integrations',
params={'api_id': Variable('websocket_api_id'),
'integrations': Variable('integrations')},
),
models.StoreValue(
name='websocket-connect-integration-lambda-path',
value=StringFormat(
'arn:aws:apigateway:{region_name}:lambda:path/'
'2015-03-31/functions/arn:aws:lambda:{region_name}:'
),
models.StoreValue(
name='websocket-message-integration-lambda-path',
value=StringFormat(
'arn:aws:apigateway:{region_name}:lambda:path/'
'2015-03-31/functions/arn:aws:lambda:{region_name}:'
'{account_id}:function:%s/'
'invocations' % 'appname-dev-function_name_message',
['region_name', 'account_id'],
),
),
models.APICall(
method_name='create_websocket_integration',
params={
'api_id': Variable('websocket_api_id'),
'lambda_function': Variable(
'websocket-message-integration-lambda-path'),
'handler_type': 'message',
},
output_var='message-integration-id',
),
models.StoreValue(
name='websocket-disconnect-integration-lambda-path',
value=StringFormat(
'arn:aws:apigateway:{region_name}:lambda:path/'
'2015-03-31/functions/arn:aws:lambda:{region_name}:'
'{account_id}:function:%s/'
'invocations' % 'appname-dev-function_name_disconnect',
['region_name', 'account_id'],
),
),
models.APICall(
'/endpointConfiguration/types/%s' % (
'{rest_api[endpointConfiguration][types][0]}'),
['rest_api'])},
],
},
),
models.APICall(
method_name='add_permission_for_apigateway',
params={'rest_api_id': Variable("rest_api_id"),
'region_name': Variable("region_name"),
'account_id': Variable("account_id"),
'function_name': 'appname-dev-function_name'},
output_var=None),
models.APICall(
method_name='deploy_rest_api',
params={'rest_api_id': Variable('rest_api_id'),
'api_gateway_stage': 'api'},
),
models.StoreValue(
name='rest_api_url',
value=StringFormat(
'https://{rest_api_id}.execute-api.{region_name}'
'.amazonaws.com/api/',
['rest_api_id', 'region_name'],
),
),
models.RecordResourceVariable(
resource_type='rest_api',
resource_name='rest_api',
name='rest_api_url',
variable_name='rest_api_url'
),
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,
)
]
return plan
def resolve_variables(self, value, variables):
# type: (Any, Dict[str, str]) -> Any
if isinstance(value, Variable):
return variables[value.name]
elif isinstance(value, StringFormat):
v = {k: variables[k] for k in value.variables}
return value.template.format(**v)
elif isinstance(value, models.Placeholder):
# The key and method_name values are added
# as the exception propagates up the stack.
raise UnresolvedValueError('', value, '')
elif isinstance(value, dict):
final = {}
for k, v in value.items():
try:
final[k] = self.resolve_variables(v, variables)
except UnresolvedValueError as e:
e.key = k
raise
name='websocket-%s-integration-lambda-path' % key,
value=StringFormat(
'arn:aws:apigateway:{region_name}:lambda:path/'
'2015-03-31/functions/arn:aws:lambda:{region_name}:'
'{account_id}:function:%s/'
'invocations' % config['name'],
['region_name', 'account_id'],
),
),
)
instructions.append(
models.APICall(
method_name='create_websocket_integration',
params={
'api_id': Variable('websocket_api_id'),
'lambda_function': Variable(
'websocket-%s-integration-lambda-path' % key),
'handler_type': key,
},
output_var='%s-integration-id' % key,
),
)
return instructions