1
# This file is part of the Juju GUI, which lets users view and manage Juju
2
# environments within a graphical interface (https://launchpad.net/juju-gui).
3
# Copyright (C) 2013 Canonical Ltd.
5
# This program is free software: you can redistribute it and/or modify it under
6
# the terms of the GNU Affero General Public License version 3, as published by
7
# the Free Software Foundation.
9
# This program is distributed in the hope that it will be useful, but WITHOUT
10
# ANY WARRANTY; without even the implied warranties of MERCHANTABILITY,
11
# SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
12
# Affero General Public License for more details.
14
# You should have received a copy of the GNU Affero General Public License
15
# along with this program. If not, see <http://www.gnu.org/licenses/>.
17
"""Juju GUI server test utilities."""
21
from tornado import websocket
23
from guiserver import auth
26
class EchoWebSocketHandler(websocket.WebSocketHandler):
27
"""A WebSocket server echoing back messages."""
29
def initialize(self, close_future, io_loop):
30
"""Echo WebSocket server initializer.
32
The handler receives a close Future and the current Tornado IO loop.
33
The close Future is fired when the connection is closed.
34
The close Future can also be used to force a connection termination by
37
self._closed_future = close_future
38
self._connected = True
39
io_loop.add_future(close_future, self.force_close)
41
def force_close(self, future):
42
"""Close the connection to the client."""
46
def on_message(self, message):
47
"""Echo back the received message."""
48
self.write_message(message, isinstance(message, bytes))
51
"""Fire the _closed_future if not already done."""
52
self._connected = False
53
if not self._closed_future.done():
54
self._closed_future.set_result(None)
57
class GoAPITestMixin(object):
58
"""Add helper methods for testing the Go API implementation."""
60
def get_auth_backend(self):
61
"""Return an authentication backend suitable for the Go API."""
62
return auth.get_backend('go')
64
def make_login_request(
65
self, request_id=42, username='user', password='passwd',
67
"""Create and return a login request message.
69
If encoded is set to True, the returned message will be JSON encoded.
72
'RequestId': request_id,
75
'Params': {'AuthTag': username, 'Password': password},
77
return json.dumps(data) if encoded else data
79
def make_login_response(
80
self, request_id=42, successful=True, encoded=False):
81
"""Create and return a login response message.
83
If encoded is set to True, the returned message will be JSON encoded.
84
By default, a successful response is returned. Set successful to False
85
to return an authentication failure.
87
data = {'RequestId': request_id, 'Response': {}}
89
data['Error'] = 'invalid entity name or password'
90
return json.dumps(data) if encoded else data
93
class PythonAPITestMixin(object):
94
"""Add helper methods for testing the Python API implementation."""
96
def get_auth_backend(self):
97
"""Return an authentication backend suitable for the Python API."""
98
return auth.get_backend('python')
100
def make_login_request(
101
self, request_id=42, username='user', password='passwd',
103
"""Create and return a login request message.
105
If encoded is set to True, the returned message will be JSON encoded.
108
'request_id': request_id,
111
'password': password,
113
return json.dumps(data) if encoded else data
115
def make_login_response(
116
self, request_id=42, successful=True, encoded=False):
117
"""Create and return a login response message.
119
If encoded is set to True, the returned message will be JSON encoded.
120
By default, a successful response is returned. Set successful to False
121
to return an authentication failure.
123
data = {'request_id': request_id, 'op': 'login'}
125
data['result'] = True
128
return json.dumps(data) if encoded else data
131
class WSSTestMixin(object):
132
"""Add some helper methods for testing secure WebSocket handlers."""
134
def get_wss_url(self, path):
135
"""Return an absolute secure WebSocket url for the given path."""
136
return 'wss://localhost:{}{}'.format(self.get_http_port(), path)
added
removed
server/guiserver/tests/helpers.py
