daphne/channels/tests/test_binding.py

from __future__ import unicode_literals
import json

from django.contrib.auth import get_user_model
from channels.binding.base import CREATE, UPDATE, DELETE
from channels.binding.websockets import WebsocketBinding
from channels.generic.websockets import WebsocketDemultiplexer
from channels.tests import ChannelTestCase, apply_routes, HttpClient
from channels import route, Group

User = get_user_model()


class TestsBinding(ChannelTestCase):

    def test_trigger_outbound_create(self):

        class TestBinding(WebsocketBinding):
            model = User
            stream = 'test'
            fields = ['username', 'email', 'password', 'last_name']

            @classmethod
            def group_names(cls, instance, action):
                return ["users"]

            def has_permission(self, user, action, pk):
                return True

        with apply_routes([route('test', TestBinding.consumer)]):
            client = HttpClient()
            client.join_group('users')

            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('last_name' in received['payload']['data'])
            self.assertTrue('model' in received['payload'])
            self.assertTrue('pk' in received['payload'])

            self.assertEqual(received['payload']['action'], 'create')
            self.assertEqual(received['payload']['model'], 'auth.user')
            self.assertEqual(received['payload']['pk'], user.pk)

            self.assertEqual(received['payload']['data']['email'], 'test@test.com')
            self.assertEqual(received['payload']['data']['username'], 'test')
            self.assertEqual(received['payload']['data']['password'], '')
            self.assertEqual(received['payload']['data']['last_name'], '')

            received = client.receive()
            self.assertIsNone(received)

    def test_trigger_outbound_update(self):
        class TestBinding(WebsocketBinding):
            model = User
            stream = 'test'
            fields = ['__all__']

            @classmethod
            def group_names(cls, instance, action):
                return ["users2"]

            def has_permission(self, user, action, pk):
                return True

        user = User.objects.create(username='test', email='test@test.com')

        with apply_routes([route('test', TestBinding.consumer)]):
            client = HttpClient()
            client.join_group('users2')

            user.username = 'test_new'
            user.save()

            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('last_name' in received['payload']['data'])
            self.assertTrue('model' in received['payload'])
            self.assertTrue('pk' in received['payload'])

            self.assertEqual(received['payload']['action'], 'update')
            self.assertEqual(received['payload']['model'], 'auth.user')
            self.assertEqual(received['payload']['pk'], user.pk)

            self.assertEqual(received['payload']['data']['email'], 'test@test.com')
            self.assertEqual(received['payload']['data']['username'], 'test_new')
            self.assertEqual(received['payload']['data']['password'], '')
            self.assertEqual(received['payload']['data']['last_name'], '')

            received = client.receive()
            self.assertIsNone(received)

    def test_trigger_outbound_delete(self):
        class TestBinding(WebsocketBinding):
            model = User
            stream = 'test'
            fields = ['username']

            @classmethod
            def group_names(cls, instance, action):
                return ["users3"]

            def has_permission(self, user, action, pk):
                return True

        user = User.objects.create(username='test', email='test@test.com')

        with apply_routes([route('test', TestBinding.consumer)]):
            client = HttpClient()
            client.join_group('users3')

            user.delete()

            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('model' in received['payload'])
            self.assertTrue('pk' in received['payload'])

            self.assertEqual(received['payload']['action'], 'delete')
            self.assertEqual(received['payload']['model'], 'auth.user')
            self.assertEqual(received['payload']['pk'], 1)
            self.assertEqual(received['payload']['data']['username'], 'test')

            received = client.receive()
            self.assertIsNone(received)

    def test_demultiplexer(self):
        class Demultiplexer(WebsocketDemultiplexer):
            mapping = {
                'users': 'binding.users',
            }

            groups = ['inbound']

        with apply_routes([Demultiplexer.as_route(path='/')]):
            client = HttpClient()
            client.send_and_consume('websocket.connect', path='/')

            # assert in group
            Group('inbound').send({'text': json.dumps({'test': 'yes'})})
            self.assertEqual(client.receive(), {'test': 'yes'})

            # assert that demultiplexer stream message
            client.send_and_consume('websocket.receive', path='/',
                                    text={'stream': 'users', 'payload': {'test': 'yes'}})
            message = client.get_next_message('binding.users')
            self.assertIsNotNone(message)
            self.assertEqual(message.content['test'], 'yes')

    def test_demultiplexer_with_wrong_stream(self):
        class Demultiplexer(WebsocketDemultiplexer):
            mapping = {
                'users': 'binding.users',
            }

            groups = ['inbound']

        with apply_routes([Demultiplexer.as_route(path='/')]):
            client = HttpClient()
            client.send_and_consume('websocket.connect', path='/')

            with self.assertRaises(ValueError) as value_error:
                client.send_and_consume('websocket.receive', path='/', text={
                    'stream': 'wrong', 'payload': {'test': 'yes'}
                })

            self.assertIn('stream not mapped', value_error.exception.args[0])

            message = client.get_next_message('binding.users')
            self.assertIsNone(message)

    def test_demultiplexer_with_wrong_payload(self):
        class Demultiplexer(WebsocketDemultiplexer):
            mapping = {
                'users': 'binding.users',
            }

            groups = ['inbound']

        with apply_routes([Demultiplexer.as_route(path='/')]):
            client = HttpClient()
            client.send_and_consume('websocket.connect', path='/')

            with self.assertRaises(ValueError) as value_error:
                client.send_and_consume('websocket.receive', path='/', text={
                    'stream': 'users', 'payload': 'test',
                })

            self.assertEqual(value_error.exception.args[0], 'Multiplexed frame payload is not a dict')

            message = client.get_next_message('binding.users')
            self.assertIsNone(message)

    def test_demultiplexer_without_payload_and_steam(self):
        class Demultiplexer(WebsocketDemultiplexer):
            mapping = {
                'users': 'binding.users',
            }

            groups = ['inbound']

        with apply_routes([Demultiplexer.as_route(path='/')]):
            client = HttpClient()
            client.send_and_consume('websocket.connect', path='/')

            with self.assertRaises(ValueError) as value_error:
                client.send_and_consume('websocket.receive', path='/', text={
                    'nostream': 'users', 'payload': 'test',
                })

            self.assertIn('no channel/payload key', value_error.exception.args[0])

            message = client.get_next_message('binding.users')
            self.assertIsNone(message)

            with self.assertRaises(ValueError) as value_error:
                client.send_and_consume('websocket.receive', path='/', text={
                    'stream': 'users',
                })

            self.assertIn('no channel/payload key', value_error.exception.args[0])

            message = client.get_next_message('binding.users')
            self.assertIsNone(message)

    def test_inbound_create(self):
        self.assertEqual(User.objects.all().count(), 0)

        class Demultiplexer(WebsocketDemultiplexer):
            mapping = {
                'users': 'binding.users',
            }

            groups = ['inbound']

        class UserBinding(WebsocketBinding):
            model = User
            stream = 'users'
            fields = ['username', 'email', 'password', 'last_name']

            @classmethod
            def group_names(cls, instance, action):
                return ['users_outbound']

            def has_permission(self, user, action, pk):
                return True

        with apply_routes([Demultiplexer.as_route(path='/'), route('binding.users', UserBinding.consumer)]):
            client = HttpClient()
            client.send_and_consume('websocket.connect', path='/')
            client.send_and_consume('websocket.receive', path='/', text={
                'stream': 'users',
                'payload': {'action': CREATE, 'data': {'username': 'test_inbound', 'email': 'test@user_steam.com'}}
            })
            # our Demultiplexer route message to the inbound consumer, so call Demultiplexer consumer
            client.consume('binding.users')

        self.assertEqual(User.objects.all().count(), 1)
        user = User.objects.all().first()
        self.assertEqual(user.username, 'test_inbound')
        self.assertEqual(user.email, 'test@user_steam.com')

    def test_inbound_update(self):
        user = User.objects.create(username='test', email='test@channels.com')

        class Demultiplexer(WebsocketDemultiplexer):
            mapping = {
                'users': 'binding.users',
            }

            groups = ['inbound']

        class UserBinding(WebsocketBinding):
            model = User
            stream = 'users'
            fields = ['username', ]

            @classmethod
            def group_names(cls, instance, action):
                return ['users_outbound']

            def has_permission(self, user, action, pk):
                return True

        with apply_routes([Demultiplexer.as_route(path='/'), route('binding.users', UserBinding.consumer)]):
            client = HttpClient()
            client.send_and_consume('websocket.connect', path='/')
            client.send_and_consume('websocket.receive', path='/', text={
                'stream': 'users',
                'payload': {'action': UPDATE, 'pk': user.pk, 'data': {'username': 'test_inbound'}}
            })
            # our Demultiplexer route message to the inbound consumer, so call Demultiplexer consumer
            client.consume('binding.users')

            user = User.objects.get(pk=user.pk)
            self.assertEqual(user.username, 'test_inbound')
            self.assertEqual(user.email, 'test@channels.com')

            # trying change field that not in binding fields
            client.send_and_consume('websocket.receive', path='/', text={
                'stream': 'users',
                'payload': {'action': UPDATE, 'pk': user.pk, 'data': {'email': 'new@test.com'}}
            })
            client.consume('binding.users')

            user = User.objects.get(pk=user.pk)
            self.assertEqual(user.username, 'test_inbound')
            self.assertEqual(user.email, 'test@channels.com')

    def test_inbound_delete(self):
        user = User.objects.create(username='test', email='test@channels.com')

        class Demultiplexer(WebsocketDemultiplexer):
            mapping = {
                'users': 'binding.users',
            }

            groups = ['inbound']

        class UserBinding(WebsocketBinding):
            model = User
            stream = 'users'
            fields = ['username', ]

            @classmethod
            def group_names(cls, instance, action):
                return ['users_outbound']

            def has_permission(self, user, action, pk):
                return True

        with apply_routes([Demultiplexer.as_route(path='/'), route('binding.users', UserBinding.consumer)]):
            client = HttpClient()
            client.send_and_consume('websocket.connect', path='/')
            client.send_and_consume('websocket.receive', path='/', text={
                'stream': 'users',
                'payload': {'action': DELETE, 'pk': user.pk}
            })
            # our Demultiplexer route message to the inbound consumer, so call Demultiplexer consumer
            client.consume('binding.users')

            self.assertIsNone(User.objects.filter(pk=user.pk).first())