From 05c41e9ad64da1f989528bf42ef65f1b837af659 Mon Sep 17 00:00:00 2001 From: Krukov D Date: Thu, 19 May 2016 21:45:25 +0300 Subject: [PATCH] More tests utils for happy users (#162) * Added Client abstraction * Added apply_routes decorator/contextmanager * Fix apply routes as decorator * Separated Http specific client and 'Simple' client * Remove Clients from ChannelTestCase * Added cookies and headers management * Fix wrong reverting * Fixs for code style * Added space before inline comment --- channels/tests/__init__.py | 3 +- channels/tests/base.py | 130 ++++++++++++++++++++++++++++++++ channels/tests/http.py | 147 +++++++++++++++++++++++++++++++++++++ 3 files changed, 279 insertions(+), 1 deletion(-) create mode 100644 channels/tests/http.py diff --git a/channels/tests/__init__.py b/channels/tests/__init__.py index a43624d..36481f0 100644 --- a/channels/tests/__init__.py +++ b/channels/tests/__init__.py @@ -1 +1,2 @@ -from .base import ChannelTestCase # NOQA isort:skip +from .base import ChannelTestCase, Client, apply_routes # NOQA isort:skip +from .http import HttpClient # NOQA isort:skip diff --git a/channels/tests/base.py b/channels/tests/base.py index 95bd9af..dff1e97 100644 --- a/channels/tests/base.py +++ b/channels/tests/base.py @@ -1,5 +1,11 @@ +import copy +import random +import string +from functools import wraps + from django.test.testcases import TestCase from channels import DEFAULT_CHANNEL_LAYER +from channels.routing import Router, include from channels.asgi import channel_layers, ChannelLayerWrapper from channels.message import Message from asgiref.inmemory import ChannelLayer as InMemoryChannelLayer @@ -59,3 +65,127 @@ class ChannelTestCase(TestCase): else: return None return Message(content, recv_channel, channel_layers[alias]) + + +class Client(object): + """ + Channel client abstraction that provides easy methods for testing full live cycle of message in channels + with determined the reply channel + """ + + def __init__(self, alias=DEFAULT_CHANNEL_LAYER): + self.reply_channel = alias + ''.join([random.choice(string.ascii_letters) for _ in range(5)]) + self.alias = alias + + @property + def channel_layer(self): + """Channel layer as lazy property""" + return channel_layers[self.alias] + + def get_next_message(self, channel): + """ + Gets the next message that was sent to the channel during the test, + or None if no message is available. + """ + recv_channel, content = channel_layers[self.alias].receive_many([channel]) + if recv_channel is None: + return + return Message(content, recv_channel, channel_layers[self.alias]) + + def send(self, to, content={}): + """ + Send a message to a channel. + Adds reply_channel name to the message. + """ + content = copy.deepcopy(content) + content.setdefault('reply_channel', self.reply_channel) + self.channel_layer.send(to, content) + + def consume(self, channel): + """ + Get next message for channel name and run appointed consumer + """ + message = self.get_next_message(channel) + if message: + consumer, kwargs = self.channel_layer.router.match(message) + return consumer(message, **kwargs) + + def send_and_consume(self, channel, content={}): + """ + Reproduce full live cycle of the message + """ + self.send(channel, content) + return self.consume(channel) + + def receive(self): + """self.get_next_message(self.reply_channel) + Get content of next message for reply channel if message exists + """ + message = self.get_next_message(self.reply_channel) + if message: + return message.content + + +class apply_routes(object): + """ + Decorator/ContextManager for rewrite layers routes in context. + Helpful for testing group routes/consumers as isolated application + + The applying routes can be list of instances of Route or list of this lists + """ + + def __init__(self, routes, aliases=[DEFAULT_CHANNEL_LAYER]): + self._aliases = aliases + self.routes = routes + self._old_routing = {} + + def enter(self): + """ + Store old routes and apply new one + """ + for alias in self._aliases: + channel_layer = channel_layers[DEFAULT_CHANNEL_LAYER] + self._old_routing[alias] = channel_layer.routing + if isinstance(self.routes, (list, tuple)): + if isinstance(self.routes[0], (list, tuple)): + routes = list(map(include, self.routes)) + else: + routes = self.routes + + channel_layer.routing = routes + channel_layer.router = Router(routes) + + def exit(self, exc_type=None, exc_val=None, exc_tb=None): + """ + Undoes rerouting + """ + for alias in self._aliases: + channel_layer = channel_layers[DEFAULT_CHANNEL_LAYER] + channel_layer.routing = self._old_routing[alias] + channel_layer.router = Router(self._old_routing[alias]) + + __enter__ = enter + __exit__ = exit + + def __call__(self, test_func): + if isinstance(test_func, type): + old_setup = test_func.setUp + old_teardown = test_func.tearDown + + def new_setup(this): + self.enter() + old_setup(this) + + def new_teardown(this): + self.exit() + old_teardown(this) + + test_func.setUp = new_setup + test_func.tearDown = new_teardown + return test_func + else: + @wraps(test_func) + def inner(*args, **kwargs): + with self: + return test_func(*args, **kwargs) + return inner diff --git a/channels/tests/http.py b/channels/tests/http.py new file mode 100644 index 0000000..fa44dfd --- /dev/null +++ b/channels/tests/http.py @@ -0,0 +1,147 @@ + +import copy + +from django.apps import apps +from django.conf import settings + + +from ..asgi import channel_layers +from ..message import Message +from ..sessions import session_for_reply_channel +from .base import Client + + +class HttpClient(Client): + """ + Channel http/ws client abstraction that provides easy methods for testing full live 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 = {} + + 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 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 + + @property + def channel_layer(self): + """Channel layer as lazy property""" + return channel_layers[self.alias] + + def get_next_message(self, channel): + """ + Gets the next message that was sent to the channel during the test, + or None if no message is available. + + If require is true, will fail the test if no message is received. + """ + recv_channel, content = channel_layers[self.alias].receive_many([channel]) + if recv_channel is None: + return + return Message(content, recv_channel, channel_layers[self.alias]) + + def send(self, to, content={}): + """ + 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', '/') + content.setdefault('headers', self.headers) + self.channel_layer.send(to, content) + + def consume(self, channel): + """ + Get next message for channel name and run appointed consumer + """ + message = self.get_next_message(channel) + if message: + consumer, kwargs = self.channel_layer.router.match(message) + return consumer(message, **kwargs) + + def send_and_consume(self, channel, content={}): + """ + Reproduce full live cycle of the message + """ + self.send(channel, content) + return self.consume(channel) + + def receive(self): + """ + Get content of next message for reply channel if message exists + """ + message = self.get_next_message(self.reply_channel) + if message: + return message.content + + 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""" + return ('&'.join('{0}={1}'.format(k, v) for k, v in cookies.items())).encode("ascii")