Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- def calculation(request):
- # doing some simple calculation depending on request
- simple_result, parameter = simple_calculation(request)
- # trigger complex asynchronous calculation depending on parameter
- Channel('background-calculations').send({
- "parameter": parameter,
- })
- # return results of simple calculation
- return render(request, 'simple_reponse.html',{'simple_result': simple_result})
- channel_routing = [
- route("background-calculations", run_background_calculations),
- route("websocket.connect", ws_connect),
- route("websocket.receive", ws_message),
- ]
- def run_background_calculations(message):
- # perform complex calculation depending on parameter from simple calculation
- result = complex_calculation(message)
- # update frontend via websocket
- Group("frontend-updates").send({
- "text": json.dumps({
- "result": result,
- })
- })
- @channel_session
- def ws_connect(message):
- message.reply_channel.send({"accept": True})
- Group("frontend-updates").add(message.reply_channel)
- @channel_session_user
- def ws_message(message):
- message.reply_channel.send({
- "text": message.content['text'],
- })
- from channels.test import Client
- from rest_framework.test import APIClient
- # setup
- rest = APIClient()
- channelclient = Client()
- Group("frontend-updates").add(u"test-channel")
- rest.login(username="username",password="password")
- class CopyTestCase(ChannelTestCase):
- def test_end_to_end(self):
- # trigger HTTP POST request
- response = rest.post("/calculate/")
- #...check response...
- # consume the background-calculation task (has to be done manually)
- channelclient.consume(u"background-calculations")
- # check data sent to the front-end as a result of this calculation
- message = testcase.get_next_message(u"test-channel")
- #...check message...
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement