| from __future__ import with_statement | |
| import sys | |
| __all__ = ['nested', 'catch_warnings', 'examine_warnings'] | |
| try: | |
| from contextlib import nested | |
| except ImportError: | |
| from contextlib import contextmanager | |
| @contextmanager | |
| def nested(*managers): | |
| exits = [] | |
| vars = [] | |
| exc = (None, None, None) | |
| try: | |
| for mgr in managers: | |
| exit = mgr.__exit__ | |
| enter = mgr.__enter__ | |
| vars.append(enter()) | |
| exits.append(exit) | |
| yield vars | |
| except: | |
| exc = sys.exc_info() | |
| finally: | |
| while exits: | |
| exit = exits.pop() | |
| try: | |
| if exit(*exc): | |
| exc = (None, None, None) | |
| except: | |
| exc = sys.exc_info() | |
| if exc != (None, None, None): | |
| raise exc[1] | |
| # copied from Python 2.6 | |
| try: | |
| from warnings import catch_warnings | |
| except ImportError: | |
| class catch_warnings(object): | |
| def __init__(self, record=False, module=None): | |
| self._record = record | |
| self._module = sys.modules['warnings'] | |
| self._entered = False | |
| def __repr__(self): | |
| args = [] | |
| if self._record: | |
| args.append("record=True") | |
| name = type(self).__name__ | |
| return "%s(%s)" % (name, ", ".join(args)) | |
| def __enter__(self): | |
| if self._entered: | |
| raise RuntimeError("Cannot enter %r twice" % self) | |
| self._entered = True | |
| self._filters = self._module.filters | |
| self._module.filters = self._filters[:] | |
| self._showwarning = self._module.showwarning | |
| if self._record: | |
| log = [] | |
| def showwarning(*args, **kwargs): | |
| log.append(WarningMessage(*args, **kwargs)) | |
| self._module.showwarning = showwarning | |
| return log | |
| else: | |
| return None | |
| def __exit__(self, *exc_info): | |
| if not self._entered: | |
| raise RuntimeError("Cannot exit %r without entering first" % self) | |
| self._module.filters = self._filters | |
| self._module.showwarning = self._showwarning | |
| class WarningMessage(object): | |
| _WARNING_DETAILS = ("message", "category", "filename", "lineno", "file", | |
| "line") | |
| def __init__(self, message, category, filename, lineno, file=None, | |
| line=None): | |
| local_values = locals() | |
| for attr in self._WARNING_DETAILS: | |
| setattr(self, attr, local_values[attr]) | |
| self._category_name = None | |
| if category.__name__: | |
| self._category_name = category.__name__ | |
| def examine_warnings(func): | |
| def wrapper(): | |
| with catch_warnings(record=True) as ws: | |
| func(ws) | |
| return wrapper |