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

Sync test suite, add notes on sync process

Browse source
This commit is contained in:
Nick Coghlan 2021-06-26 17:49:42 +10:00
parent 94a3b86586
commit 4b39470cbb
11 changed files with 980 additions and 149 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 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.

14
contextlib2.py
View file

@ -517,7 +517,12 @@ class _BaseExitStack:
Cannot suppress exceptions.
"""
# Python 3.6/3.7 compatibility: no native positional-only args syntax
self, callback, *args = args
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
@ -666,7 +671,12 @@ class AsyncExitStack(_BaseExitStack, AbstractAsyncContextManager):
Cannot suppress exceptions.
"""
# Python 3.6/3.7 compatibility: no native positional-only args syntax
self, callback, *args = args
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

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

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

@ -1,16 +1,14 @@
"""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
class TestAbstractContextManager(unittest.TestCase):
@ -31,8 +29,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 +43,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 +131,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 +140,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 +161,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 +202,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 +220,56 @@ 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)
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 +301,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 +395,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 +552,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 +575,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)
@ -455,12 +592,22 @@ class TestExitStack(unittest.TestCase):
else:
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__)
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 = []
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,7 +625,7 @@ 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][1], _expect_ok)
cm = ExitCM(_expect_ok)
@ -504,7 +651,7 @@ 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)
@ -516,7 +663,7 @@ class TestExitStack(unittest.TestCase):
def test_close(self):
result = []
with ExitStack() as stack:
with self.exit_stack() as stack:
@stack.callback
def _exit():
result.append(1)
@ -527,7 +674,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 +687,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 +723,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 +749,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 +771,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 +784,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 +793,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 +820,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 +831,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 +847,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 +857,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 +865,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,44 +887,11 @@ 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][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][1].__self__, cm)
stack.push(cb)
stack.push(cm)
self.assertIs(stack._exit_callbacks[-1][1].__self__, cm)
result.append("Running")
stack.callback(cb)
self.assertIs(stack._exit_callbacks[-1][1].__wrapped__, cb)
self.assertEqual(result, ["Enter", "Running",
"Callback", "Exit",
"Callback", "Exit",
])
with ExitStack():
pass
def test_dont_reraise_RuntimeError(self):
# https://bugs.python.org/issue27122
class UniqueException(Exception): pass
@ -787,10 +902,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 +914,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 +936,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 +987,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 +1002,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 +1054,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()

551
test/test_contextlib_async.py Normal file
View file

@ -0,0 +1,551 @@
import asyncio
from contextlib2 import (
asynccontextmanager, AbstractAsyncContextManager,
AsyncExitStack, nullcontext, aclosing)
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 = []
class Resource:
def __del__(self):
state.append(1)
async def agenfunc():
r = 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 =