commit 0ee3864a784ce8e116466800329863d37ec35400 Author: Damian Johnson atagar@torproject.org Date: Thu Apr 6 11:19:05 2017 -0700
Use decorators for common test requirements
Expanding our use of decorators for testing requirements. This doesn't eliminate direct use of skipTest(), but restrict it to uncommon testing needs. --- test/integ/process.py | 7 +- test/integ/util/proc.py | 39 +++----- test/integ/util/system.py | 247 ++++++++++++++-------------------------------- test/integ/version.py | 10 +- test/unit/manual.py | 19 ++-- test/util.py | 14 +++ 6 files changed, 117 insertions(+), 219 deletions(-)
diff --git a/test/integ/process.py b/test/integ/process.py index e77fdfb..b81a3bf 100644 --- a/test/integ/process.py +++ b/test/integ/process.py @@ -23,6 +23,7 @@ import stem.version import test.runner
from test.util import ( + require_command, require_controller, require_version, ) @@ -198,6 +199,7 @@ class TestProcess(unittest.TestCase): self.assertTrue('UseBridges' in output) self.assertTrue('SocksPort' in output)
+ @require_command('sleep') @patch('re.compile', Mock(side_effect = KeyboardInterrupt('nope'))) def test_no_orphaned_process(self): """ @@ -430,6 +432,7 @@ class TestProcess(unittest.TestCase): if not (runtime > 0.05 and runtime < 1): self.fail('Test should have taken 0.05-1 seconds, took %0.1f instead' % runtime)
+ @require_command('sleep') @require_version(stem.version.Requirement.TAKEOWNERSHIP) @only_run_once @patch('os.getpid') @@ -439,10 +442,6 @@ class TestProcess(unittest.TestCase): test this we spawn a process and trick tor into thinking that it is us. """
- if not stem.util.system.is_available('sleep'): - self.skipTest("('sleep' command is unavailable)") - return - sleep_process = subprocess.Popen(['sleep', '60']) getpid_mock.return_value = str(sleep_process.pid)
diff --git a/test/integ/util/proc.py b/test/integ/util/proc.py index a7e798d..ef3a7c0 100644 --- a/test/integ/util/proc.py +++ b/test/integ/util/proc.py @@ -10,45 +10,39 @@ import test.runner
from stem.util import proc
+from test.util import ( + require_proc, + require_ptrace, +) +
class TestProc(unittest.TestCase): + @require_proc + @require_ptrace def test_cwd(self): """ Checks that stem.util.proc.cwd matches our tor instance's cwd. """
- if not proc.is_available(): - self.skipTest('(proc unavailable)') - return - elif not test.runner.get_runner().is_ptraceable(): - self.skipTest('(DisableDebuggerAttachment is set)') - return - runner = test.runner.get_runner() runner_pid, tor_cwd = runner.get_pid(), runner.get_tor_cwd() self.assertEqual(tor_cwd, proc.cwd(runner_pid))
+ @require_proc def test_uid(self): """ Checks that stem.util.proc.uid matches our tor instance's uid. """
- if not proc.is_available(): - self.skipTest('(proc unavailable)') - return - tor_pid = test.runner.get_runner().get_pid() self.assertEqual(os.geteuid(), proc.uid(tor_pid))
+ @require_proc def test_memory_usage(self): """ Checks that stem.util.proc.memory_usage looks somewhat reasonable. """
- if not proc.is_available(): - self.skipTest('(proc unavailable)') - return - tor_pid = test.runner.get_runner().get_pid() res_size, vir_size = proc.memory_usage(tor_pid)
@@ -56,15 +50,12 @@ class TestProc(unittest.TestCase): self.assertTrue(res_size > 1024) self.assertTrue(vir_size > 1024)
+ @require_proc def test_stats(self): """ Checks that stem.util.proc.stats looks somewhat reasonable. """
- if not proc.is_available(): - self.skipTest('(proc unavailable)') - return - tor_cmd = test.runner.get_runner().get_tor_command(True) tor_pid = test.runner.get_runner().get_pid() command, utime, stime, start_time = proc.stats(tor_pid, 'command', 'utime', 'stime', 'start time') @@ -74,6 +65,8 @@ class TestProc(unittest.TestCase): self.assertTrue(float(stime) >= 0) self.assertTrue(float(start_time) > proc.system_start_time())
+ @require_proc + @require_ptrace def test_connections(self): """ Checks for our control port in the stem.util.proc.connections output if @@ -82,15 +75,9 @@ class TestProc(unittest.TestCase):
runner = test.runner.get_runner()
- if not proc.is_available(): - self.skipTest('(proc unavailable)') - return - elif test.runner.Torrc.PORT not in runner.get_options(): + if test.runner.Torrc.PORT not in runner.get_options(): self.skiTestp('(no control port)') return - elif not test.runner.get_runner().is_ptraceable(): - self.skipTest('(DisableDebuggerAttachment is set)') - return elif not os.access('/proc/net/tcp', os.R_OK) or not os.access('/proc/net/udp', os.R_OK): self.skipTest('(proc lacks read permissions)') return diff --git a/test/integ/util/system.py b/test/integ/util/system.py index 973c3cb..88a5529 100644 --- a/test/integ/util/system.py +++ b/test/integ/util/system.py @@ -12,6 +12,13 @@ import stem.util.proc import stem.util.system import test.runner
+from test.util import ( + require, + require_command, + require_proc, + require_ptrace, +) + try: # added in python 3.3 from unittest.mock import Mock, patch @@ -36,6 +43,23 @@ def filter_system_call(prefixes): return _filter_system_call
+def _is_single_tor_running(): + if stem.util.system.is_windows(): + return True # TODO: not sure how to check for this on windows + elif not stem.util.system.is_bsd(): + tor_cmd = test.runner.get_runner().get_tor_command(True) + pgrep_results = stem.util.system.call(stem.util.system.GET_PID_BY_NAME_PGREP % tor_cmd) + return len(pgrep_results) == 1 + else: + ps_results = stem.util.system.call(stem.util.system.GET_PID_BY_NAME_PS_BSD) + results = [r for r in ps_results if r.endswith(' tor')] + return len(results) == 1 + + +def _is_linux(): + return not stem.util.system.is_bsd() and not stem.util.system.is_windows() + + def _has_port(): """ True if our test runner has a control port, False otherwise. @@ -44,6 +68,12 @@ def _has_port(): return test.runner.Torrc.PORT in test.runner.get_runner().get_options()
+require_single_tor_instance = require(_is_single_tor_running, 'multiple tor instances') +require_control_port = require(_has_port, 'test instance has no port') +require_linux = require(_is_linux, 'linux only') +require_bsd = require(stem.util.system.is_bsd, 'bsd only') + + class TestSystem(unittest.TestCase): def test_is_available(self): """ @@ -61,15 +91,12 @@ class TestSystem(unittest.TestCase):
self.assertFalse(stem.util.system.is_available('blarg_and_stuff'))
+ @require_command('ps') def test_is_running(self): """ Checks the stem.util.system.is_running function. """
- if not stem.util.system.is_available('ps'): - self.skipTest('(ps unavailable)') - return - # Check to see if the command we started tor with is running. The process # might be running under another name so need to check for 'tor.real' too # (#15449). @@ -78,35 +105,26 @@ class TestSystem(unittest.TestCase): self.assertTrue(stem.util.system.is_running(tor_cmd) or stem.util.system.is_running('tor.real')) self.assertFalse(stem.util.system.is_running('blarg_and_stuff'))
+ @require_single_tor_instance def test_pid_by_name(self): """ Checks general usage of the stem.util.system.pid_by_name function. This will fail if there's other tor instances running. """
- if self._is_extra_tor_running(): - self.skipTest('(multiple tor instances)') - return - tor_pid = test.runner.get_runner().get_pid() tor_cmd = test.runner.get_runner().get_tor_command(True) self.assertEqual(tor_pid, stem.util.system.pid_by_name(tor_cmd)) self.assertEqual(None, stem.util.system.pid_by_name('blarg_and_stuff'))
+ @require_command('pgrep') + @require_single_tor_instance def test_pid_by_name_pgrep(self): """ Tests the pid_by_name function with a pgrep response. """
- if self._is_extra_tor_running(): - self.skipTest('(multiple tor instances)') - return - elif not stem.util.system.is_available('pgrep'): - self.skipTest('(pgrep unavailable)') - return - pgrep_prefix = stem.util.system.GET_PID_BY_NAME_PGREP % '' - call_replacement = filter_system_call([pgrep_prefix])
with patch('stem.util.system.call') as call_mock: @@ -116,20 +134,14 @@ class TestSystem(unittest.TestCase): tor_cmd = test.runner.get_runner().get_tor_command(True) self.assertEqual(tor_pid, stem.util.system.pid_by_name(tor_cmd))
+ @require_command('pidof') + @require_single_tor_instance def test_pid_by_name_pidof(self): """ Tests the pid_by_name function with a pidof response. """
- if self._is_extra_tor_running(): - self.skipTest('(multiple tor instances)') - return - elif not stem.util.system.is_available('pidof'): - self.skipTest('(pidof unavailable)') - return - pidof_prefix = stem.util.system.GET_PID_BY_NAME_PIDOF % '' - call_replacement = filter_system_call([pidof_prefix])
with patch('stem.util.system.call') as call_mock: @@ -139,23 +151,15 @@ class TestSystem(unittest.TestCase): tor_cmd = test.runner.get_runner().get_tor_command() self.assertEqual(tor_pid, stem.util.system.pid_by_name(tor_cmd))
+ @require_linux + @require_command('ps') + @require_single_tor_instance def test_pid_by_name_ps_linux(self): """ Tests the pid_by_name function with the linux variant of ps. """
- if self._is_extra_tor_running(): - self.skipTest('(multiple tor instances)') - return - elif not stem.util.system.is_available('ps'): - self.skipTest('(ps unavailable)') - return - elif stem.util.system.is_bsd(): - self.skipTest('(linux only)') - return - ps_prefix = stem.util.system.GET_PID_BY_NAME_PS_LINUX % '' - call_replacement = filter_system_call([ps_prefix])
with patch('stem.util.system.call') as call_mock: @@ -165,23 +169,15 @@ class TestSystem(unittest.TestCase): tor_cmd = test.runner.get_runner().get_tor_command(True) self.assertEqual(tor_pid, stem.util.system.pid_by_name(tor_cmd))
+ @require_bsd + @require_command('ps') + @require_single_tor_instance def test_pid_by_name_ps_bsd(self): """ Tests the pid_by_name function with the bsd variant of ps. """
- if self._is_extra_tor_running(): - self.skipTest('(multiple tor instances)') - return - elif not stem.util.system.is_available('ps'): - self.skipTest('(ps unavailable)') - return - elif not stem.util.system.is_bsd(): - self.skipTest('(bsd only)') - return - ps_prefix = stem.util.system.GET_PID_BY_NAME_PS_BSD - call_replacement = filter_system_call([ps_prefix])
with patch('stem.util.system.call') as call_mock: @@ -191,24 +187,15 @@ class TestSystem(unittest.TestCase): tor_cmd = test.runner.get_runner().get_tor_command(True) self.assertEqual(tor_pid, stem.util.system.pid_by_name(tor_cmd))
+ @require_ptrace + @require_command('lsof') + @require_single_tor_instance def test_pid_by_name_lsof(self): """ Tests the pid_by_name function with a lsof response. """
- runner = test.runner.get_runner() - if self._is_extra_tor_running(): - self.skipTest('(multiple tor instances)') - return - elif not stem.util.system.is_available('lsof'): - self.skipTest('(lsof unavailable)') - return - elif not runner.is_ptraceable(): - self.skipTest('(DisableDebuggerAttachment is set)') - return - lsof_prefix = stem.util.system.GET_PID_BY_NAME_LSOF % '' - call_replacement = filter_system_call([lsof_prefix])
with patch('stem.util.system.call') as call_mock: @@ -221,71 +208,52 @@ class TestSystem(unittest.TestCase): if len(all_tor_pids) == 1: self.assertEqual(our_tor_pid, all_tor_pids[0])
+ @require_command('tasklist') + @require_single_tor_instance def test_pid_by_name_tasklist(self): """ Tests the pid_by_name function with a tasklist response. """
- if self._is_extra_tor_running(): - self.skipTest('(multiple tor instances)') - return - elif not stem.util.system.is_available('tasklist'): - self.skipTest('(tasklist unavailable)') - return - runner = test.runner.get_runner() self.assertEqual(runner.get_pid(), stem.util.system.pid_by_name(runner.get_tor_command(True)))
+ @require_ptrace + @require_control_port def test_pid_by_port(self): """ Checks general usage of the stem.util.system.pid_by_port function. """
- runner = test.runner.get_runner() if stem.util.system.is_windows(): self.skipTest('(unavailable on windows)') return - elif not _has_port(): - self.skipTest('(test instance has no port)') - return elif stem.util.system.is_mac() or stem.util.system.is_gentoo(): self.skipTest('(resolvers unavailable)') return - elif not runner.is_ptraceable(): - self.skipTest('(DisableDebuggerAttachment is set)') - return elif not (stem.util.system.is_available('netstat') or stem.util.system.is_available('sockstat') or stem.util.system.is_available('lsof')): self.skipTest('(connection resolvers unavailable)') return
+ runner = test.runner.get_runner() tor_pid, tor_port = runner.get_pid(), test.runner.CONTROL_PORT self.assertEqual(tor_pid, stem.util.system.pid_by_port(tor_port)) self.assertEqual(None, stem.util.system.pid_by_port(99999))
+ @require_linux + @require_ptrace + @require_control_port + @require_command('netstat') def test_pid_by_port_netstat(self): """ Tests the pid_by_port function with a netstat response. """
- runner = test.runner.get_runner() - - if not _has_port(): - self.skipTest('(test instance has no port)') - return - elif not stem.util.system.is_available('netstat'): - self.skipTest('(netstat unavailable)') - return - elif stem.util.system.is_bsd() or stem.util.system.is_windows(): - self.skipTest('(linux only)') - return - elif stem.util.system.is_gentoo(): + if stem.util.system.is_gentoo(): self.skipTest('(unavailable on gentoo)') return - elif not runner.is_ptraceable(): - self.skipTest('(DisableDebuggerAttachment is set)') - return
netstat_prefix = stem.util.system.GET_PID_BY_PORT_NETSTAT
@@ -297,28 +265,16 @@ class TestSystem(unittest.TestCase): tor_pid = test.runner.get_runner().get_pid() self.assertEqual(tor_pid, stem.util.system.pid_by_port(test.runner.CONTROL_PORT))
+ @require_bsd + @require_ptrace + @require_control_port + @require_command('sockstat') def test_pid_by_port_sockstat(self): """ Tests the pid_by_port function with a sockstat response. """
- runner = test.runner.get_runner() - - if not _has_port(): - self.skipTest('(test instance has no port)') - return - elif not stem.util.system.is_available('sockstat'): - self.skipTest('(sockstat unavailable)') - return - elif not stem.util.system.is_bsd(): - self.skipTest('(bsd only)') - return - elif not runner.is_ptraceable(): - self.skipTest('(DisableDebuggerAttachment is set)') - return - sockstat_prefix = stem.util.system.GET_PID_BY_PORT_SOCKSTAT % '' - call_replacement = filter_system_call([sockstat_prefix])
with patch('stem.util.system.call') as call_mock: @@ -327,25 +283,17 @@ class TestSystem(unittest.TestCase): tor_pid = test.runner.get_runner().get_pid() self.assertEqual(tor_pid, stem.util.system.pid_by_port(test.runner.CONTROL_PORT))
+ @require_ptrace + @require_control_port + @require_command('lsof') def test_pid_by_port_lsof(self): """ Tests the pid_by_port function with a lsof response. """
- runner = test.runner.get_runner() - - if not _has_port(): - self.skipTest('(test instance has no port)') - return - elif not stem.util.system.is_available('lsof'): - self.skipTest('(lsof unavailable)') - return - elif stem.util.system.is_mac() or stem.util.system.is_gentoo(): + if stem.util.system.is_mac() or stem.util.system.is_gentoo(): self.skipTest('(resolvers unavailable)') return - elif not runner.is_ptraceable(): - self.skipTest('(DisableDebuggerAttachment is set)') - return
lsof_prefix = stem.util.system.GET_PID_BY_PORT_LSOF
@@ -380,38 +328,28 @@ class TestSystem(unittest.TestCase): pids = stem.util.system.pids_by_user(getpass.getuser()) self.assertTrue(os.getpid() in pids)
+ @require_ptrace def test_cwd(self): """ Checks general usage of the stem.util.system.cwd function. """
- runner = test.runner.get_runner() - if stem.util.system.is_windows(): self.skipTest('(unavailable on windows)') return - elif not runner.is_ptraceable(): - self.skipTest('(DisableDebuggerAttachment is set)') - return
+ runner = test.runner.get_runner() runner_pid, tor_cwd = runner.get_pid(), runner.get_tor_cwd() self.assertEqual(tor_cwd, stem.util.system.cwd(runner_pid)) self.assertEqual(None, stem.util.system.cwd(99999))
+ @require_ptrace + @require_command('pwdx') def test_cwd_pwdx(self): """ Tests the pid_by_cwd function with a pwdx response. """
- runner = test.runner.get_runner() - - if not stem.util.system.is_available('pwdx'): - self.skipTest('(pwdx unavailable)') - return - elif not runner.is_ptraceable(): - self.skipTest('(DisableDebuggerAttachment is set)') - return - # filter the call function to only allow this command
pwdx_prefix = stem.util.system.GET_CWD_PWDX % '' @@ -421,23 +359,17 @@ class TestSystem(unittest.TestCase): with patch('stem.util.system.call') as call_mock: call_mock.side_effect = call_replacement
+ runner = test.runner.get_runner() runner_pid, tor_cwd = runner.get_pid(), runner.get_tor_cwd() self.assertEqual(tor_cwd, stem.util.system.cwd(runner_pid))
+ @require_ptrace + @require_command('lsof') def test_cwd_lsof(self): """ Tests the pid_by_cwd function with a lsof response. """
- runner = test.runner.get_runner() - - if not stem.util.system.is_available('lsof'): - self.skipTest('(lsof unavailable)') - return - elif not runner.is_ptraceable(): - self.skipTest('(DisableDebuggerAttachment is set)') - return - # filter the call function to only allow this command
lsof_prefix = 'lsof -a -p ' @@ -447,6 +379,7 @@ class TestSystem(unittest.TestCase): with patch('stem.util.system.call') as call_mock: call_mock.side_effect = call_replacement
+ runner = test.runner.get_runner() runner_pid, tor_cwd = runner.get_pid(), runner.get_tor_cwd() self.assertEqual(tor_cwd, stem.util.system.cwd(runner_pid))
@@ -459,15 +392,12 @@ class TestSystem(unittest.TestCase): self.assertEqual(None, stem.util.system.user(-5)) self.assertEqual(None, stem.util.system.start_time(98765))
+ @require_proc def test_user_proc(self): """ Tests the user function with a proc response. """
- if not stem.util.proc.is_available(): - self.skipTest('(proc unavailable)') - return - call_replacement = filter_system_call(['ps '])
with patch('stem.util.system.call') as call_mock: @@ -478,16 +408,13 @@ class TestSystem(unittest.TestCase): pid = test.runner.get_runner().get_pid() self.assertTrue(getpass.getuser(), stem.util.system.user(pid))
+ @require_command('ps') @patch('stem.util.proc.is_available', Mock(return_value = False)) def test_user_ps(self): """ Tests the user function with a ps response. """
- if not stem.util.system.is_available('ps'): - self.skipTest('(ps unavailable)') - return - pid = test.runner.get_runner().get_pid() self.assertTrue(getpass.getuser(), stem.util.system.user(pid))
@@ -500,15 +427,12 @@ class TestSystem(unittest.TestCase): self.assertEqual(None, stem.util.system.start_time(-5)) self.assertEqual(None, stem.util.system.start_time(98765))
+ @require_proc def test_start_time_proc(self): """ Tests the start_time function with a proc response. """
- if not stem.util.proc.is_available(): - self.skipTest('(proc unavailable)') - return - call_replacement = filter_system_call(['ps '])
with patch('stem.util.system.call') as call_mock: @@ -517,16 +441,13 @@ class TestSystem(unittest.TestCase): pid = test.runner.get_runner().get_pid() self.assertTrue(stem.util.system.start_time(pid) >= 0)
+ @require_command('ps') @patch('stem.util.proc.is_available', Mock(return_value = False)) def test_start_time_ps(self): """ Tests the start_time function with a ps response. """
- if not stem.util.system.is_available('ps'): - self.skipTest('(ps unavailable)') - return - pid = test.runner.get_runner().get_pid() self.assertTrue(stem.util.system.start_time(pid) >= 0)
@@ -592,21 +513,3 @@ class TestSystem(unittest.TestCase): self.assertEqual('stem_integ', stem.util.system.get_process_name()) finally: stem.util.system.set_process_name(initial_name) - - def _is_extra_tor_running(self): - # Try to figure out if there's more than one tor instance running. This - # check will fail if pgrep is unavailable (for instance on bsd) but this - # isn't the end of the world. It's just used to skip tests if they should - # legitemately fail. - - if stem.util.system.is_windows(): - # TODO: not sure how to check for this on windows - return False - elif not stem.util.system.is_bsd(): - tor_cmd = test.runner.get_runner().get_tor_command(True) - pgrep_results = stem.util.system.call(stem.util.system.GET_PID_BY_NAME_PGREP % tor_cmd) - return len(pgrep_results) > 1 - else: - ps_results = stem.util.system.call(stem.util.system.GET_PID_BY_NAME_PS_BSD) - results = [r for r in ps_results if r.endswith(' tor')] - return len(results) > 1 diff --git a/test/integ/version.py b/test/integ/version.py index 35e016a..3cef9b2 100644 --- a/test/integ/version.py +++ b/test/integ/version.py @@ -9,19 +9,19 @@ import stem.prereq import stem.version import test.runner
-from test.util import require_controller +from test.util import ( + require_command, + require_controller, +)
class TestVersion(unittest.TestCase): + @require_command('tor') def test_get_system_tor_version(self): """ Basic verification checks for the get_system_tor_version() function. """
- if not stem.util.system.is_available('tor'): - self.skipTest("(tor isn't in our path)") - return - # Since tor is in our path we should expect to be able to get the version # that way, though this might not belong to our test instance (if we're # running against a specific tor binary). diff --git a/test/unit/manual.py b/test/unit/manual.py index 6d0c174..9e6904e 100644 --- a/test/unit/manual.py +++ b/test/unit/manual.py @@ -12,6 +12,8 @@ import stem.prereq import stem.manual import stem.util.system
+from test.util import require_command + try: # account for urllib's change between python 2.x and 3.x import urllib.request as urllib @@ -150,6 +152,7 @@ class TestManual(unittest.TestCase): self.assertEqual('', blank.summary) self.assertEqual('', blank.description)
+ @require_command('man') def test_parsing_with_example(self): """ Read a trimmed copy of tor's man page. This gives a good exercise of our @@ -157,10 +160,7 @@ class TestManual(unittest.TestCase): expand our example (or add another). """
- if not stem.util.system.is_available('man'): - self.skipTest('(require man command)') - return - elif stem.util.system.is_mac(): + if stem.util.system.is_mac(): self.skipTest('(man lacks --encoding arg on OSX, #18660)') return
@@ -174,16 +174,14 @@ class TestManual(unittest.TestCase): self.assertEqual(EXPECTED_FILES, manual.files) self.assertEqual(EXPECTED_CONFIG_OPTIONS, manual.config_options)
+ @require_command('man') def test_parsing_with_unknown_options(self): """ Check that we can read a local mock man page that contains unrecognized options. Unlike most other tests this doesn't require network access. """
- if not stem.util.system.is_available('man'): - self.skipTest('(require man command)') - return - elif stem.util.system.is_mac(): + if stem.util.system.is_mac(): self.skipTest('(man lacks --encoding arg on OSX, #18660)') return
@@ -205,15 +203,12 @@ class TestManual(unittest.TestCase): self.assertEqual('', option.summary) self.assertEqual('Description of this new option.', option.description)
+ @require_command('man') def test_saving_manual(self): """ Check that we can save and reload manuals. """
- if not stem.util.system.is_available('man'): - self.skipTest('(require man command)') - return - manual = stem.manual.Manual.from_man(EXAMPLE_MAN_PATH)
with tempfile.NamedTemporaryFile(prefix = 'saved_test_manual.') as tmp: diff --git a/test/util.py b/test/util.py index 2468b55..864baab 100644 --- a/test/util.py +++ b/test/util.py @@ -24,9 +24,12 @@ Tasks are... | |- require_cryptography - skips test unless the cryptography module is present |- require_pynacl - skips test unless the pynacl module is present + |- require_command - requires a command to be on the path + |- require_proc - requires the platform to have recognized /proc contents | |- require_controller - skips test unless tor provides a controller endpoint |- require_version - skips test unless we meet a tor version requirement + |- require_ptrace - requires 'DisableDebuggerAttachment' to be set +- require_online - skips unless targets allow for online tests
Initialization @@ -255,6 +258,7 @@ def require(condition, message):
require_cryptography = require(stem.prereq.is_crypto_available, 'requires cryptography') require_pynacl = require(stem.prereq._is_pynacl_available, 'requires pynacl module') +require_proc = require(stem.util.proc.is_available, 'proc unavailable')
def require_controller(func): @@ -271,6 +275,14 @@ def require_controller(func): return wrapped
+def require_command(cmd): + """ + Skips the test unless a command is available on the path. + """ + + return require(lambda: stem.util.system.is_available(cmd), '%s unavailable' % cmd) + + def require_version(req_version): """ Skips the test unless we meet the required version. @@ -526,3 +538,5 @@ class Task(object):
import test.runner # needs to be imported at the end to avoid a circular dependency + +require_ptrace = require(test.runner.get_runner().is_ptraceable, 'DisableDebuggerAttachment is set')