Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_invalid_message(self):
"""
Test basic validation of event messages.
"""
shark = SocketShark(TEST_CONFIG)
client = MockClient(shark)
session = client.session
await session.on_client_event(None)
assert client.log.pop() == {
'status': 'error',
'error': c.ERR_INVALID_EVENT,
}
await session.on_client_event('hello')
assert client.log.pop() == {
'status': 'error',
'error': c.ERR_INVALID_EVENT,
}
await session.on_client_event({})
assert client.log.pop() == {
'status': 'error',
'error': c.ERR_INVALID_EVENT,
}
await session.on_client_event({'event': None})
assert client.log.pop() == {
def __init__(self, session, data):
super().__init__(session, data)
self.auth_config = self.config['AUTHENTICATION']
self.method = data.get('method', c.DEFAULT_AUTH_METHOD)
def from_data(cls, session, data):
if not isinstance(data, dict) or 'event' not in data:
return InvalidEvent(session)
event = data['event']
# Make sure we don't echo back large messages.
if not isinstance(event, str) or len(event) > c.MAX_EVENT_LENGTH:
return InvalidEvent(session)
cls = {
'auth': AuthEvent,
'message': MessageEvent,
'subscribe': SubscribeEvent,
'unsubscribe': UnsubscribeEvent,
}.get(event)
if cls:
return cls(session, data)
else:
return UnknownEvent(session, data)