diff --git a/test-requirements.txt b/test-requirements.txt index a63a15e..8e01437 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -1,4 +1,4 @@ -mox3 +mock nose jwcrypto;python_version>="2.7" redis;python_version>="2.7" diff --git a/tests/test_websocketproxy.py b/tests/test_websocketproxy.py index c0a8d93..d8a4916 100644 --- a/tests/test_websocketproxy.py +++ b/tests/test_websocketproxy.py @@ -20,10 +20,11 @@ import sys import unittest import unittest import socket +try: + from mock import patch +except ImportError: + from unittest.mock import patch -from mox3 import stubout - -from websockify import websockifyserver from websockify import websocketproxy from websockify import token_plugins from websockify import auth_plugins @@ -74,16 +75,14 @@ class FakeServer(object): class ProxyRequestHandlerTestCase(unittest.TestCase): def setUp(self): super(ProxyRequestHandlerTestCase, self).setUp() - self.stubs = stubout.StubOutForTesting() self.handler = websocketproxy.ProxyRequestHandler( FakeSocket(''), "127.0.0.1", FakeServer()) self.handler.path = "https://localhost:6080/websockify?token=blah" self.handler.headers = None - self.stubs.Set(websockifyserver.WebSockifyServer, 'socket', - staticmethod(lambda *args, **kwargs: None)) + patch('websockify.websockifyserver.WebSockifyServer.socket').start() def tearDown(self): - self.stubs.UnsetAll() + patch.stopall() super(ProxyRequestHandlerTestCase, self).tearDown() def test_get_target(self): @@ -120,8 +119,7 @@ class ProxyRequestHandlerTestCase(unittest.TestCase): def lookup(self, token): return (self.source + token).split(',') - self.stubs.Set(websocketproxy.ProxyRequestHandler, 'send_auth_error', - staticmethod(lambda *args, **kwargs: None)) + patcher = patch('websockify.websocketproxy.ProxyRequestHandler.send_auth_error').start() self.handler.server.token_plugin = TestPlugin("somehost,") self.handler.validate_connection() @@ -138,8 +136,7 @@ class ProxyRequestHandlerTestCase(unittest.TestCase): jwt_token.make_signed_token(key) self.handler.path = "https://localhost:6080/websockify?token={jwt_token}".format(jwt_token=jwt_token.serialize()) - self.stubs.Set(websocketproxy.ProxyRequestHandler, 'send_auth_error', - staticmethod(lambda *args, **kwargs: None)) + patcher = patch('websockify.websocketproxy.ProxyRequestHandler.send_auth_error').start() self.handler.server.token_plugin = token_plugins.JWTTokenApi("./tests/fixtures/public.pem") self.handler.validate_connection() @@ -155,8 +152,7 @@ class ProxyRequestHandlerTestCase(unittest.TestCase): jwt_token.make_signed_token(key) self.handler.path = "https://localhost:6080/websockify?token={jwt_token}".format(jwt_token=jwt_token.serialize()) - self.stubs.Set(websocketproxy.ProxyRequestHandler, 'send_auth_error', - staticmethod(lambda *args, **kwargs: None)) + patcher = patch('websockify.websocketproxy.ProxyRequestHandler.send_auth_error').start() self.handler.server.token_plugin = token_plugins.JWTTokenApi("wrong.pub") self.assertRaises(self.handler.server.EClose, @@ -171,8 +167,7 @@ class ProxyRequestHandlerTestCase(unittest.TestCase): jwt_token.make_signed_token(key) self.handler.path = "https://localhost:6080/websockify?token={jwt_token}".format(jwt_token=jwt_token.serialize()) - self.stubs.Set(websocketproxy.ProxyRequestHandler, 'send_auth_error', - staticmethod(lambda *args, **kwargs: None)) + patcher = patch('websockify.websocketproxy.ProxyRequestHandler.send_auth_error').start() self.handler.server.token_plugin = token_plugins.JWTTokenApi("./tests/fixtures/symmetric.key") self.handler.validate_connection() @@ -188,8 +183,7 @@ class ProxyRequestHandlerTestCase(unittest.TestCase): jwt_token.make_signed_token(key) self.handler.path = "https://localhost:6080/websockify?token={jwt_token}".format(jwt_token=jwt_token.serialize()) - self.stubs.Set(websocketproxy.ProxyRequestHandler, 'send_auth_error', - staticmethod(lambda *args, **kwargs: None)) + patcher = patch('websockify.websocketproxy.ProxyRequestHandler.send_auth_error').start() self.handler.server.token_plugin = token_plugins.JWTTokenApi("wrong_sauce") self.assertRaises(self.handler.server.EClose, @@ -210,8 +204,7 @@ class ProxyRequestHandlerTestCase(unittest.TestCase): self.handler.path = "https://localhost:6080/websockify?token={jwt_token}".format(jwt_token=jwe_token.serialize()) - self.stubs.Set(websocketproxy.ProxyRequestHandler, 'send_auth_error', - staticmethod(lambda *args, **kwargs: None)) + patcher = patch('websockify.websocketproxy.ProxyRequestHandler.send_auth_error').start() self.handler.server.token_plugin = token_plugins.JWTTokenApi("./tests/fixtures/private.pem") self.handler.validate_connection() @@ -225,8 +218,7 @@ class ProxyRequestHandlerTestCase(unittest.TestCase): if target_host == self.source: raise auth_plugins.AuthenticationError(response_msg="some_error") - self.stubs.Set(websocketproxy.ProxyRequestHandler, 'send_auth_error', - staticmethod(lambda *args, **kwargs: None)) + patcher = patch('websockify.websocketproxy.ProxyRequestHandler.send_auth_error').start() self.handler.server.auth_plugin = TestPlugin("somehost") self.handler.server.target_host = "somehost" diff --git a/tests/test_websockifyserver.py b/tests/test_websockifyserver.py index b9312dc..a089f55 100644 --- a/tests/test_websockifyserver.py +++ b/tests/test_websockifyserver.py @@ -22,7 +22,10 @@ import select import shutil import socket import ssl -from mox3 import stubout +try: + from mock import patch, MagicMock, ANY +except ImportError: + from unittest.mock import patch, MagicMock, ANY import sys import tempfile import unittest @@ -73,22 +76,13 @@ class FakeSocket(object): class WebSockifyRequestHandlerTestCase(unittest.TestCase): def setUp(self): super(WebSockifyRequestHandlerTestCase, self).setUp() - self.stubs = stubout.StubOutForTesting() self.tmpdir = tempfile.mkdtemp('-websockify-tests') # Mock this out cause it screws tests up - self.stubs.Set(os, 'chdir', lambda *args, **kwargs: None) - self.stubs.Set(BaseHTTPRequestHandler, 'send_response', - lambda *args, **kwargs: None) - - def fake_send_error(self, code, message=None, explain=None): - self.last_code = code - - self.stubs.Set(BaseHTTPRequestHandler, 'send_error', - fake_send_error) + patch('os.chdir').start() def tearDown(self): """Called automatically after each test.""" - self.stubs.UnsetAll() + patch.stopall() os.rmdir(self.tmpdir) super(WebSockifyRequestHandlerTestCase, self).tearDown() @@ -101,47 +95,36 @@ class WebSockifyRequestHandlerTestCase(unittest.TestCase): record=self.tmpdir, daemon=False, ssl_only=0, idle_timeout=1, **kwargs) - def test_normal_get_with_only_upgrade_returns_error(self): + @patch('websockify.websockifyserver.WebSockifyRequestHandler.send_error') + def test_normal_get_with_only_upgrade_returns_error(self, send_error): server = self._get_server(web=None) handler = websockifyserver.WebSockifyRequestHandler( FakeSocket('GET /tmp.txt HTTP/1.1'), '127.0.0.1', server) - def fake_send_response(self, code, message=None): - self.last_code = code - - self.stubs.Set(BaseHTTPRequestHandler, 'send_response', - fake_send_response) - handler.do_GET() - self.assertEqual(handler.last_code, 405) + send_error.assert_called_with(405, ANY) - def test_list_dir_with_file_only_returns_error(self): + @patch('websockify.websockifyserver.WebSockifyRequestHandler.send_error') + def test_list_dir_with_file_only_returns_error(self, send_error): server = self._get_server(file_only=True) handler = websockifyserver.WebSockifyRequestHandler( FakeSocket('GET / HTTP/1.1'), '127.0.0.1', server) - def fake_send_response(self, code, message=None): - self.last_code = code - - self.stubs.Set(BaseHTTPRequestHandler, 'send_response', - fake_send_response) - handler.path = '/' handler.do_GET() - self.assertEqual(handler.last_code, 404) + send_error.assert_called_with(404, ANY) class WebSockifyServerTestCase(unittest.TestCase): def setUp(self): super(WebSockifyServerTestCase, self).setUp() - self.stubs = stubout.StubOutForTesting() self.tmpdir = tempfile.mkdtemp('-websockify-tests') # Mock this out cause it screws tests up - self.stubs.Set(os, 'chdir', lambda *args, **kwargs: None) + patch('os.chdir').start() def tearDown(self): """Called automatically after each test.""" - self.stubs.UnsetAll() + patch.stopall() os.rmdir(self.tmpdir) super(WebSockifyServerTestCase, self).tearDown() @@ -154,10 +137,10 @@ class WebSockifyServerTestCase(unittest.TestCase): def test_daemonize_raises_error_while_closing_fds(self): server = self._get_server(daemon=True, ssl_only=1, idle_timeout=1) - self.stubs.Set(os, 'fork', lambda *args: 0) - self.stubs.Set(signal, 'signal', lambda *args: None) - self.stubs.Set(os, 'setsid', lambda *args: None) - self.stubs.Set(os, 'close', raise_oserror) + patch('os.fork').start().return_value = 0 + patch('signal.signal').start() + patch('os.setsid').start() + patch('os.close').start().side_effect = raise_oserror self.assertRaises(OSError, server.daemonize, keepfd=None, chdir='./') def test_daemonize_ignores_ebadf_error_while_closing_fds(self): @@ -165,11 +148,11 @@ class WebSockifyServerTestCase(unittest.TestCase): raise OSError(errno.EBADF, 'fake error') server = self._get_server(daemon=True, ssl_only=1, idle_timeout=1) - self.stubs.Set(os, 'fork', lambda *args: 0) - self.stubs.Set(os, 'setsid', lambda *args: None) - self.stubs.Set(signal, 'signal', lambda *args: None) - self.stubs.Set(os, 'close', raise_oserror_ebadf) - self.stubs.Set(os, 'open', raise_oserror) + patch('os.fork').start().return_value = 0 + patch('signal.signal').start() + patch('os.setsid').start() + patch('os.close').start().side_effect = raise_oserror_ebadf + patch('os.open').start().side_effect = raise_oserror self.assertRaises(OSError, server.daemonize, keepfd=None, chdir='./') def test_handshake_fails_on_not_ready(self): @@ -178,7 +161,7 @@ class WebSockifyServerTestCase(unittest.TestCase): def fake_select(rlist, wlist, xlist, timeout=None): return ([], [], []) - self.stubs.Set(select, 'select', fake_select) + patch('select.select').start().side_effect = fake_select self.assertRaises( websockifyserver.WebSockifyServer.EClose, server.do_handshake, FakeSocket(), '127.0.0.1') @@ -191,7 +174,7 @@ class WebSockifyServerTestCase(unittest.TestCase): def fake_select(rlist, wlist, xlist, timeout=None): return ([sock], [], []) - self.stubs.Set(select, 'select', fake_select) + patch('select.select').start().side_effect = fake_select self.assertRaises( websockifyserver.WebSockifyServer.EClose, server.do_handshake, sock, '127.0.0.1') @@ -208,7 +191,7 @@ class WebSockifyServerTestCase(unittest.TestCase): def fake_select(rlist, wlist, xlist, timeout=None): return ([sock], [], []) - self.stubs.Set(select, 'select', fake_select) + patch('select.select').start().side_effect = fake_select self.assertRaises( websockifyserver.WebSockifyServer.EClose, server.do_handshake, sock, '127.0.0.1') @@ -230,7 +213,7 @@ class WebSockifyServerTestCase(unittest.TestCase): def fake_select(rlist, wlist, xlist, timeout=None): return ([sock], [], []) - self.stubs.Set(select, 'select', fake_select) + patch('select.select').start().side_effect = fake_select self.assertEqual(server.do_handshake(sock, '127.0.0.1'), sock) self.assertTrue(FakeHandler.CALLED, True) @@ -251,7 +234,7 @@ class WebSockifyServerTestCase(unittest.TestCase): def fake_select(rlist, wlist, xlist, timeout=None): return ([sock], [], []) - self.stubs.Set(select, 'select', fake_select) + patch('select.select').start().side_effect = fake_select self.assertRaises( websockifyserver.WebSockifyServer.EClose, server.do_handshake, sock, '127.0.0.1') @@ -280,13 +263,13 @@ class WebSockifyServerTestCase(unittest.TestCase): def wrap_socket(self, *args, **kwargs): raise ssl.SSLError(ssl.SSL_ERROR_EOF) - self.stubs.Set(select, 'select', fake_select) + patch('select.select').start().side_effect = fake_select if (hasattr(ssl, 'create_default_context')): # for recent versions of python - self.stubs.Set(ssl, 'create_default_context', fake_create_default_context) + patch('ssl.create_default_context').start().side_effect = fake_create_default_context else: # for fallback for old versions of python - self.stubs.Set(ssl, 'wrap_socket', fake_wrap_socket) + patch('ssl.warp_socket').start().side_effect = fake_wrap_socket self.assertRaises( websockifyserver.WebSockifyServer.EClose, server.do_handshake, sock, '127.0.0.1') @@ -321,10 +304,10 @@ class WebSockifyServerTestCase(unittest.TestCase): def set_ciphers(self, ciphers_to_set): fake_create_default_context.CIPHERS = ciphers_to_set - self.stubs.Set(select, 'select', fake_select) + patch('select.select').start().side_effect = fake_select if (hasattr(ssl, 'create_default_context')): # for recent versions of python - self.stubs.Set(ssl, 'create_default_context', fake_create_default_context) + patch('ssl.create_default_context').start().side_effect = fake_create_default_context server.do_handshake(sock, '127.0.0.1') self.assertEqual(fake_create_default_context.CIPHERS, test_ciphers) else: @@ -365,10 +348,10 @@ class WebSockifyServerTestCase(unittest.TestCase): fake_create_default_context.OPTIONS = val options = property(get_options, set_options) - self.stubs.Set(select, 'select', fake_select) + patch('select.select').start().side_effect = fake_select if (hasattr(ssl, 'create_default_context')): # for recent versions of python - self.stubs.Set(ssl, 'create_default_context', fake_create_default_context) + patch('ssl.create_default_context').start().side_effect = fake_create_default_context server.do_handshake(sock, '127.0.0.1') self.assertEqual(fake_create_default_context.OPTIONS, test_options) else: @@ -387,11 +370,9 @@ class WebSockifyServerTestCase(unittest.TestCase): def fake_select(rlist, wlist, xlist, timeout=None): raise Exception("fake error") - self.stubs.Set(websockifyserver.WebSockifyServer, 'socket', - lambda *args, **kwargs: sock) - self.stubs.Set(websockifyserver.WebSockifyServer, 'daemonize', - lambda *args, **kwargs: None) - self.stubs.Set(select, 'select', fake_select) + patch('websockify.websockifyserver.WebSockifyServer.socket').start() + patch('websockify.websockifyserver.WebSockifyServer.daemonize').start() + patch('select.select').start().side_effect = fake_select server.start_server() def test_start_server_keyboardinterrupt(self): @@ -401,11 +382,9 @@ class WebSockifyServerTestCase(unittest.TestCase): def fake_select(rlist, wlist, xlist, timeout=None): raise KeyboardInterrupt - self.stubs.Set(websockifyserver.WebSockifyServer, 'socket', - lambda *args, **kwargs: sock) - self.stubs.Set(websockifyserver.WebSockifyServer, 'daemonize', - lambda *args, **kwargs: None) - self.stubs.Set(select, 'select', fake_select) + patch('websockify.websockifyserver.WebSockifyServer.socket').start() + patch('websockify.websockifyserver.WebSockifyServer.daemonize').start() + patch('select.select').start().side_effect = fake_select server.start_server() def test_start_server_systemexit(self): @@ -415,11 +394,9 @@ class WebSockifyServerTestCase(unittest.TestCase): def fake_select(rlist, wlist, xlist, timeout=None): sys.exit() - self.stubs.Set(websockifyserver.WebSockifyServer, 'socket', - lambda *args, **kwargs: sock) - self.stubs.Set(websockifyserver.WebSockifyServer, 'daemonize', - lambda *args, **kwargs: None) - self.stubs.Set(select, 'select', fake_select) + patch('websockify.websockifyserver.WebSockifyServer.socket').start() + patch('websockify.websockifyserver.WebSockifyServer.daemonize').start() + patch('select.select').start().side_effect = fake_select server.start_server() def test_socket_set_keepalive_options(self):