Documentation of Client/HttpClient and data binding unit tests (#417)

* Added as_route documentation

* Added documentation for client

* Improve tests for binding

* Changes for client docs

* Fix docs indentations at client part

* Added missed imports

* Small fixes and refs

* Fix typos

* Fix errors and typos.
This commit is contained in:
Krukov D 2016-10-31 14:42:10 +03:00 committed by Andrew Godwin
parent c16de0e1e3
commit e24bc17bbf
4 changed files with 285 additions and 69 deletions

View File

@ -28,35 +28,34 @@ class TestsBinding(ChannelTestCase):
def has_permission(self, user, action, pk):
return True
with apply_routes([route('test', TestBinding.consumer)]):
client = HttpClient()
client.join_group('users')
client = HttpClient()
client.join_group('users')
user = User.objects.create(username='test', email='test@test.com')
user = User.objects.create(username='test', email='test@test.com')
consumer_finished.send(sender=None)
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'])
consumer_finished.send(sender=None)
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']['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'], '')
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)
received = client.receive()
self.assertIsNone(received)
def test_trigger_outbound_create_exclude(self):
class TestBinding(WebsocketBinding):
@ -134,36 +133,35 @@ class TestsBinding(ChannelTestCase):
user = User.objects.create(username='test', email='test@test.com')
consumer_finished.send(sender=None)
with apply_routes([route('test', TestBinding.consumer)]):
client = HttpClient()
client.join_group('users2')
client = HttpClient()
client.join_group('users2')
user.username = 'test_new'
user.save()
user.username = 'test_new'
user.save()
consumer_finished.send(sender=None)
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'])
consumer_finished.send(sender=None)
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']['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'], '')
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)
received = client.receive()
self.assertIsNone(received)
def test_trigger_outbound_delete(self):
class TestBinding(WebsocketBinding):
@ -182,28 +180,27 @@ class TestsBinding(ChannelTestCase):
user = User.objects.create(username='test', email='test@test.com')
consumer_finished.send(sender=None)
with apply_routes([route('test', TestBinding.consumer)]):
client = HttpClient()
client.join_group('users3')
client = HttpClient()
client.join_group('users3')
user.delete()
user.delete()
consumer_finished.send(sender=None)
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'])
consumer_finished.send(sender=None)
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')
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)
received = client.receive()
self.assertIsNone(received)
def test_demultiplexer(self):
class Demultiplexer(WebsocketDemultiplexer):
@ -341,6 +338,8 @@ class TestsBinding(ChannelTestCase):
self.assertEqual(user.username, 'test_inbound')
self.assertEqual(user.email, 'test@user_steam.com')
self.assertIsNone(client.receive())
def test_inbound_update(self):
user = User.objects.create(username='test', email='test@channels.com')
@ -388,6 +387,8 @@ class TestsBinding(ChannelTestCase):
self.assertEqual(user.username, 'test_inbound')
self.assertEqual(user.email, 'test@channels.com')
self.assertIsNone(client.receive())
def test_inbound_delete(self):
user = User.objects.create(username='test', email='test@channels.com')
@ -420,4 +421,5 @@ class TestsBinding(ChannelTestCase):
# 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())
self.assertIsNone(User.objects.filter(pk=user.pk).first())
self.assertIsNone(client.receive())

View File

@ -250,3 +250,40 @@ want to override; like so::
You can also use the Django ``method_decorator`` utility to wrap methods that
have ``message`` as their first positional argument - note that it won't work
for more high-level methods, like ``WebsocketConsumer.receive``.
As route
--------
Instead of making routes using ``route_class`` you may use the ``as_route`` shortcut.
This function takes route filters (:ref:`filters`) as kwargs and returns
``route_class``. For example::
from . import consumers
channel_routing = [
consumers.ChatServer.as_route(path=r"^/chat/"),
]
Use the ``attrs`` dict keyword for dynamic class attributes. For example you have
the generic consumer::
class MyGenericConsumer(WebsocketConsumer):
group = 'default'
group_prefix = ''
def connection_groups(self, **kwargs):
return ['_'.join(self.group_prefix, self.group)]
You can create consumers with different ``group`` and ``group_prefix`` with ``attrs``,
like so::
from . import consumers
channel_routing = [
consumers.MyGenericConsumer.as_route(path=r"^/path/1/",
attrs={'group': 'one', 'group_prefix': 'pre'}),
consumers.MyGenericConsumer.as_route(path=r"^/path/2/",
attrs={'group': 'two', 'group_prefix': 'public'}),
]

View File

@ -33,6 +33,7 @@ The three default routing objects are:
* ``include``: Takes either a list or string import path to a routing list,
and optional filter keyword arguments.
.. _filters:
Filters
-------

View File

@ -108,6 +108,182 @@ do group adds and sends during a test. For example::
self.assertEqual(result['value'], 42)
Clients
-------
For more complicated test suites you can use the ``Client`` abstraction that
provides an easy way to test the full life cycle of messages with a couple of methods:
``send`` to sending message with given content to the given channel, ``consume``
to run appointed consumer for the next message, ``receive`` to getting replies for client.
Very often you may need to ``send`` and than call a consumer one by one, for this
purpose use ``send_and_consume`` method::
from channels.tests import ChannelTestCase, Client
class MyTests(ChannelTestCase):
def test_my_consumer(self):
client = Client()
client.send_and_consume('my_internal_channel', {'value': 'my_value'})
self.assertEqual(client.receive(), {'all is': 'done'})
You can use ``HttpClient`` for websocket related consumers. It automatically serializes JSON content,
manage cookies and headers, give easy access to the session and add ability to authorize your requests.
For example::
# consumers.py
class RoomConsumer(JsonWebsocketConsumer):
http_user = True
groups = ['rooms_watchers']
def receive(self, content, **kwargs):
self.send({'rooms': self.message.http_session.get("rooms", [])})
Channel("rooms_receive").send({'user': self.message.user.id,
'message': content['message']}
# tests.py
from channels import Group
from channels.tests import ChannelTestCase, HttpClient
class RoomsTests(ChannelTestCase):
def test_rooms(self):
client = HttpClient()
user = User.objects.create_user(username='test', email='test@test.com',
password='123456') # fuck you security
client.login(username='test', password='123456')
client.send_and_consume('websocket.connect', '/rooms/')
# check that there is nothing to receive
self.assertIsNone(client.receive())
# test that the client in the group
Group(RoomConsumer.groups[0]).send({'text': 'ok'}, immediately=True)
self.assertEqual(client.receive(json=False), 'ok')
client.session['rooms'] = ['test', '1']
client.session.save()
client.send_and_consume('websocket.receive',
text={'message': 'hey'},
path='/rooms/')
# test 'response'
self.assertEqual(client.receive(), {'rooms': ['test', '1']})
self.assertEqual(self.get_next_message('rooms_receive').content,
{'user': user.id, 'message': 'hey'})
# There is nothing to receive
self.assertIsNone(client.receive())
Instead of ``HttpClient.login`` method with credentials at arguments you
may call ``HttpClient.force_login`` (like at django client) with the user object.
``receive`` method by default trying to deserialize json text content of a message,
so if you need to pass decoding use ``receive(json=False)``, like in the example.
Applying routes
---------------
When you need to testing you consumers without routes in settings or you
want to testing your consumers in more isolate and atomic way, it will be
simpler with ``apply_routes`` contextmanager and decorator for your ``ChannelTestCase``.
It takes list of routes that you want to use and overwrite existing routes::
from channels.tests import ChannelTestCase, HttpClient, apply_routes
class MyTests(ChannelTestCase):
def test_myconsumer(self):
client = HttpClient()
with apply_routes([MyConsumer.as_route(path='/new')]):
client.send_and_consume('websocket.connect', '/new')
self.assertEqual(client.receive(), {'key': 'value'})
Test Data binding with ``HttpClient``
-------------------------------------
As you know data binding in channels works in outbound and inbound ways,
so that ways tests in different ways and ``HttpClient`` and ``apply_routes``
will help to do this.
When you testing outbound consumers you need just import your ``Binding``
subclass with specified ``group_names``. At test you can join to one of them,
make some changes with target model and check received message.
Lets test ``IntegerValueBinding`` from :doc:`data binding <binding>`
with creating::
from channels.tests import ChannelTestCase, HttpClient
from channels.signals import consumer_finished
class TestIntegerValueBinding(ChannelTestCase):
def test_outbound_create(self):
# We use HttpClient because of json encoding messages
client = HttpClient()
client.join_group("intval-updates") # join outbound binding
# create target entity
value = IntegerValue.objects.create(name='fifty', value=50)
consumer_finished.send(sender=None)
received = client.receive() # receive outbound binding message
self.assertIsNotNone(received)
self.assertTrue('payload' in received)
self.assertTrue('action' in received['payload'])
self.assertTrue('data' in received['payload'])
self.assertTrue('name' in received['payload']['data'])
self.assertTrue('value' in received['payload']['data'])
self.assertEqual(received['payload']['action'], 'create')
self.assertEqual(received['payload']['model'], 'values.integervalue')
self.assertEqual(received['payload']['pk'], value.pk)
self.assertEqual(received['payload']['data']['name'], 'fifty')
self.assertEqual(received['payload']['data']['value'], 50)
# assert that is nothing to receive
self.assertIsNone(client.receive())
There is another situation with inbound binding. It is used with :ref:`multiplexing`,
So we apply two routes: websocket route for demultiplexer and route with internal
consumer for binding itself, connect to websocket entrypoint and test different actions.
For example::
class TestIntegerValueBinding(ChannelTestCase):
def test_inbound_create(self):
# check that initial state is empty
self.assertEqual(IntegerValue.objects.all().count(), 0)
with apply_routes([Demultiplexer.as_route(path='/'),
route("binding.intval", IntegerValueBinding.consumer)]):
client = HttpClient()
client.send_and_consume('websocket.connect', path='/')
client.send_and_consume('websocket.receive', path='/', text={
'stream': 'intval',
'payload': {'action': CREATE, 'data': {'name': 'one', 'value': 1}}
})
# our Demultiplexer route message to the inbound consumer,
# so we need to call this consumer
client.consume('binding.users')
self.assertEqual(IntegerValue.objects.all().count(), 1)
value = IntegerValue.objects.all().first()
self.assertEqual(value.name, 'one')
self.assertEqual(value.value, 1)
Multiple Channel Layers
-----------------------
@ -116,5 +292,5 @@ of the layers you want to mock as the ``test_channel_aliases`` attribute on
the ``ChannelTestCase`` subclass; by default, only the ``default`` layer is
mocked.
You can pass an ``alias`` argument to ``get_next_message`` and ``Channel``
You can pass an ``alias`` argument to ``get_next_message``, ``Client`` and ``Channel``
to use a different layer too.