commit cfb2c383fc487f4a129345c5724b6a73d96c712d
Author: Damian Johnson <atagar(a)torproject.org>
Date: Sun Jan 22 23:03:41 2012 -0800
Implementing testing --no-color arg
An option for suppressing color escape sequences is important for when test
output is piped to a file (the escape characters then just look like
gibberish). Still... output is sooooo ugly without color. :(
Routing most of the testing stdout writes through the output module, and
either respecting or discarding formatting there based on if we had a
'--no-color' argument.
---
run_tests.py | 16 ++++++----
test/output.py | 43 ++++++++++++++++++---------
test/runner.py | 87 +++++++++++++++++++++----------------------------------
3 files changed, 70 insertions(+), 76 deletions(-)
diff --git a/run_tests.py b/run_tests.py
index b344111..c7dd91e 100755
--- a/run_tests.py
+++ b/run_tests.py
@@ -147,6 +147,8 @@ def load_user_configuration(test_config):
arg_overrides["argument.log"] = arg.upper()
elif opt in ("--tor"):
arg_overrides["argument.tor"] = arg
+ elif opt in ("--no-color"):
+ arg_overrides["argument.no_color"] = "true"
elif opt in ("-h", "--help"):
# Prints usage information and quits. This includes a listing of the
# valid integration targets.
@@ -202,7 +204,7 @@ if __name__ == '__main__':
load_user_configuration(test_config)
if not CONFIG["argument.unit"] and not CONFIG["argument.integ"]:
- print "Nothing to run (for usage provide --help)\n"
+ test.output.print_line("Nothing to run (for usage provide --help)\n")
sys.exit()
# if we have verbose logging then provide the testing config
@@ -289,12 +291,12 @@ if __name__ == '__main__':
if opt in test.runner.Torrc.keys():
torrc_opts.append(test.runner.Torrc[opt])
else:
- print "'%s' isn't a test.runner.Torrc enumeration" % opt
+ test.output.print_line("'%s' isn't a test.runner.Torrc enumeration" % opt)
sys.exit(1)
integ_runner.start(CONFIG["argument.tor"], extra_torrc_opts = torrc_opts)
- print term.format("Running tests...", term.Color.BLUE, term.Attr.BOLD)
+ test.output.print_line("Running tests...", term.Color.BLUE, term.Attr.BOLD)
print
for test_class in INTEG_TESTS:
@@ -317,7 +319,7 @@ if __name__ == '__main__':
for target in skip_targets:
req_version = stem.version.Requirement[CONFIG["target.prereq"][target]]
- print term.format("Unable to run target %s, this requires tor version %s" % (target, req_version), term.Color.RED, term.Attr.BOLD)
+ test.output.print_line("Unable to run target %s, this requires tor version %s" % (target, req_version), term.Color.RED, term.Attr.BOLD)
print
@@ -326,11 +328,11 @@ if __name__ == '__main__':
runtime_label = "(%i seconds)" % (time.time() - start_time)
if error_tracker.has_error_occured():
- print term.format("TESTING FAILED %s" % runtime_label, term.Color.RED, term.Attr.BOLD)
+ test.output.print_line("TESTING FAILED %s" % runtime_label, term.Color.RED, term.Attr.BOLD)
for line in error_tracker:
- print term.format(" %s" % line, term.Color.RED, term.Attr.BOLD)
+ test.output.print_line(" %s" % line, term.Color.RED, term.Attr.BOLD)
else:
- print term.format("TESTING PASSED %s" % runtime_label, term.Color.GREEN, term.Attr.BOLD)
+ test.output.print_line("TESTING PASSED %s" % runtime_label, term.Color.GREEN, term.Attr.BOLD)
print
diff --git a/test/output.py b/test/output.py
index 327482a..5ba9ca4 100644
--- a/test/output.py
+++ b/test/output.py
@@ -4,11 +4,19 @@ together for improved readability.
"""
import re
+import sys
import logging
+import stem.util.conf
import stem.util.enum
import stem.util.term as term
+CONFIG = {
+ "argument.no_color": False,
+}
+
+stem.util.conf.get_config("test").sync(CONFIG)
+
DIVIDER = "=" * 70
HEADER_ATTR = (term.Color.CYAN, term.Attr.BOLD)
CATEGORY_ATTR = (term.Color.GREEN, term.Attr.BOLD)
@@ -30,33 +38,37 @@ LINE_ATTR = {
LineType.CONTENT: (term.Color.CYAN,),
}
+def print_line(msg, *attr):
+ if CONFIG["argument.no_color"]: print msg
+ else: print term.format(msg, *attr)
+
+def print_noline(msg, *attr):
+ if CONFIG["argument.no_color"]: sys.stdout.write(msg)
+ else: sys.stdout.write(term.format(msg, *attr))
+
def print_divider(msg, is_header = False):
attr = HEADER_ATTR if is_header else CATEGORY_ATTR
- print term.format("%s\n%s\n%s\n" % (DIVIDER, msg.center(70), DIVIDER), *attr)
+ print_line("%s\n%s\n%s\n" % (DIVIDER, msg.center(70), DIVIDER), *attr)
def print_logging(logging_buffer):
if not logging_buffer.is_empty():
for entry in logging_buffer:
- print term.format(entry.replace("\n", "\n "), term.Color.MAGENTA)
+ print_line(entry.replace("\n", "\n "), term.Color.MAGENTA)
print
def print_config(test_config):
print_divider("TESTING CONFIG", True)
+ print_line("Test configuration... ", term.Color.BLUE, term.Attr.BOLD)
- try:
- print term.format("Test configuration... ", term.Color.BLUE, term.Attr.BOLD)
+ for config_key in test_config.keys():
+ key_entry = " %s => " % config_key
- for config_key in test_config.keys():
- key_entry = " %s => " % config_key
-
- # if there's multiple values then list them on separate lines
- value_div = ",\n" + (" " * len(key_entry))
- value_entry = value_div.join(test_config.get_value(config_key, multiple = True))
-
- print term.format(key_entry + value_entry, term.Color.BLUE)
- except IOError, exc:
- sys.stdout.write(term.format("failed (%s)\n" % exc, term.Color.RED, term.Attr.BOLD))
+ # if there's multiple values then list them on separate lines
+ value_div = ",\n" + (" " * len(key_entry))
+ value_entry = value_div.join(test_config.get_value(config_key, multiple = True))
+
+ print_line(key_entry + value_entry, term.Color.BLUE)
print
@@ -102,7 +114,8 @@ def colorize(line_type, line_content):
Applies escape sequences so each line is colored according to its type.
"""
- return term.format(line_content, *LINE_ATTR[line_type])
+ if CONFIG["argument.no_color"]: return line_content
+ else: return term.format(line_content, *LINE_ATTR[line_type])
def strip_module(line_type, line_content):
"""
diff --git a/test/runner.py b/test/runner.py
index f7a78dc..405f96c 100644
--- a/test/runner.py
+++ b/test/runner.py
@@ -39,6 +39,7 @@ import stem.version
import stem.util.conf
import stem.util.enum
import stem.util.term as term
+import test.output
CONFIG = {
"integ.test_directory": "./test/data",
@@ -121,7 +122,7 @@ class Runner:
self._custom_opts = None
self._tor_process = None
- def start(self, tor_cmd, extra_torrc_opts, quiet = False):
+ def start(self, tor_cmd, extra_torrc_opts):
"""
Makes temporary testing resources and starts tor, blocking until it
completes.
@@ -129,8 +130,6 @@ class Runner:
Arguments:
tor_cmd (str) - command to start tor with
extra_torrc_opts (list) - additional torrc options for our test instance
- quiet (bool) - if False then this prints status information as we start
- up to stdout
Raises:
OSError if unable to run test preparations or start tor
@@ -140,9 +139,9 @@ class Runner:
# if we're holding on to a tor process (running or not) then clean up after
# it so we can start a fresh instance
- if self._tor_process: self.stop(quiet)
+ if self._tor_process: self.stop()
- _print_status("Setting up a test instance...\n", STATUS_ATTR, quiet)
+ test.output.print_line("Setting up a test instance...", *STATUS_ATTR)
# if 'test_directory' is unset then we make a new data directory in /tmp
# and clean it up when we're done
@@ -171,28 +170,25 @@ class Runner:
try:
self._tor_cwd = os.getcwd()
- self._run_setup(quiet)
- self._start_tor(tor_cmd, quiet)
+ self._run_setup()
+ self._start_tor(tor_cmd)
# revert our cwd back to normal
if CONFIG["test.target.relative_data_dir"]:
os.chdir(original_cwd)
except OSError, exc:
- self.stop(quiet)
+ self.stop()
raise exc
finally:
self._runner_lock.release()
- def stop(self, quiet = False):
+ def stop(self):
"""
Stops our tor test instance and cleans up any temporary resources.
-
- Argument:
- quiet (bool) - prints status information to stdout if False
"""
self._runner_lock.acquire()
- _print_status("Shutting down tor... ", STATUS_ATTR, quiet)
+ test.output.print_noline("Shutting down tor... ", *STATUS_ATTR)
if self._tor_process:
self._tor_process.kill()
@@ -208,7 +204,7 @@ class Runner:
self._custom_opts = None
self._tor_process = None
- _print_status("done\n", STATUS_ATTR, quiet)
+ test.output.print_line("done", *STATUS_ATTR)
self._runner_lock.release()
def is_running(self):
@@ -423,28 +419,25 @@ class Runner:
finally:
self._runner_lock.release()
- def _run_setup(self, quiet):
+ def _run_setup(self):
"""
Makes a temporary runtime resources of our integration test instance.
- Arguments:
- quiet (bool) - prints status information to stdout if False
-
Raises:
OSError if unsuccessful
"""
# makes a temporary data directory if needed
try:
- _print_status(" making test directory (%s)... " % self._test_dir, STATUS_ATTR, quiet)
+ test.output.print_noline(" making test directory (%s)... " % self._test_dir, *STATUS_ATTR)
if os.path.exists(self._test_dir):
- _print_status("skipped\n", STATUS_ATTR, quiet)
+ test.output.print_line("skipped", *STATUS_ATTR)
else:
os.makedirs(self._test_dir)
- _print_status("done\n", STATUS_ATTR, quiet)
+ test.output.print_line("done", *STATUS_ATTR)
except OSError, exc:
- _print_status("failed (%s)\n" % exc, ERROR_ATTR, quiet)
+ test.output.print_line("failed (%s)" % exc, *ERROR_ATTR)
raise exc
# Makes a directory for the control socket if needed. As of, at least, Tor
@@ -455,18 +448,18 @@ class Runner:
if Torrc.SOCKET in self._custom_opts:
try:
socket_dir = os.path.dirname(CONTROL_SOCKET_PATH)
- _print_status(" making control socket directory (%s)... " % socket_dir, STATUS_ATTR, quiet)
+ test.output.print_noline(" making control socket directory (%s)... " % socket_dir, *STATUS_ATTR)
if os.path.exists(socket_dir) and stat.S_IMODE(os.stat(socket_dir).st_mode) == 0700:
- _print_status("skipped\n", STATUS_ATTR, quiet)
+ test.output.print_line("skipped", *STATUS_ATTR)
else:
if not os.path.exists(socket_dir):
os.makedirs(socket_dir)
os.chmod(socket_dir, 0700)
- _print_status("done\n", STATUS_ATTR, quiet)
+ test.output.print_line("done", *STATUS_ATTR)
except OSError, exc:
- _print_status("failed (%s)\n" % exc, ERROR_ATTR, quiet)
+ test.output.print_line("failed (%s)" % exc, *ERROR_ATTR)
raise exc
# configures logging
@@ -474,7 +467,7 @@ class Runner:
if logging_path:
logging_path = stem.util.system.expand_path(logging_path, STEM_BASE)
- _print_status(" configuring logger (%s)... " % logging_path, STATUS_ATTR, quiet)
+ test.output.print_noline(" configuring logger (%s)... " % logging_path, *STATUS_ATTR)
# delete the old log
if os.path.exists(logging_path):
@@ -487,44 +480,43 @@ class Runner:
datefmt = '%D %H:%M:%S',
)
- _print_status("done\n", STATUS_ATTR, quiet)
+ test.output.print_line("done", *STATUS_ATTR)
else:
- _print_status(" configuring logger... skipped\n", STATUS_ATTR, quiet)
+ test.output.print_line(" configuring logger... skipped", *STATUS_ATTR)
# writes our testing torrc
torrc_dst = os.path.join(self._test_dir, "torrc")
try:
- _print_status(" writing torrc (%s)... " % torrc_dst, STATUS_ATTR, quiet)
+ test.output.print_noline(" writing torrc (%s)... " % torrc_dst, *STATUS_ATTR)
torrc_file = open(torrc_dst, "w")
torrc_file.write(self._torrc_contents)
torrc_file.close()
- _print_status("done\n", STATUS_ATTR, quiet)
+ test.output.print_line("done", *STATUS_ATTR)
for line in self._torrc_contents.strip().splitlines():
- _print_status(" %s\n" % line.strip(), SUBSTATUS_ATTR, quiet)
+ test.output.print_line(" %s" % line.strip(), *SUBSTATUS_ATTR)
- _print_status("\n", (), quiet)
+ print
except Exception, exc:
- _print_status("failed (%s)\n\n" % exc, ERROR_ATTR, quiet)
+ test.output.print_line("failed (%s)\n" % exc, *ERROR_ATTR)
raise OSError(exc)
- def _start_tor(self, tor_cmd, quiet):
+ def _start_tor(self, tor_cmd):
"""
Initializes a tor process. This blocks until initialization completes or we
error out.
Arguments:
tor_cmd (str) - command to start tor with
- quiet (bool) - prints status information to stdout if False
Raises:
OSError if we either fail to create the tor process or reached a timeout
without success
"""
- _print_status("Starting tor...\n", STATUS_ATTR, quiet)
+ test.output.print_line("Starting tor...\n", *STATUS_ATTR)
start_time = time.time()
try:
@@ -533,30 +525,17 @@ class Runner:
complete_percent = 100 if CONFIG["test.target.online"] else 5
# prints output from tor's stdout while it starts up
- print_init_line = lambda line: _print_status(" %s\n" % line, SUBSTATUS_ATTR, quiet)
+ print_init_line = lambda line: test.output.print_line(" %s" % line, *SUBSTATUS_ATTR)
torrc_dst = os.path.join(self._test_dir, "torrc")
self._tor_process = stem.process.launch_tor(tor_cmd, torrc_dst, complete_percent, print_init_line)
runtime = time.time() - start_time
- _print_status(" done (%i seconds)\n\n" % runtime, STATUS_ATTR, quiet)
+ test.output.print_line(" done (%i seconds)\n" % runtime, *STATUS_ATTR)
except KeyboardInterrupt:
- _print_status(" aborted starting tor: keyboard interrupt\n\n", ERROR_ATTR, quiet)
+ test.output.print_line(" aborted starting tor: keyboard interrupt\n", *ERROR_ATTR)
raise OSError("keyboard interrupt")
except OSError, exc:
- _print_status(" failed to start tor: %s\n\n" % exc, ERROR_ATTR, quiet)
+ test.output.print_line(" failed to start tor: %s\n" % exc, *ERROR_ATTR)
raise exc
-def _print_status(msg, attr = (), quiet = False):
- """
- Short alias for printing status messages.
-
- Arguments:
- msg (str) - text to be printed
- attr (tuple) - list of term attributes to be applied to the text
- quiet (bool) - no-op if true, prints otherwise
- """
-
- if not quiet:
- sys.stdout.write(term.format(msg, *attr))
-
