You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
393 lines
18 KiB
393 lines
18 KiB
From 992e09eac484f25871b7fcfc6d11b8e5beac9edb Mon Sep 17 00:00:00 2001
|
|
From: Pierre Ossman <ossman@cendio.se>
|
|
Date: Fri, 21 Aug 2020 10:50:11 +0200
|
|
Subject: [PATCH] Convert tests from mox to mock
|
|
|
|
mox is deprecated upstream in favour of mock
|
|
---
|
|
test-requirements.txt | 2 +-
|
|
tests/test_websocketproxy.py | 34 ++++------
|
|
tests/test_websockifyserver.py | 111 +++++++++++++--------------------
|
|
3 files changed, 58 insertions(+), 89 deletions(-)
|
|
|
|
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 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 @@ def __init__(self):
|
|
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 TestPlugin(token_plugins.BasePlugin):
|
|
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 @@ def test_asymmetric_jws_token_plugin(self):
|
|
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 @@ def test_asymmetric_jws_token_plugin_with_illigal_key_exception(self):
|
|
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 @@ def test_symmetric_jws_token_plugin(self):
|
|
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 @@ def test_symmetric_jws_token_plugin_with_illigal_key_exception(self):
|
|
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 @@ def test_asymmetric_jwe_token_plugin(self):
|
|
|
|
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 @@ def authenticate(self, headers, target_host, target_port):
|
|
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 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 @@ def makefile(self, mode='r', buffsize=None):
|
|
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 @@ def _get_server(self, handler_class=websockifyserver.WebSockifyRequestHandler,
|
|
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 @@ def _get_server(self, handler_class=websockifyserver.WebSockifyRequestHandler,
|
|
|
|
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 @@ def raise_oserror_ebadf(fd):
|
|
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 @@ def test_handshake_fails_on_not_ready(self):
|
|
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 @@ def test_empty_handshake_fails(self):
|
|
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 @@ def test_handshake_ssl_only_without_ssl_raises_error(self):
|
|
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 @@ def __init__(self, *args, **kwargs):
|
|
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 @@ def test_do_handshake_ssl_without_cert_raises_error(self):
|
|
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 @@ def load_verify_locations(self, cafile):
|
|
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 @@ def wrap_socket(self, *args, **kwargs):
|
|
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 @@ def set_options(self, val):
|
|
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 @@ def test_start_server_error(self):
|
|
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 @@ def test_start_server_keyboardinterrupt(self):
|
|
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 @@ def test_start_server_systemexit(self):
|
|
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):
|