Stem Docs


Source code for stem.util.test_tools

# Copyright 2015-2019, Damian Johnson and The Tor Project
# See LICENSE for licensing information

Helper functions for testing.

Our **stylistic_issues**, **pyflakes_issues**, and **type_check_issues**
respect a 'exclude_paths' in our test config, excluding any absolute paths
matching those regexes. Issue strings can start or end with an asterisk
to match just against the prefix or suffix. For instance...


  exclude_paths .*/stem/test/data/.*

.. versionadded:: 1.2.0


  TimedTestRunner - test runner that tracks test runtimes
  test_runtimes - provides runtime of tests excuted through TimedTestRunners
  clean_orphaned_pyc - delete *.pyc files without corresponding *.py

  is_pyflakes_available - checks if pyflakes is available
  is_pycodestyle_available - checks if pycodestyle is available

  pyflakes_issues - static checks for problems via pyflakes
  stylistic_issues - checks for PEP8 and other stylistic issues

import collections
import linecache
import multiprocessing
import os
import re
import threading
import time
import traceback
import unittest

import stem.prereq
import stem.util.conf
import stem.util.enum
import stem.util.system

CONFIG = stem.util.conf.config_dict('test', {
  'pep8.ignore': [],  # TODO: drop with stem 2.x, legacy alias for pycodestyle.ignore
  'pycodestyle.ignore': [],
  'pyflakes.ignore': [],
  'exclude_paths': [],


AsyncStatus = stem.util.enum.UppercaseEnum('PENDING', 'RUNNING', 'FINISHED')
AsyncResult = collections.namedtuple('AsyncResult', 'type msg')

# TODO: Providing a copy of SkipTest that works with python 2.6. This will be
# dropped when we remove python 2.6 support.

if stem.prereq._is_python_26():
  class SkipTest(Exception):
    'Notes that the test was skipped.'
  SkipTest =

[docs]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)
[docs]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)
[docs]def asynchronous(func): test = stem.util.test_tools.AsyncTest(func) ASYNC_TESTS[] = test return test.method
[docs]class AsyncTest(object): """ Test that's run asychronously. These are functions (no self reference) performed like the following... :: class MyTest(unittest.TestCase): @staticmethod def run_tests(): MyTest.test_addition = stem.util.test_tools.AsyncTest(MyTest.test_addition).method @staticmethod def test_addition(): if 1 + 1 != 2: raise AssertionError('tisk, tisk') .. versionadded:: 1.6.0 """ def __init__(self, runner, args = None, threaded = False): = '%s.%s' % (runner.__module__, runner.__name__) self._runner = runner self._runner_args = args self._threaded = threaded self.method = lambda test: self.result(test) # method that can be mixed into TestCases self._process = None self._process_pipe = None self._process_lock = threading.RLock() self._result = None self._status = AsyncStatus.PENDING
[docs] def run(self, *runner_args, **kwargs): if stem.prereq._is_python_26(): return # not supported under python 2.6 def _wrapper(conn, runner, args): os.nice(12) try: runner(*args) if args else runner() conn.send(AsyncResult('success', None)) except AssertionError as exc: conn.send(AsyncResult('failure', str(exc))) except SkipTest as exc: conn.send(AsyncResult('skipped', str(exc))) except: conn.send(AsyncResult('error', traceback.format_exc())) finally: conn.close() with self._process_lock: if self._status == AsyncStatus.PENDING: if runner_args: self._runner_args = runner_args if 'threaded' in kwargs: self._threaded = kwargs['threaded'] self._process_pipe, child_pipe = multiprocessing.Pipe() if self._threaded: self._process = threading.Thread( target = _wrapper, args = (child_pipe, self._runner, self._runner_args), name = 'Background test of %s' %, ) self._process.setDaemon(True) else: self._process = multiprocessing.Process(target = _wrapper, args = (child_pipe, self._runner, self._runner_args)) self._process.start() self._status = AsyncStatus.RUNNING
[docs] def pid(self): with self._process_lock: return if (self._process and not self._threaded) else None
[docs] def join(self): self.result(None)
[docs] def result(self, test): if stem.prereq._is_python_26(): return # not supported under python 2.6 with self._process_lock: if self._status == AsyncStatus.PENDING: if self._status == AsyncStatus.RUNNING: self._result = self._process_pipe.recv() self._process.join() self._status = AsyncStatus.FINISHED if test and self._result.type == 'failure': elif test and self._result.type == 'error': elif test and self._result.type == 'skipped': test.skipTest(self._result.msg)
[docs]class Issue(collections.namedtuple('Issue', ['line_number', 'message', 'line'])): """ Issue encountered by pyflakes or pycodestyle. :var int line_number: line number the issue occured on :var str message: description of the issue :var str line: content of the line the issue is about """
[docs]class TimedTestRunner(unittest.TextTestRunner): """ Test runner that tracks the runtime of individual tests. When tests are run with this their runtimes are made available through :func:`stem.util.test_tools.test_runtimes`. .. versionadded:: 1.6.0 """
[docs] def run(self, test): for t in test._tests: original_type = type(t) class _TestWrapper(original_type): def run(self, result = None): start_time = time.time() result = super(type(self), self).run(result) TEST_RUNTIMES[] = time.time() - start_time return result # TODO: remove and drop unnecessary 'returns' when dropping python 2.6 # support def skipTest(self, message): if not stem.prereq._is_python_26(): return super(original_type, self).skipTest(message) # TODO: remove when dropping python 2.6 support def assertItemsEqual(self, expected, actual): if stem.prereq._is_python_26(): self.assertEqual(set(expected), set(actual)) else: return super(original_type, self).assertItemsEqual(expected, actual) def assertRaisesWith(self, exc_type, exc_msg, func, *args, **kwargs): """ Asserts the given invokation raises the expected excepiton. This is similar to unittest's assertRaises and assertRaisesRegexp, but checks for an exact match. This method is **not** being vended to external users and may be changed without notice. If you want this method to be part of our vended API then please let us know. """ return self.assertRaisesRegexp(exc_type, '^%s$' % re.escape(exc_msg), func, *args, **kwargs) def assertRaisesRegexp(self, exc_type, exc_msg, func, *args, **kwargs): if stem.prereq._is_python_26(): try: func(*args, **kwargs)'Expected a %s to be raised but nothing was' % exc_type) except exc_type as exc: self.assertTrue(, str(exc), re.MULTILINE)) else: return super(original_type, self).assertRaisesRegexp(exc_type, exc_msg, func, *args, **kwargs) def id(self): return '%s.%s.%s' % (original_type.__module__, original_type.__name__, self._testMethodName) def __str__(self): return '%s (%s.%s)' % (self._testMethodName, original_type.__module__, original_type.__name__) t.__class__ = _TestWrapper return super(TimedTestRunner, self).run(test)
[docs]def test_runtimes(): """ Provides the runtimes of tests executed through TimedTestRunners. :returns: **dict** of fully qualified test names to floats for the runtime in seconds .. versionadded:: 1.6.0 """ return dict(TEST_RUNTIMES)
[docs]def clean_orphaned_pyc(paths): """ Deletes any file with a \*.pyc extention without a corresponding \*.py. This helps to address a common gotcha when deleting python files... * You delete module '' and run the tests to ensure that you haven't broken anything. They pass, however there *are* still some 'import foo' statements that still work because the bytecode (foo.pyc) is still around. * You push your change. * Another developer clones our repository and is confused because we have a bunch of ImportErrors. :param list paths: paths to search for orphaned pyc files :returns: list of absolute paths that were deleted """ orphaned_pyc = [] for path in paths: for pyc_path in stem.util.system.files_with_suffix(path, '.pyc'): py_path = pyc_path[:-1] # If we're running python 3 then the *.pyc files are no longer bundled # with the *.py. Rather, they're in a __pycache__ directory. pycache = '%s__pycache__%s' % (os.path.sep, os.path.sep) if pycache in pyc_path: directory, pycache_filename = pyc_path.split(pycache, 1) if not pycache_filename.endswith('.pyc'): continue # should look like 'test_tools.cpython-32.pyc' py_path = os.path.join(directory, pycache_filename.split('.')[0] + '.py') if not os.path.exists(py_path): orphaned_pyc.append(pyc_path) os.remove(pyc_path) return orphaned_pyc
[docs]def is_pyflakes_available(): """ Checks if pyflakes is availalbe. :returns: **True** if we can use pyflakes and **False** otherwise """ return _module_exists('pyflakes.api') and _module_exists('pyflakes.reporter')
[docs]def is_pycodestyle_available(): """ Checks if pycodestyle is availalbe. :returns: **True** if we can use pycodestyle and **False** otherwise """ if _module_exists('pycodestyle'): import pycodestyle elif _module_exists('pep8'): import pep8 as pycodestyle else: return False return hasattr(pycodestyle, 'BaseReport')
[docs]def stylistic_issues(paths, check_newlines = False, check_exception_keyword = False, prefer_single_quotes = False): """ Checks for stylistic issues that are an issue according to the parts of PEP8 we conform to. You can suppress pycodestyle issues by making a 'test' configuration that sets 'pycodestyle.ignore'. For example, with a 'test/settings.cfg' of... :: # pycodestyle compliance issues that we're ignoreing... # # * E111 and E121 four space indentations # * E501 line is over 79 characters pycodestyle.ignore E111 pycodestyle.ignore E121 pycodestyle.ignore E501 pycodestyle.ignore => E402: import stem.util.enum ... you can then run tests with... :: import stem.util.conf test_config = stem.util.conf.get_config('test') test_config.load('test/settings.cfg') issues = stylistic_issues('my_project') .. versionchanged:: 1.3.0 Renamed from get_stylistic_issues() to stylistic_issues(). The old name still works as an alias, but will be dropped in Stem version 2.0.0. .. versionchanged:: 1.4.0 Changing tuples in return value to be namedtuple instances, and adding the line that had the issue. .. versionchanged:: 1.4.0 Added the prefer_single_quotes option. .. versionchanged:: 1.6.0 Changed 'pycodestyle.ignore' code snippets to only need to match against the prefix. :param list paths: paths to search for stylistic issues :param bool check_newlines: check that we have standard newlines (\\n), not windows (\\r\\n) nor classic mac (\\r) :param bool check_exception_keyword: checks that we're using 'as' for exceptions rather than a comma :param bool prefer_single_quotes: standardize on using single rather than double quotes for strings, when reasonable :returns: dict of paths list of :class:`stem.util.test_tools.Issue` instances """ issues = {} ignore_rules = [] ignore_for_file = [] for rule in CONFIG['pycodestyle.ignore'] + CONFIG['pep8.ignore']: if '=>' in rule: path, rule_entry = rule.split('=>', 1) if ':' in rule_entry: rule, code = rule_entry.split(':', 1) ignore_for_file.append((path.strip(), rule.strip(), code.strip())) else: ignore_rules.append(rule) def is_ignored(path, rule, code): for ignored_path, ignored_rule, ignored_code in ignore_for_file: if path.endswith(ignored_path) and ignored_rule == rule and code.strip().startswith(ignored_code): return True return False if is_pycodestyle_available(): if _module_exists('pep8'): import pep8 as pycodestyle else: import pycodestyle class StyleReport(pycodestyle.BaseReport): def init_file(self, filename, lines, expected, line_offset): super(StyleReport, self).init_file(filename, lines, expected, line_offset) if not check_newlines and not check_exception_keyword and not prefer_single_quotes: return is_block_comment = False for index, line in enumerate(lines): content = line.split('#', 1)[0].strip() if check_newlines and '\r' in line: issues.setdefault(filename, []).append(Issue(index + 1, 'contains a windows newline', line)) if not content: continue # blank line if '"""' in content: is_block_comment = not is_block_comment if check_exception_keyword and content.startswith('except') and content.endswith(', exc:'): # Python 2.6 - 2.7 supports two forms for exceptions... # # except ValueError, exc: # except ValueError as exc: # # The former is the old method and no longer supported in python 3 # going forward. # TODO: This check only works if the exception variable is called # 'exc'. We should generalize this via a regex so other names work # too. issues.setdefault(filename, []).append(Issue(index + 1, "except clause should use 'as', not comma", line)) if prefer_single_quotes and not is_block_comment: if '"' in content and "'" not in content and '"""' not in content and not content.endswith('\\'): # Checking if the line already has any single quotes since that # usually means double quotes are preferable for the content (for # instance "I'm hungry"). Also checking for '\' at the end since # that can indicate a multi-line string. issues.setdefault(filename, []).append(Issue(index + 1, 'use single rather than double quotes', line)) def error(self, line_number, offset, text, check): code = super(StyleReport, self).error(line_number, offset, text, check) if code: line = linecache.getline(self.filename, line_number) if not is_ignored(self.filename, code, line): issues.setdefault(self.filename, []).append(Issue(line_number, text, line)) style_checker = pycodestyle.StyleGuide(ignore = ignore_rules, reporter = StyleReport) style_checker.check_files(list(_python_files(paths))) return issues
[docs]def pyflakes_issues(paths): """ Performs static checks via pyflakes. False positives can be ignored via 'pyflakes.ignore' entries in our 'test' config. For instance... :: pyflakes.ignore stem/util/ => 'pyflakes' imported but unused pyflakes.ignore stem/util/ => 'pycodestyle' imported but unused .. versionchanged:: 1.3.0 Renamed from get_pyflakes_issues() to pyflakes_issues(). The old name still works as an alias, but will be dropped in Stem version 2.0.0. .. versionchanged:: 1.4.0 Changing tuples in return value to be namedtuple instances, and adding the line that had the issue. .. versionchanged:: 1.5.0 Support matching against prefix or suffix issue strings. :param list paths: paths to search for problems :returns: dict of paths list of :class:`stem.util.test_tools.Issue` instances """ issues = {} if is_pyflakes_available(): import pyflakes.api import pyflakes.reporter class Reporter(pyflakes.reporter.Reporter): def __init__(self): self._ignored_issues = {} for line in CONFIG['pyflakes.ignore']: path, issue = line.split('=>') self._ignored_issues.setdefault(path.strip(), []).append(issue.strip()) def unexpectedError(self, filename, msg): self._register_issue(filename, None, msg, None) def syntaxError(self, filename, msg, lineno, offset, text): self._register_issue(filename, lineno, msg, text) def flake(self, msg): self._register_issue(msg.filename, msg.lineno, msg.message % msg.message_args, None) def _is_ignored(self, path, issue): # Paths in pyflakes_ignore are relative, so we need to check to see if our # path ends with any of them. for ignored_path, ignored_issues in self._ignored_issues.items(): if path.endswith(ignored_path): if issue in ignored_issues: return True for prefix in [i[:1] for i in ignored_issues if i.endswith('*')]: if issue.startswith(prefix): return True for suffix in [i[1:] for i in ignored_issues if i.startswith('*')]: if issue.endswith(suffix): return True return False def _register_issue(self, path, line_number, issue, line): if not self._is_ignored(path, issue): if path and line_number and not line: line = linecache.getline(path, line_number).strip() issues.setdefault(path, []).append(Issue(line_number, issue, line)) reporter = Reporter() for path in _python_files(paths): pyflakes.api.checkPath(path, reporter) return issues
def _module_exists(module_name): """ Checks if a module exists. :param str module_name: module to check existance of :returns: **True** if module exists and **False** otherwise """ try: __import__(module_name) return True except ImportError: return False def _python_files(paths): for path in paths: for file_path in stem.util.system.files_with_suffix(path, '.py'): skip = False for exclude_path in CONFIG['exclude_paths']: if re.match(exclude_path, file_path): skip = True break if not skip: yield file_path # TODO: drop with stem 2.x # We renamed our methods to drop a redundant 'get_*' prefix, so alias the old # names for backward compatability, and account for pep8 being renamed to # pycodestyle. get_stylistic_issues = stylistic_issues get_pyflakes_issues = pyflakes_issues is_pep8_available = is_pycodestyle_available