mirror of
https://github.com/django/daphne.git
synced 2025-11-04 09:37:32 +03:00
104 lines
3.3 KiB
Python
104 lines
3.3 KiB
Python
|
|
import copy
|
|
|
|
from django.apps import apps
|
|
from django.conf import settings
|
|
|
|
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
|
|
|
|
def send(self, to, content={}, 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)
|
|
self.channel_layer.send(to, 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")
|