Hopiu/contextlib2
1
0
Fork 0
mirror of https://github.com/jazzband/contextlib2.git synced 2026-04-04 23:10:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #32 from ncoghlan/issue-12-sync-module-from-python-3.10

Browse source 
Issue #12:  sync module from CPython 3.10
This commit is contained in:
Nick Coghlan 2021-06-26 18:45:29 +10:00 committed by GitHub
commit b99ed09dfd
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
11 changed files with 1437 additions and 270 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
MANIFEST.in
View file

@ -1,2 +1,2 @@
include *.py *.txt *.rst *.md MANIFEST.in
recursive-include docs *.rst *.py make.bat Makefile
include *.py *.txt *.rst *.md *.ini MANIFEST.in
recursive-include dev test docs *.rst *.py make.bat Makefile

23
NEWS.rst
View file

@ -1,25 +1,23 @@
Release History
---------------
21.6.0 (2021-06-TBD)
^^^^^^^^^^^^^^^^^^^^^^^^
21.6.0 (2021-06-27)
^^^^^^^^^^^^^^^^^^^
* Switched to calendar based versioning rather than continuing with pre-1.0
semantic versioning (`#29 <https://github.com/jazzband/contextlib2/issues/29>`__)
* Due to the inclusion of asynchronous features from Python 3.7+, the
minimum supported Python version is now Python 3.6
(`#29 <https://github.com/jazzband/contextlib2/issues/29>`__)
* (WIP) Synchronised with the Python 3.10 version of contextlib, bringing the
following new features to Python 3.6+ (
`#12 <https://github.com/jazzband/contextlib2/issues/12>`__,
`#19 <https://github.com/jazzband/contextlib2/issues/19>`__,
`#27 <https://github.com/jazzband/contextlib2/issues/27>`__):
* Synchronised with the Python 3.10 version of contextlib
(`#12 <https://github.com/jazzband/contextlib2/issues/12>`__), making the
following new features available on Python 3.6+:
* ``asyncontextmanager`` (Python 3.7)
* ``aclosing`` (Python 3.10)
* ``AbstractAsyncContextManager`` (Python 3.7)
* ``AsyncContextDecorator`` (Python 3.10)
* ``AsyncExitStack`` (Python 3.7)
* ``asyncontextmanager`` (added in Python 3.7, enhanced in Python 3.10)
* ``aclosing`` (added in Python 3.10)
* ``AbstractAsyncContextManager`` (added in Python 3.7)
* ``AsyncContextDecorator`` (added in Python 3.10)
* ``AsyncExitStack`` (added in Python 3.7)
* async support in ``nullcontext`` (Python 3.10)
* Updates to the default compatibility testing matrix:
@ -27,7 +25,6 @@ Release History
* Added: CPython 3.9, CPython 3.10
* Dropped: CPython 2.7, CPython 3.5, PyPy2
0.6.0.post1 (2019-10-10)
^^^^^^^^^^^^^^^^^^^^^^^^

30
README.rst
View file

@ -15,7 +15,7 @@
:alt: Latest Docs
contextlib2 is a backport of the `standard library's contextlib
module <https://docs.python.org/3.5/library/contextlib.html>`_ to
module <https://docs.python.org/3/library/contextlib.html>`_ to
earlier Python versions.
It also serves as a real world proving ground for possible future
@ -28,7 +28,9 @@ contextlib2 has no runtime dependencies, but requires ``unittest2`` for testing
on Python 2.x, as well as ``setuptools`` and ``wheel`` to generate universal
wheel archives.
Local testing is just a matter of running ``python test_contextlib2.py``.
Local testing is a matter of running::
python3 -m unittest discover -t . -s test
You can test against multiple versions of Python with
`tox <https://tox.testrun.org/>`_::
@ -38,10 +40,28 @@ You can test against multiple versions of Python with
Versions currently tested in both tox and GitHub Actions are:
* CPython 2.7
* CPython 3.5
* CPython 3.6
* CPython 3.7
* CPython 3.8
* PyPy
* CPython 3.9
* CPython 3.10
* PyPy3
Updating to a new stdlib reference version
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
As of Python 3.10, 3 files needed to be copied from the CPython reference
implementation to contextlib2:
* ``Lib/contextlib.py`` -> ``contextlib2.py``
* ``Lib/test/test_contextlib.py`` -> ``test/test_contextlib.py``
* ``Lib/test/test_contextlib_async.py`` -> ``test/test_contextlib_async.py``
For the 3.10 sync, the only changes needed to the test files were to import from
``contextlib2`` rather than ``contextlib``. The test directory is laid out so
that the test suite's imports from ``test.support`` work the same way they do in
the main CPython test suite.
The changes made to the ``contextlib2.py`` file to get it to run on the older
versions (and to add back in the deprecated APIs that never graduated to the
standard library version) are saved as a patch file in the ``dev`` directory.

572
contextlib2.py
View file

@ -3,15 +3,25 @@
import abc
import sys
import warnings
import _collections_abc
from collections import deque
from functools import wraps
from types import MethodType
from _collections_abc import _check_methods
# Python 3.6/3.7/3.8 compatibility: GenericAlias may not be defined
try:
from types import GenericAlias
except ImportError:
# If the real GenericAlias type doesn't exist, __class_getitem__ won't be used,
# so the fallback placeholder doesn't need to provide any meaningful behaviour
class GenericAlias:
pass
__all__ = ["contextmanager", "closing", "nullcontext",
"AbstractContextManager",
"ContextDecorator", "ExitStack",
"redirect_stdout", "redirect_stderr", "suppress"]
__all__ = ["asynccontextmanager", "contextmanager", "closing", "nullcontext",
"AbstractContextManager", "AbstractAsyncContextManager",
"AsyncExitStack", "ContextDecorator", "ExitStack",
"redirect_stdout", "redirect_stderr", "suppress", "aclosing"]
# Backwards compatibility
__all__ += ["ContextStack"]
@ -19,6 +29,8 @@ __all__ += ["ContextStack"]
class AbstractContextManager(abc.ABC):
"""An abstract base class for context managers."""
__class_getitem__ = classmethod(GenericAlias)
def __enter__(self):
"""Return `self` upon entering the runtime context."""
return self
@ -30,14 +42,36 @@ class AbstractContextManager(abc.ABC):
@classmethod
def __subclasshook__(cls, C):
"""Check whether subclass is considered a subclass of this ABC."""
if cls is AbstractContextManager:
return _check_methods(C, "__enter__", "__exit__")
return _collections_abc._check_methods(C, "__enter__", "__exit__")
return NotImplemented
class AbstractAsyncContextManager(abc.ABC):
"""An abstract base class for asynchronous context managers."""
__class_getitem__ = classmethod(GenericAlias)
async def __aenter__(self):
"""Return `self` upon entering the runtime context."""
return self
@abc.abstractmethod
async def __aexit__(self, exc_type, exc_value, traceback):
"""Raise any exception triggered within the runtime context."""
return None
@classmethod
def __subclasshook__(cls, C):
if cls is AbstractAsyncContextManager:
return _collections_abc._check_methods(C, "__aenter__",
"__aexit__")
return NotImplemented
class ContextDecorator(object):
"""A base class or mixin that enables context managers to work as decorators."""
"A base class or mixin that enables context managers to work as decorators."
def refresh_cm(self):
"""Returns the context manager used to actually wrap the call to the
@ -76,8 +110,24 @@ class ContextDecorator(object):
return inner
class _GeneratorContextManager(ContextDecorator):
"""Helper for @contextmanager decorator."""
class AsyncContextDecorator(object):
"A base class or mixin that enables async context managers to work as decorators."
def _recreate_cm(self):
"""Return a recreated instance of self.
"""
return self
def __call__(self, func):
@wraps(func)
async def inner(*args, **kwds):
async with self._recreate_cm():
return await func(*args, **kwds)
return inner
class _GeneratorContextManagerBase:
"""Shared functionality for @contextmanager and @asynccontextmanager."""
def __init__(self, func, args, kwds):
self.gen = func(*args, **kwds)
@ -93,6 +143,12 @@ class _GeneratorContextManager(ContextDecorator):
# for the class instead.
# See http://bugs.python.org/issue19404 for more details.
class _GeneratorContextManager(_GeneratorContextManagerBase,
AbstractContextManager,
ContextDecorator):
"""Helper for @contextmanager decorator."""
def _recreate_cm(self):
# _GCM instances are one-shot context managers, so the
# CM must be recreated each time a decorated function is
@ -100,17 +156,20 @@ class _GeneratorContextManager(ContextDecorator):
return self.__class__(self.func, self.args, self.kwds)
def __enter__(self):
# do not keep args and kwds alive unnecessarily
# they are only needed for recreation, which is not possible anymore
del self.args, self.kwds, self.func
try:
return next(self.gen)
except StopIteration:
raise RuntimeError("generator didn't yield")
raise RuntimeError("generator didn't yield") from None
def __exit__(self, type, value, traceback):
if type is None:
try:
next(self.gen)
except StopIteration:
return
return False
else:
raise RuntimeError("generator didn't stop")
else:
@ -120,20 +179,19 @@ class _GeneratorContextManager(ContextDecorator):
value = type()
try:
self.gen.throw(type, value, traceback)
raise RuntimeError("generator didn't stop after throw()")
except StopIteration as exc:
# Suppress StopIteration *unless* it's the same exception that
# was passed to throw(). This prevents a StopIteration
# raised inside the "with" statement from being suppressed.
return exc is not value
except RuntimeError as exc:
# Don't re-raise the passed in exception
# Don't re-raise the passed in exception. (issue27122)
if exc is value:
return False
# Likewise, avoid suppressing if a StopIteration exception
# was passed to throw() and later wrapped into a RuntimeError
# (see PEP 479).
if exc.__cause__ is value:
if type is StopIteration and exc.__cause__ is value:
return False
raise
except:
@ -144,7 +202,66 @@ class _GeneratorContextManager(ContextDecorator):
# fixes the impedance mismatch between the throw() protocol
# and the __exit__() protocol.
#
if sys.exc_info()[1] is not value:
# This cannot use 'except BaseException as exc' (as in the
# async implementation) to maintain compatibility with
# Python 2, where old-style class exceptions are not caught
# by 'except BaseException'.
if sys.exc_info()[1] is value:
return False
raise
raise RuntimeError("generator didn't stop after throw()")
class _AsyncGeneratorContextManager(_GeneratorContextManagerBase,
AbstractAsyncContextManager,
AsyncContextDecorator):
"""Helper for @asynccontextmanager."""
def _recreate_cm(self):
# _AGCM instances are one-shot context managers, so the
# ACM must be recreated each time a decorated function is
# called
return self.__class__(self.func, self.args, self.kwds)
async def __aenter__(self):
try:
return await self.gen.__anext__()
except StopAsyncIteration:
raise RuntimeError("generator didn't yield") from None
async def __aexit__(self, typ, value, traceback):
if typ is None:
try:
await self.gen.__anext__()
except StopAsyncIteration:
return
else:
raise RuntimeError("generator didn't stop")
else:
if value is None:
value = typ()
# See _GeneratorContextManager.__exit__ for comments on subtleties
# in this implementation
try:
await self.gen.athrow(typ, value, traceback)
raise RuntimeError("generator didn't stop after athrow()")
except StopAsyncIteration as exc:
return exc is not value
except RuntimeError as exc:
if exc is value:
return False
# Avoid suppressing if a StopIteration exception
# was passed to throw() and later wrapped into a RuntimeError
# (see PEP 479 for sync generators; async generators also
# have this behavior). But do this only if the exception wrapped
# by the RuntimeError is actually Stop(Async)Iteration (see
# issue29692).
if isinstance(value, (StopIteration, StopAsyncIteration)):
if exc.__cause__ is value:
return False
raise
except BaseException as exc:
if exc is not value:
raise
@ -174,7 +291,6 @@ def contextmanager(func):
<body>
finally:
<cleanup>
"""
@wraps(func)
def helper(*args, **kwds):
@ -182,7 +298,40 @@ def contextmanager(func):
return helper
class closing(object):
def asynccontextmanager(func):
"""@asynccontextmanager decorator.
Typical usage:
@asynccontextmanager
async def some_async_generator(<arguments>):
<setup>
try:
yield <value>
finally:
<cleanup>
This makes this:
async with some_async_generator(<arguments>) as <variable>:
<body>
equivalent to this:
<setup>
try:
<variable> = <value>
<body>
finally:
<cleanup>
"""
@wraps(func)
def helper(*args, **kwds):
return _AsyncGeneratorContextManager(func, args, kwds)
return helper
class closing(AbstractContextManager):
"""Context to automatically close something at the end of a block.
Code like this:
@ -201,15 +350,39 @@ class closing(object):
"""
def __init__(self, thing):
self.thing = thing
def __enter__(self):
return self.thing
def __exit__(self, *exc_info):
self.thing.close()
class _RedirectStream(object):
class aclosing(AbstractAsyncContextManager):
"""Async context manager for safely finalizing an asynchronously cleaned-up
resource such as an async generator, calling its ``aclose()`` method.
Code like this:
async with aclosing(<module>.fetch(<arguments>)) as agen:
<block>
is equivalent to this:
agen = <module>.fetch(<arguments>)
try:
<block>
finally:
await agen.aclose()
"""
def __init__(self, thing):
self.thing = thing
async def __aenter__(self):
return self.thing
async def __aexit__(self, *exc_info):
await self.thing.aclose()
class _RedirectStream(AbstractContextManager):
_stream = None
@ -249,7 +422,7 @@ class redirect_stderr(_RedirectStream):
_stream = "stderr"
class suppress(object):
class suppress(AbstractContextManager):
"""Context manager to suppress specified exceptions
After the exception is suppressed, execution proceeds with the next
@ -279,113 +452,105 @@ class suppress(object):
return exctype is not None and issubclass(exctype, self._exceptions)
# Context manipulation helpers
def _make_context_fixer(frame_exc):
def _fix_exception_context(new_exc, old_exc):
# Context may not be correct, so find the end of the chain
while 1:
exc_context = new_exc.__context__
if exc_context is old_exc:
# Context is already set correctly (see issue 20317)
return
if exc_context is None or exc_context is frame_exc:
break
new_exc = exc_context
# Change the end of the chain to point to the exception
# we expect it to reference
new_exc.__context__ = old_exc
return _fix_exception_context
class _BaseExitStack:
"""A base class for ExitStack and AsyncExitStack."""
def _reraise_with_existing_context(exc_details):
try:
# bare "raise exc_details[1]" replaces our carefully
# set-up context
fixed_ctx = exc_details[1].__context__
raise exc_details[1]
except BaseException:
exc_details[1].__context__ = fixed_ctx
raise
@staticmethod
def _create_exit_wrapper(cm, cm_exit):
return MethodType(cm_exit, cm)
@staticmethod
def _create_cb_wrapper(*args, **kwds):
# Python 3.6/3.7 compatibility: no native positional-only args syntax
callback, *args = args
def _exit_wrapper(exc_type, exc, tb):
callback(*args, **kwds)
return _exit_wrapper
# Inspired by discussions on http://bugs.python.org/issue13585
class ExitStack(object):
"""Context manager for dynamic management of a stack of exit callbacks
For example:
with ExitStack() as stack:
files = [stack.enter_context(open(fname)) for fname in filenames]
# All opened files will automatically be closed at the end of
# the with statement, even if attempts to open files later
# in the list raise an exception
"""
def __init__(self):
self._exit_callbacks = deque()
def pop_all(self):
"""Preserve the context stack by transferring it to a new instance"""
"""Preserve the context stack by transferring it to a new instance."""
new_stack = type(self)()
new_stack._exit_callbacks = self._exit_callbacks
self._exit_callbacks = deque()
return new_stack
def _push_cm_exit(self, cm, cm_exit):
"""Helper to correctly register callbacks to __exit__ methods"""
def _exit_wrapper(*exc_details):
return cm_exit(cm, *exc_details)
_exit_wrapper.__self__ = cm
self.push(_exit_wrapper)
def push(self, exit):
"""Registers a callback with the standard __exit__ method signature
Can suppress exceptions the same way __exit__ methods can.
"""Registers a callback with the standard __exit__ method signature.
Can suppress exceptions the same way __exit__ method can.
Also accepts any object with an __exit__ method (registering a call
to the method instead of the object itself)
to the method instead of the object itself).
"""
# We use an unbound method rather than a bound method to follow
# the standard lookup behaviour for special methods
# the standard lookup behaviour for special methods.
_cb_type = type(exit)
try:
exit_method = _cb_type.__exit__
except AttributeError:
# Not a context manager, so assume its a callable
self._exit_callbacks.append(exit)
# Not a context manager, so assume it's a callable.
self._push_exit_callback(exit)
else:
self._push_cm_exit(exit, exit_method)
return exit # Allow use as a decorator
def callback(self, callback, *args, **kwds):
"""Registers an arbitrary callback and arguments.
Cannot suppress exceptions.
"""
def _exit_wrapper(exc_type, exc, tb):
callback(*args, **kwds)
# We changed the signature, so using @wraps is not appropriate, but
# setting __wrapped__ may still help with introspection
_exit_wrapper.__wrapped__ = callback
self.push(_exit_wrapper)
return callback # Allow use as a decorator
return exit # Allow use as a decorator.
def enter_context(self, cm):
"""Enters the supplied context manager
"""Enters the supplied context manager.
If successful, also pushes its __exit__ method as a callback and
returns the result of the __enter__ method.
"""
# We look up the special methods on the type to match the with statement
# We look up the special methods on the type to match the with
# statement.
_cm_type = type(cm)
_exit = _cm_type.__exit__
result = _cm_type.__enter__(cm)
self._push_cm_exit(cm, _exit)
return result
def close(self):
"""Immediately unwind the context stack"""
self.__exit__(None, None, None)
def callback(*args, **kwds):
"""Registers an arbitrary callback and arguments.
Cannot suppress exceptions.
"""
# Python 3.6/3.7 compatibility: no native positional-only args syntax
try:
self, callback, *args = args
except ValueError as exc:
exc_details = str(exc).partition("(")[2]
msg = "Not enough positional arguments {}".format(exc_details)
raise TypeError(msg) from None
_exit_wrapper = self._create_cb_wrapper(callback, *args, **kwds)
# We changed the signature, so using @wraps is not appropriate, but
# setting __wrapped__ may still help with introspection.
_exit_wrapper.__wrapped__ = callback
self._push_exit_callback(_exit_wrapper)
return callback # Allow use as a decorator
def _push_cm_exit(self, cm, cm_exit):
"""Helper to correctly register callbacks to __exit__ methods."""
_exit_wrapper = self._create_exit_wrapper(cm, cm_exit)
self._push_exit_callback(_exit_wrapper, True)
def _push_exit_callback(self, callback, is_sync=True):
self._exit_callbacks.append((is_sync, callback))
# Inspired by discussions on http://bugs.python.org/issue13585
class ExitStack(_BaseExitStack, AbstractContextManager):
"""Context manager for dynamic management of a stack of exit callbacks.
For example:
with ExitStack() as stack:
files = [stack.enter_context(open(fname)) for fname in filenames]
# All opened files will automatically be closed at the end of
# the with statement, even if attempts to open files later
# in the list raise an exception.
"""
def __enter__(self):
return self
@ -396,14 +561,27 @@ class ExitStack(object):
# We manipulate the exception state so it behaves as though
# we were actually nesting multiple with statements
frame_exc = sys.exc_info()[1]
_fix_exception_context = _make_context_fixer(frame_exc)
def _fix_exception_context(new_exc, old_exc):
# Context may not be correct, so find the end of the chain
while 1:
exc_context = new_exc.__context__
if exc_context is old_exc:
# Context is already set correctly (see issue 20317)
return
if exc_context is None or exc_context is frame_exc:
break
new_exc = exc_context
# Change the end of the chain to point to the exception
# we expect it to reference
new_exc.__context__ = old_exc
# Callbacks are invoked in LIFO order to match the behaviour of
# nested context managers
suppressed_exc = False
pending_raise = False
while self._exit_callbacks:
cb = self._exit_callbacks.pop()
is_sync, cb = self._exit_callbacks.pop()
assert is_sync
try:
if cb(*exc_details):
suppressed_exc = True
@ -416,9 +594,190 @@ class ExitStack(object):
pending_raise = True
exc_details = new_exc_details
if pending_raise:
_reraise_with_existing_context(exc_details)
try:
# bare "raise exc_details[1]" replaces our carefully
# set-up context
fixed_ctx = exc_details[1].__context__
raise exc_details[1]
except BaseException:
exc_details[1].__context__ = fixed_ctx
raise
return received_exc and suppressed_exc
def close(self):
"""Immediately unwind the context stack."""
self.__exit__(None, None, None)
# Inspired by discussions on https://bugs.python.org/issue29302
class AsyncExitStack(_BaseExitStack, AbstractAsyncContextManager):
"""Async context manager for dynamic management of a stack of exit
callbacks.
For example:
async with AsyncExitStack() as stack:
connections = [await stack.enter_async_context(get_connection())
for i in range(5)]
# All opened connections will automatically be released at the
# end of the async with statement, even if attempts to open a
# connection later in the list raise an exception.
"""
@staticmethod
def _create_async_exit_wrapper(cm, cm_exit):
return MethodType(cm_exit, cm)
@staticmethod
def _create_async_cb_wrapper(*args, **kwds):
# Python 3.6/3.7 compatibility: no native positional-only args syntax
callback, *args = args
async def _exit_wrapper(exc_type, exc, tb):
await callback(*args, **kwds)
return _exit_wrapper
async def enter_async_context(self, cm):
"""Enters the supplied async context manager.
If successful, also pushes its __aexit__ method as a callback and
returns the result of the __aenter__ method.
"""
_cm_type = type(cm)
_exit = _cm_type.__aexit__
result = await _cm_type.__aenter__(cm)
self._push_async_cm_exit(cm, _exit)
return result
def push_async_exit(self, exit):
"""Registers a coroutine function with the standard __aexit__ method
signature.
Can suppress exceptions the same way __aexit__ method can.
Also accepts any object with an __aexit__ method (registering a call
to the method instead of the object itself).
"""
_cb_type = type(exit)
try:
exit_method = _cb_type.__aexit__
except AttributeError:
# Not an async context manager, so assume it's a coroutine function
self._push_exit_callback(exit, False)
else:
self._push_async_cm_exit(exit, exit_method)
return exit # Allow use as a decorator
def push_async_callback(*args, **kwds):
"""Registers an arbitrary coroutine function and arguments.
Cannot suppress exceptions.
"""
# Python 3.6/3.7 compatibility: no native positional-only args syntax
try:
self, callback, *args = args
except ValueError as exc:
exc_details = str(exc).partition("(")[2]
msg = "Not enough positional arguments {}".format(exc_details)
raise TypeError(msg) from None
_exit_wrapper = self._create_async_cb_wrapper(callback, *args, **kwds)
# We changed the signature, so using @wraps is not appropriate, but
# setting __wrapped__ may still help with introspection.
_exit_wrapper.__wrapped__ = callback
self._push_exit_callback(_exit_wrapper, False)
return callback # Allow use as a decorator
async def aclose(self):
"""Immediately unwind the context stack."""
await self.__aexit__(None, None, None)
def _push_async_cm_exit(self, cm, cm_exit):
"""Helper to correctly register coroutine function to __aexit__
method."""
_exit_wrapper = self._create_async_exit_wrapper(cm, cm_exit)
self._push_exit_callback(_exit_wrapper, False)
async def __aenter__(self):
return self
async def __aexit__(self, *exc_details):
received_exc = exc_details[0] is not None
# We manipulate the exception state so it behaves as though
# we were actually nesting multiple with statements
frame_exc = sys.exc_info()[1]
def _fix_exception_context(new_exc, old_exc):
# Context may not be correct, so find the end of the chain
while 1:
exc_context = new_exc.__context__
if exc_context is old_exc:
# Context is already set correctly (see issue 20317)
return
if exc_context is None or exc_context is frame_exc:
break
new_exc = exc_context
# Change the end of the chain to point to the exception
# we expect it to reference
new_exc.__context__ = old_exc
# Callbacks are invoked in LIFO order to match the behaviour of
# nested context managers
suppressed_exc = False
pending_raise = False
while self._exit_callbacks:
is_sync, cb = self._exit_callbacks.pop()
try:
if is_sync:
cb_suppress = cb(*exc_details)
else:
cb_suppress = await cb(*exc_details)
if cb_suppress:
suppressed_exc = True
pending_raise = False
exc_details = (None, None, None)
except:
new_exc_details = sys.exc_info()
# simulate the stack of exceptions by setting the context
_fix_exception_context(new_exc_details[1], exc_details[1])
pending_raise = True
exc_details = new_exc_details
if pending_raise:
try:
# bare "raise exc_details[1]" replaces our carefully
# set-up context
fixed_ctx = exc_details[1].__context__
raise exc_details[1]
except BaseException:
exc_details[1].__context__ = fixed_ctx
raise
return received_exc and suppressed_exc
class nullcontext(AbstractContextManager, AbstractAsyncContextManager):
"""Context manager that does no additional processing.
Used as a stand-in for a normal context manager, when a particular
block of code is only sometimes used with a normal context manager:
cm = optional_cm if condition else nullcontext()
with cm:
# Perform operation, using optional_cm if condition is True
"""
def __init__(self, enter_result=None):
self.enter_result = enter_result
def __enter__(self):
return self.enter_result
def __exit__(self, *excinfo):
pass
async def __aenter__(self):
return self.enter_result
async def __aexit__(self, *excinfo):
pass
# Preserve backwards compatibility
class ContextStack(ExitStack):
@ -437,22 +796,3 @@ class ContextStack(ExitStack):
def preserve(self):
return self.pop_all()
class nullcontext(AbstractContextManager):
"""Context manager that does no additional processing.
Used as a stand-in for a normal context manager, when a particular
block of code is only sometimes used with a normal context manager:
cm = optional_cm if condition else nullcontext()
with cm:
# Perform operation, using optional_cm if condition is True
"""
def __init__(self, enter_result=None):
self.enter_result = enter_result
def __enter__(self):
return self.enter_result
def __exit__(self, *excinfo):
pass

147
dev/py3_10_contextlib_to_contextlib2.patch Normal file
View file

@ -0,0 +1,147 @@
--- ../cpython/Lib/contextlib.py 2021-06-26 16:28:03.835372955 +1000
+++ contextlib2.py 2021-06-26 17:40:30.047079570 +1000
@@ -1,19 +1,32 @@
-"""Utilities for with-statement contexts. See PEP 343."""
+"""contextlib2 - backports and enhancements to the contextlib module"""
+
import abc
import sys
+import warnings
import _collections_abc
from collections import deque
from functools import wraps
-from types import MethodType, GenericAlias
+from types import MethodType
+
+# Python 3.6/3.7/3.8 compatibility: GenericAlias may not be defined
+try:
+ from types import GenericAlias
+except ImportError:
+ # If the real GenericAlias type doesn't exist, __class_getitem__ won't be used,
+ # so the fallback placeholder doesn't need to provide any meaningful behaviour
+ class GenericAlias:
+ pass
+
__all__ = ["asynccontextmanager", "contextmanager", "closing", "nullcontext",
"AbstractContextManager", "AbstractAsyncContextManager",
"AsyncExitStack", "ContextDecorator", "ExitStack",
"redirect_stdout", "redirect_stderr", "suppress", "aclosing"]
+# Backwards compatibility
+__all__ += ["ContextStack"]
class AbstractContextManager(abc.ABC):
-
"""An abstract base class for context managers."""
__class_getitem__ = classmethod(GenericAlias)
@@ -60,6 +73,23 @@
class ContextDecorator(object):
"A base class or mixin that enables context managers to work as decorators."
+ def refresh_cm(self):
+ """Returns the context manager used to actually wrap the call to the
+ decorated function.
+
+ The default implementation just returns *self*.
+
+ Overriding this method allows otherwise one-shot context managers
+ like _GeneratorContextManager to support use as decorators via
+ implicit recreation.
+
+ DEPRECATED: refresh_cm was never added to the standard library's
+ ContextDecorator API
+ """
+ warnings.warn("refresh_cm was never added to the standard library",
+ DeprecationWarning)
+ return self._recreate_cm()
+
def _recreate_cm(self):
"""Return a recreated instance of self.
@@ -430,7 +460,9 @@
return MethodType(cm_exit, cm)
@staticmethod
- def _create_cb_wrapper(callback, /, *args, **kwds):
+ def _create_cb_wrapper(*args, **kwds):
+ # Python 3.6/3.7 compatibility: no native positional-only args syntax
+ callback, *args = args
def _exit_wrapper(exc_type, exc, tb):
callback(*args, **kwds)
return _exit_wrapper
@@ -479,11 +511,18 @@
self._push_cm_exit(cm, _exit)
return result
- def callback(self, callback, /, *args, **kwds):
+ def callback(*args, **kwds):
"""Registers an arbitrary callback and arguments.
Cannot suppress exceptions.
"""
+ # Python 3.6/3.7 compatibility: no native positional-only args syntax
+ try:
+ self, callback, *args = args
+ except ValueError as exc:
+ exc_details = str(exc).partition("(")[2]
+ msg = "Not enough positional arguments {}".format(exc_details)
+ raise TypeError(msg) from None
_exit_wrapper = self._create_cb_wrapper(callback, *args, **kwds)
# We changed the signature, so using @wraps is not appropriate, but
@@ -589,7 +628,9 @@
return MethodType(cm_exit, cm)
@staticmethod
- def _create_async_cb_wrapper(callback, /, *args, **kwds):
+ def _create_async_cb_wrapper(*args, **kwds):
+ # Python 3.6/3.7 compatibility: no native positional-only args syntax
+ callback, *args = args
async def _exit_wrapper(exc_type, exc, tb):
await callback(*args, **kwds)
return _exit_wrapper
@@ -624,11 +665,18 @@
self._push_async_cm_exit(exit, exit_method)
return exit # Allow use as a decorator
- def push_async_callback(self, callback, /, *args, **kwds):
+ def push_async_callback(*args, **kwds):
"""Registers an arbitrary coroutine function and arguments.
Cannot suppress exceptions.
"""
+ # Python 3.6/3.7 compatibility: no native positional-only args syntax
+ try:
+ self, callback, *args = args
+ except ValueError as exc:
+ exc_details = str(exc).partition("(")[2]
+ msg = "Not enough positional arguments {}".format(exc_details)
+ raise TypeError(msg) from None
_exit_wrapper = self._create_async_cb_wrapper(callback, *args, **kwds)
# We changed the signature, so using @wraps is not appropriate, but
@@ -729,3 +777,22 @@
async def __aexit__(self, *excinfo):
pass
+
+
+# Preserve backwards compatibility
+class ContextStack(ExitStack):
+ """Backwards compatibility alias for ExitStack"""
+
+ def __init__(self):
+ warnings.warn("ContextStack has been renamed to ExitStack",
+ DeprecationWarning)
+ super(ContextStack, self).__init__()
+
+ def register_exit(self, callback):
+ return self.push(callback)
+
+ def register(self, callback, *args, **kwds):
+ return self.callback(callback, *args, **kwds)
+
+ def preserve(self):
+ return self.pop_all()

1
test/__init__.py Normal file
View file

@ -0,0 +1 @@
# unittest test discovery requires an __init__.py file in the test directory

6
test/support/__init__.py Normal file
View file

@ -0,0 +1,6 @@
"""Enough of the test.support APIs to run the contextlib test suite"""
import sys
import unittest
requires_docstrings = unittest.skipIf(sys.flags.optimize >= 2,
"Test requires docstrings")

4
test/support/os_helper.py Normal file
View file

@ -0,0 +1,4 @@
"""Enough of the test.support.os_helper APIs to run the contextlib test suite"""
import os
unlink = os.unlink

364
test_contextlib2.py → test/test_contextlib.py Executable file → Normal file
View file

@ -1,16 +1,15 @@
"""Unit tests for contextlib2.py"""
from __future__ import print_function
from __future__ import unicode_literals
"""Unit tests for contextlib.py, and other context managers."""
import io
import sys
import tempfile
import threading
import unittest
import __future__ # For PEP 479 conditional test
import contextlib2
from contextlib2 import * # Tests __all__
requires_docstrings = unittest.skipIf(sys.flags.optimize >= 2,
"Test requires docstrings")
from test import support
from test.support import os_helper
import weakref
import gc
class TestAbstractContextManager(unittest.TestCase):
@ -31,8 +30,7 @@ class TestAbstractContextManager(unittest.TestCase):
MissingExit()
def test_structural_subclassing(self):
# New style classes used here
class ManagerFromScratch(object):
class ManagerFromScratch:
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
@ -46,22 +44,15 @@ class TestAbstractContextManager(unittest.TestCase):
self.assertTrue(issubclass(DefaultEnter, AbstractContextManager))
if sys.version_info[:2] <= (3, 0):
def test_structural_subclassing_classic(self):
# Old style classes used here
class ManagerFromScratch:
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
return None
class NoEnter(ManagerFromScratch):
__enter__ = None
self.assertTrue(issubclass(ManagerFromScratch, AbstractContextManager))
self.assertFalse(issubclass(NoEnter, AbstractContextManager))
class DefaultEnter(AbstractContextManager):
def __exit__(self, *args):
super().__exit__(*args)
class NoExit(ManagerFromScratch):
__exit__ = None
self.assertTrue(issubclass(DefaultEnter, AbstractContextManager))
self.assertFalse(issubclass(NoExit, AbstractContextManager))
class ContextManagerTestCase(unittest.TestCase):
@ -141,7 +132,7 @@ class ContextManagerTestCase(unittest.TestCase):
def woohoo():
yield
try:
with self.assertWarnsRegex(PendingDeprecationWarning,
with self.assertWarnsRegex(DeprecationWarning,
"StopIteration"):
with woohoo():
raise stop_exc
@ -150,8 +141,6 @@ class ContextManagerTestCase(unittest.TestCase):
else:
self.fail('StopIteration was suppressed')
@unittest.skipUnless(hasattr(__future__, "generator_stop"),
"Test only valid for versions implementing PEP 479")
def test_contextmanager_except_pep479(self):
code = """\
from __future__ import generator_stop
@ -173,6 +162,29 @@ def woohoo():
else:
self.fail('StopIteration was suppressed')
def test_contextmanager_do_not_unchain_non_stopiteration_exceptions(self):
@contextmanager
def test_issue29692():
try:
yield
except Exception as exc:
raise RuntimeError('issue29692:Chained') from exc
try:
with test_issue29692():
raise ZeroDivisionError
except Exception as ex:
self.assertIs(type(ex), RuntimeError)
self.assertEqual(ex.args[0], 'issue29692:Chained')
self.assertIsInstance(ex.__cause__, ZeroDivisionError)
try:
with test_issue29692():
raise StopIteration('issue29692:Unchained')
except Exception as ex:
self.assertIs(type(ex), StopIteration)
self.assertEqual(ex.args[0], 'issue29692:Unchained')
self.assertIsNone(ex.__cause__)
def _create_contextmanager_attribs(self):
def attribs(**kw):
def decorate(func):
@ -191,12 +203,12 @@ def woohoo():
self.assertEqual(baz.__name__,'baz')
self.assertEqual(baz.foo, 'bar')
@requires_docstrings
@support.requires_docstrings
def test_contextmanager_doc_attrib(self):
baz = self._create_contextmanager_attribs()
self.assertEqual(baz.__doc__, "Whee!")
@requires_docstrings
@support.requires_docstrings
def test_instance_docstring_given_cm_docstring(self):
baz = self._create_contextmanager_attribs()(None)
self.assertEqual(baz.__doc__, "Whee!")
@ -209,10 +221,58 @@ def woohoo():
with woohoo(self=11, func=22, args=33, kwds=44) as target:
self.assertEqual(target, (11, 22, 33, 44))
def test_nokeepref(self):
class A:
pass
@contextmanager
def woohoo(a, b):
a = weakref.ref(a)
b = weakref.ref(b)
# Allow test to work with a non-refcounted GC
gc.collect(); gc.collect(); gc.collect()
self.assertIsNone(a())
self.assertIsNone(b())
yield
with woohoo(A(), b=A()):
pass
def test_param_errors(self):
@contextmanager
def woohoo(a, *, b):
yield
with self.assertRaises(TypeError):
woohoo()
with self.assertRaises(TypeError):
woohoo(3, 5)
with self.assertRaises(TypeError):
woohoo(b=3)
def test_recursive(self):
depth = 0
@contextmanager
def woohoo():
nonlocal depth
before = depth
depth += 1
yield
depth -= 1
self.assertEqual(depth, before)
@woohoo()
def recursive():
if depth < 10:
recursive()
recursive()
self.assertEqual(depth, 0)
class ClosingTestCase(unittest.TestCase):
@requires_docstrings
@support.requires_docstrings
def test_instance_docs(self):
# Issue 19330: ensure context manager instances have good docstrings
cm_docstring = closing.__doc__
@ -244,6 +304,83 @@ class ClosingTestCase(unittest.TestCase):
self.assertEqual(state, [1])
class NullcontextTestCase(unittest.TestCase):
def test_nullcontext(self):
class C:
pass
c = C()
with nullcontext(c) as c_in:
self.assertIs(c_in, c)
class FileContextTestCase(unittest.TestCase):
def testWithOpen(self):
tfn = tempfile.mktemp()
try:
f = None
with open(tfn, "w", encoding="utf-8") as f:
self.assertFalse(f.closed)
f.write("Booh\n")
self.assertTrue(f.closed)
f = None
with self.assertRaises(ZeroDivisionError):
with open(tfn, "r", encoding="utf-8") as f:
self.assertFalse(f.closed)
self.assertEqual(f.read(), "Booh\n")
1 / 0
self.assertTrue(f.closed)
finally:
os_helper.unlink(tfn)
class LockContextTestCase(unittest.TestCase):
def boilerPlate(self, lock, locked):
self.assertFalse(locked())
with lock:
self.assertTrue(locked())
self.assertFalse(locked())
with self.assertRaises(ZeroDivisionError):
with lock:
self.assertTrue(locked())
1 / 0
self.assertFalse(locked())
def testWithLock(self):
lock = threading.Lock()
self.boilerPlate(lock, lock.locked)
def testWithRLock(self):
lock = threading.RLock()
self.boilerPlate(lock, lock._is_owned)
def testWithCondition(self):
lock = threading.Condition()
def locked():
return lock._is_owned()
self.boilerPlate(lock, locked)
def testWithSemaphore(self):
lock = threading.Semaphore()
def locked():
if lock.acquire(False):
lock.release()
return False
else:
return True
self.boilerPlate(lock, locked)
def testWithBoundedSemaphore(self):
lock = threading.BoundedSemaphore()
def locked():
if lock.acquire(False):
lock.release()
return False
else:
return True
self.boilerPlate(lock, locked)
class mycontext(ContextDecorator):
"""Example decoration-compatible context manager for testing"""
started = False
@ -261,7 +398,7 @@ class mycontext(ContextDecorator):
class TestContextDecorator(unittest.TestCase):
@requires_docstrings
@support.requires_docstrings
def test_instance_docs(self):
# Issue 19330: ensure context manager instances have good docstrings
cm_docstring = mycontext.__doc__
@ -418,17 +555,19 @@ class TestContextDecorator(unittest.TestCase):
test('something else')
self.assertEqual(state, [1, 'something else', 999])
class TestExitStack(unittest.TestCase):
@requires_docstrings
class TestBaseExitStack:
exit_stack = None
@support.requires_docstrings
def test_instance_docs(self):
# Issue 19330: ensure context manager instances have good docstrings
cm_docstring = ExitStack.__doc__
obj = ExitStack()
cm_docstring = self.exit_stack.__doc__
obj = self.exit_stack()
self.assertEqual(obj.__doc__, cm_docstring)
def test_no_resources(self):
with ExitStack():
with self.exit_stack():
pass
def test_callback(self):
@ -439,12 +578,13 @@ class TestExitStack(unittest.TestCase):
((), dict(example=1)),
((1,), dict(example=1)),
((1,2), dict(example=1)),
((1,2), dict(self=3, callback=4)),
]
result = []
def _exit(*args, **kwds):
"""Test metadata propagation"""
result.append((args, kwds))
with ExitStack() as stack:
with self.exit_stack() as stack:
for args, kwds in reversed(expected):
if args and kwds:
f = stack.callback(_exit, *args, **kwds)
@ -456,11 +596,21 @@ class TestExitStack(unittest.TestCase):
f = stack.callback(_exit)
self.assertIs(f, _exit)
for wrapper in stack._exit_callbacks:
self.assertIs(wrapper.__wrapped__, _exit)
self.assertNotEqual(wrapper.__name__, _exit.__name__)
self.assertIsNone(wrapper.__doc__, _exit.__doc__)
self.assertIs(wrapper[1].__wrapped__, _exit)
self.assertNotEqual(wrapper[1].__name__, _exit.__name__)
self.assertIsNone(wrapper[1].__doc__, _exit.__doc__)
self.assertEqual(result, expected)
result = []
with self.exit_stack() as stack:
with self.assertRaises(TypeError):
stack.callback(arg=1)
with self.assertRaises(TypeError):
self.exit_stack.callback(arg=2)
with self.assertRaises(TypeError):
stack.callback(callback=_exit, arg=3)
self.assertEqual(result, [])
def test_push(self):
exc_raised = ZeroDivisionError
def _expect_exc(exc_type, exc, exc_tb):
@ -478,21 +628,21 @@ class TestExitStack(unittest.TestCase):
self.fail("Should not be called!")
def __exit__(self, *exc_details):
self.check_exc(*exc_details)
with ExitStack() as stack:
with self.exit_stack() as stack:
stack.push(_expect_ok)
self.assertIs(stack._exit_callbacks[-1], _expect_ok)
self.assertIs(stack._exit_callbacks[-1][1], _expect_ok)
cm = ExitCM(_expect_ok)
stack.push(cm)
self.assertIs(stack._exit_callbacks[-1].__self__, cm)
self.assertIs(stack._exit_callbacks[-1][1].__self__, cm)
stack.push(_suppress_exc)
self.assertIs(stack._exit_callbacks[-1], _suppress_exc)
self.assertIs(stack._exit_callbacks[-1][1], _suppress_exc)
cm = ExitCM(_expect_exc)
stack.push(cm)
self.assertIs(stack._exit_callbacks[-1].__self__, cm)
self.assertIs(stack._exit_callbacks[-1][1].__self__, cm)
stack.push(_expect_exc)
self.assertIs(stack._exit_callbacks[-1], _expect_exc)
self.assertIs(stack._exit_callbacks[-1][1], _expect_exc)
stack.push(_expect_exc)
self.assertIs(stack._exit_callbacks[-1], _expect_exc)
self.assertIs(stack._exit_callbacks[-1][1], _expect_exc)
1/0
def test_enter_context(self):
@ -504,19 +654,19 @@ class TestExitStack(unittest.TestCase):
result = []
cm = TestCM()
with ExitStack() as stack:
with self.exit_stack() as stack:
@stack.callback # Registered first => cleaned up last
def _exit():
result.append(4)
self.assertIsNotNone(_exit)
stack.enter_context(cm)
self.assertIs(stack._exit_callbacks[-1].__self__, cm)
self.assertIs(stack._exit_callbacks[-1][1].__self__, cm)
result.append(2)
self.assertEqual(result, [1, 2, 3, 4])
def test_close(self):
result = []
with ExitStack() as stack:
with self.exit_stack() as stack:
@stack.callback
def _exit():
result.append(1)
@ -527,7 +677,7 @@ class TestExitStack(unittest.TestCase):
def test_pop_all(self):
result = []
with ExitStack() as stack:
with self.exit_stack() as stack:
@stack.callback
def _exit():
result.append(3)
@ -540,12 +690,12 @@ class TestExitStack(unittest.TestCase):
def test_exit_raise(self):
with self.assertRaises(ZeroDivisionError):
with ExitStack() as stack:
with self.exit_stack() as stack:
stack.push(lambda *exc: False)
1/0
def test_exit_suppress(self):
with ExitStack() as stack:
with self.exit_stack() as stack:
stack.push(lambda *exc: True)
1/0
@ -576,7 +726,7 @@ class TestExitStack(unittest.TestCase):
def __enter__(self):
return self
def __exit__(self, *exc_details):
self.__class__.saved_details = exc_details
type(self).saved_details = exc_details
return True
try:
@ -602,13 +752,14 @@ class TestExitStack(unittest.TestCase):
def raise_exc(exc):
raise exc
saved_details = [None]
saved_details = None
def suppress_exc(*exc_details):
saved_details[0] = exc_details
nonlocal saved_details
saved_details = exc_details
return True
try:
with ExitStack() as stack:
with self.exit_stack() as stack:
stack.callback(raise_exc, IndexError)
stack.callback(raise_exc, KeyError)
stack.callback(raise_exc, AttributeError)
@ -623,7 +774,7 @@ class TestExitStack(unittest.TestCase):
else:
self.fail("Expected IndexError, but no exception was raised")
# Check the inner exceptions
inner_exc = saved_details[0][1]
inner_exc = saved_details[1]
self.assertIsInstance(inner_exc, ValueError)
self.assertIsInstance(inner_exc.__context__, ZeroDivisionError)
@ -636,7 +787,7 @@ class TestExitStack(unittest.TestCase):
return True
try:
with ExitStack() as stack:
with self.exit_stack() as stack:
stack.callback(lambda: None)
stack.callback(raise_exc, IndexError)
except Exception as exc:
@ -645,7 +796,7 @@ class TestExitStack(unittest.TestCase):
self.fail("Expected IndexError, but no exception was raised")
try:
with ExitStack() as stack:
with self.exit_stack() as stack:
stack.callback(raise_exc, KeyError)
stack.push(suppress_exc)
stack.callback(raise_exc, IndexError)
@ -672,7 +823,7 @@ class TestExitStack(unittest.TestCase):
# fix, ExitStack would try to fix it *again* and get into an
# infinite self-referential loop
try:
with ExitStack() as stack:
with self.exit_stack() as stack:
stack.enter_context(gets_the_context_right(exc4))
stack.enter_context(gets_the_context_right(exc3))
stack.enter_context(gets_the_context_right(exc2))
@ -683,7 +834,7 @@ class TestExitStack(unittest.TestCase):
self.assertIs(exc.__context__.__context__, exc2)
self.assertIs(exc.__context__.__context__.__context__, exc1)
self.assertIsNone(
exc.__context__.__context__.__context__.__context__)
exc.__context__.__context__.__context__.__context__)
def test_exit_exception_with_existing_context(self):
# Addresses a lack of test coverage discovered after checking in a
@ -699,7 +850,7 @@ class TestExitStack(unittest.TestCase):
exc4 = Exception(4)
exc5 = Exception(5)
try:
with ExitStack() as stack:
with self.exit_stack() as stack:
stack.callback(raise_nested, exc4, exc5)
stack.callback(raise_nested, exc2, exc3)
raise exc1
@ -709,7 +860,7 @@ class TestExitStack(unittest.TestCase):
self.assertIs(exc.__context__.__context__, exc3)
self.assertIs(exc.__context__.__context__.__context__, exc2)
self.assertIs(
exc.__context__.__context__.__context__.__context__, exc1)
exc.__context__.__context__.__context__.__context__, exc1)
self.assertIsNone(
exc.__context__.__context__.__context__.__context__.__context__)
@ -717,21 +868,21 @@ class TestExitStack(unittest.TestCase):
def suppress_exc(*exc_details):
return True
try:
with ExitStack() as stack:
with self.exit_stack() as stack:
stack.push(suppress_exc)
1/0
except IndexError as exc:
self.fail("Expected no exception, got IndexError")
def test_exit_exception_chaining_suppress(self):
with ExitStack() as stack:
with self.exit_stack() as stack:
stack.push(lambda *exc: True)
stack.push(lambda *exc: 1/0)
stack.push(lambda *exc: {}[1])
def test_excessive_nesting(self):
# The original implementation would die with RecursionError here
with ExitStack() as stack:
with self.exit_stack() as stack:
for i in range(10000):
stack.callback(int)
@ -739,43 +890,10 @@ class TestExitStack(unittest.TestCase):
class Example(object): pass
cm = Example()
cm.__exit__ = object()
stack = ExitStack()
stack = self.exit_stack()
self.assertRaises(AttributeError, stack.enter_context, cm)
stack.push(cm)
self.assertIs(stack._exit_callbacks[-1], cm)
def test_default_class_semantics(self):
# For Python 2.x, this ensures compatibility with old-style classes
# For Python 3.x, it just reruns some of the other tests
class DefaultCM:
def __enter__(self):
result.append("Enter")
def __exit__(self, *exc_details):
result.append("Exit")
class DefaultCallable:
def __call__(self, *exc_details):
result.append("Callback")
result = []
cm = DefaultCM()
cb = DefaultCallable()
with ExitStack() as stack:
stack.enter_context(cm)
self.assertIs(stack._exit_callbacks[-1].__self__, cm)
stack.push(cb)
stack.push(cm)
self.assertIs(stack._exit_callbacks[-1].__self__, cm)
result.append("Running")
stack.callback(cb)
self.assertIs(stack._exit_callbacks[-1].__wrapped__, cb)
self.assertEqual(result, ["Enter", "Running",
"Callback", "Exit",
"Callback", "Exit",
])
with ExitStack():
pass
self.assertIs(stack._exit_callbacks[-1][1], cm)
def test_dont_reraise_RuntimeError(self):
# https://bugs.python.org/issue27122
@ -787,10 +905,7 @@ class TestExitStack(unittest.TestCase):
try:
yield 1
except Exception as exc:
# Py2 compatible explicit exception chaining
new_exc = UniqueException("new exception")
new_exc.__cause__ = exc
raise new_exc
raise UniqueException("new exception") from exc
@contextmanager
def first():
@ -802,17 +917,21 @@ class TestExitStack(unittest.TestCase):
# The UniqueRuntimeError should be caught by second()'s exception
# handler which chain raised a new UniqueException.
with self.assertRaises(UniqueException) as err_ctx:
with ExitStack() as es_ctx:
with self.exit_stack() as es_ctx:
es_ctx.enter_context(second())
es_ctx.enter_context(first())
raise UniqueRuntimeError("please no infinite loop.")
exc = err_ctx.exception
self.assertIsInstance(exc, UniqueException)
self.assertIsInstance(exc.__cause__, UniqueRuntimeError)
self.assertIs(exc.__context__, exc.__cause__)
self.assertIsNone(exc.__cause__.__context__)
self.assertIsNone(exc.__cause__.__cause__)
self.assertIsInstance(exc.__context__, UniqueRuntimeError)
self.assertIsNone(exc.__context__.__context__)
self.assertIsNone(exc.__context__.__cause__)
self.assertIs(exc.__cause__, exc.__context__)
class TestExitStack(TestBaseExitStack, unittest.TestCase):
exit_stack = ExitStack
class TestRedirectStream:
@ -820,7 +939,7 @@ class TestRedirectStream:
redirect_stream = None
orig_stream = None
@requires_docstrings
@support.requires_docstrings
def test_instance_docs(self):
# Issue 19330: ensure context manager instances have good docstrings
cm_docstring = self.redirect_stream.__doc__
@ -871,11 +990,6 @@ class TestRedirectStream:
s = f.getvalue()
self.assertEqual(s, "Hello World!\n")
def test_cm_is_exitstack_compatible(self):
with ExitStack() as stack:
# This shouldn't raise an exception.
stack.enter_context(self.redirect_stream(io.StringIO()))
class TestRedirectStdout(TestRedirectStream, unittest.TestCase):
@ -891,7 +1005,7 @@ class TestRedirectStderr(TestRedirectStream, unittest.TestCase):
class TestSuppress(unittest.TestCase):
@requires_docstrings
@support.requires_docstrings
def test_instance_docs(self):
# Issue 19330: ensure context manager instances have good docstrings
cm_docstring = suppress.__doc__
@ -943,21 +1057,5 @@ class TestSuppress(unittest.TestCase):
1/0
self.assertTrue(outer_continued)
def test_cm_is_exitstack_compatible(self):
with ExitStack() as stack:
# This shouldn't raise an exception.
stack.enter_context(suppress())
class NullcontextTestCase(unittest.TestCase):
def test_nullcontext(self):
class C:
pass
c = C()
with nullcontext(c) as c_in:
self.assertIs(c_in, c)
if __name__ == "__main__":
import unittest
unittest.main()

554
test/test_contextlib_async.py Normal file
View file

@ -0,0 +1,554 @@
import asyncio
from contextlib2 import (
asynccontextmanager, AbstractAsyncContextManager,
AsyncExitStack, nullcontext, aclosing, contextmanager)
import functools
from test import support
import unittest
from test.test_contextlib import TestBaseExitStack
def _async_test(func):
"""Decorator to turn an async function into a test case."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
coro = func(*args, **kwargs)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
return loop.run_until_complete(coro)
finally:
loop.close()
asyncio.set_event_loop_policy(None)
return wrapper
class TestAbstractAsyncContextManager(unittest.TestCase):
@_async_test
async def test_enter(self):
class DefaultEnter(AbstractAsyncContextManager):
async def __aexit__(self, *args):
await super().__aexit__(*args)
manager = DefaultEnter()
self.assertIs(await manager.__aenter__(), manager)
async with manager as context:
self.assertIs(manager, context)
@_async_test
async def test_async_gen_propagates_generator_exit(self):
# A regression test for https://bugs.python.org/issue33786.
@asynccontextmanager
async def ctx():
yield
async def gen():
async with ctx():
yield 11
ret = []
exc = ValueError(22)
with self.assertRaises(ValueError):
async with ctx():
async for val in gen():
ret.append(val)
raise exc
self.assertEqual(ret, [11])
def test_exit_is_abstract(self):
class MissingAexit(AbstractAsyncContextManager):
pass
with self.assertRaises(TypeError):
MissingAexit()
def test_structural_subclassing(self):
class ManagerFromScratch:
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_value, traceback):
return None
self.assertTrue(issubclass(ManagerFromScratch, AbstractAsyncContextManager))
class DefaultEnter(AbstractAsyncContextManager):
async def __aexit__(self, *args):
await super().__aexit__(*args)
self.assertTrue(issubclass(DefaultEnter, AbstractAsyncContextManager))
class NoneAenter(ManagerFromScratch):
__aenter__ = None
self.assertFalse(issubclass(NoneAenter, AbstractAsyncContextManager))
class NoneAexit(ManagerFromScratch):
__aexit__ = None
self.assertFalse(issubclass(NoneAexit, AbstractAsyncContextManager))
class AsyncContextManagerTestCase(unittest.TestCase):
@_async_test
async def test_contextmanager_plain(self):
state = []
@asynccontextmanager
async def woohoo():
state.append(1)
yield 42
state.append(999)
async with woohoo() as x:
self.assertEqual(state, [1])
self.assertEqual(x, 42)
state.append(x)
self.assertEqual(state, [1, 42, 999])
@_async_test
async def test_contextmanager_finally(self):
state = []
@asynccontextmanager
async def woohoo():
state.append(1)
try:
yield 42
finally:
state.append(999)
with self.assertRaises(ZeroDivisionError):
async with woohoo() as x:
self.assertEqual(state, [1])
self.assertEqual(x, 42)
state.append(x)
raise ZeroDivisionError()
self.assertEqual(state, [1, 42, 999])
@_async_test
async def test_contextmanager_no_reraise(self):
@asynccontextmanager
async def whee():
yield
ctx = whee()
await ctx.__aenter__()
# Calling __aexit__ should not result in an exception
self.assertFalse(await ctx.__aexit__(TypeError, TypeError("foo"), None))
@_async_test
async def test_contextmanager_trap_yield_after_throw(self):
@asynccontextmanager
async def whoo():
try:
yield
except:
yield
ctx = whoo()
await ctx.__aenter__()
with self.assertRaises(RuntimeError):
await ctx.__aexit__(TypeError, TypeError('foo'), None)
@_async_test
async def test_contextmanager_trap_no_yield(self):
@asynccontextmanager
async def whoo():
if False:
yield
ctx = whoo()
with self.assertRaises(RuntimeError):
await ctx.__aenter__()
@_async_test
async def test_contextmanager_trap_second_yield(self):
@asynccontextmanager
async def whoo():
yield
yield
ctx = whoo()
await ctx.__aenter__()
with self.assertRaises(RuntimeError):
await ctx.__aexit__(None, None, None)
@_async_test
async def test_contextmanager_non_normalised(self):
@asynccontextmanager
async def whoo():
try:
yield
except RuntimeError:
raise SyntaxError
ctx = whoo()
await ctx.__aenter__()
with self.assertRaises(SyntaxError):
await ctx.__aexit__(RuntimeError, None, None)
@_async_test
async def test_contextmanager_except(self):
state = []
@asynccontextmanager
async def woohoo():
state.append(1)
try:
yield 42
except ZeroDivisionError as e:
state.append(e.args[0])
self.assertEqual(state, [1, 42, 999])
async with woohoo() as x:
self.assertEqual(state, [1])
self.assertEqual(x, 42)
state.append(x)
raise ZeroDivisionError(999)
self.assertEqual(state, [1, 42, 999])
@_async_test
async def test_contextmanager_except_stopiter(self):
@asynccontextmanager
async def woohoo():
yield
for stop_exc in (StopIteration('spam'), StopAsyncIteration('ham')):
with self.subTest(type=type(stop_exc)):
try:
async with woohoo():
raise stop_exc
except Exception as ex:
self.assertIs(ex, stop_exc)
else:
self.fail(f'{stop_exc} was suppressed')
@_async_test
async def test_contextmanager_wrap_runtimeerror(self):
@asynccontextmanager
async def woohoo():
try:
yield
except Exception as exc:
raise RuntimeError(f'caught {exc}') from exc
with self.assertRaises(RuntimeError):
async with woohoo():
1 / 0
# If the context manager wrapped StopAsyncIteration in a RuntimeError,
# we also unwrap it, because we can't tell whether the wrapping was
# done by the generator machinery or by the generator itself.
with self.assertRaises(StopAsyncIteration):
async with woohoo():
raise StopAsyncIteration
def _create_contextmanager_attribs(self):
def attribs(**kw):
def decorate(func):
for k,v in kw.items():
setattr(func,k,v)
return func
return decorate
@asynccontextmanager
@attribs(foo='bar')
async def baz(spam):
"""Whee!"""
yield
return baz
def test_contextmanager_attribs(self):
baz = self._create_contextmanager_attribs()
self.assertEqual(baz.__name__,'baz')
self.assertEqual(baz.foo, 'bar')
@support.requires_docstrings
def test_contextmanager_doc_attrib(self):
baz = self._create_contextmanager_attribs()
self.assertEqual(baz.__doc__, "Whee!")
@support.requires_docstrings
@_async_test
async def test_instance_docstring_given_cm_docstring(self):
baz = self._create_contextmanager_attribs()(None)
self.assertEqual(baz.__doc__, "Whee!")
async with baz:
pass # suppress warning
@_async_test
async def test_keywords(self):
# Ensure no keyword arguments are inhibited
@asynccontextmanager
async def woohoo(self, func, args, kwds):
yield (self, func, args, kwds)
async with woohoo(self=11, func=22, args=33, kwds=44) as target:
self.assertEqual(target, (11, 22, 33, 44))
@_async_test
async def test_recursive(self):
depth = 0
ncols = 0
@asynccontextmanager
async def woohoo():
nonlocal ncols
ncols += 1
nonlocal depth
before = depth
depth += 1
yield
depth -= 1
self.assertEqual(depth, before)
@woohoo()
async def recursive():
if depth < 10:
await recursive()
await recursive()
self.assertEqual(ncols, 10)
self.assertEqual(depth, 0)
class AclosingTestCase(unittest.TestCase):
@support.requires_docstrings
def test_instance_docs(self):
cm_docstring = aclosing.__doc__
obj = aclosing(None)
self.assertEqual(obj.__doc__, cm_docstring)
@_async_test
async def test_aclosing(self):
state = []
class C:
async def aclose(self):
state.append(1)
x = C()
self.assertEqual(state, [])
async with aclosing(x) as y:
self.assertEqual(x, y)
self.assertEqual(state, [1])
@_async_test
async def test_aclosing_error(self):
state = []
class C:
async def aclose(self):
state.append(1)
x = C()
self.assertEqual(state, [])
with self.assertRaises(ZeroDivisionError):
async with aclosing(x) as y:
self.assertEqual(x, y)
1 / 0
self.assertEqual(state, [1])
@_async_test
async def test_aclosing_bpo41229(self):
state = []
@contextmanager
def sync_resource():
try:
yield
finally:
state.append(1)
async def agenfunc():
with sync_resource():
yield -1
yield -2
x = agenfunc()
self.assertEqual(state, [])
with self.assertRaises(ZeroDivisionError):
async with aclosing(x) as y:
self.assertEqual(x, y)
self.assertEqual(-1, await x.__anext__())
1 / 0
self.assertEqual(state, [1])
class TestAsyncExitStack(TestBaseExitStack, unittest.TestCase):
class SyncAsyncExitStack(AsyncExitStack):
@staticmethod
def run_coroutine(coro):
loop = asyncio.get_event_loop_policy().get_event_loop()
t = loop.create_task(coro)
t.add_done_callback(lambda f: loop.stop())
loop.run_forever()
exc = t.exception()
if not exc:
return t.result()
else:
context = exc.__context__
try:
raise exc
except:
exc.__context__ = context
raise exc
def close(self):
return self.run_coroutine(self.aclose())
def __enter__(self):
return self.run_coroutine(self.__aenter__())
def __exit__(self, *exc_details):
return self.run_coroutine(self.__aexit__(*exc_details))
exit_stack = SyncAsyncExitStack
def setUp(self):
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
self.addCleanup(self.loop.close)
self.addCleanup(asyncio.set_event_loop_policy, None)
@_async_test
async def test_async_callback(self):
expected = [
((), {}),
((1,), {}),
((1,2), {}),
((), dict(example=1)),
((1,), dict(example=1)),
((1,2), dict(example=1)),
]
result = []
async def _exit(*args, **kwds):
"""Test metadata propagation"""
result.append((args, kwds))
async with AsyncExitStack() as stack:
for args, kwds in reversed(expected):
if args and kwds:
f = stack.push_async_callback(_exit, *args, **kwds)
elif args:
f = stack.push_async_callback(_exit, *args)
elif kwds:
f = stack.push_async_callback(_exit, **kwds)
else:
f = stack.push_async_callback(_exit)
self.assertIs(f, _exit)
for wrapper in stack._exit_callbacks:
self.assertIs(wrapper[1].__wrapped__, _exit)
self.assertNotEqual(wrapper[1].__name__, _exit.__name__)
self.assertIsNone(wrapper[1].__doc__, _exit.__doc__)
self.assertEqual(result, expected)
result = []
async with AsyncExitStack() as stack:
with self.assertRaises(TypeError):
stack.push_async_callback(arg=1)
with self.assertRaises(TypeError):
self.exit_stack.push_async_callback(arg=2)
with self.assertRaises(TypeError):
stack.push_async_callback(callback=_exit, arg=3)
self.assertEqual(result, [])
@_async_test
async def test_async_push(self):
exc_raised = ZeroDivisionError
async def _expect_exc(exc_type, exc, exc_tb):
self.assertIs(exc_type, exc_raised)
async def _suppress_exc(*exc_details):
return True
async def _expect_ok(exc_type, exc, exc_tb):
self.assertIsNone(exc_type)
self.assertIsNone(exc)
self.assertIsNone(exc_tb)
class ExitCM(object):
def __init__(self, check_exc):
self.check_exc = check_exc
async def __aenter__(self):
self.fail("Should not be called!")
async def __aexit__(self, *exc_details):
await self.check_exc(*exc_details)
async with self.exit_stack() as stack:
stack.push_async_exit(_expect_ok)
self.assertIs(stack._exit_callbacks[-1][1], _expect_ok)
cm = ExitCM(_expect_ok)
stack.push_async_exit(cm)
self.assertIs(stack._exit_callbacks[-1][1].__self__, cm)
stack.push_async_exit(_suppress_exc)
self.assertIs(stack._exit_callbacks[-1][1], _suppress_exc)
cm = ExitCM(_expect_exc)
stack.push_async_exit(cm)
self.assertIs(stack._exit_callbacks[-1][1].__self__, cm)
stack.push_async_exit(_expect_exc)
self.assertIs(stack._exit_callbacks[-1][1], _expect_exc)
stack.push_async_exit(_expect_exc)
self.assertIs(stack._exit_callbacks[-1][1], _expect_exc)
1/0
@_async_test
async def test_async_enter_context(self):
class TestCM(object):
async def __aenter__(self):
result.append(1)
async def __aexit__(self, *exc_details):
result.append(3)
result = []
cm = TestCM()
async with AsyncExitStack() as stack:
@stack.push_async_callback # Registered first => cleaned up last
async def _exit():
result.append(4)
self.assertIsNotNone(_exit)
await stack.enter_async_context(cm)
self.assertIs(stack._exit_callbacks[-1][1].__self__, cm)
result.append(2)
self.assertEqual(result, [1, 2, 3, 4])
@_async_test
async def test_async_exit_exception_chaining(self):
# Ensure exception chaining matches the reference behaviour
async def raise_exc(exc):
raise exc
saved_details = None
async def suppress_exc(*exc_details):
nonlocal saved_details
saved_details = exc_details
return True
try:
async with self.exit_stack() as stack:
stack.push_async_callback(raise_exc, IndexError)
stack.push_async_callback(raise_exc, KeyError)
stack.push_async_callback(raise_exc, AttributeError)
stack.push_async_exit(suppress_exc)
stack.push_async_callback(raise_exc, ValueError)
1 / 0
except IndexError as exc:
self.assertIsInstance(exc.__context__, KeyError)
self.assertIsInstance(exc.__context__.__context__, AttributeError)
# Inner exceptions were suppressed
self.assertIsNone(exc.__context__.__context__.__context__)
else:
self.fail("Expected IndexError, but no exception was raised")
# Check the inner exceptions
inner_exc = saved_details[1]
self.assertIsInstance(inner_exc, ValueError)
self.assertIsInstance(inner_exc.__context__, ZeroDivisionError)
class TestAsyncNullcontext(unittest.TestCase):
@_async_test
async def test_async_nullcontext(self):
class C:
pass
c = C()
async with nullcontext(c) as c_in:
self.assertIs(c_in, c)
if __name__ == '__main__':
unittest.main()

2
tox.ini
View file

@ -4,7 +4,7 @@ skip_missing_interpreters = True
[testenv]
commands =
coverage run test_contextlib2.py
coverage run -m unittest discover -t . -s test
coverage report
coverage xml
deps =