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_mount_event_sources_with_blueprint():
myapp = app.Chalice('myapp')
foo = app.Blueprint('app.chalicelib.blueprints.foo')
@foo.schedule('rate(5 minutes)')
def myfunction(event):
return event
myapp.register_blueprint(foo, name_prefix='myprefix_')
assert len(myapp.event_sources) == 1
event_source = myapp.event_sources[0]
assert event_source.name == 'myprefix_myfunction'
assert event_source.schedule_expression == 'rate(5 minutes)'
assert event_source.handler_string == (
'app.chalicelib.blueprints.foo.myfunction')
def test_content_type_with_charset(create_event):
demo = app.Chalice('demo-app')
@demo.route('/', content_types=['application/json'])
def index():
return {'foo': 'bar'}
event = create_event('/', 'GET', {}, 'application/json; charset=utf-8')
response = json_response_body(demo(event, context=None))
assert response == {'foo': 'bar'}
def sample_app():
demo = app.Chalice('demo-app')
@demo.route('/', methods=['GET'])
def index():
return {'hello': 'world'}
@demo.route('/test-cors', methods=['POST'], cors=True)
def test_cors():
return {'hello': 'world'}
return demo
def test_can_close_websocket_connection(create_websocket_event):
demo = app.Chalice('app-name')
client = FakeClient()
demo.websocket_api.session = FakeSession(client)
@demo.on_ws_message()
def message_handler(event):
demo.websocket_api.close('connection_id')
event = create_websocket_event('$default', body='foo bar')
handler = websocket_handler_for_route('$default', demo)
handler(event, context=None)
calls = client.calls['close']
assert len(calls) == 1
call = calls[0]
connection_id = call[0]
assert connection_id == 'connection_id'
def test_info_does_fail_if_already_disconnected(create_websocket_event):
demo = app.Chalice('app-name')
client = FakeClient(errors=[FakeGoneException])
demo.websocket_api.session = FakeSession(client)
@demo.on_ws_message()
def message_handler(event):
with pytest.raises(WebsocketDisconnectedError) as e:
demo.websocket_api.info('connection_id')
assert e.value.connection_id == 'connection_id'
event = create_websocket_event('$default', body='foo bar')
handler = websocket_handler_for_route('$default', demo)
handler(event, context=None)
calls = client.calls['info']
assert len(calls) == 1
call = calls[0]
def test_raw_body_cache_returns_same_result(create_event):
demo = app.Chalice('app-name')
@demo.route('/index')
def index_view():
# The first raw_body decodes base64,
# the second value should return the cached value.
# Both should be the same value
return {'rawbody': demo.current_request.raw_body.decode('utf-8'),
'rawbody2': demo.current_request.raw_body.decode('utf-8')}
event = create_event('/index', 'GET', {})
event['base64-body'] = base64.b64encode(
b'{"hello": "world"}').decode('ascii')
result = demo(event, context=None)
result = json_response_body(result)
assert result['rawbody'] == result['rawbody2']
def test_aws_execution_env_set():
env = {'AWS_EXECUTION_ENV': 'AWS_Lambda_python2.7'}
app.Chalice('app-name', env=env)
assert env['AWS_EXECUTION_ENV'] == (
'AWS_Lambda_python2.7 aws-chalice/%s' % chalice_version
)
def test_can_access_websocket_json_body_twice(create_websocket_event):
demo = app.Chalice('app-name')
client = FakeClient()
demo.websocket_api.session = FakeSession(client)
@demo.on_ws_message()
def message(event):
assert event.json_body == {'foo': 'bar'}
assert event.json_body == {'foo': 'bar'}
event = create_websocket_event('$default', body='{"foo": "bar"}')
demo(event, context=None)
def scheduled_event_app():
app = Chalice('scheduled-event')
@app.schedule('rate(5 minutes)')
def foo(event):
return {}
return app
def test_can_register_blueprint_on_app():
myapp = app.Chalice('myapp')
foo = app.Blueprint('foo')
@foo.route('/foo')
def first():
pass
myapp.register_blueprint(foo)
assert sorted(list(myapp.routes.keys())) == ['/foo']