Issue python#136: Add get/set_debug() methods to BaseEventLoopTests. Add also a

PYTHONASYNCIODEBUG environment variable to debug coroutines since Python
startup, to be able to debug coroutines defined directly in the asyncio module.

Author: vstinner

asyncio/base_events.py

@@ -123,6 +123,7 @@ def __init__(self):
         self._running = False
         self._clock_resolution = time.get_clock_info('monotonic').resolution
         self._exception_handler = None
+        self._debug = False
 
     def _make_socket_transport(self, sock, protocol, waiter=None, *,
                                extra=None, server=None):
@@ -795,3 +796,9 @@ def _run_once(self):
             if not handle._cancelled:
                 handle._run()
         handle = None  # Needed to break cycles when an exception occurs.
+
+    def get_debug(self):
+        return self._debug
+
+    def set_debug(self, enabled):
+        self._debug = enabled

asyncio/events.py

@@ -345,6 +345,14 @@ def default_exception_handler(self, context):
     def call_exception_handler(self, context):
         raise NotImplementedError
 
+    # Debug flag management.
+
+    def get_debug(self):
+        raise NotImplementedError
+
+    def set_debug(self, enabled):
+        raise NotImplementedError
+
 
 class AbstractEventLoopPolicy:
     """Abstract policy for accessing the event loop."""

asyncio/tasks.py

@@ -12,6 +12,8 @@
 import functools
 import inspect
 import linecache
+import os
+import sys
 import traceback
 import weakref
 
@@ -28,7 +30,8 @@
 # before you define your coroutines.  A downside of using this feature
 # is that tracebacks show entries for the CoroWrapper.__next__ method
 # when _DEBUG is true.
-_DEBUG = False
+_DEBUG = (not sys.flags.ignore_environment
+          and bool(os.environ.get('PYTHONASYNCIODEBUG')))
 
 
 class CoroWrapper:

tests/test_base_events.py

@@ -197,6 +197,12 @@ def test__run_once(self):
         self.assertEqual([h2], self.loop._scheduled)
         self.assertTrue(self.loop._process_events.called)
 
+    def test_set_debug(self):
+        self.loop.set_debug(True)
+        self.assertTrue(self.loop.get_debug())
+        self.loop.set_debug(False)
+        self.assertFalse(self.loop.get_debug())
+
     @unittest.mock.patch('asyncio.base_events.time')
     @unittest.mock.patch('asyncio.base_events.logger')
     def test__run_once_logging(self, m_logger, m_time):

tests/test_tasks.py

@@ -1,7 +1,9 @@
 """Tests for tasks.py."""
 
 import gc
+import os.path
 import unittest
+from test.script_helper import assert_python_ok
 
 import asyncio
 from asyncio import test_utils
@@ -1461,6 +1463,32 @@ def test_return_exceptions(self):
         cb.assert_called_once_with(fut)
         self.assertEqual(fut.result(), [3, 1, exc, exc2])
 
+    def test_env_var_debug(self):
+        path = os.path.dirname(asyncio.__file__)
+        path = os.path.normpath(os.path.join(path, '..'))
+        code = '\n'.join((
+            'import sys',
+            'sys.path.insert(0, %r)' % path,
+            'import asyncio.tasks',
+            'print(asyncio.tasks._DEBUG)'))
+
+        # Test with -E to not fail if the unit test was run with
+        # PYTHONASYNCIODEBUG set to a non-empty string
+        sts, stdout, stderr = assert_python_ok('-E', '-c', code)
+        self.assertEqual(stdout.rstrip(), b'False')
+
+        sts, stdout, stderr = assert_python_ok('-c', code,
+                                               PYTHONASYNCIODEBUG='')
+        self.assertEqual(stdout.rstrip(), b'False')
+
+        sts, stdout, stderr = assert_python_ok('-c', code,
+                                               PYTHONASYNCIODEBUG='1')
+        self.assertEqual(stdout.rstrip(), b'True')
+
+        sts, stdout, stderr = assert_python_ok('-E', '-c', code,
+                                               PYTHONASYNCIODEBUG='1')
+        self.assertEqual(stdout.rstrip(), b'False')
+
 
 class FutureGatherTests(GatherTestsBase, unittest.TestCase):