[tor-commits] [stem/master] Assertion helpers for async tests
atagar at torproject.org
atagar at torproject.org
Thu Jun 8 17:17:55 UTC 2017
commit 5b18cd5958ab248c4376d0df4e68bba8fe651229
Author: Damian Johnson <atagar at torproject.org>
Date: Thu Jun 8 10:00:41 2017 -0700
Assertion helpers for async tests
Asynchronously run tests don't have access to the TestCase self reference which
provides assertion methods. As such providing functions that do the same thing.
---
stem/util/test_tools.py | 48 ++++++++++++++++++++++
test/integ/process.py | 103 ++++++++++++++++--------------------------------
2 files changed, 81 insertions(+), 70 deletions(-)
diff --git a/stem/util/test_tools.py b/stem/util/test_tools.py
index 9f4fff4..6aecbd8 100644
--- a/stem/util/test_tools.py
+++ b/stem/util/test_tools.py
@@ -65,6 +65,54 @@ else:
SkipTest = unittest.case.SkipTest
+def assert_equal(expected, actual, msg = None):
+ """
+ Function form of a TestCase's assertEqual.
+
+ .. versionadded:: 1.6.0
+
+ :param object expected: expected value
+ :param object actual: actual value
+ :param str msg: message if assertion fails
+
+ :raises: **AssertionError** if values aren't equal
+ """
+
+ if expected != actual:
+ raise AssertionError("Expected '%s' but was '%s'" % (expected, actual) if msg is None else msg)
+
+
+def assert_in(expected, actual, msg = None):
+ """
+ Asserts that a given value is within this content.
+
+ .. versionadded:: 1.6.0
+
+ :param object expected: expected value
+ :param object actual: actual value
+ :param str msg: message if assertion fails
+
+ :raises: **AssertionError** if the expected value isn't in the actual
+ """
+
+ if expected not in actual:
+ raise AssertionError("Expected '%s' to be within '%s'" % (expected, actual) if msg is None else msg)
+
+
+def skip(msg):
+ """
+ Function form of a TestCase's skipTest.
+
+ .. versionadded:: 1.6.0
+
+ :param str msg: reason test is being skipped
+
+ :raises: **unittest.case.SkipTest** for this reason
+ """
+
+ raise SkipTest(msg)
+
+
def asynchronous(func):
test = stem.util.test_tools.AsyncTest(func)
ASYNC_TESTS['%s.%s' % (func.__module__, func.__name__)] = test
diff --git a/test/integ/process.py b/test/integ/process.py
index 9c23077..9ee96ad 100644
--- a/test/integ/process.py
+++ b/test/integ/process.py
@@ -26,7 +26,7 @@ import test
import test.require
from contextlib import contextmanager
-from stem.util.test_tools import asynchronous
+from stem.util.test_tools import asynchronous, assert_equal, assert_in, skip
try:
# added in python 3.3
@@ -114,10 +114,7 @@ class TestProcess(unittest.TestCase):
Check that 'tor --version' matches 'GETINFO version'.
"""
- version_output = run_tor(tor_cmd, '--version')
-
- if 'Tor version %s.\n' % test.tor_version() != version_output:
- raise AssertionError('Unexpected response: %s' % version_output)
+ assert_equal('Tor version %s.\n' % test.tor_version(), run_tor(tor_cmd, '--version'))
@asynchronous
def test_help_argument(tor_cmd):
@@ -130,8 +127,7 @@ class TestProcess(unittest.TestCase):
if not help_output.startswith('Copyright (c) 2001') or not help_output.endswith('tor -f <torrc> [args]\nSee man page for options, or https://www.torproject.org/ for documentation.\n'):
raise AssertionError("Help output didn't have the expected strings: %s" % help_output)
- if help_output != run_tor(tor_cmd, '-h'):
- raise AssertionError("'tor -h' should simply be an alias for 'tor --help'")
+ assert_equal(help_output, run_tor(tor_cmd, '-h'), "'tor -h' should simply be an alias for 'tor --help'")
@asynchronous
def test_quiet_argument(tor_cmd):
@@ -139,8 +135,8 @@ class TestProcess(unittest.TestCase):
Check that we don't provide anything on stdout when running 'tor --quiet'.
"""
- if '' != run_tor(tor_cmd, '--quiet', '--invalid_argument', 'true', expect_failure = True):
- raise AssertionError('No output should be provided with the --quiet argument')
+ quiet_output = run_tor(tor_cmd, '--quiet', '--invalid_argument', 'true', expect_failure = True)
+ assert_equal('', quiet_output, 'No output should be provided with the --quiet argument')
@asynchronous
def test_hush_argument(tor_cmd):
@@ -149,14 +145,10 @@ class TestProcess(unittest.TestCase):
"""
output = run_tor(tor_cmd, '--hush', '--invalid_argument', expect_failure = True)
-
- if "[warn] Command-line option '--invalid_argument' with no value. Failing." not in output:
- raise AssertionError('Unexpected response: %s' % output)
+ assert_in("[warn] Command-line option '--invalid_argument' with no value. Failing.", output)
output = run_tor(tor_cmd, '--hush', '--invalid_argument', 'true', expect_failure = True)
-
- if "[warn] Failed to parse/validate config: Unknown option 'invalid_argument'. Failing." not in output:
- raise AssertionError('Unexpected response: %s' % output)
+ assert_in("[warn] Failed to parse/validate config: Unknown option 'invalid_argument'. Failing.", output)
@asynchronous
def test_hash_password(tor_cmd):
@@ -185,9 +177,7 @@ class TestProcess(unittest.TestCase):
stuff = salt + b'my_password'
repetitions = count // len(stuff) + 1
inp = (stuff * repetitions)[:count]
-
- if hashlib.sha1(inp).digest() != hashed:
- raise AssertionError('Password hash not what we expected (%s rather than %s)' % (hashlib.sha1(inp).digest(), hashed))
+ assert_equal(hashlib.sha1(inp).digest(), hashed)
@asynchronous
def test_hash_password_requires_argument(tor_cmd):
@@ -197,9 +187,7 @@ class TestProcess(unittest.TestCase):
"""
output = run_tor(tor_cmd, '--hash-password', expect_failure = True)
-
- if "[warn] Command-line option '--hash-password' with no value. Failing." not in output:
- raise AssertionError("'tor --hash-password' should require an argument")
+ assert_in("[warn] Command-line option '--hash-password' with no value. Failing.", output)
@asynchronous
def test_dump_config_argument(tor_cmd):
@@ -212,14 +200,9 @@ class TestProcess(unittest.TestCase):
full_output = run_tor(tor_cmd, '--dump-config', 'full', with_torrc = True)
run_tor(tor_cmd, '--dump-config', 'invalid_option', with_torrc = True, expect_failure = True)
- if 'Nickname stemIntegTest' not in short_output:
- raise AssertionError("Dumping short config options didn't include our nickname: %s" % short_output)
-
- if 'Nickname stemIntegTest' not in non_builtin_output:
- raise AssertionError("Dumping non-builtin config options didn't include our nickname: %s" % non_builtin_output)
-
- if 'Nickname stemIntegTest' not in full_output:
- raise AssertionError("Dumping full config options didn't include our nickname: %s" % full_output)
+ assert_in('Nickname stemIntegTest', short_output)
+ assert_in('Nickname stemIntegTest', non_builtin_output)
+ assert_in('Nickname stemIntegTest', full_output)
@asynchronous
def test_validate_config_argument(tor_cmd):
@@ -228,10 +211,7 @@ class TestProcess(unittest.TestCase):
"""
valid_output = run_tor(tor_cmd, '--verify-config', with_torrc = True)
-
- if 'Configuration was valid\n' not in valid_output:
- raise AssertionError('Expected configuration to be valid')
-
+ assert_in('Configuration was valid\n', valid_output, 'Expected configuration to be valid')
run_tor(tor_cmd, '--verify-config', '-f', __file__, expect_failure = True)
@asynchronous
@@ -243,9 +223,7 @@ class TestProcess(unittest.TestCase):
# This command should only work with a relay (which our test instance isn't).
output = run_tor(tor_cmd, '--list-fingerprint', with_torrc = True, expect_failure = True)
-
- if "Clients don't have long-term identity keys. Exiting." not in output:
- raise AssertionError('Should fail to start due to lacking an ORPort')
+ assert_in("Clients don't have long-term identity keys. Exiting.", output, 'Should fail to start due to lacking an ORPort')
with tmp_directory() as data_directory:
torrc_path = os.path.join(data_directory, 'torrc')
@@ -257,19 +235,15 @@ class TestProcess(unittest.TestCase):
nickname, fingerprint_with_spaces = output.splitlines()[-1].split(' ', 1)
fingerprint = fingerprint_with_spaces.replace(' ', '')
- if 'stemIntegTest' != nickname:
- raise AssertionError("Nickname should be 'stemIntegTest': %s" % nickname)
- elif 49 != len(fingerprint_with_spaces):
- raise AssertionError('There should be 49 components in our fingerprint: %i' % len(fingerprint_with_spaces))
- elif not stem.util.tor_tools.is_valid_fingerprint(fingerprint):
+ assert_equal('stemIntegTest', nickname)
+ assert_equal(49, len(fingerprint_with_spaces))
+
+ if not stem.util.tor_tools.is_valid_fingerprint(fingerprint):
raise AssertionError('We should have a valid fingerprint: %s' % fingerprint)
with open(os.path.join(data_directory, 'fingerprint')) as fingerprint_file:
expected = 'stemIntegTest %s\n' % fingerprint
- fingerprint_file_content = fingerprint_file.read()
-
- if expected != fingerprint_file_content:
- raise AssertionError('Unexpected fingerprint file: %s' % fingerprint_file_content)
+ assert_equal(expected, fingerprint_file.read())
@asynchronous
def test_list_torrc_options_argument(tor_cmd):
@@ -333,8 +307,7 @@ class TestProcess(unittest.TestCase):
'SocksPort 9090',
]
- if expected != result:
- raise AssertionError("Unexpected output from 'tor -f torrc --dump-config short': %s" % result)
+ assert_equal(expected, result)
@asynchronous
def test_torrc_arguments_via_stdin(tor_cmd):
@@ -343,14 +316,12 @@ class TestProcess(unittest.TestCase):
"""
if test.tor_version() < stem.version.Requirement.TORRC_VIA_STDIN:
- raise stem.util.test_tools.SkipTest('(requires )' % stem.version.Requirement.TORRC_VIA_STDIN)
+ skip('(requires )' % stem.version.Requirement.TORRC_VIA_STDIN)
with tmp_directory() as data_directory:
torrc = BASIC_RELAY_TORRC % data_directory
output = run_tor(tor_cmd, '-f', '-', '--dump-config', 'short', stdin = torrc)
-
- if sorted(torrc.splitlines()) != sorted(output.splitlines()):
- raise AssertionError("Unexpected output from 'tor -f - --dump-config short': %s" % output)
+ assert_equal(sorted(torrc.splitlines()), sorted(output.splitlines()))
@asynchronous
def test_with_missing_torrc(tor_cmd):
@@ -359,14 +330,10 @@ class TestProcess(unittest.TestCase):
"""
output = run_tor(tor_cmd, '-f', '/path/that/really/shouldnt/exist', '--verify-config', expect_failure = True)
-
- if '[warn] Unable to open configuration file "/path/that/really/shouldnt/exist".' not in output:
- raise AssertionError('Tor refuse to read a non-existant torrc file')
+ assert_in('[warn] Unable to open configuration file "/path/that/really/shouldnt/exist".', output, 'Tor should refuse to read a non-existant torrc file')
output = run_tor(tor_cmd, '-f', '/path/that/really/shouldnt/exist', '--verify-config', '--ignore-missing-torrc')
-
- if '[notice] Configuration file "/path/that/really/shouldnt/exist" not present, using reasonable defaults.' not in output:
- raise AssertionError('Missing torrc should be allowed with --ignore-missing-torrc')
+ assert_in('[notice] Configuration file "/path/that/really/shouldnt/exist" not present, using reasonable defaults.', output, 'Missing torrc should be allowed with --ignore-missing-torrc')
@asynchronous
def test_can_run_multithreaded(tor_cmd):
@@ -411,9 +378,8 @@ class TestProcess(unittest.TestCase):
return raised_exc[0]
exc = launch_async_with_timeout(0.5)
-
- if type(exc) != OSError or str(exc) != 'Launching tor with a timeout can only be done in the main thread':
- raise AssertionError("Exception isn't what we expected: %s" % exc)
+ assert_equal(OSError, type(exc))
+ assert_equal('Launching tor with a timeout can only be done in the main thread', str(exc))
# We should launch successfully if no timeout is specified or we specify it
# to be 'None'.
@@ -456,8 +422,7 @@ class TestProcess(unittest.TestCase):
control_socket.send('GETCONF ControlPort')
getconf_response = control_socket.recv()
- if 'ControlPort=%s' % control_port != str(getconf_response):
- raise AssertionError('Expected tor to report its ControlPort as %s but was: %s' % (control_port, getconf_response))
+ assert_equal('ControlPort=%s' % control_port, str(getconf_response))
finally:
if control_socket:
control_socket.close()
@@ -473,7 +438,7 @@ class TestProcess(unittest.TestCase):
"""
if test.tor_version() < stem.version.Requirement.TORRC_VIA_STDIN:
- raise stem.util.test_tools.SkipTest('(requires )' % stem.version.Requirement.TORRC_VIA_STDIN)
+ skip('(requires )' % stem.version.Requirement.TORRC_VIA_STDIN)
with tmp_directory() as data_directory:
control_port = random_port()
@@ -497,8 +462,7 @@ class TestProcess(unittest.TestCase):
control_socket.send('GETCONF ControlPort')
getconf_response = control_socket.recv()
- if 'ControlPort=%s' % control_port != str(getconf_response):
- raise AssertionError('Expected tor to report its ControlPort as %s but was: %s' % (control_port, getconf_response))
+ assert_equal('ControlPort=%s' % control_port, str(getconf_response))
finally:
if control_socket:
control_socket.close()
@@ -533,8 +497,7 @@ class TestProcess(unittest.TestCase):
raise AssertionError('Tor should fail to launch')
except OSError as exc:
- if str(exc) != 'Process terminated: Failed to bind one of the listener ports.':
- raise AssertionError('Unexpected error response from tor: %s' % exc)
+ assert_equal('Process terminated: Failed to bind one of the listener ports.', str(exc))
@asynchronous
def test_launch_tor_with_timeout(tor_cmd):
@@ -570,9 +533,9 @@ class TestProcess(unittest.TestCase):
"""
if not stem.util.system.is_available('sleep'):
- raise stem.util.test_tools.SkipTest('(sleep unavailable)')
+ skip('(sleep unavailable)')
elif test.tor_version() < stem.version.Requirement.TAKEOWNERSHIP:
- raise stem.util.test_tools.SkipTest('(requires )' % stem.version.Requirement.TAKEOWNERSHIP)
+ skip('(requires )' % stem.version.Requirement.TAKEOWNERSHIP)
with tmp_directory() as data_directory:
sleep_process = subprocess.Popen(['sleep', '60'])
@@ -616,7 +579,7 @@ class TestProcess(unittest.TestCase):
"""
if test.tor_version() < stem.version.Requirement.TAKEOWNERSHIP:
- raise stem.util.test_tools.SkipTest('(requires )' % stem.version.Requirement.TAKEOWNERSHIP)
+ skip('(requires )' % stem.version.Requirement.TAKEOWNERSHIP)
with tmp_directory() as data_directory:
control_port = random_port()
More information about the tor-commits
mailing list