[tor-commits] [stem/master] Use decorators for common test requirements

atagar at torproject.org atagar at torproject.org
Thu Apr 6 18:38:22 UTC 2017


commit 0ee3864a784ce8e116466800329863d37ec35400
Author: Damian Johnson <atagar at 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')





More information about the tor-commits mailing list