diff options
Diffstat (limited to 'testing/web-platform/tests/tools/pywebsocket/src/test/test_dispatch.py')
| -rwxr-xr-x | testing/web-platform/tests/tools/pywebsocket/src/test/test_dispatch.py | 288 |
1 files changed, 288 insertions, 0 deletions
diff --git a/testing/web-platform/tests/tools/pywebsocket/src/test/test_dispatch.py b/testing/web-platform/tests/tools/pywebsocket/src/test/test_dispatch.py new file mode 100755 index 000000000..9ca3d4f3a --- /dev/null +++ b/testing/web-platform/tests/tools/pywebsocket/src/test/test_dispatch.py @@ -0,0 +1,288 @@ +#!/usr/bin/env python +# +# Copyright 2011, Google Inc. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are +# met: +# +# * Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above +# copyright notice, this list of conditions and the following disclaimer +# in the documentation and/or other materials provided with the +# distribution. +# * Neither the name of Google Inc. nor the names of its +# contributors may be used to endorse or promote products derived from +# this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + + +"""Tests for dispatch module.""" + + +import os +import unittest + +import set_sys_path # Update sys.path to locate mod_pywebsocket module. + +from mod_pywebsocket import dispatch +from mod_pywebsocket import handshake +from test import mock + + +_TEST_HANDLERS_DIR = os.path.join( + os.path.split(__file__)[0], 'testdata', 'handlers') + +_TEST_HANDLERS_SUB_DIR = os.path.join(_TEST_HANDLERS_DIR, 'sub') + + +class DispatcherTest(unittest.TestCase): + """A unittest for dispatch module.""" + + def test_normalize_path(self): + self.assertEqual(os.path.abspath('/a/b').replace('\\', '/'), + dispatch._normalize_path('/a/b')) + self.assertEqual(os.path.abspath('/a/b').replace('\\', '/'), + dispatch._normalize_path('\\a\\b')) + self.assertEqual(os.path.abspath('/a/b').replace('\\', '/'), + dispatch._normalize_path('/a/c/../b')) + self.assertEqual(os.path.abspath('abc').replace('\\', '/'), + dispatch._normalize_path('abc')) + + def test_converter(self): + converter = dispatch._create_path_to_resource_converter('/a/b') + # Python built by MSC inserts a drive name like 'C:\' via realpath(). + # Converter Generator expands provided path using realpath() and uses + # the path including a drive name to verify the prefix. + os_root = os.path.realpath('/') + self.assertEqual('/h', converter(os_root + 'a/b/h_wsh.py')) + self.assertEqual('/c/h', converter(os_root + 'a/b/c/h_wsh.py')) + self.assertEqual(None, converter(os_root + 'a/b/h.py')) + self.assertEqual(None, converter('a/b/h_wsh.py')) + + converter = dispatch._create_path_to_resource_converter('a/b') + self.assertEqual('/h', converter(dispatch._normalize_path( + 'a/b/h_wsh.py'))) + + converter = dispatch._create_path_to_resource_converter('/a/b///') + self.assertEqual('/h', converter(os_root + 'a/b/h_wsh.py')) + self.assertEqual('/h', converter(dispatch._normalize_path( + '/a/b/../b/h_wsh.py'))) + + converter = dispatch._create_path_to_resource_converter( + '/a/../a/b/../b/') + self.assertEqual('/h', converter(os_root + 'a/b/h_wsh.py')) + + converter = dispatch._create_path_to_resource_converter(r'\a\b') + self.assertEqual('/h', converter(os_root + r'a\b\h_wsh.py')) + self.assertEqual('/h', converter(os_root + r'a/b/h_wsh.py')) + + def test_enumerate_handler_file_paths(self): + paths = list( + dispatch._enumerate_handler_file_paths(_TEST_HANDLERS_DIR)) + paths.sort() + self.assertEqual(8, len(paths)) + expected_paths = [ + os.path.join(_TEST_HANDLERS_DIR, 'abort_by_user_wsh.py'), + os.path.join(_TEST_HANDLERS_DIR, 'blank_wsh.py'), + os.path.join(_TEST_HANDLERS_DIR, 'origin_check_wsh.py'), + os.path.join(_TEST_HANDLERS_DIR, 'sub', + 'exception_in_transfer_wsh.py'), + os.path.join(_TEST_HANDLERS_DIR, 'sub', 'non_callable_wsh.py'), + os.path.join(_TEST_HANDLERS_DIR, 'sub', 'plain_wsh.py'), + os.path.join(_TEST_HANDLERS_DIR, 'sub', + 'wrong_handshake_sig_wsh.py'), + os.path.join(_TEST_HANDLERS_DIR, 'sub', + 'wrong_transfer_sig_wsh.py'), + ] + for expected, actual in zip(expected_paths, paths): + self.assertEqual(expected, actual) + + def test_source_handler_file(self): + self.assertRaises( + dispatch.DispatchException, dispatch._source_handler_file, '') + self.assertRaises( + dispatch.DispatchException, dispatch._source_handler_file, 'def') + self.assertRaises( + dispatch.DispatchException, dispatch._source_handler_file, '1/0') + self.failUnless(dispatch._source_handler_file( + 'def web_socket_do_extra_handshake(request):pass\n' + 'def web_socket_transfer_data(request):pass\n')) + + def test_source_warnings(self): + dispatcher = dispatch.Dispatcher(_TEST_HANDLERS_DIR, None) + warnings = dispatcher.source_warnings() + warnings.sort() + expected_warnings = [ + (os.path.realpath(os.path.join( + _TEST_HANDLERS_DIR, 'blank_wsh.py')) + + ': web_socket_do_extra_handshake is not defined.'), + (os.path.realpath(os.path.join( + _TEST_HANDLERS_DIR, 'sub', 'non_callable_wsh.py')) + + ': web_socket_do_extra_handshake is not callable.'), + (os.path.realpath(os.path.join( + _TEST_HANDLERS_DIR, 'sub', 'wrong_handshake_sig_wsh.py')) + + ': web_socket_do_extra_handshake is not defined.'), + (os.path.realpath(os.path.join( + _TEST_HANDLERS_DIR, 'sub', 'wrong_transfer_sig_wsh.py')) + + ': web_socket_transfer_data is not defined.'), + ] + self.assertEquals(4, len(warnings)) + for expected, actual in zip(expected_warnings, warnings): + self.assertEquals(expected, actual) + + def test_do_extra_handshake(self): + dispatcher = dispatch.Dispatcher(_TEST_HANDLERS_DIR, None) + request = mock.MockRequest() + request.ws_resource = '/origin_check' + request.ws_origin = 'http://example.com' + dispatcher.do_extra_handshake(request) # Must not raise exception. + + request.ws_origin = 'http://bad.example.com' + try: + dispatcher.do_extra_handshake(request) + self.fail('Could not catch HandshakeException with 403 status') + except handshake.HandshakeException, e: + self.assertEquals(403, e.status) + except Exception, e: + self.fail('Unexpected exception: %r' % e) + + def test_abort_extra_handshake(self): + dispatcher = dispatch.Dispatcher(_TEST_HANDLERS_DIR, None) + request = mock.MockRequest() + request.ws_resource = '/abort_by_user' + self.assertRaises(handshake.AbortedByUserException, + dispatcher.do_extra_handshake, request) + + def test_transfer_data(self): + dispatcher = dispatch.Dispatcher(_TEST_HANDLERS_DIR, None) + + request = mock.MockRequest(connection=mock.MockConn('\xff\x00')) + request.ws_resource = '/origin_check' + request.ws_protocol = 'p1' + dispatcher.transfer_data(request) + self.assertEqual('origin_check_wsh.py is called for /origin_check, p1' + '\xff\x00', + request.connection.written_data()) + + request = mock.MockRequest(connection=mock.MockConn('\xff\x00')) + request.ws_resource = '/sub/plain' + request.ws_protocol = None + dispatcher.transfer_data(request) + self.assertEqual('sub/plain_wsh.py is called for /sub/plain, None' + '\xff\x00', + request.connection.written_data()) + + request = mock.MockRequest(connection=mock.MockConn('\xff\x00')) + request.ws_resource = '/sub/plain?' + request.ws_protocol = None + dispatcher.transfer_data(request) + self.assertEqual('sub/plain_wsh.py is called for /sub/plain?, None' + '\xff\x00', + request.connection.written_data()) + + request = mock.MockRequest(connection=mock.MockConn('\xff\x00')) + request.ws_resource = '/sub/plain?q=v' + request.ws_protocol = None + dispatcher.transfer_data(request) + self.assertEqual('sub/plain_wsh.py is called for /sub/plain?q=v, None' + '\xff\x00', + request.connection.written_data()) + + def test_transfer_data_no_handler(self): + dispatcher = dispatch.Dispatcher(_TEST_HANDLERS_DIR, None) + for resource in ['/blank', '/sub/non_callable', + '/sub/no_wsh_at_the_end', '/does/not/exist']: + request = mock.MockRequest(connection=mock.MockConn('')) + request.ws_resource = resource + request.ws_protocol = 'p2' + try: + dispatcher.transfer_data(request) + self.fail() + except dispatch.DispatchException, e: + self.failUnless(str(e).find('No handler') != -1) + except Exception: + self.fail() + + def test_transfer_data_handler_exception(self): + dispatcher = dispatch.Dispatcher(_TEST_HANDLERS_DIR, None) + request = mock.MockRequest(connection=mock.MockConn('')) + request.ws_resource = '/sub/exception_in_transfer' + request.ws_protocol = 'p3' + try: + dispatcher.transfer_data(request) + self.fail() + except Exception, e: + self.failUnless(str(e).find('Intentional') != -1, + 'Unexpected exception: %s' % e) + + def test_abort_transfer_data(self): + dispatcher = dispatch.Dispatcher(_TEST_HANDLERS_DIR, None) + request = mock.MockRequest() + request.ws_resource = '/abort_by_user' + self.assertRaises(handshake.AbortedByUserException, + dispatcher.transfer_data, request) + + def test_scan_dir(self): + disp = dispatch.Dispatcher(_TEST_HANDLERS_DIR, None) + self.assertEqual(4, len(disp._handler_suite_map)) + self.failUnless('/origin_check' in disp._handler_suite_map) + self.failUnless( + '/sub/exception_in_transfer' in disp._handler_suite_map) + self.failUnless('/sub/plain' in disp._handler_suite_map) + + def test_scan_sub_dir(self): + disp = dispatch.Dispatcher(_TEST_HANDLERS_DIR, _TEST_HANDLERS_SUB_DIR) + self.assertEqual(2, len(disp._handler_suite_map)) + self.failIf('/origin_check' in disp._handler_suite_map) + self.failUnless( + '/sub/exception_in_transfer' in disp._handler_suite_map) + self.failUnless('/sub/plain' in disp._handler_suite_map) + + def test_scan_sub_dir_as_root(self): + disp = dispatch.Dispatcher(_TEST_HANDLERS_SUB_DIR, + _TEST_HANDLERS_SUB_DIR) + self.assertEqual(2, len(disp._handler_suite_map)) + self.failIf('/origin_check' in disp._handler_suite_map) + self.failIf('/sub/exception_in_transfer' in disp._handler_suite_map) + self.failIf('/sub/plain' in disp._handler_suite_map) + self.failUnless('/exception_in_transfer' in disp._handler_suite_map) + self.failUnless('/plain' in disp._handler_suite_map) + + def test_scan_dir_must_under_root(self): + dispatch.Dispatcher('a/b', 'a/b/c') # OK + dispatch.Dispatcher('a/b///', 'a/b') # OK + self.assertRaises(dispatch.DispatchException, + dispatch.Dispatcher, 'a/b/c', 'a/b') + + def test_resource_path_alias(self): + disp = dispatch.Dispatcher(_TEST_HANDLERS_DIR, None) + disp.add_resource_path_alias('/', '/origin_check') + self.assertEqual(5, len(disp._handler_suite_map)) + self.failUnless('/origin_check' in disp._handler_suite_map) + self.failUnless( + '/sub/exception_in_transfer' in disp._handler_suite_map) + self.failUnless('/sub/plain' in disp._handler_suite_map) + self.failUnless('/' in disp._handler_suite_map) + self.assertRaises(dispatch.DispatchException, + disp.add_resource_path_alias, '/alias', '/not-exist') + + +if __name__ == '__main__': + unittest.main() + + +# vi:sts=4 sw=4 et |