commit 342503b71b1ccbdb13b975808c7348a88ebac634
Author: Damian Johnson <atagar(a)torproject.org>
Date: Sun Dec 15 17:49:11 2013 -0800
Breaking up starter's unit tests
Our test/starter.py was starting to get a bit big so breaking it up.
---
test/starter.py | 204 ----------------------------------------
test/starter/arg_parsing.py | 67 +++++++++++++
test/starter/authenticate.py | 83 ++++++++++++++++
test/starter/get_controller.py | 65 +++++++++++++
4 files changed, 215 insertions(+), 204 deletions(-)
diff --git a/test/__init__.py b/test/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/test/starter.py b/test/starter.py
deleted file mode 100644
index 724a318..0000000
--- a/test/starter.py
+++ /dev/null
@@ -1,204 +0,0 @@
-"""
-Unit tests for arm's initialization module.
-"""
-
-import getopt
-import unittest
-
-from mock import Mock, patch
-
-from arm.starter import (
- _load_settings,
- _get_args,
- _get_controller,
- _authenticate,
- ARGS,
-)
-
-import stem
-import stem.connection
-import stem.socket
-
-_load_settings()
-
-class TestArgumentParsing(unittest.TestCase):
- def test_that_we_get_default_values(self):
- args = _get_args([])
-
- for attr in ARGS:
- self.assertEqual(ARGS[attr], getattr(args, attr))
-
- def test_that_we_load_arguments(self):
- args = _get_args(['--interface', '10.0.0.25:80'])
- self.assertEqual('10.0.0.25', args.control_address)
- self.assertEqual(80, args.control_port)
-
- args = _get_args(['--interface', '80'])
- self.assertEqual(ARGS['control_address'], args.control_address)
- self.assertEqual(80, args.control_port)
-
- args = _get_args(['--socket', '/tmp/my_socket', '--config', '/tmp/my_config'])
- self.assertEqual('/tmp/my_socket', args.control_socket)
- self.assertEqual('/tmp/my_config', args.config)
-
- args = _get_args(['--debug', '--blind'])
- self.assertEqual(True, args.debug)
- self.assertEqual(True, args.blind)
-
- args = _get_args(['--event', 'D1'])
- self.assertEqual('D1', args.logged_events)
-
- args = _get_args(['--version'])
- self.assertEqual(True, args.print_version)
-
- args = _get_args(['--help'])
- self.assertEqual(True, args.print_help)
-
- def test_examples(self):
- args = _get_args(['-b', '-i', '1643'])
- self.assertEqual(True, args.blind)
- self.assertEqual(1643, args.control_port)
-
- args = _get_args(['-e', 'we', '-c', '/tmp/cfg'])
- self.assertEqual('we', args.logged_events)
- self.assertEqual('/tmp/cfg', args.config)
-
- def test_that_we_reject_unrecognized_arguments(self):
- self.assertRaises(getopt.GetoptError, _get_args, ['--blarg', 'stuff'])
-
- def test_that_we_reject_invalid_interfaces(self):
- invalid_inputs = (
- '',
- ' ',
- 'blarg',
- '127.0.0.1',
- '127.0.0.1:',
- ':80',
- '400.0.0.1:80',
- '127.0.0.1:-5',
- '127.0.0.1:500000',
- )
-
- for invalid_input in invalid_inputs:
- self.assertRaises(ValueError, _get_args, ['--interface', invalid_input])
-
-class TestGetController(unittest.TestCase):
- @patch('os.path.exists', Mock(return_value = True))
- @patch('stem.util.system.is_running')
- @patch('stem.control.Controller.from_socket_file', Mock(side_effect = stem.SocketError('failed')))
- @patch('stem.control.Controller.from_port', Mock(side_effect = stem.SocketError('failed')))
- def test_failue_with_the_default_endpoint(self, is_running_mock):
- is_running_mock.return_value = False
- self._assert_get_controller_fails_with([], "Unable to connect to tor. Are you sure it's running?")
-
- is_running_mock.return_value = True
- self._assert_get_controller_fails_with([], "Unable to connect to tor. Maybe it's running without a ControlPort?")
-
- @patch('os.path.exists')
- @patch('stem.util.system.is_running', Mock(return_value = True))
- @patch('stem.control.Controller.from_socket_file', Mock(side_effect = stem.SocketError('failed')))
- @patch('stem.control.Controller.from_port', Mock(side_effect = stem.SocketError('failed')))
- def test_failure_with_a_custom_endpoint(self, path_exists_mock):
- path_exists_mock.return_value = True
- self._assert_get_controller_fails_with(['--interface', '80'], "Unable to connect to 127.0.0.1:80: failed")
- self._assert_get_controller_fails_with(['--socket', '/tmp/my_socket'], "Unable to connect to '/tmp/my_socket': failed")
-
- path_exists_mock.return_value = False
- self._assert_get_controller_fails_with(['--interface', '80'], "Unable to connect to 127.0.0.1:80: failed")
- self._assert_get_controller_fails_with(['--socket', '/tmp/my_socket'], "The socket file you specified (/tmp/my_socket) doesn't exist")
-
- @patch('os.path.exists', Mock(return_value = False))
- @patch('stem.control.Controller.from_port')
- def test_getting_a_control_port(self, from_port_mock):
- from_port_mock.return_value = 'success'
-
- self.assertEqual('success', _get_controller(_get_args([])))
- from_port_mock.assert_called_once_with('127.0.0.1', 9051)
- from_port_mock.reset_mock()
-
- self.assertEqual('success', _get_controller(_get_args(['--interface', '255.0.0.10:80'])))
- from_port_mock.assert_called_once_with('255.0.0.10', 80)
-
- @patch('os.path.exists', Mock(return_value = True))
- @patch('stem.control.Controller.from_socket_file')
- def test_getting_a_control_socket(self, from_socket_file_mock):
- from_socket_file_mock.return_value = 'success'
-
- self.assertEqual('success', _get_controller(_get_args([])))
- from_socket_file_mock.assert_called_once_with('/var/run/tor/control')
- from_socket_file_mock.reset_mock()
-
- self.assertEqual('success', _get_controller(_get_args(['--socket', '/tmp/my_socket'])))
- from_socket_file_mock.assert_called_once_with('/tmp/my_socket')
-
- def _assert_get_controller_fails_with(self, args, msg):
- try:
- _get_controller(_get_args(args))
- self.fail()
- except ValueError, exc:
- self.assertEqual(msg, str(exc))
-
-class TestAuthenticate(unittest.TestCase):
- @patch('arm.util.torTools.get_chroot')
- def test_success(self, get_chroot_mock):
- controller = Mock()
-
- get_chroot_mock.return_value = '' # no chroot
- _authenticate(controller, None)
- controller.authenticate.assert_called_with(password = None, chroot_path = '')
- controller.authenticate.reset_mock()
-
- get_chroot_mock.return_value = '/my/chroot'
- _authenticate(controller, 's3krit!!!')
- controller.authenticate.assert_called_with(password = 's3krit!!!', chroot_path = '/my/chroot')
-
- @patch('arm.util.torTools.get_chroot', Mock(return_value = ''))
- @patch('getpass.getpass')
- def test_success_with_password_prompt(self, getpass_mock):
- controller = Mock()
-
- def authenticate_mock(password, **kwargs):
- 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)
-
- controller.authenticate.side_effect = authenticate_mock
- getpass_mock.return_value = 'my_password'
-
- _authenticate(controller, None)
- controller.authenticate.assert_any_call(password = None, chroot_path = '')
- controller.authenticate.assert_any_call(password = 'my_password', chroot_path = '')
-
- @patch('arm.util.torTools.get_chroot', Mock(return_value = ''))
- def test_failure(self):
- controller = Mock()
-
- controller.authenticate.side_effect = stem.connection.IncorrectSocketType('unable to connect to socket')
- controller.get_socket.return_value = stem.socket.ControlPort(connect = False)
- self._assert_authenticate_fails_with(controller, 'Please check in your torrc that 9051 is the ControlPort.')
-
- controller.get_socket.return_value = stem.socket.ControlSocketFile(connect = False)
- self._assert_authenticate_fails_with(controller, 'Are you sure the interface you specified belongs to')
-
- controller.authenticate.side_effect = stem.connection.UnrecognizedAuthMethods('unable to connect', ['telepathy'])
- self._assert_authenticate_fails_with(controller, 'Tor is using a type of authentication we do not recognize...\n\n telepathy')
-
- controller.authenticate.side_effect = stem.connection.IncorrectPassword('password rejected')
- self._assert_authenticate_fails_with(controller, 'Incorrect password')
-
- controller.authenticate.side_effect = stem.connection.UnreadableCookieFile('permission denied', '/tmp/my_cookie', False)
- self._assert_authenticate_fails_with(controller, "We were unable to read tor's authentication cookie...\n\n Path: /tmp/my_cookie\n Issue: permission denied")
-
- controller.authenticate.side_effect = stem.connection.OpenAuthRejected('crazy failure')
- self._assert_authenticate_fails_with(controller, 'Unable to authenticate: crazy failure')
-
- def _assert_authenticate_fails_with(self, controller, msg):
- try:
- _get_controller(_authenticate(controller, None))
- self.fail()
- except ValueError, exc:
- if not msg in str(exc):
- self.fail("Expected...\n\n%s\n\n... which couldn't be found in...\n\n%s" % (msg, exc))
diff --git a/test/starter/__init__.py b/test/starter/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/test/starter/arg_parsing.py b/test/starter/arg_parsing.py
new file mode 100644
index 0000000..26ed3bf
--- /dev/null
+++ b/test/starter/arg_parsing.py
@@ -0,0 +1,67 @@
+import getopt
+import unittest
+
+from arm.starter import _get_args, ARGS
+
+class TestArgumentParsing(unittest.TestCase):
+ def test_that_we_get_default_values(self):
+ args = _get_args([])
+
+ for attr in ARGS:
+ self.assertEqual(ARGS[attr], getattr(args, attr))
+
+ def test_that_we_load_arguments(self):
+ args = _get_args(['--interface', '10.0.0.25:80'])
+ self.assertEqual('10.0.0.25', args.control_address)
+ self.assertEqual(80, args.control_port)
+
+ args = _get_args(['--interface', '80'])
+ self.assertEqual(ARGS['control_address'], args.control_address)
+ self.assertEqual(80, args.control_port)
+
+ args = _get_args(['--socket', '/tmp/my_socket', '--config', '/tmp/my_config'])
+ self.assertEqual('/tmp/my_socket', args.control_socket)
+ self.assertEqual('/tmp/my_config', args.config)
+
+ args = _get_args(['--debug', '--blind'])
+ self.assertEqual(True, args.debug)
+ self.assertEqual(True, args.blind)
+
+ args = _get_args(['--event', 'D1'])
+ self.assertEqual('D1', args.logged_events)
+
+ args = _get_args(['--version'])
+ self.assertEqual(True, args.print_version)
+
+ args = _get_args(['--help'])
+ self.assertEqual(True, args.print_help)
+
+ def test_examples(self):
+ args = _get_args(['-b', '-i', '1643'])
+ self.assertEqual(True, args.blind)
+ self.assertEqual(1643, args.control_port)
+
+ args = _get_args(['-e', 'we', '-c', '/tmp/cfg'])
+ self.assertEqual('we', args.logged_events)
+ self.assertEqual('/tmp/cfg', args.config)
+
+ def test_that_we_reject_unrecognized_arguments(self):
+ self.assertRaises(getopt.GetoptError, _get_args, ['--blarg', 'stuff'])
+
+ def test_that_we_reject_invalid_interfaces(self):
+ invalid_inputs = (
+ '',
+ ' ',
+ 'blarg',
+ '127.0.0.1',
+ '127.0.0.1:',
+ ':80',
+ '400.0.0.1:80',
+ '127.0.0.1:-5',
+ '127.0.0.1:500000',
+ )
+
+ for invalid_input in invalid_inputs:
+ self.assertRaises(ValueError, _get_args, ['--interface', invalid_input])
+
+
diff --git a/test/starter/authenticate.py b/test/starter/authenticate.py
new file mode 100644
index 0000000..5f9e0aa
--- /dev/null
+++ b/test/starter/authenticate.py
@@ -0,0 +1,83 @@
+import unittest
+
+from mock import Mock, patch
+
+from arm.starter import (
+ _load_settings,
+ _get_args,
+ _get_controller,
+ _authenticate,
+ ARGS,
+)
+
+import stem
+import stem.connection
+import stem.socket
+
+_load_settings()
+
+class TestAuthenticate(unittest.TestCase):
+ @patch('arm.util.torTools.get_chroot')
+ def test_success(self, get_chroot_mock):
+ controller = Mock()
+
+ get_chroot_mock.return_value = '' # no chroot
+ _authenticate(controller, None)
+ controller.authenticate.assert_called_with(password = None, chroot_path = '')
+ controller.authenticate.reset_mock()
+
+ get_chroot_mock.return_value = '/my/chroot'
+ _authenticate(controller, 's3krit!!!')
+ controller.authenticate.assert_called_with(password = 's3krit!!!', chroot_path = '/my/chroot')
+
+ @patch('arm.util.torTools.get_chroot', Mock(return_value = ''))
+ @patch('getpass.getpass')
+ def test_success_with_password_prompt(self, getpass_mock):
+ controller = Mock()
+
+ def authenticate_mock(password, **kwargs):
+ 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)
+
+ controller.authenticate.side_effect = authenticate_mock
+ getpass_mock.return_value = 'my_password'
+
+ _authenticate(controller, None)
+ controller.authenticate.assert_any_call(password = None, chroot_path = '')
+ controller.authenticate.assert_any_call(password = 'my_password', chroot_path = '')
+
+ @patch('arm.util.torTools.get_chroot', Mock(return_value = ''))
+ def test_failure(self):
+ controller = Mock()
+
+ controller.authenticate.side_effect = stem.connection.IncorrectSocketType('unable to connect to socket')
+ controller.get_socket.return_value = stem.socket.ControlPort(connect = False)
+ self._assert_authenticate_fails_with(controller, 'Please check in your torrc that 9051 is the ControlPort.')
+
+ controller.get_socket.return_value = stem.socket.ControlSocketFile(connect = False)
+ self._assert_authenticate_fails_with(controller, 'Are you sure the interface you specified belongs to')
+
+ controller.authenticate.side_effect = stem.connection.UnrecognizedAuthMethods('unable to connect', ['telepathy'])
+ self._assert_authenticate_fails_with(controller, 'Tor is using a type of authentication we do not recognize...\n\n telepathy')
+
+ controller.authenticate.side_effect = stem.connection.IncorrectPassword('password rejected')
+ self._assert_authenticate_fails_with(controller, 'Incorrect password')
+
+ controller.authenticate.side_effect = stem.connection.UnreadableCookieFile('permission denied', '/tmp/my_cookie', False)
+ self._assert_authenticate_fails_with(controller, "We were unable to read tor's authentication cookie...\n\n Path: /tmp/my_cookie\n Issue: permission denied")
+
+ controller.authenticate.side_effect = stem.connection.OpenAuthRejected('crazy failure')
+ self._assert_authenticate_fails_with(controller, 'Unable to authenticate: crazy failure')
+
+ def _assert_authenticate_fails_with(self, controller, msg):
+ try:
+ _get_controller(_authenticate(controller, None))
+ self.fail()
+ except ValueError, exc:
+ if not msg in str(exc):
+ self.fail("Expected...\n\n%s\n\n... which couldn't be found in...\n\n%s" % (msg, exc))
+
diff --git a/test/starter/get_controller.py b/test/starter/get_controller.py
new file mode 100644
index 0000000..5890dba
--- /dev/null
+++ b/test/starter/get_controller.py
@@ -0,0 +1,65 @@
+import unittest
+
+from mock import Mock, patch
+
+from arm.starter import _get_args, _get_controller
+
+import stem
+import stem.connection
+import stem.socket
+
+class TestGetController(unittest.TestCase):
+ @patch('os.path.exists', Mock(return_value = True))
+ @patch('stem.util.system.is_running')
+ @patch('stem.control.Controller.from_socket_file', Mock(side_effect = stem.SocketError('failed')))
+ @patch('stem.control.Controller.from_port', Mock(side_effect = stem.SocketError('failed')))
+ def test_failue_with_the_default_endpoint(self, is_running_mock):
+ is_running_mock.return_value = False
+ self._assert_get_controller_fails_with([], "Unable to connect to tor. Are you sure it's running?")
+
+ is_running_mock.return_value = True
+ self._assert_get_controller_fails_with([], "Unable to connect to tor. Maybe it's running without a ControlPort?")
+
+ @patch('os.path.exists')
+ @patch('stem.util.system.is_running', Mock(return_value = True))
+ @patch('stem.control.Controller.from_socket_file', Mock(side_effect = stem.SocketError('failed')))
+ @patch('stem.control.Controller.from_port', Mock(side_effect = stem.SocketError('failed')))
+ def test_failure_with_a_custom_endpoint(self, path_exists_mock):
+ path_exists_mock.return_value = True
+ self._assert_get_controller_fails_with(['--interface', '80'], "Unable to connect to 127.0.0.1:80: failed")
+ self._assert_get_controller_fails_with(['--socket', '/tmp/my_socket'], "Unable to connect to '/tmp/my_socket': failed")
+
+ path_exists_mock.return_value = False
+ self._assert_get_controller_fails_with(['--interface', '80'], "Unable to connect to 127.0.0.1:80: failed")
+ self._assert_get_controller_fails_with(['--socket', '/tmp/my_socket'], "The socket file you specified (/tmp/my_socket) doesn't exist")
+
+ @patch('os.path.exists', Mock(return_value = False))
+ @patch('stem.control.Controller.from_port')
+ def test_getting_a_control_port(self, from_port_mock):
+ from_port_mock.return_value = 'success'
+
+ self.assertEqual('success', _get_controller(_get_args([])))
+ from_port_mock.assert_called_once_with('127.0.0.1', 9051)
+ from_port_mock.reset_mock()
+
+ self.assertEqual('success', _get_controller(_get_args(['--interface', '255.0.0.10:80'])))
+ from_port_mock.assert_called_once_with('255.0.0.10', 80)
+
+ @patch('os.path.exists', Mock(return_value = True))
+ @patch('stem.control.Controller.from_socket_file')
+ def test_getting_a_control_socket(self, from_socket_file_mock):
+ from_socket_file_mock.return_value = 'success'
+
+ self.assertEqual('success', _get_controller(_get_args([])))
+ from_socket_file_mock.assert_called_once_with('/var/run/tor/control')
+ from_socket_file_mock.reset_mock()
+
+ self.assertEqual('success', _get_controller(_get_args(['--socket', '/tmp/my_socket'])))
+ from_socket_file_mock.assert_called_once_with('/tmp/my_socket')
+
+ def _assert_get_controller_fails_with(self, args, msg):
+ try:
+ _get_controller(_get_args(args))
+ self.fail()
+ except ValueError, exc:
+ self.assertEqual(msg, str(exc))
