| #!/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 |
