Mercurial > repos > shellac > guppy_basecaller
diff env/lib/python3.7/site-packages/humanfriendly/testing.py @ 5:9b1c78e6ba9c draft default tip
"planemo upload commit 6c0a8142489327ece472c84e558c47da711a9142"
| author | shellac |
|---|---|
| date | Mon, 01 Jun 2020 08:59:25 -0400 |
| parents | 79f47841a781 |
| children |
line wrap: on
line diff
--- a/env/lib/python3.7/site-packages/humanfriendly/testing.py Thu May 14 16:47:39 2020 -0400 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,669 +0,0 @@ -# Human friendly input/output in Python. -# -# Author: Peter Odding <peter@peterodding.com> -# Last Change: March 6, 2020 -# URL: https://humanfriendly.readthedocs.io - -""" -Utility classes and functions that make it easy to write :mod:`unittest` compatible test suites. - -Over the years I've developed the habit of writing test suites for Python -projects using the :mod:`unittest` module. During those years I've come to know -:pypi:`pytest` and in fact I use :pypi:`pytest` to run my test suites (due to -its much better error reporting) but I've yet to publish a test suite that -*requires* :pypi:`pytest`. I have several reasons for doing so: - -- It's nice to keep my test suites as simple and accessible as possible and - not requiring a specific test runner is part of that attitude. - -- Whereas :mod:`unittest` is quite explicit, :pypi:`pytest` contains a lot of - magic, which kind of contradicts the Python mantra "explicit is better than - implicit" (IMHO). -""" - -# Standard library module -import functools -import logging -import os -import pipes -import shutil -import sys -import tempfile -import time -import unittest - -# Modules included in our package. -from humanfriendly.compat import StringIO -from humanfriendly.text import random_string - -# Initialize a logger for this module. -logger = logging.getLogger(__name__) - -# A unique object reference used to detect missing attributes. -NOTHING = object() - -# Public identifiers that require documentation. -__all__ = ( - 'CallableTimedOut', - 'CaptureBuffer', - 'CaptureOutput', - 'ContextManager', - 'CustomSearchPath', - 'MockedProgram', - 'PatchedAttribute', - 'PatchedItem', - 'TemporaryDirectory', - 'TestCase', - 'configure_logging', - 'make_dirs', - 'retry', - 'run_cli', - 'skip_on_raise', - 'touch', -) - - -def configure_logging(log_level=logging.DEBUG): - """configure_logging(log_level=logging.DEBUG) - Automatically configure logging to the terminal. - - :param log_level: The log verbosity (a number, defaults - to :mod:`logging.DEBUG <logging>`). - - When :mod:`coloredlogs` is installed :func:`coloredlogs.install()` will be - used to configure logging to the terminal. When this fails with an - :exc:`~exceptions.ImportError` then :func:`logging.basicConfig()` is used - as a fall back. - """ - try: - import coloredlogs - coloredlogs.install(level=log_level) - except ImportError: - logging.basicConfig( - level=log_level, - format='%(asctime)s %(name)s[%(process)d] %(levelname)s %(message)s', - datefmt='%Y-%m-%d %H:%M:%S') - - -def make_dirs(pathname): - """ - Create missing directories. - - :param pathname: The pathname of a directory (a string). - """ - if not os.path.isdir(pathname): - os.makedirs(pathname) - - -def retry(func, timeout=60, exc_type=AssertionError): - """retry(func, timeout=60, exc_type=AssertionError) - Retry a function until assertions no longer fail. - - :param func: A callable. When the callable returns - :data:`False` it will also be retried. - :param timeout: The number of seconds after which to abort (a number, - defaults to 60). - :param exc_type: The type of exceptions to retry (defaults - to :exc:`~exceptions.AssertionError`). - :returns: The value returned by `func`. - :raises: Once the timeout has expired :func:`retry()` will raise the - previously retried assertion error. When `func` keeps returning - :data:`False` until `timeout` expires :exc:`CallableTimedOut` - will be raised. - - This function sleeps between retries to avoid claiming CPU cycles we don't - need. It starts by sleeping for 0.1 second but adjusts this to one second - as the number of retries grows. - """ - pause = 0.1 - timeout += time.time() - while True: - try: - result = func() - if result is not False: - return result - except exc_type: - if time.time() > timeout: - raise - else: - if time.time() > timeout: - raise CallableTimedOut() - time.sleep(pause) - if pause < 1: - pause *= 2 - - -def run_cli(entry_point, *arguments, **options): - """ - Test a command line entry point. - - :param entry_point: The function that implements the command line interface - (a callable). - :param arguments: Any positional arguments (strings) become the command - line arguments (:data:`sys.argv` items 1-N). - :param options: The following keyword arguments are supported: - - **capture** - Whether to use :class:`CaptureOutput`. Defaults - to :data:`True` but can be disabled by passing - :data:`False` instead. - **input** - Refer to :class:`CaptureOutput`. - **merged** - Refer to :class:`CaptureOutput`. - **program_name** - Used to set :data:`sys.argv` item 0. - :returns: A tuple with two values: - - 1. The return code (an integer). - 2. The captured output (a string). - """ - # Add the `program_name' option to the arguments. - arguments = list(arguments) - arguments.insert(0, options.pop('program_name', sys.executable)) - # Log the command line arguments (and the fact that we're about to call the - # command line entry point function). - logger.debug("Calling command line entry point with arguments: %s", arguments) - # Prepare to capture the return code and output even if the command line - # interface raises an exception (whether the exception type is SystemExit - # or something else). - returncode = 0 - stdout = None - stderr = None - try: - # Temporarily override sys.argv. - with PatchedAttribute(sys, 'argv', arguments): - # Manipulate the standard input/output/error streams? - options['enabled'] = options.pop('capture', True) - with CaptureOutput(**options) as capturer: - try: - # Call the command line interface. - entry_point() - finally: - # Get the output even if an exception is raised. - stdout = capturer.stdout.getvalue() - stderr = capturer.stderr.getvalue() - # Reconfigure logging to the terminal because it is very - # likely that the entry point function has changed the - # configured log level. - configure_logging() - except BaseException as e: - if isinstance(e, SystemExit): - logger.debug("Intercepting return code %s from SystemExit exception.", e.code) - returncode = e.code - else: - logger.warning("Defaulting return code to 1 due to raised exception.", exc_info=True) - returncode = 1 - else: - logger.debug("Command line entry point returned successfully!") - # Always log the output captured on stdout/stderr, to make it easier to - # diagnose test failures (but avoid duplicate logging when merged=True). - is_merged = options.get('merged', False) - merged_streams = [('merged streams', stdout)] - separate_streams = [('stdout', stdout), ('stderr', stderr)] - streams = merged_streams if is_merged else separate_streams - for name, value in streams: - if value: - logger.debug("Output on %s:\n%s", name, value) - else: - logger.debug("No output on %s.", name) - return returncode, stdout - - -def skip_on_raise(*exc_types): - """ - Decorate a test function to translation specific exception types to :exc:`unittest.SkipTest`. - - :param exc_types: One or more positional arguments give the exception - types to be translated to :exc:`unittest.SkipTest`. - :returns: A decorator function specialized to `exc_types`. - """ - def decorator(function): - @functools.wraps(function) - def wrapper(*args, **kw): - try: - return function(*args, **kw) - except exc_types as e: - logger.debug("Translating exception to unittest.SkipTest ..", exc_info=True) - raise unittest.SkipTest("skipping test because %s was raised" % type(e)) - return wrapper - return decorator - - -def touch(filename): - """ - The equivalent of the UNIX :man:`touch` program in Python. - - :param filename: The pathname of the file to touch (a string). - - Note that missing directories are automatically created using - :func:`make_dirs()`. - """ - make_dirs(os.path.dirname(filename)) - with open(filename, 'a'): - os.utime(filename, None) - - -class CallableTimedOut(Exception): - - """Raised by :func:`retry()` when the timeout expires.""" - - -class ContextManager(object): - - """Base class to enable composition of context managers.""" - - def __enter__(self): - """Enable use as context managers.""" - return self - - def __exit__(self, exc_type=None, exc_value=None, traceback=None): - """Enable use as context managers.""" - - -class PatchedAttribute(ContextManager): - - """Context manager that temporary replaces an object attribute using :func:`setattr()`.""" - - def __init__(self, obj, name, value): - """ - Initialize a :class:`PatchedAttribute` object. - - :param obj: The object to patch. - :param name: An attribute name. - :param value: The value to set. - """ - self.object_to_patch = obj - self.attribute_to_patch = name - self.patched_value = value - self.original_value = NOTHING - - def __enter__(self): - """ - Replace (patch) the attribute. - - :returns: The object whose attribute was patched. - """ - # Enable composition of context managers. - super(PatchedAttribute, self).__enter__() - # Patch the object's attribute. - self.original_value = getattr(self.object_to_patch, self.attribute_to_patch, NOTHING) - setattr(self.object_to_patch, self.attribute_to_patch, self.patched_value) - return self.object_to_patch - - def __exit__(self, exc_type=None, exc_value=None, traceback=None): - """Restore the attribute to its original value.""" - # Enable composition of context managers. - super(PatchedAttribute, self).__exit__(exc_type, exc_value, traceback) - # Restore the object's attribute. - if self.original_value is NOTHING: - delattr(self.object_to_patch, self.attribute_to_patch) - else: - setattr(self.object_to_patch, self.attribute_to_patch, self.original_value) - - -class PatchedItem(ContextManager): - - """Context manager that temporary replaces an object item using :meth:`~object.__setitem__()`.""" - - def __init__(self, obj, item, value): - """ - Initialize a :class:`PatchedItem` object. - - :param obj: The object to patch. - :param item: The item to patch. - :param value: The value to set. - """ - self.object_to_patch = obj - self.item_to_patch = item - self.patched_value = value - self.original_value = NOTHING - - def __enter__(self): - """ - Replace (patch) the item. - - :returns: The object whose item was patched. - """ - # Enable composition of context managers. - super(PatchedItem, self).__enter__() - # Patch the object's item. - try: - self.original_value = self.object_to_patch[self.item_to_patch] - except KeyError: - self.original_value = NOTHING - self.object_to_patch[self.item_to_patch] = self.patched_value - return self.object_to_patch - - def __exit__(self, exc_type=None, exc_value=None, traceback=None): - """Restore the item to its original value.""" - # Enable composition of context managers. - super(PatchedItem, self).__exit__(exc_type, exc_value, traceback) - # Restore the object's item. - if self.original_value is NOTHING: - del self.object_to_patch[self.item_to_patch] - else: - self.object_to_patch[self.item_to_patch] = self.original_value - - -class TemporaryDirectory(ContextManager): - - """ - Easy temporary directory creation & cleanup using the :keyword:`with` statement. - - Here's an example of how to use this: - - .. code-block:: python - - with TemporaryDirectory() as directory: - # Do something useful here. - assert os.path.isdir(directory) - """ - - def __init__(self, **options): - """ - Initialize a :class:`TemporaryDirectory` object. - - :param options: Any keyword arguments are passed on to - :func:`tempfile.mkdtemp()`. - """ - self.mkdtemp_options = options - self.temporary_directory = None - - def __enter__(self): - """ - Create the temporary directory using :func:`tempfile.mkdtemp()`. - - :returns: The pathname of the directory (a string). - """ - # Enable composition of context managers. - super(TemporaryDirectory, self).__enter__() - # Create the temporary directory. - self.temporary_directory = tempfile.mkdtemp(**self.mkdtemp_options) - return self.temporary_directory - - def __exit__(self, exc_type=None, exc_value=None, traceback=None): - """Cleanup the temporary directory using :func:`shutil.rmtree()`.""" - # Enable composition of context managers. - super(TemporaryDirectory, self).__exit__(exc_type, exc_value, traceback) - # Cleanup the temporary directory. - if self.temporary_directory is not None: - shutil.rmtree(self.temporary_directory) - self.temporary_directory = None - - -class MockedHomeDirectory(PatchedItem, TemporaryDirectory): - - """ - Context manager to temporarily change ``$HOME`` (the current user's profile directory). - - This class is a composition of the :class:`PatchedItem` and - :class:`TemporaryDirectory` context managers. - """ - - def __init__(self): - """Initialize a :class:`MockedHomeDirectory` object.""" - PatchedItem.__init__(self, os.environ, 'HOME', os.environ.get('HOME')) - TemporaryDirectory.__init__(self) - - def __enter__(self): - """ - Activate the custom ``$PATH``. - - :returns: The pathname of the directory that has - been added to ``$PATH`` (a string). - """ - # Get the temporary directory. - directory = TemporaryDirectory.__enter__(self) - # Override the value to patch now that we have - # the pathname of the temporary directory. - self.patched_value = directory - # Temporary patch $HOME. - PatchedItem.__enter__(self) - # Pass the pathname of the temporary directory to the caller. - return directory - - def __exit__(self, exc_type=None, exc_value=None, traceback=None): - """Deactivate the custom ``$HOME``.""" - super(MockedHomeDirectory, self).__exit__(exc_type, exc_value, traceback) - - -class CustomSearchPath(PatchedItem, TemporaryDirectory): - - """ - Context manager to temporarily customize ``$PATH`` (the executable search path). - - This class is a composition of the :class:`PatchedItem` and - :class:`TemporaryDirectory` context managers. - """ - - def __init__(self, isolated=False): - """ - Initialize a :class:`CustomSearchPath` object. - - :param isolated: :data:`True` to clear the original search path, - :data:`False` to add the temporary directory to the - start of the search path. - """ - # Initialize our own instance variables. - self.isolated_search_path = isolated - # Selectively initialize our superclasses. - PatchedItem.__init__(self, os.environ, 'PATH', self.current_search_path) - TemporaryDirectory.__init__(self) - - def __enter__(self): - """ - Activate the custom ``$PATH``. - - :returns: The pathname of the directory that has - been added to ``$PATH`` (a string). - """ - # Get the temporary directory. - directory = TemporaryDirectory.__enter__(self) - # Override the value to patch now that we have - # the pathname of the temporary directory. - self.patched_value = ( - directory if self.isolated_search_path - else os.pathsep.join([directory] + self.current_search_path.split(os.pathsep)) - ) - # Temporary patch the $PATH. - PatchedItem.__enter__(self) - # Pass the pathname of the temporary directory to the caller - # because they may want to `install' custom executables. - return directory - - def __exit__(self, exc_type=None, exc_value=None, traceback=None): - """Deactivate the custom ``$PATH``.""" - super(CustomSearchPath, self).__exit__(exc_type, exc_value, traceback) - - @property - def current_search_path(self): - """The value of ``$PATH`` or :data:`os.defpath` (a string).""" - return os.environ.get('PATH', os.defpath) - - -class MockedProgram(CustomSearchPath): - - """ - Context manager to mock the existence of a program (executable). - - This class extends the functionality of :class:`CustomSearchPath`. - """ - - def __init__(self, name, returncode=0, script=None): - """ - Initialize a :class:`MockedProgram` object. - - :param name: The name of the program (a string). - :param returncode: The return code that the program should emit (a - number, defaults to zero). - :param script: Shell script code to include in the mocked program (a - string or :data:`None`). This can be used to mock a - program that is expected to generate specific output. - """ - # Initialize our own instance variables. - self.program_name = name - self.program_returncode = returncode - self.program_script = script - self.program_signal_file = None - # Initialize our superclasses. - super(MockedProgram, self).__init__() - - def __enter__(self): - """ - Create the mock program. - - :returns: The pathname of the directory that has - been added to ``$PATH`` (a string). - """ - directory = super(MockedProgram, self).__enter__() - self.program_signal_file = os.path.join(directory, 'program-was-run-%s' % random_string(10)) - pathname = os.path.join(directory, self.program_name) - with open(pathname, 'w') as handle: - handle.write('#!/bin/sh\n') - handle.write('echo > %s\n' % pipes.quote(self.program_signal_file)) - if self.program_script: - handle.write('%s\n' % self.program_script.strip()) - handle.write('exit %i\n' % self.program_returncode) - os.chmod(pathname, 0o755) - return directory - - def __exit__(self, *args, **kw): - """ - Ensure that the mock program was run. - - :raises: :exc:`~exceptions.AssertionError` when - the mock program hasn't been run. - """ - try: - assert self.program_signal_file and os.path.isfile(self.program_signal_file), \ - ("It looks like %r was never run!" % self.program_name) - finally: - return super(MockedProgram, self).__exit__(*args, **kw) - - -class CaptureOutput(ContextManager): - - """ - Context manager that captures what's written to :data:`sys.stdout` and :data:`sys.stderr`. - - .. attribute:: stdin - - The :class:`~humanfriendly.compat.StringIO` object used to feed the standard input stream. - - .. attribute:: stdout - - The :class:`CaptureBuffer` object used to capture the standard output stream. - - .. attribute:: stderr - - The :class:`CaptureBuffer` object used to capture the standard error stream. - """ - - def __init__(self, merged=False, input='', enabled=True): - """ - Initialize a :class:`CaptureOutput` object. - - :param merged: :data:`True` to merge the streams, - :data:`False` to capture them separately. - :param input: The data that reads from :data:`sys.stdin` - should return (a string). - :param enabled: :data:`True` to enable capturing (the default), - :data:`False` otherwise. This makes it easy to - unconditionally use :class:`CaptureOutput` in - a :keyword:`with` block while preserving the - choice to opt out of capturing output. - """ - self.stdin = StringIO(input) - self.stdout = CaptureBuffer() - self.stderr = self.stdout if merged else CaptureBuffer() - self.patched_attributes = [] - if enabled: - self.patched_attributes.extend( - PatchedAttribute(sys, name, getattr(self, name)) - for name in ('stdin', 'stdout', 'stderr') - ) - - def __enter__(self): - """Start capturing what's written to :data:`sys.stdout` and :data:`sys.stderr`.""" - super(CaptureOutput, self).__enter__() - for context in self.patched_attributes: - context.__enter__() - return self - - def __exit__(self, exc_type=None, exc_value=None, traceback=None): - """Stop capturing what's written to :data:`sys.stdout` and :data:`sys.stderr`.""" - super(CaptureOutput, self).__exit__(exc_type, exc_value, traceback) - for context in self.patched_attributes: - context.__exit__(exc_type, exc_value, traceback) - - def get_lines(self): - """Get the contents of :attr:`stdout` split into separate lines.""" - return self.get_text().splitlines() - - def get_text(self): - """Get the contents of :attr:`stdout` as a Unicode string.""" - return self.stdout.get_text() - - def getvalue(self): - """Get the text written to :data:`sys.stdout`.""" - return self.stdout.getvalue() - - -class CaptureBuffer(StringIO): - - """ - Helper for :class:`CaptureOutput` to provide an easy to use API. - - The two methods defined by this subclass were specifically chosen to match - the names of the methods provided by my :pypi:`capturer` package which - serves a similar role as :class:`CaptureOutput` but knows how to simulate - an interactive terminal (tty). - """ - - def get_lines(self): - """Get the contents of the buffer split into separate lines.""" - return self.get_text().splitlines() - - def get_text(self): - """Get the contents of the buffer as a Unicode string.""" - return self.getvalue() - - -class TestCase(unittest.TestCase): - - """Subclass of :class:`unittest.TestCase` with automatic logging and other miscellaneous features.""" - - def __init__(self, *args, **kw): - """ - Initialize a :class:`TestCase` object. - - Any positional and/or keyword arguments are passed on to the - initializer of the superclass. - """ - super(TestCase, self).__init__(*args, **kw) - - def setUp(self, log_level=logging.DEBUG): - """setUp(log_level=logging.DEBUG) - Automatically configure logging to the terminal. - - :param log_level: Refer to :func:`configure_logging()`. - - The :func:`setUp()` method is automatically called by - :class:`unittest.TestCase` before each test method starts. - It does two things: - - - Logging to the terminal is configured using - :func:`configure_logging()`. - - - Before the test method starts a newline is emitted, to separate the - name of the test method (which will be printed to the terminal by - :mod:`unittest` or :pypi:`pytest`) from the first line of logging - output that the test method is likely going to generate. - """ - # Configure logging to the terminal. - configure_logging(log_level) - # Separate the name of the test method (printed by the superclass - # and/or py.test without a newline at the end) from the first line of - # logging output that the test method is likely going to generate. - sys.stderr.write("\n")
