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_websocket_close_exception(self):
def consumer(message):
raise WebsocketCloseException
client = WSClient(reply_channel='daphne.response.1')
with apply_routes(route('websocket.receive', consumer)):
client.send_and_consume('websocket.receive')
self.assertEqual(client.get_next_message(client.reply_channel).content, {'close': True})
def test_trigger_outbound_create_exclude(self):
class TestBinding(WebsocketBinding):
model = User
stream = 'test'
exclude = ['first_name', 'last_name']
@classmethod
def group_names(cls, instance):
return ["users_exclude"]
def has_permission(self, user, action, pk):
return True
with apply_routes([route('test', TestBinding.consumer)]):
client = WSClient()
client.join_group('users_exclude')
user = User.objects.create(username='test', email='test@test.com')
received = client.receive()
self.assertTrue('payload' in received)
self.assertTrue('action' in received['payload'])
self.assertTrue('data' in received['payload'])
self.assertTrue('username' in received['payload']['data'])
self.assertTrue('email' in received['payload']['data'])
self.assertTrue('password' in received['payload']['data'])
self.assertTrue('model' in received['payload'])
self.assertTrue('pk' in received['payload'])
self.assertFalse('last_name' in received['payload']['data'])
from channels import include, route
from .consumers import ws_connect, ws_receive, ws_disconnect
websocket_routing = [
route('websocket.connect', ws_connect),
route('websocket.receive', ws_receive),
route('websocket.disconnect', ws_disconnect),
]
channel_routing = [
# /Room/
include('coding_night_live.routing.websocket_routing', path=r'^/room/'),
include('manage_room.routing.custom_routing'),
# /Chat/
include('manage_chat.routing.custom_routing'),
]
from channels import route
from .consumers import ws_connect, ws_receive, ws_disconnect, chat_join, chat_leave, chat_send
# There's no path matching on these routes; we just rely on the matching
# from the top-level routing. We _could_ path match here if we wanted.
websocket_routing = [
# Called when WebSockets connect
route("websocket.connect", ws_connect),
# Called when WebSockets get sent a data frame
route("websocket.receive", ws_receive),
# Called when WebSockets disconnect
route("websocket.disconnect", ws_disconnect),
]
# You can have as many lists here as you like, and choose any name.
# Just refer to the individual names in the include() function.
custom_routing = [
# Handling different chat commands (websocket.receive is decoded and put
# onto this channel) - routed on the "command" attribute of the decoded
# message.
route("chat.receive", chat_join, command="^join$"),
route("chat.receive", chat_leave, command="^leave$"),
route("chat.receive", chat_send, command="^send$"),
]
from channels import route, route_class
from .consumers import ChatServer, events, Demultiplexer, RoomBinding
chat_routing = [
route_class(ChatServer, path=r'^(?P[^/]+)/stream/$'),
]
event_routing = [
route('chat.receive', events.user_join, event=r'^user-join$'),
route('chat.receive', events.user_leave, event=r'^user-leave$'),
route('chat.receive', events.client_send, event=r'^message-send$'),
]
binding_routing = [
route_class(Demultiplexer, path=r'^/binding/'),
route('binding.room', RoomBinding.consumer),
]
from channels import route, route_class
from .consumers import ChatServer, events, Demultiplexer, RoomBinding
chat_routing = [
route_class(ChatServer, path=r'^(?P[^/]+)/stream/$'),
]
event_routing = [
route('chat.receive', events.user_join, event=r'^user-join$'),
route('chat.receive', events.user_leave, event=r'^user-leave$'),
route('chat.receive', events.client_send, event=r'^message-send$'),
]
binding_routing = [
route_class(Demultiplexer, path=r'^/binding/'),
route('binding.room', RoomBinding.consumer),
]
# -*- coding: utf-8 -*-
from __future__ import absolute_import, print_function, unicode_literals
from channels import route
from .consumers import liveblog_connect, liveblog_disconnect
channel_routing = [
route(
'websocket.connect', liveblog_connect,
path=r'^/liveblog/(?P[a-zA-Z0-9_-]+)/'
r'(?P[a-zA-Z_-]+)/(?P[a-zA-Z0-9_-]+)/$'
),
route(
'websocket.disconnect', liveblog_disconnect,
path=r'^/liveblog/(?P[a-zA-Z0-9_-]+)/'
r'(?P[a-zA-Z_-]+)/(?P[a-zA-Z0-9_-]+)/$'
from channels import route
from manage_room import consumers
# Websocket command : join => call room_join
custom_routing = [
route("room.receive", consumers.room_join, command="^join$"),
route("room.receive", consumers.room_leave, command="^leave$"),
route("room.receive", consumers.rename_room, command="^rename_room$"),
route("room.receive", consumers.new_slide, command="^new_slide$"),
route("room.receive", consumers.get_slide, command="^get_slide$"),
route("room.receive", consumers.del_slide, command="^del_slide$"),
route("room.receive", consumers.curr_slide, command="^curr_slide$"),
route("room.receive", consumers.change_slide_order, command="^change_slide_order$"),
route("room.receive", consumers.rename_slide, command="^rename_slide_title$"),
route("room.receive", consumers.change_slide, command="^change_slide$"),
route("room.receive", consumers.get_slide_diff, command="^get_slide_diff$"),
]
from channels import route
from manage_room import consumers
# Websocket command : join => call room_join
custom_routing = [
route("room.receive", consumers.room_join, command="^join$"),
route("room.receive", consumers.room_leave, command="^leave$"),
route("room.receive", consumers.rename_room, command="^rename_room$"),
route("room.receive", consumers.new_slide, command="^new_slide$"),
route("room.receive", consumers.get_slide, command="^get_slide$"),
route("room.receive", consumers.del_slide, command="^del_slide$"),
route("room.receive", consumers.curr_slide, command="^curr_slide$"),
route("room.receive", consumers.change_slide_order, command="^change_slide_order$"),
route("room.receive", consumers.rename_slide, command="^rename_slide_title$"),
route("room.receive", consumers.change_slide, command="^change_slide$"),
route("room.receive", consumers.get_slide_diff, command="^get_slide_diff$"),
]
from channels import route
from .consumers import ws_connect, ws_receive, ws_disconnect, chat_join, chat_leave, chat_send
# There's no path matching on these routes; we just rely on the matching
# from the top-level routing. We _could_ path match here if we wanted.
websocket_routing = [
# Called when WebSockets connect
route("websocket.connect", ws_connect),
# Called when WebSockets get sent a data frame
route("websocket.receive", ws_receive),
# Called when WebSockets disconnect
route("websocket.disconnect", ws_disconnect),
]
# You can have as many lists here as you like, and choose any name.
# Just refer to the individual names in the include() function.
custom_routing = [
# Handling different chat commands (websocket.receive is decoded and put
# onto this channel) - routed on the "command" attribute of the decoded
# message.
route("chat.receive", chat_join, command="^join$"),