commit 435de196896e5b170777edfc1b2331e733997172 Author: Damian Johnson atagar@torproject.org Date: Sun Sep 8 14:06:50 2013 -0700
Alternate function for parsing arguments
Adding a _get_args() function for parsing commandline arguments and providing back a named tuple. This is similar to what I'm doing for stem's run_tests.py argument parsing, and allows for much cleaner code than the present config mess.
This also includes unit tests for the new function. My plan is to add tests as I incrementally rewrite arm, but being a curses app I'm not sure if we'll be able to achieve too much coverage. Still, any bit helps.
The _get_args() function is not yet used. It's gonna take a bit of work to untangle the present config starter parameters. --- arm/starter.py | 82 +++++++++++++++++++++++++++++++++++++++++++++++++++---- run_tests.py | 14 ++++++++++ test/starter.py | 66 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 157 insertions(+), 5 deletions(-)
diff --git a/arm/starter.py b/arm/starter.py index 0131e97..7fe8d92 100644 --- a/arm/starter.py +++ b/arm/starter.py @@ -6,10 +6,14 @@ information. This is the starter for the application, handling and validating command line parameters. """
+import collections +import getopt + +import stem.util.connection + import os import sys import time -import getopt import getpass import locale import logging @@ -46,13 +50,9 @@ CONFIG = stem.util.conf.config_dict("arm", { "features.config.descriptions.persist": True, })
-OPT = "gi:s:c:dbe:vh" -OPT_EXPANDED = ["interface=", "socket=", "config=", "debug", "blind", "event=", "version", "help"] - HELP_MSG = """Usage arm [OPTION] Terminal status monitor for Tor relays.
- -p, --prompt only start the control interpretor -i, --interface [ADDRESS:]PORT change control interface from %s:%i -s, --socket SOCKET_PATH attach using unix domain socket if present, SOCKET_PATH defaults to: %s @@ -99,6 +99,78 @@ ARM_ROOT_NOTICE = "Arm is currently running with root permissions. This is not a
os.putenv("LANG", "C")
+# 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, + 'control_socket': '/var/run/tor/control', + 'config': None, + 'debug': False, + 'blind': False, + 'logged_events': 'N3', + 'print_version': False, + 'print_help': False, +} + +OPT = "gi:s:c:dbe:vh" +OPT_EXPANDED = ["interface=", "socket=", "config=", "debug", "blind", "event=", "version", "help"] + + +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) + elif opt in ("-s", "--socket"): + args['control_socket'] = arg + 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 allowConnectionTypes(): """ This provides a tuple with booleans indicating if we should or shouldn't diff --git a/run_tests.py b/run_tests.py new file mode 100755 index 0000000..2fb88de --- /dev/null +++ b/run_tests.py @@ -0,0 +1,14 @@ +#!/usr/bin/env python +# Copyright 2013, Damian Johnson and The Tor Project +# See LICENSE for licensing information + +""" +Runs arm's unit tests. This is a curses application so we're pretty limited on +the test coverage we can achieve, but exercising what we can. +""" + +import unittest + +tests = unittest.defaultTestLoader.discover('test', pattern='*.py') +test_runner = unittest.TextTestRunner() +test_runner.run(tests) diff --git a/test/starter.py b/test/starter.py new file mode 100644 index 0000000..d4720b1 --- /dev/null +++ b/test/starter.py @@ -0,0 +1,66 @@ +""" +Unit tests for arm's initialization module. +""" + +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_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]) +