"""Support code for packaging test cases. A few helper classes are provided: LoggingCatcher, TempdirManager and EnvironRestorer. They are written to be used as mixins:: from packaging.tests import unittest from packaging.tests.support import LoggingCatcher class SomeTestCase(LoggingCatcher, unittest.TestCase): If you need to define a setUp method on your test class, you have to call the mixin class' setUp method or it won't work (same thing for tearDown): def setUp(self): super(SomeTestCase, self).setUp() ... # other setup code Also provided is a DummyCommand class, useful to mock commands in the tests of another command that needs them, a create_distribution function and a skip_unless_symlink decorator. Also provided is a DummyCommand class, useful to mock commands in the tests of another command that needs them, a create_distribution function and a skip_unless_symlink decorator. Each class or function has a docstring to explain its purpose and usage. """ import os import shutil import logging import weakref import tempfile from packaging.dist import Distribution from packaging.tests import unittest from test.support import requires_zlib, unlink __all__ = ['LoggingCatcher', 'TempdirManager', 'EnvironRestorer', 'DummyCommand', 'unittest', 'create_distribution', 'skip_unless_symlink', 'requires_zlib'] logger = logging.getLogger('packaging') logger2to3 = logging.getLogger('RefactoringTool') class _TestHandler(logging.handlers.BufferingHandler): # stolen and adapted from test.support def __init__(self): logging.handlers.BufferingHandler.__init__(self, 0) self.setLevel(logging.DEBUG) def shouldFlush(self): return False def emit(self, record): self.buffer.append(record) class LoggingCatcher: """TestCase-compatible mixin to receive logging calls. Upon setUp, instances of this classes get a BufferingHandler that's configured to record all messages logged to the 'packaging' logger. Use get_logs to retrieve messages and self.loghandler.flush to discard them. get_logs automatically flushes the logs; if you test code that generates logging messages but don't use get_logs, you have to flush manually before doing other checks on logging message, otherwise you will get irrelevant results. See example in test_command_check. """ def setUp(self): super(LoggingCatcher, self).setUp() self.loghandler = handler = _TestHandler() self._old_levels = logger.level, logger2to3.level logger.addHandler(handler) logger.setLevel(logging.DEBUG) # we want all messages logger2to3.setLevel(logging.CRITICAL) # we don't want 2to3 messages def tearDown(self): handler = self.loghandler # All this is necessary to properly shut down the logging system and # avoid a regrtest complaint. Thanks to Vinay Sajip for the help. handler.close() logger.removeHandler(handler) for ref in weakref.getweakrefs(handler): logging._removeHandlerRef(ref) del self.loghandler logger.setLevel(self._old_levels[0]) logger2to3.setLevel(self._old_levels[1]) super(LoggingCatcher, self).tearDown() def get_logs(self, *levels): """Return all log messages with level in *levels*. Without explicit levels given, returns all messages. *levels* defaults to all levels. For log calls with arguments (i.e. logger.info('bla bla %r', arg)), the messages will be formatted before being returned (e.g. "bla bla 'thing'"). Returns a list. Automatically flushes the loghandler after being called. Example: self.get_logs(logging.WARN, logging.DEBUG). """ if not levels: messages = [log.getMessage() for log in self.loghandler.buffer] else: messages = [log.getMessage() for log in self.loghandler.buffer if log.levelno in levels] self.loghandler.flush() return messages class TempdirManager: """TestCase-compatible mixin to create temporary directories and files. Directories and files created in a test_* method will be removed after it has run. """ def setUp(self): super(TempdirManager, self).setUp() self._olddir = os.getcwd() self._basetempdir = tempfile.mkdtemp() self._files = [] def tearDown(self): for handle, name in self._files: handle.close() unlink(name) os.chdir(self._olddir) shutil.rmtree(self._basetempdir) super(TempdirManager, self).tearDown() def mktempfile(self): """Create a read-write temporary file and return it.""" fd, fn = tempfile.mkstemp(dir=self._basetempdir) os.close(fd) fp = open(fn, 'w+') self._files.append((fp, fn)) return fp def mkdtemp(self): """Create a temporary directory and return its path.""" d = tempfile.mkdtemp(dir=self._basetempdir) return d def write_file(self, path, content='xxx', encoding=None): """Write a file at the given path. path can be a string, a tuple or a list; if it's a tuple or list, os.path.join will be used to produce a path. """ if isinstance(path, (list, tuple)): path = os.path.join(*path) with open(path, 'w', encoding=encoding) as f: f.write(content) def create_dist(self, **kw): """Create a stub distribution object and files. This function creates a Distribution instance (use keyword arguments to customize it) and a temporary directory with a project structure (currently an empty directory). It returns the path to the directory and the Distribution instance. You can use self.write_file to write any file in that directory, e.g. setup scripts or Python modules. """ if 'name' not in kw: kw['name'] = 'foo' tmp_dir = self.mkdtemp() project_dir = os.path.join(tmp_dir, kw['name']) os.mkdir(project_dir) dist = Distribution(attrs=kw) return project_dir, dist def assertIsFile(self, *args): path = os.path.join(*args) dirname = os.path.dirname(path) file = os.path.basename(path) if os.path.isdir(dirname): files = os.listdir(dirname) msg = "%s not found in %s: %s" % (file, dirname, files) assert os.path.isfile(path), msg else: raise AssertionError( '%s not found. %s does not exist' % (file, dirname)) def assertIsNotFile(self, *args): path = os.path.join(*args) self.assertFalse(os.path.isfile(path), "%r exists" % path) class EnvironRestorer: """TestCase-compatible mixin to restore or delete environment variables. The variables to restore (or delete if they were not originally present) must be explicitly listed in self.restore_environ. It's better to be aware of what we're modifying instead of saving and restoring the whole environment. """ def setUp(self): super(EnvironRestorer, self).setUp() self._saved = [] self._added = [] for key in self.restore_environ: if key in os.environ: self._saved.append((key, os.environ[key])) else: self._added.append(key) def tearDown(self): for key, value in self._saved: os.environ[key] = value for key in self._added: os.environ.pop(key, None) super(EnvironRestorer, self).tearDown() class DummyCommand: """Class to store options for retrieval via set_undefined_options(). Useful for mocking one dependency command in the tests for another command, see e.g. the dummy build command in test_build_scripts. """ def __init__(self, **kwargs): for kw, val in kwargs.items(): setattr(self, kw, val) def ensure_finalized(self): pass class TestDistribution(Distribution): """Distribution subclasses that avoids the default search for configuration files. The ._config_files attribute must be set before .parse_config_files() is called. """ def find_config_files(self): return self._config_files def create_distribution(configfiles=()): """Prepares a distribution with given config files parsed.""" d = TestDistribution() d.config.find_config_files = d.find_config_files d._config_files = configfiles d.parse_config_files() d.parse_command_line() return d def fake_dec(*args, **kw): """Fake decorator""" def _wrap(func): def __wrap(*args, **kw): return func(*args, **kw) return __wrap return _wrap try: from test.support import skip_unless_symlink except ImportError: skip_unless_symlink = unittest.skip( 'requires test.support.skip_unless_symlink')