commit d61e330a715fbf266f7c9471f1a5d35073480a65 Author: Damian Johnson atagar@torproject.org Date: Sat Dec 21 16:38:12 2013 -0800
Moving argument parsing to its own module
Our starter is way too large. Argument parsing is a good candidate for its own submodule. --- arm/arguments.py | 117 ++++++++++++++++++++++++++++++++++++++++ arm/starter.py | 104 ++++------------------------------- test/arguments.py | 66 +++++++++++++++++++++++ test/starter/arg_parsing.py | 66 ----------------------- test/starter/get_controller.py | 13 ++--- 5 files changed, 200 insertions(+), 166 deletions(-)
diff --git a/arm/arguments.py b/arm/arguments.py new file mode 100644 index 0000000..4f796cc --- /dev/null +++ b/arm/arguments.py @@ -0,0 +1,117 @@ +""" +Commandline argument parsing for arm. +""" + +import collections +import getopt +import os + +import arm.logPanel + +import stem.connection +import stem.util.conf + +CONFIG = stem.util.conf.config_dict("arm", { + 'attribute.debug_log_path': '', + 'msg.help': '', +}) + +DEFAULT_ARGS = { + 'control_address': '127.0.0.1', + 'control_port': 9051, + 'user_provided_port': False, + 'control_socket': '/var/run/tor/control', + 'user_provided_socket': False, + 'config': os.path.expanduser("~/.arm/armrc"), + 'debug': False, + 'blind': False, + 'logged_events': 'N3', + 'print_version': False, + 'print_help': False, +} + +OPT = "i:s:c:dbe:vh" + +OPT_EXPANDED = [ + "interface=", + "socket=", + "config=", + "debug", + "blind", + "event=", + "version", + "help", +] + + +def parse(argv): + """ + Parses our arguments, providing a named tuple with their values. + + :param list argv: input arguments to be parsed + + :returns: a **named tuple** with our parsed arguments + + :raises: **ValueError** if we got an invalid argument + :raises: **getopt.GetoptError** if the arguments don't conform with what we + accept + """ + + args = dict(DEFAULT_ARGS) + + for opt, arg in getopt.getopt(argv, OPT, OPT_EXPANDED)[0]: + if opt in ("-i", "--interface"): + if ':' in arg: + address, port = arg.split(':', 1) + else: + address, port = None, arg + + if address is not None: + if not stem.util.connection.is_valid_ipv4_address(address): + raise ValueError("'%s' isn't a valid IPv4 address" % address) + + args['control_address'] = address + + if not stem.util.connection.is_valid_port(port): + raise ValueError("'%s' isn't a valid port number" % port) + + args['control_port'] = int(port) + args['user_provided_port'] = True + elif opt in ("-s", "--socket"): + args['control_socket'] = arg + args['user_provided_socket'] = True + elif opt in ("-c", "--config"): + args['config'] = arg + elif opt in ("-d", "--debug"): + args['debug'] = True + elif opt in ("-b", "--blind"): + args['blind'] = True + elif opt in ("-e", "--event"): + args['logged_events'] = arg + elif opt in ("-v", "--version"): + args['print_version'] = True + elif opt in ("-h", "--help"): + args['print_help'] = True + + # translates our args dict into a named tuple + + Args = collections.namedtuple('Args', args.keys()) + return Args(**args) + + +def get_help(): + """ + Provides our --help usage information. + + :returns: **str** with our usage information + """ + + return CONFIG['msg.help'].format( + address = DEFAULT_ARGS['control_address'], + port = DEFAULT_ARGS['control_port'], + socket = DEFAULT_ARGS['control_socket'], + config = DEFAULT_ARGS['config'], + debug_path = CONFIG['attribute.debug_log_path'], + events = DEFAULT_ARGS['logged_events'], + event_flags = arm.logPanel.EVENT_LISTING, + ) diff --git a/arm/starter.py b/arm/starter.py index e497c72..fd96236 100644 --- a/arm/starter.py +++ b/arm/starter.py @@ -4,7 +4,6 @@ information. This starts the applicatin, getting a tor connection and parsing arguments. """
-import collections import curses import getopt import getpass @@ -17,6 +16,7 @@ import time import threading
import arm +import arm.arguments import arm.controller import arm.logPanel import arm.util.panel @@ -33,13 +33,12 @@ import stem.util.connection import stem.util.log import stem.util.system
-LOG_DUMP_PATH = os.path.expanduser("~/.arm/log") SETTINGS_PATH = os.path.join(os.path.dirname(__file__), 'settings.cfg')
CONFIG = stem.util.conf.config_dict("arm", { + 'attribute.debug_log_path': '', 'tor.password': None, 'startup.events': 'N3', - 'msg.help': '', 'msg.debug_header': '', 'msg.wrong_port_type': '', 'msg.wrong_socket_type': '', @@ -54,26 +53,6 @@ CONFIG = stem.util.conf.config_dict("arm", { 'msg.unknown_term': '', })
-# Our default arguments. The _get_args() function provides a named tuple of -# this merged with our argv. - -ARGS = { - 'control_address': '127.0.0.1', - 'control_port': 9051, - 'user_provided_port': False, - 'control_socket': '/var/run/tor/control', - 'user_provided_socket': False, - 'config': os.path.expanduser("~/.arm/armrc"), - 'debug': False, - 'blind': False, - 'logged_events': 'N3', - 'print_version': False, - 'print_help': False, -} - -OPT = "i:s:c:dbe:vh" -OPT_EXPANDED = ["interface=", "socket=", "config=", "debug", "blind", "event=", "version", "help"] -
def _load_settings(config = 'arm'): """ @@ -99,61 +78,6 @@ def _load_settings(config = 'arm'): return config
-def _get_args(argv): - """ - Parses our arguments, providing a named tuple with their values. - - :param list argv: input arguments to be parsed - - :returns: a **named tuple** with our parsed arguments - - :raises: **ValueError** if we got an invalid argument - :raises: **getopt.GetoptError** if the arguments don't conform with what we - accept - """ - - args = dict(ARGS) - - for opt, arg in getopt.getopt(argv, OPT, OPT_EXPANDED)[0]: - if opt in ("-i", "--interface"): - if ':' in arg: - address, port = arg.split(':', 1) - else: - address, port = None, arg - - if address is not None: - if not stem.util.connection.is_valid_ipv4_address(address): - raise ValueError("'%s' isn't a valid IPv4 address" % address) - - args['control_address'] = address - - if not stem.util.connection.is_valid_port(port): - raise ValueError("'%s' isn't a valid port number" % port) - - args['control_port'] = int(port) - args['user_provided_port'] = True - elif opt in ("-s", "--socket"): - args['control_socket'] = arg - args['user_provided_socket'] = True - elif opt in ("-c", "--config"): - args['config'] = arg - elif opt in ("-d", "--debug"): - args['debug'] = True - elif opt in ("-b", "--blind"): - args['blind'] = True - elif opt in ("-e", "--event"): - args['logged_events'] = arg - elif opt in ("-v", "--version"): - args['print_version'] = True - elif opt in ("-h", "--help"): - args['print_help'] = True - - # translates our args dict into a named tuple - - Args = collections.namedtuple('Args', args.keys()) - return Args(**args) - - def _get_controller(args): """ Provides a Controller for the endpoint specified in the given arguments. @@ -225,17 +149,17 @@ def _authenticate(controller, password):
def _setup_debug_logging(): """ - Configures us to log at stem's trace level to LOG_DUMP_PATH. + Configures us to log at stem's trace level to 'attribute.debug_log_path'.
:raises: **IOError** if we can't log to this location """
- debug_dir = os.path.dirname(LOG_DUMP_PATH) + debug_dir = os.path.dirname(CONFIG['attribute.debug_log_path'])
if not os.path.exists(debug_dir): os.makedirs(debug_dir)
- debug_handler = logging.FileHandler(LOG_DUMP_PATH, mode = 'w') + debug_handler = logging.FileHandler(CONFIG['attribute.debug_log_path'], mode = 'w') debug_handler.setLevel(stem.util.log.logging_level(stem.util.log.TRACE)) debug_handler.setFormatter(logging.Formatter( fmt = '%(asctime)s [%(levelname)s] %(message)s', @@ -285,10 +209,11 @@ def _shutdown_daemons(): def main(): config = stem.util.conf.get_config("arm") config.set('attribute.start_time', str(int(time.time()))) + config.set('attribute.debug_log_path', os.path.expanduser("~/.arm/log"))
try: _load_settings() - args = _get_args(sys.argv[1:]) + args = arm.arguments.parse(sys.argv[1:]) except getopt.GetoptError as exc: print "%s (for usage provide --help)" % exc sys.exit(1) @@ -297,16 +222,7 @@ def main(): sys.exit(1)
if args.print_help: - print CONFIG['msg.help'].format( - address = ARGS['control_address'], - port = ARGS['control_port'], - socket = ARGS['control_socket'], - config = ARGS['config'], - debug_path = LOG_DUMP_PATH, - events = ARGS['logged_events'], - event_flags = arm.logPanel.EVENT_LISTING, - ) - + print arm.arguments.get_help() sys.exit()
if args.print_version: @@ -317,7 +233,7 @@ def main(): try: _setup_debug_logging() except IOError as exc: - print "Unable to write to our debug log file (%s): %s" % (LOG_DUMP_PATH, exc.strerror) + print "Unable to write to our debug log file (%s): %s" % (CONFIG['attribute.debug_log_path'], exc.strerror) sys.exit(1)
stem.util.log.trace(CONFIG['msg.debug_header'].format( @@ -330,7 +246,7 @@ def main(): armrc_content = _armrc_dump(args.config), ))
- print "Saving a debug log to %s, please check it for sensitive information before sharing" % LOG_DUMP_PATH + print "Saving a debug log to %s, please check it for sensitive information before sharing" % CONFIG['attribute.debug_log_path']
# loads user's personal armrc if available
diff --git a/test/arguments.py b/test/arguments.py new file mode 100644 index 0000000..c7e1168 --- /dev/null +++ b/test/arguments.py @@ -0,0 +1,66 @@ +import getopt +import unittest + +from arm.arguments import parse, DEFAULT_ARGS + + +class TestArgumentParsing(unittest.TestCase): + def test_that_we_get_default_values(self): + args = parse([]) + + for attr in DEFAULT_ARGS: + self.assertEqual(DEFAULT_ARGS[attr], getattr(args, attr)) + + def test_that_we_load_arguments(self): + args = parse(['--interface', '10.0.0.25:80']) + self.assertEqual('10.0.0.25', args.control_address) + self.assertEqual(80, args.control_port) + + args = parse(['--interface', '80']) + self.assertEqual(DEFAULT_ARGS['control_address'], args.control_address) + self.assertEqual(80, args.control_port) + + args = parse(['--socket', '/tmp/my_socket', '--config', '/tmp/my_config']) + self.assertEqual('/tmp/my_socket', args.control_socket) + self.assertEqual('/tmp/my_config', args.config) + + args = parse(['--debug', '--blind']) + self.assertEqual(True, args.debug) + self.assertEqual(True, args.blind) + + args = parse(['--event', 'D1']) + self.assertEqual('D1', args.logged_events) + + args = parse(['--version']) + self.assertEqual(True, args.print_version) + + args = parse(['--help']) + self.assertEqual(True, args.print_help) + + def test_examples(self): + args = parse(['-b', '-i', '1643']) + self.assertEqual(True, args.blind) + self.assertEqual(1643, args.control_port) + + args = parse(['-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, parse, ['--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, parse, ['--interface', invalid_input]) diff --git a/test/starter/arg_parsing.py b/test/starter/arg_parsing.py deleted file mode 100644 index f6c60f5..0000000 --- a/test/starter/arg_parsing.py +++ /dev/null @@ -1,66 +0,0 @@ -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/get_controller.py b/test/starter/get_controller.py index 974b4b5..dcd5f0c 100644 --- a/test/starter/get_controller.py +++ b/test/starter/get_controller.py @@ -2,7 +2,8 @@ import unittest
from mock import Mock, patch
-from arm.starter import _get_args, _get_controller +from arm.arguments import parse +from arm.starter import _get_controller
import stem import stem.connection @@ -39,11 +40,11 @@ class TestGetController(unittest.TestCase): def test_getting_a_control_port(self, from_port_mock): from_port_mock.return_value = 'success'
- self.assertEqual('success', _get_controller(_get_args([]))) + self.assertEqual('success', _get_controller(parse([]))) 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']))) + self.assertEqual('success', _get_controller(parse(['--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)) @@ -51,16 +52,16 @@ class TestGetController(unittest.TestCase): 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([]))) + self.assertEqual('success', _get_controller(parse([]))) 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']))) + self.assertEqual('success', _get_controller(parse(['--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)) + _get_controller(parse(args)) self.fail() except ValueError, exc: self.assertEqual(msg, str(exc))