[tor-commits] [stem/master] Unit tests for the connect() function
atagar at torproject.org
atagar at torproject.org
Thu Apr 10 16:12:26 UTC 2014
commit 0efd180327c96a75d25744374e72bf11b5a5b586
Author: Damian Johnson <atagar at torproject.org>
Date: Sat Apr 5 21:07:51 2014 -0700
Unit tests for the connect() function
One really neat thing about stealing arm's code is that I'd actually written a
lot of good unit tests, so we can now move those to stem! Our connect* methods
previously only had integ coverage, so this is a really nice improvement.
This also uncovered a few bugs from porting the function.
---
stem/connection.py | 52 +++++++++++++--
test/settings.cfg | 1 +
test/unit/connection/connect.py | 141 +++++++++++++++++++++++++++++++++++++++
3 files changed, 188 insertions(+), 6 deletions(-)
diff --git a/stem/connection.py b/stem/connection.py
index 1304357..35524d7 100644
--- a/stem/connection.py
+++ b/stem/connection.py
@@ -147,6 +147,40 @@ AuthMethod = stem.util.enum.Enum("NONE", "PASSWORD", "COOKIE", "SAFECOOKIE", "UN
CLIENT_HASH_CONSTANT = b"Tor safe cookie authentication controller-to-server hash"
SERVER_HASH_CONSTANT = b"Tor safe cookie authentication server-to-controller hash"
+MISSING_PASSWORD_BUG_MSG = """
+BUG: You provided a password but despite this stem reported that it was
+missing. This shouldn't happen - please let us know about it!
+
+ http://bugs.torproject.org
+"""
+
+UNRECOGNIZED_AUTH_TYPE_MSG = """
+Tor is using a type of authentication we do not recognize...
+
+ {auth_methods}
+
+Please check that arm is up to date and if there is an existing issue on
+'http://bugs.torproject.org'. If there isn't one then let us know!
+"""
+
+
+UNREADABLE_COOKIE_FILE_MSG = """
+We were unable to read tor's authentication cookie...
+
+ Path: {path}
+ Issue: {issue}
+"""
+
+WRONG_PORT_TYPE_MSG = """
+Please check in your torrc that {port} is the ControlPort. Maybe you
+configured it to be the ORPort or SocksPort instead?
+"""
+
+WRONG_SOCKET_TYPE_MSG = """
+Unable to connect to tor. Are you sure the interface you specified belongs to
+tor?
+"""
+
CONNECT_MESSAGES = {
'general_auth_failure': "Unable to authenticate: {error}",
'incorrect_password': "Incorrect password",
@@ -156,6 +190,11 @@ CONNECT_MESSAGES = {
'tor_isnt_running': "Unable to connect to tor. Are you sure it's running?",
'unable_to_use_port': "Unable to connect to {address}:{port}: {error}",
'unable_to_use_socket': "Unable to connect to '{path}': {error}",
+ 'missing_password_bug': MISSING_PASSWORD_BUG_MSG.strip(),
+ 'uncrcognized_auth_type': UNRECOGNIZED_AUTH_TYPE_MSG.strip(),
+ 'unreadable_cookie_file': UNREADABLE_COOKIE_FILE_MSG.strip(),
+ 'wrong_port_type': WRONG_PORT_TYPE_MSG.strip(),
+ 'wrong_socket_type': WRONG_SOCKET_TYPE_MSG.strip(),
}
@@ -232,7 +271,7 @@ def connect(control_port = ('127.0.0.1', 9051), control_socket = '/var/run/tor/c
print error_msg
return None
- return _connect(control_connection, password, chroot_path, controller)
+ return _connect_auth(control_connection, password, chroot_path, controller)
def connect_port(address = "127.0.0.1", port = 9051, password = None, chroot_path = None, controller = stem.control.Controller):
@@ -261,7 +300,7 @@ def connect_port(address = "127.0.0.1", port = 9051, password = None, chroot_pat
print exc
return None
- return _connect(control_port, password, chroot_path, controller)
+ return _connect_auth(control_port, password, chroot_path, controller)
def connect_socket_file(path = "/var/run/tor/control", password = None, chroot_path = None, controller = stem.control.Controller):
@@ -291,12 +330,13 @@ def connect_socket_file(path = "/var/run/tor/control", password = None, chroot_p
print exc
return None
- return _connect(control_socket, password, chroot_path, controller)
+ return _connect_auth(control_socket, password, chroot_path, controller)
-def _connect(control_socket, password, chroot_path, controller):
+def _connect_auth(control_socket, password, chroot_path, controller):
"""
- Common implementation for the connect_* functions.
+ Helper for the connect_* functions that authenticates the socket and
+ constructs the controller.
:param stem.socket.ControlSocket control_socket: socket being authenticated to
:param str password: passphrase to authenticate to the socket
@@ -341,7 +381,7 @@ def _connect(control_socket, password, chroot_path, controller):
control_socket.close()
return None
- return _connect(control_socket, password, chroot_path, controller)
+ return _connect_auth(control_socket, password, chroot_path, controller)
except UnreadableCookieFile as exc:
print CONNECT_MESSAGES['unreadable_cookie_file'].format(path = exc.cookie_path, issue = str(exc))
control_socket.close()
diff --git a/test/settings.cfg b/test/settings.cfg
index f12311d..f0eb758 100644
--- a/test/settings.cfg
+++ b/test/settings.cfg
@@ -181,6 +181,7 @@ test.unit_tests
|test.unit.response.protocolinfo.TestProtocolInfoResponse
|test.unit.response.mapaddress.TestMapAddressResponse
|test.unit.connection.authentication.TestAuthenticate
+|test.unit.connection.connect.TestConnect
|test.unit.control.controller.TestControl
|test.unit.doctest.TestDocumentation
diff --git a/test/unit/connection/connect.py b/test/unit/connection/connect.py
new file mode 100644
index 0000000..52039f2
--- /dev/null
+++ b/test/unit/connection/connect.py
@@ -0,0 +1,141 @@
+"""
+Unit tests for the stem.connection.connect function.
+"""
+
+import StringIO
+import unittest
+
+from mock import Mock, patch
+
+import stem
+import stem.connection
+import stem.socket
+
+
+class TestConnect(unittest.TestCase):
+ @patch('sys.stdout', new_callable = StringIO.StringIO)
+ @patch('stem.util.system.is_running')
+ @patch('os.path.exists', Mock(return_value = True))
+ @patch('stem.socket.ControlSocketFile', Mock(side_effect = stem.SocketError('failed')))
+ @patch('stem.socket.ControlPort', Mock(side_effect = stem.SocketError('failed')))
+ @patch('stem.connection._connect_auth', Mock())
+ def test_failue_with_the_default_endpoint(self, is_running_mock, stdout_mock):
+ is_running_mock.return_value = False
+ self._assert_connect_fails_with({}, stdout_mock, "Unable to connect to tor. Are you sure it's running?")
+
+ is_running_mock.return_value = True
+ self._assert_connect_fails_with({}, stdout_mock, "Unable to connect to tor. Maybe it's running without a ControlPort?")
+
+ @patch('sys.stdout', new_callable = StringIO.StringIO)
+ @patch('os.path.exists')
+ @patch('stem.util.system.is_running', Mock(return_value = True))
+ @patch('stem.socket.ControlSocketFile', Mock(side_effect = stem.SocketError('failed')))
+ @patch('stem.socket.ControlPort', Mock(side_effect = stem.SocketError('failed')))
+ @patch('stem.connection._connect_auth', Mock())
+ def test_failure_with_a_custom_endpoint(self, path_exists_mock, stdout_mock):
+ path_exists_mock.return_value = True
+ self._assert_connect_fails_with({'control_port': ('127.0.0.1', 80), 'control_socket': None}, stdout_mock, "Unable to connect to 127.0.0.1:80: failed")
+ self._assert_connect_fails_with({'control_port': None, 'control_socket': '/tmp/my_socket'}, stdout_mock, "Unable to connect to '/tmp/my_socket': failed")
+
+ path_exists_mock.return_value = False
+ self._assert_connect_fails_with({'control_port': ('127.0.0.1', 80), 'control_socket': None}, stdout_mock, "Unable to connect to 127.0.0.1:80: failed")
+ self._assert_connect_fails_with({'control_port': None, 'control_socket': '/tmp/my_socket'}, stdout_mock, "The socket file you specified (/tmp/my_socket) doesn't exist")
+
+ @patch('stem.socket.ControlPort')
+ @patch('os.path.exists', Mock(return_value = False))
+ @patch('stem.connection._connect_auth', Mock())
+ def test_getting_a_control_port(self, port_mock):
+ stem.connection.connect()
+ port_mock.assert_called_once_with('127.0.0.1', 9051)
+ port_mock.reset_mock()
+
+ stem.connection.connect(control_port = ('255.0.0.10', 80), control_socket = None)
+ port_mock.assert_called_once_with('255.0.0.10', 80)
+
+ @patch('stem.socket.ControlSocketFile')
+ @patch('os.path.exists', Mock(return_value = True))
+ @patch('stem.connection._connect_auth', Mock())
+ def test_getting_a_control_socket(self, socket_mock):
+ stem.connection.connect()
+ socket_mock.assert_called_once_with('/var/run/tor/control')
+ socket_mock.reset_mock()
+
+ stem.connection.connect(control_port = None, control_socket = '/tmp/my_socket')
+ socket_mock.assert_called_once_with('/tmp/my_socket')
+
+ def _assert_connect_fails_with(self, args, stdout_mock, msg):
+ result = stem.connection.connect(**args)
+
+ if result is not None:
+ self.fail()
+
+ stdout_output = stdout_mock.getvalue()
+ stdout_mock.truncate(0)
+ self.assertEqual(msg, stdout_output.strip())
+
+ @patch('stem.connection.authenticate')
+ def test_auth_success(self, authenticate_mock):
+ control_socket = Mock()
+
+ stem.connection._connect_auth(control_socket, None, None, None)
+ authenticate_mock.assert_called_with(control_socket, None, None)
+ authenticate_mock.reset_mock()
+
+ stem.connection._connect_auth(control_socket, 's3krit!!!', '/my/chroot', None)
+ authenticate_mock.assert_called_with(control_socket, 's3krit!!!', '/my/chroot')
+
+ @patch('getpass.getpass')
+ @patch('stem.connection.authenticate')
+ def test_auth_success_with_password_prompt(self, authenticate_mock, getpass_mock):
+ control_socket = Mock()
+
+ def authenticate_mock_func(controller, password, *args):
+ if password is None:
+ raise stem.connection.MissingPassword('no password')
+ elif password == 'my_password':
+ return None # success
+ else:
+ raise ValueError("Unexpected authenticate_mock input: %s" % password)
+
+ authenticate_mock.side_effect = authenticate_mock_func
+ getpass_mock.return_value = 'my_password'
+
+ stem.connection._connect_auth(control_socket, None, None, None)
+ authenticate_mock.assert_any_call(control_socket, None, None)
+ authenticate_mock.assert_any_call(control_socket, 'my_password', None)
+
+ @patch('sys.stdout', new_callable = StringIO.StringIO)
+ @patch('stem.connection.authenticate')
+ def test_auth_failure(self, authenticate_mock, stdout_mock):
+ control_socket = stem.socket.ControlPort(connect = False)
+
+ authenticate_mock.side_effect = stem.connection.IncorrectSocketType('unable to connect to socket')
+ self._assert_authenticate_fails_with(control_socket, stdout_mock, 'Please check in your torrc that 9051 is the ControlPort.')
+
+ control_socket = stem.socket.ControlSocketFile(connect = False)
+
+ self._assert_authenticate_fails_with(control_socket, stdout_mock, 'Are you sure the interface you specified belongs to')
+
+ authenticate_mock.side_effect = stem.connection.UnrecognizedAuthMethods('unable to connect', ['telepathy'])
+ self._assert_authenticate_fails_with(control_socket, stdout_mock, 'Tor is using a type of authentication we do not recognize...\n\n telepathy')
+
+ authenticate_mock.side_effect = stem.connection.IncorrectPassword('password rejected')
+ self._assert_authenticate_fails_with(control_socket, stdout_mock, 'Incorrect password')
+
+ authenticate_mock.side_effect = stem.connection.UnreadableCookieFile('permission denied', '/tmp/my_cookie', False)
+ self._assert_authenticate_fails_with(control_socket, stdout_mock, "We were unable to read tor's authentication cookie...\n\n Path: /tmp/my_cookie\n Issue: permission denied")
+
+ authenticate_mock.side_effect = stem.connection.OpenAuthRejected('crazy failure')
+ self._assert_authenticate_fails_with(control_socket, stdout_mock, 'Unable to authenticate: crazy failure')
+
+ def _assert_authenticate_fails_with(self, control_socket, stdout_mock, msg):
+ result = stem.connection._connect_auth(control_socket, None, None, None)
+
+ if result is not None:
+ self.fail() # _connect_auth() was successful
+
+ stdout_output = stdout_mock.getvalue()
+ stdout_mock.truncate(0)
+
+ if not msg in stdout_output:
+ self.fail("Expected...\n\n%s\n\n... which couldn't be found in...\n\n%s" % (msg, stdout_output))
More information about the tor-commits
mailing list