Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_can_delete_all_websocket_routes(self, stubbed_session):
stubbed_session.stub('apigatewayv2').delete_route(
ApiId='api-id',
RouteId='route-id',
).returns({})
stubbed_session.stub('apigatewayv2').delete_route(
ApiId='api-id',
RouteId='old-route-id',
).returns({})
stubbed_session.activate_stubs()
client = TypedAWSClient(stubbed_session)
client.delete_websocket_routes(
api_id='api-id',
routes=['route-id', 'old-route-id'],
)
stubbed_session.verify_stubs()
def test_update_function_with_no_tag_updates_needed(self, stubbed_session):
function_arn = 'arn'
lambda_client = stubbed_session.stub('lambda')
lambda_client.update_function_code(
FunctionName='name', ZipFile=b'foo').returns(
{'FunctionArn': function_arn})
lambda_client.list_tags(
Resource=function_arn).returns({'Tags': {'MyKey': 'SameValue'}})
stubbed_session.activate_stubs()
awsclient = TypedAWSClient(stubbed_session)
awsclient.update_function('name', b'foo', tags={'MyKey': 'SameValue'})
stubbed_session.verify_stubs()
def test_update_function_with_vpc_config(self, stubbed_session):
lambda_client = stubbed_session.stub('lambda')
lambda_client.update_function_code(
FunctionName='name', ZipFile=b'foo').returns({})
lambda_client.update_function_configuration(
FunctionName='name', VpcConfig={
'SecurityGroupIds': ['sg1', 'sg2'],
'SubnetIds': ['sn1', 'sn2']
}
).returns({})
stubbed_session.activate_stubs()
awsclient = TypedAWSClient(stubbed_session)
awsclient.update_function(
'name', b'foo',
subnet_ids=['sn1', 'sn2'],
security_group_ids=['sg1', 'sg2'],
)
stubbed_session.verify_stubs()
'FunctionName': 'name',
'Runtime': 'python2.7',
'Code': {'ZipFile': b'foo'},
'Handler': 'app.app',
'Role': 'myarn',
}
for _ in range(TypedAWSClient.LAMBDA_CREATE_ATTEMPTS):
stubbed_session.stub('lambda').create_function(
**kwargs).raises_error(
error_code='InvalidParameterValueException',
message=('The role defined for the function cannot '
'be assumed by Lambda.')
)
stubbed_session.activate_stubs()
awsclient = TypedAWSClient(stubbed_session, mock.Mock(spec=time.sleep))
with pytest.raises(LambdaClientError) as excinfo:
awsclient.create_function('name', 'myarn', b'foo', 'python2.7',
'app.app')
assert isinstance(
excinfo.value.original_error, botocore.exceptions.ClientError)
stubbed_session.verify_stubs()
def test_raises_large_deployment_error_for_connection_error(
self, stubbed_session):
too_large_content = b'a' * 60 * (1024 ** 2)
stubbed_session.stub('lambda').update_function_code(
FunctionName='name', ZipFile=too_large_content).raises_error(
error=RequestsConnectionError())
stubbed_session.activate_stubs()
awsclient = TypedAWSClient(stubbed_session, mock.Mock(spec=time.sleep))
with pytest.raises(DeploymentPackageTooLargeError) as excinfo:
awsclient.update_function('name', too_large_content)
stubbed_session.verify_stubs()
assert excinfo.value.context.function_name == 'name'
assert (
excinfo.value.context.client_method_name == 'update_function_code')
assert excinfo.value.context.deployment_size == 60 * (1024 ** 2)
def test_can_query_lambda_function_does_not_exist(self, stubbed_session):
stubbed_session.stub('lambda').get_function(FunctionName='myappname')\
.raises_error(error_code='ResourceNotFoundException',
message='ResourceNotFound')
stubbed_session.activate_stubs()
awsclient = TypedAWSClient(stubbed_session)
assert not awsclient.lambda_function_exists(name='myappname')
stubbed_session.verify_stubs()
def test_invoke_no_payload_no_context(self, stubbed_session):
stubbed_session.stub('lambda').invoke(
FunctionName='name',
InvocationType='RequestResponse',
).returns({})
stubbed_session.activate_stubs()
awsclient = TypedAWSClient(stubbed_session)
assert awsclient.invoke_function('name') == {}
stubbed_session.verify_stubs()
def test_create_function_fails_after_max_retries(self, stubbed_session):
kwargs = {
'FunctionName': 'name',
'Runtime': 'python2.7',
'Code': {'ZipFile': b'foo'},
'Handler': 'app.app',
'Role': 'myarn',
}
for _ in range(TypedAWSClient.LAMBDA_CREATE_ATTEMPTS):
stubbed_session.stub('lambda').create_function(
**kwargs).raises_error(
error_code='InvalidParameterValueException',
message=('The role defined for the function cannot '
'be assumed by Lambda.')
)
stubbed_session.activate_stubs()
awsclient = TypedAWSClient(stubbed_session, mock.Mock(spec=time.sleep))
with pytest.raises(LambdaClientError) as excinfo:
awsclient.create_function('name', 'myarn', b'foo', 'python2.7',
'app.app')
assert isinstance(
excinfo.value.original_error, botocore.exceptions.ClientError)
stubbed_session.verify_stubs()
def test_can_delete_rule(stubbed_session):
events = stubbed_session.stub('events')
events.remove_targets(
Rule='rule-name',
Ids=['1']).returns({})
events.delete_rule(Name='rule-name').returns({})
stubbed_session.activate_stubs()
awsclient = TypedAWSClient(stubbed_session)
awsclient.delete_rule('rule-name')
stubbed_session.verify_stubs()
def test_invoke_can_call_api_handler(self, stubbed_session):
arn = 'arn:aws:lambda:region:id:function:name-dev'
stubbed_session.stub('lambda').invoke(
FunctionName=arn,
InvocationType='RequestResponse'
).returns({})
stubbed_session.activate_stubs()
client = TypedAWSClient(stubbed_session)
invoker = LambdaInvoker(arn, client)
invoker.invoke()
stubbed_session.verify_stubs()