commit d61e330a715fbf266f7c9471f1a5d35073480a65
Author: Damian Johnson <atagar(a)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))
