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
"""Tests for the Juju GUI server handlers."""
28
from tornado.testing import (
35
from guiserver import (
39
from guiserver.tests import helpers
42
class TestWebSocketHandler(AsyncHTTPSTestCase, helpers.WSSTestMixin):
45
# In this test case a WebSocket server is created. The server creates a
46
# new client on each request. This client should forward messages to a
47
# WebSocket echo server. In order to test the communication, some of
48
# the tests create another client that connects to the server, e.g.:
49
# ws-client -> ws-server -> ws-forwarding-client -> ws-echo-server
50
# Messages arriving to the echo server are returned back to the client:
51
# ws-echo-server -> ws-forwarding-client -> ws-server -> ws-client
52
self.echo_server_address = self.get_wss_url('/echo')
53
self.echo_server_closed_future = concurrent.Future()
54
echo_options = {'close_future': self.echo_server_closed_future}
55
ws_options = {'jujuapi': self.echo_server_address}
56
return web.Application([
57
(r'/echo', helpers.EchoWebSocketHandler, echo_options),
58
(r'/ws', handlers.WebSocketHandler, ws_options),
61
def make_client(self):
62
"""Return a WebSocket client ready to be connected to the server."""
63
url = self.get_wss_url('/ws')
64
# The client callback is tested elsewhere.
65
callback = lambda message: None
66
return helpers.WebSocketClient(url, callback, io_loop=self.io_loop)
68
def make_handler(self):
69
"""Create and return a WebSocketHandler instance."""
71
return handlers.WebSocketHandler(self.get_app(), request)
74
def test_initialization(self):
75
# A WebSocket client is created and connected when the handler is
77
handler = self.make_handler()
78
yield handler.initialize(self.echo_server_address)
79
self.assertIsInstance(handler.jujuconn, clients.WebSocketClient)
80
self.assertTrue(handler.jujuconn.connected)
82
def test_client_callback(self):
83
# The WebSocket client is created passing the proper callback.
84
handler = self.make_handler()
85
with mock.patch('guiserver.handlers.WebSocketClient') as mock_client:
86
handler.initialize(self.echo_server_address)
87
mock_client.assert_called_once_with(
88
self.echo_server_address, handler.on_juju_message)
90
def test_from_browser_to_juju(self):
91
# A message from the browser is forwarded to the remote server.
92
handler = self.make_handler()
93
with mock.patch('guiserver.handlers.WebSocketClient'):
94
handler.initialize(self.echo_server_address)
95
handler.on_message('hello')
96
handler.jujuconn.write_message.assert_called_once_with('hello')
98
def test_from_juju_to_browser(self):
99
# A message from the remote server is returned to the browser.
100
handler = self.make_handler()
101
handler.initialize(self.echo_server_address)
102
with mock.patch('guiserver.handlers.WebSocketHandler.write_message'):
103
handler.on_juju_message('hello')
104
handler.write_message.assert_called_once_with('hello')
107
def test_end_to_end_proxy(self):
108
# Messages are correctly forwarded from the client to the echo server
109
# and back to the client.
110
client = self.make_client()
111
yield client.connect()
112
message = yield client.send('hello')
113
self.assertEqual('hello', message)
116
def test_connection_close(self):
117
# The proxy connection is terminated when the client disconnects.
118
client = self.make_client()
119
yield client.connect()
121
self.assertFalse(client.connected)
122
yield self.echo_server_closed_future
125
class TestIndexHandler(AsyncHTTPTestCase, LogTrapTestCase):
128
# Set up a static path with an index.html in it.
129
self.path = tempfile.mkdtemp()
130
self.addCleanup(shutil.rmtree, self.path)
131
self.index_contents = 'We are the Borg!'
132
index_path = os.path.join(self.path, 'index.html')
133
with open(index_path, 'w') as index_file:
134
index_file.write(self.index_contents)
135
super(TestIndexHandler, self).setUp()
138
return web.Application([
139
(r'/(.*)', handlers.IndexHandler, {'path': self.path}),
142
def ensure_index(self, path):
143
"""Ensure the index contents are returned requesting the given path."""
144
response = self.fetch(path)
145
self.assertEqual(200, response.code)
146
self.assertEqual(self.index_contents, response.body)
149
# Requests for the root path are served by the index file.
150
self.ensure_index('/')
153
# Requests for internal pages are served by the index file.
154
self.ensure_index('/resistance/is/futile')
156
def test_page_with_flags_and_queries(self):
157
# Requests including flags and queries are served by the index file.
158
self.ensure_index('/:flag:/activated/?my=query')
161
class TestHttpsRedirectHandler(AsyncHTTPTestCase, LogTrapTestCase):
164
return web.Application([(r'.*', handlers.HttpsRedirectHandler)])
166
def assert_redirected(self, response, path):
167
"""Ensure the given response is a permanent redirect to the given path.
169
Also check that the URL schema is HTTPS.
171
self.assertEqual(301, response.code)
172
expected = 'https://localhost:{}{}'.format(self.get_http_port(), path)
173
self.assertEqual(expected, response.headers['location'])
175
def test_redirection(self):
176
# The HTTP traffic is redirected to HTTPS.
177
response = self.fetch('/', follow_redirects=False)
178
self.assert_redirected(response, '/')
180
def test_page_redirection(self):
181
# The path and query parts of the URL are preserved,
182
path_and_query = '/my/page?my=query'
183
response = self.fetch(path_and_query, follow_redirects=False)
184
self.assert_redirected(response, path_and_query)
added
removed
server/guiserver/tests/test_handlers.py
