django/daphne
1
1
Fork 0
mirror of https://github.com/django/daphne.git synced 2025-07-29 00:19:45 +03:00
Code Issues Packages Projects Releases Wiki Activity

Rename HTTPClient to WSClient (#609)

Browse Source 
This is what it was actually intended as. HTTP testing methods may follow later.
This commit is contained in:
Krukov D 2017-04-17 19:21:10 +03:00 committed by Andrew Godwin
parent a0cbccfebc
commit b7ea0b9287
7 changed files with 186 additions and 177 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
channels/test/__init__.py
View File

@ -1,3 +1,4 @@
from .base import TransactionChannelTestCase, ChannelTestCase, Client, apply_routes # NOQA isort:skip
from .http import HttpClient # NOQA isort:skip
from .websocket import WSClient # NOQA isort:skip
from .liveserver import ChannelLiveServerTestCase # NOQA isort:skip

150
channels/test/http.py
View File

@ -1,146 +1,8 @@
import copy
import json
import warnings
import six
from django.apps import apps
from django.conf import settings
warnings.warn(
"test.http.HttpClient is deprecated. Use test.websocket.WSClient",
DeprecationWarning,
)
from django.http.cookie import SimpleCookie
from ..sessions import session_for_reply_channel
from .base import Client
json_module = json # alias for using at functions with json kwarg
class HttpClient(Client):
"""
Channel http/ws client abstraction that provides easy methods for testing full life cycle of message in channels
with determined reply channel, auth opportunity, cookies, headers and so on
"""
def __init__(self, **kwargs):
super(HttpClient, self).__init__(**kwargs)
self._session = None
self._headers = {}
self._cookies = {}
self._session_cookie = True
def set_cookie(self, key, value):
"""
Set cookie
"""
self._cookies[key] = value
def set_header(self, key, value):
"""
Set header
"""
if key == 'cookie':
raise ValueError('Use set_cookie method for cookie header')
self._headers[key] = value
def get_cookies(self):
"""Return cookies"""
cookies = copy.copy(self._cookies)
if self._session_cookie and apps.is_installed('django.contrib.sessions'):
cookies[settings.SESSION_COOKIE_NAME] = self.session.session_key
return cookies
@property
def headers(self):
headers = copy.deepcopy(self._headers)
headers.setdefault('cookie', _encoded_cookies(self.get_cookies()))
return headers
@property
def session(self):
"""Session as Lazy property: check that django.contrib.sessions is installed"""
if not apps.is_installed('django.contrib.sessions'):
raise EnvironmentError('Add django.contrib.sessions to the INSTALLED_APPS to use session')
if not self._session:
self._session = session_for_reply_channel(self.reply_channel)
return self._session
def receive(self, json=True):
"""
Return text content of a message for client channel and decoding it if json kwarg is set
"""
content = super(HttpClient, self).receive()
if content and json and 'text' in content and isinstance(content['text'], six.string_types):
return json_module.loads(content['text'])
return content.get('text', content) if content else None
def send(self, to, content={}, text=None, path='/'):
"""
Send a message to a channel.
Adds reply_channel name and channel_session to the message.
"""
content = copy.deepcopy(content)
content.setdefault('reply_channel', self.reply_channel)
content.setdefault('path', path)
content.setdefault('headers', self.headers)
text = text or content.get('text', None)
if text is not None:
if not isinstance(text, six.string_types):
content['text'] = json.dumps(text)
else:
content['text'] = text
self.channel_layer.send(to, content)
self._session_cookie = False
def send_and_consume(self, channel, content={}, text=None, path='/', fail_on_none=True, check_accept=True):
"""
Reproduce full life cycle of the message
"""
self.send(channel, content, text, path)
return self.consume(channel, fail_on_none=fail_on_none, check_accept=check_accept)
def consume(self, channel, fail_on_none=True, check_accept=True):
result = super(HttpClient, self).consume(channel, fail_on_none=fail_on_none)
if channel == "websocket.connect" and check_accept:
received = self.receive(json=False)
if received != {"accept": True}:
raise AssertionError("Connection rejected: %s != '{accept: True}'" % received)
return result
def login(self, **credentials):
"""
Returns True if login is possible; False if the provided credentials
are incorrect, or the user is inactive, or if the sessions framework is
not available.
"""
from django.contrib.auth import authenticate
user = authenticate(**credentials)
if user and user.is_active and apps.is_installed('django.contrib.sessions'):
self._login(user)
return True
else:
return False
def force_login(self, user, backend=None):
if backend is None:
backend = settings.AUTHENTICATION_BACKENDS[0]
user.backend = backend
self._login(user)
def _login(self, user):
from django.contrib.auth import login
# Fake http request
request = type('FakeRequest', (object, ), {'session': self.session, 'META': {}})
login(request, user)
# Save the session values.
self.session.save()
def _encoded_cookies(cookies):
"""Encode dict of cookies to ascii string"""
cookie_encoder = SimpleCookie()
for k, v in cookies.items():
cookie_encoder[k] = v
return cookie_encoder.output(header='', sep=';').encode("ascii")
from .websocket import WSClient as HttpClient # NOQA isort:skip

146
channels/test/websocket.py Normal file
View File

@ -0,0 +1,146 @@
import copy
import json
import six
from django.apps import apps
from django.conf import settings
from django.http.cookie import SimpleCookie
from ..sessions import session_for_reply_channel
from .base import Client
json_module = json # alias for using at functions with json kwarg
class WSClient(Client):
"""
Channel http/ws client abstraction that provides easy methods for testing full life cycle of message in channels
with determined reply channel, auth opportunity, cookies, headers and so on
"""
def __init__(self, **kwargs):
super(WSClient, self).__init__(**kwargs)
self._session = None
self._headers = {}
self._cookies = {}
self._session_cookie = True
def set_cookie(self, key, value):
"""
Set cookie
"""
self._cookies[key] = value
def set_header(self, key, value):
"""
Set header
"""
if key == 'cookie':
raise ValueError('Use set_cookie method for cookie header')
self._headers[key] = value
def get_cookies(self):
"""Return cookies"""
cookies = copy.copy(self._cookies)
if self._session_cookie and apps.is_installed('django.contrib.sessions'):
cookies[settings.SESSION_COOKIE_NAME] = self.session.session_key
return cookies
@property
def headers(self):
headers = copy.deepcopy(self._headers)
headers.setdefault('cookie', _encoded_cookies(self.get_cookies()))
return headers
@property
def session(self):
"""Session as Lazy property: check that django.contrib.sessions is installed"""
if not apps.is_installed('django.contrib.sessions'):
raise EnvironmentError('Add django.contrib.sessions to the INSTALLED_APPS to use session')
if not self._session:
self._session = session_for_reply_channel(self.reply_channel)
return self._session
def receive(self, json=True):
"""
Return text content of a message for client channel and decoding it if json kwarg is set
"""
content = super(WSClient, self).receive()
if content and json and 'text' in content and isinstance(content['text'], six.string_types):
return json_module.loads(content['text'])
return content.get('text', content) if content else None
def send(self, to, content={}, text=None, path='/'):
"""
Send a message to a channel.
Adds reply_channel name and channel_session to the message.
"""
content = copy.deepcopy(content)
content.setdefault('reply_channel', self.reply_channel)
content.setdefault('path', path)
content.setdefault('headers', self.headers)
text = text or content.get('text', None)
if text is not None:
if not isinstance(text, six.string_types):
content['text'] = json.dumps(text)
else:
content['text'] = text
self.channel_layer.send(to, content)
self._session_cookie = False
def send_and_consume(self, channel, content={}, text=None, path='/', fail_on_none=True, check_accept=True):
"""
Reproduce full life cycle of the message
"""
self.send(channel, content, text, path)
return self.consume(channel, fail_on_none=fail_on_none, check_accept=check_accept)
def consume(self, channel, fail_on_none=True, check_accept=True):
result = super(WSClient, self).consume(channel, fail_on_none=fail_on_none)
if channel == "websocket.connect" and check_accept:
received = self.receive(json=False)
if received != {"accept": True}:
raise AssertionError("Connection rejected: %s != '{accept: True}'" % received)
return result
def login(self, **credentials):
"""
Returns True if login is possible; False if the provided credentials
are incorrect, or the user is inactive, or if the sessions framework is
not available.
"""
from django.contrib.auth import authenticate
user = authenticate(**credentials)
if user and user.is_active and apps.is_installed('django.contrib.sessions'):
self._login(user)
return True
else:
return False
def force_login(self, user, backend=None):
if backend is None:
backend = settings.AUTHENTICATION_BACKENDS[0]
user.backend = backend
self._login(user)
def _login(self, user):
from django.contrib.auth import login
# Fake http request
request = type('FakeRequest', (object, ), {'session': self.session, 'META': {}})
login(request, user)
# Save the session values.
self.session.save()
def _encoded_cookies(cookies):
"""Encode dict of cookies to ascii string"""
cookie_encoder = SimpleCookie()
for k, v in cookies.items():
cookie_encoder[k] = v
return cookie_encoder.output(header='', sep=';').encode("ascii")

26
docs/testing.rst
View File

@ -128,7 +128,7 @@ purpose use ``send_and_consume`` method::
self.assertEqual(client.receive(), {'all is': 'done'})
You can use ``HttpClient`` for websocket related consumers. It automatically serializes JSON content,
You can use ``WSClient`` 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::
@ -146,13 +146,13 @@ For example::
# tests.py
from channels import Group
from channels.test import ChannelTestCase, HttpClient
from channels.test import ChannelTestCase, WSClient
class RoomsTests(ChannelTestCase):
def test_rooms(self):
client = HttpClient()
client = WSClient()
user = User.objects.create_user(
username='test', email='test@test.com', password='123456')
client.login(username='test', password='123456')
@ -181,8 +181,8 @@ For example::
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.
Instead of ``WSClient.login`` method with credentials at arguments you
may call ``WSClient.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.
@ -196,23 +196,23 @@ want to test your consumers in a more isolate and atomic way, it will be
simpler with ``apply_routes`` contextmanager and decorator for your ``ChannelTestCase``.
It takes a list of routes that you want to use and overwrites existing routes::
from channels.test import ChannelTestCase, HttpClient, apply_routes
from channels.test import ChannelTestCase, WSClient, apply_routes
class MyTests(ChannelTestCase):
def test_myconsumer(self):
client = HttpClient()
client = WSClient()
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``
Test Data binding with ``WSClient``
-------------------------------------
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``
so that ways tests in different ways and ``WSClient`` 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,
@ -220,14 +220,14 @@ make some changes with target model and check received message.
Lets test ``IntegerValueBinding`` from :doc:`data binding <binding>`
with creating::
from channels.test import ChannelTestCase, HttpClient
from channels.test import ChannelTestCase, WSClient
from channels.signals import consumer_finished
class TestIntegerValueBinding(ChannelTestCase):
def test_outbound_create(self):
# We use HttpClient because of json encoding messages
client = HttpClient()
# We use WSClient because of json encoding messages
client = WSClient()
client.join_group("intval-updates") # join outbound binding
# create target entity
@ -266,7 +266,7 @@ For example::
with apply_routes([Demultiplexer.as_route(path='/'),
route("binding.intval", IntegerValueBinding.consumer)]):
client = HttpClient()
client = WSClient()
client.send_and_consume('websocket.connect', path='/')
client.send_and_consume('websocket.receive', path='/', text={
'stream': 'intval',

20
tests/test_binding.py
View File

@ -6,7 +6,7 @@ from channels import route
from channels.binding.base import CREATE, DELETE, UPDATE
from channels.binding.websockets import WebsocketBinding
from channels.generic.websockets import WebsocketDemultiplexer
from channels.test import ChannelTestCase, HttpClient, apply_routes
from channels.test import ChannelTestCase, WSClient, apply_routes
from tests import models
User = get_user_model()
@ -28,7 +28,7 @@ class TestsBinding(ChannelTestCase):
def has_permission(self, user, action, pk):
return True
client = HttpClient()
client = WSClient()
client.join_group('users')
user = User.objects.create(username='test', email='test@test.com')
@ -70,7 +70,7 @@ class TestsBinding(ChannelTestCase):
def has_permission(self, user, action, pk):
return True
client = HttpClient()
client = WSClient()
client.join_group('testuuidmodels')
instance = models.TestUUIDModel.objects.create(name='testname')
@ -106,7 +106,7 @@ class TestsBinding(ChannelTestCase):
return True
with apply_routes([route('test', TestBinding.consumer)]):
client = HttpClient()
client = WSClient()
client.join_group('users_exclude')
user = User.objects.create(username='test', email='test@test.com')
@ -165,7 +165,7 @@ class TestsBinding(ChannelTestCase):
# Make model and clear out pending sends
user = User.objects.create(username='test', email='test@test.com')
client = HttpClient()
client = WSClient()
client.join_group('users2')
user.username = 'test_new'
@ -210,7 +210,7 @@ class TestsBinding(ChannelTestCase):
# Make model and clear out pending sends
user = User.objects.create(username='test', email='test@test.com')
client = HttpClient()
client = WSClient()
client.join_group('users3')
user.delete()
@ -254,7 +254,7 @@ class TestsBinding(ChannelTestCase):
groups = ['inbound']
with apply_routes([Demultiplexer.as_route(path='/')]):
client = HttpClient()
client = WSClient()
client.send_and_consume('websocket.connect', path='/')
client.send_and_consume('websocket.receive', path='/', text={
'stream': 'users',
@ -294,7 +294,7 @@ class TestsBinding(ChannelTestCase):
groups = ['inbound']
with apply_routes([Demultiplexer.as_route(path='/')]):
client = HttpClient()
client = WSClient()
client.send_and_consume('websocket.connect', path='/')
client.send_and_consume('websocket.receive', path='/', text={
'stream': 'users',
@ -340,7 +340,7 @@ class TestsBinding(ChannelTestCase):
groups = ['inbound']
with apply_routes([Demultiplexer.as_route(path='/')]):
client = HttpClient()
client = WSClient()
client.send_and_consume('websocket.connect', path='/')
client.send_and_consume('websocket.receive', path='/', text={
'stream': 'users',
@ -372,6 +372,6 @@ class TestsBinding(ChannelTestCase):
groups = ['inbound']
with apply_routes([Demultiplexer.as_route(path='/path/(?P<id>\d+)')]):
client = HttpClient()
client = WSClient()
consumer = client.send_and_consume('websocket.connect', path='/path/789')
self.assertEqual(consumer.kwargs['id'], '789')

14
tests/test_generic.py
View File

@ -8,7 +8,7 @@ from django.contrib.auth import get_user_model
from channels import route_class
from channels.exceptions import SendNotAvailableOnDemultiplexer
from channels.generic import BaseConsumer, websockets
from channels.test import ChannelTestCase, Client, HttpClient, apply_routes
from channels.test import ChannelTestCase, Client, WSClient, apply_routes
@override_settings(SESSION_ENGINE="django.contrib.sessions.backends.cache")
@ -96,7 +96,7 @@ class GenericTests(ChannelTestCase):
user_model = get_user_model()
user = user_model.objects.create_user(username='test', email='test@test.com', password='123456')
client = HttpClient()
client = WSClient()
client.force_login(user)
with apply_routes([route_class(WebsocketConsumer, path='/path')]):
connect = client.send_and_consume('websocket.connect', {'path': '/path'})
@ -128,7 +128,7 @@ class GenericTests(ChannelTestCase):
self.assertIs(routes[1].consumer, WebsocketConsumer)
with apply_routes(routes):
client = HttpClient()
client = WSClient()
client.send('websocket.connect', {'path': '/path', 'order': 1})
client.send('websocket.connect', {'path': '/path', 'order': 0})
@ -180,7 +180,7 @@ class GenericTests(ChannelTestCase):
}
with apply_routes([route_class(Demultiplexer, path='/path/(?P<id>\d+)')]):
client = HttpClient()
client = WSClient()
client.send_and_consume('websocket.connect', path='/path/1')
self.assertEqual(client.receive(), {
@ -216,7 +216,7 @@ class GenericTests(ChannelTestCase):
}
with apply_routes([route_class(Demultiplexer, path='/path/(?P<id>\d+)')]):
client = HttpClient()
client = WSClient()
with self.assertRaises(SendNotAvailableOnDemultiplexer):
client.send_and_consume('websocket.receive', path='/path/1', text={
@ -250,7 +250,7 @@ class GenericTests(ChannelTestCase):
return json.dumps(lowered)
with apply_routes([route_class(WebsocketConsumer, path='/path')]):
client = HttpClient()
client = WSClient()
consumer = client.send_and_consume('websocket.receive', path='/path', text={"key": "value"})
self.assertEqual(consumer.content_received, {"KEY": "value"})
@ -284,7 +284,7 @@ class GenericTests(ChannelTestCase):
}
with apply_routes([route_class(Demultiplexer, path='/path/(?P<ID>\d+)')]):
client = HttpClient()
client = WSClient()
client.send_and_consume('websocket.connect', path='/path/1')
self.assertEqual(client.receive(), {

6
tests/test_http.py → tests/test_wsclient.py
View File

@ -2,12 +2,12 @@ from __future__ import unicode_literals
from django.http.cookie import parse_cookie
from channels.test import ChannelTestCase, HttpClient
from channels.test import ChannelTestCase, WSClient
class HttpClientTests(ChannelTestCase):
class WSClientTests(ChannelTestCase):
def test_cookies(self):
client = HttpClient()
client = WSClient()
client.set_cookie('foo', 'not-bar')
client.set_cookie('foo', 'bar')
client.set_cookie('qux', 'qu;x')