This issue tracker has been migrated to GitHub, and is currently read-only.
For more information, see the GitHub FAQs in the Python's Developer Guide.

Author JDLH
Recipients JDLH, asvetlov, yselivanov
Date 2018-07-08.23:32:51
SpamBayes Score -1.0
Marked as misclassified Yes
Message-id <1531092771.84.0.56676864532.issue34071@psf.upfronthosting.co.za>
In-reply-to
Content
In Python 3.6.5, Task.__repr__() with raise an AssertionError for 
certain, arguably incorrect, coroutines which the instantiation of Task() 
accepts. repr(task) thus fails with an AssertionError, instead of
returning a string which describes the task object and why its coroutine 
is wrong. Complications ensue. 

In particular, if this Task is used in a unittest.TestCase, and appears in test failure diagnostics, the test
diagnostics are discarded when the AssertionError occurs, masking the 
problem in the code under test. 

In Python 3.7.0, Task.__init__() checks the supplied coroutine argument,
using the same asyncio.iscoroutine() method which Task.__repr__() calls.
Thus, while repr(task) can still raise an AssertionError, it is much harder to reach this situation. Task.__init__() is likely to raise an exception first, and the diagnostics will be pretty clear.

To reproduce:

1. Here is a simple, perhaps misguided, attempt at a coroutine, and a Task which uses it:

import asyncio

async def ag():
    yield None

print("type(ag()) is ", type(ag()))
print("asyncio.iscoroutine(ag)? ", asyncio.iscoroutine(ag))
print("repr(ag()) is ", repr(ag()))

task = asyncio.get_event_loop().create_task(ag())
print("type(task) is ", type(task))
print("asyncio.iscoroutine(task._coro)? ", asyncio.iscoroutine(task._coro))
print("repr(task) is ")
print(repr(task))

The output from Python 3.6.5 is:
type(ag()) is  <class 'async_generator'>
asyncio.iscoroutine(ag)?  False
repr(ag()) is  <async_generator object ag at 0x108331938>
type(task) is  <class '_asyncio.Task'>
asyncio.iscoroutine(task._coro)?  False
repr(task) is 
Traceback (most recent call last):
  File "bug_task_repr_1.py", line 31, in <module>
    print(repr(task))
  File "/opt/local/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/base_tasks.py", line 15, in _task_repr_info
    coro = coroutines._format_coroutine(task._coro)
  File "/opt/local/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/coroutines.py", line 276, in _format_coroutine
    assert iscoroutine(coro)
AssertionError

Observed behaviour: 

Here is a unit test which attempts to exercise the above Task. The test fails, but the AssertionError in Task.__repr__() discards the error diagnostics:

from asyncio import test_utils
 
class Test(test_utils.TestCase):

    def setUp(self):
        super().setUp()
        self.loop = self.new_test_loop()

    def test_wrongly_scheduled(self):
        # this is a simple, and maybe mistaken, coroutine
        async def ag():
            yield None
        
        _ = self.loop.create_task(ag())
        test_utils.run_once(self.loop)

The output from Python 3.6.5 is:

% python -m unittest bug_task_repr_2.py
Exception in default exception handler
Traceback (most recent call last):
  File "/opt/local/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/base_events.py", line 1291, in call_exception_handler
    self.default_exception_handler(context)
  File "/opt/local/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/base_events.py", line 1263, in default_exception_handler
    value = repr(value)
  File "/opt/local/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/base_tasks.py", line 15, in _task_repr_info
    coro = coroutines._format_coroutine(task._coro)
  File "/opt/local/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/coroutines.py", line 276, in _format_coroutine
    assert iscoroutine(coro)
AssertionError
.
----------------------------------------------------------------------
Ran 1 test in 0.001s
 
OK

Note that the error traceback is all about the exception from Task.__repr__(), and the diagnostics about why the unit test failed is discarded.

Expected behaviour:

Here is the same unit test, but this time with a replacement method for Task._task_repr_info() which avoids the AssertionError. 

import asyncio
from asyncio import test_utils
import types

class Test(test_utils.TestCase):

    def setUp(self):
        super().setUp()
        self.loop = self.new_test_loop()

    def test_wrongly_scheduled(self):
        # this is a simple, and maybe mistaken, coroutine
        async def ag():
            yield None
        
        task = self.loop.create_task(ag())
        task._repr_info = types.MethodType( task_safe_repr_info, task )
        test_utils.run_once(self.loop)


def task_safe_repr_info(self):
    '''task.task_safe_repr_info(): string list representation of task. Won't raise AssertionError
    
    If you have a Task object which is raising an AssertionError from 
    "assert iscoroutine(coro)", then assign to instantiated object:
    task._repr_info = task_safe_repr_info
    Patched task object should be able to survive the AssertionError and 
    deliver a useful representation.
    '''
    try:
        # from Task._repr_info(self), which calls base_tasks._task_repr_info(self)
        info = asyncio.base_futures._future_repr_info(self)
    
        if self._must_cancel:
            # replace status
            info[0] = 'cancelling'
    
        # Expand: coro = coroutines._format_coroutine(task._coro)
        # coro_repr = coroutines._format_coroutine(task._coro):276 does assert iscoroutine(coro)
        # This fails. Expand that routine, without the assert.
        coro = self._coro
        if not hasattr(coro, 'cr_code') and not hasattr(coro, 'gi_code'):
            # Most likely a built-in type or a Cython coroutine.
    
            # Built-in types might not have __qualname__ or __name__.
            coro_name = getattr(
                coro, '__qualname__',
                getattr(coro, '__name__', type(coro).__name__))
            coro_name = '{}()'.format(coro_name)
    
            running = False
            try:
                running = coro.cr_running
            except AttributeError:
                try:
                    running = coro.gi_running
                except AttributeError:
                    pass
    
            if running:
                coro_repr = '{} running'.format(coro_name)
            else:
                coro_repr = coro_name
        else:
            coro_name = None
            if isinstance(coro, asyncio.coroutines.CoroWrapper):
                func = coro.func
                coro_name = coro.__qualname__
                if coro_name is not None:
                    coro_name = '{}()'.format(coro_name)
            else:
                func = coro
        
            if coro_name is None:
                coro_name = asyncio.events._format_callback(func, (), {})
        
            try:
                coro_code = coro.gi_code
            except AttributeError:
                coro_code = coro.cr_code
        
            try:
                coro_frame = coro.gi_frame
            except AttributeError:
                coro_frame = coro.cr_frame
        
            filename = coro_code.co_filename
            lineno = 0
            if (isinstance(coro, asyncio.coroutines.CoroWrapper) and
                    not inspect.isgeneratorfunction(coro.func) and
                    coro.func is not None):
                source = asyncio.events._get_function_source(coro.func)
                if source is not None:
                    filename, lineno = source
                if coro_frame is None:
                    coro_repr = ('%s done, defined at %s:%s'
                                 % (coro_name, filename, lineno))
                else:
                    coro_repr = ('%s running, defined at %s:%s'
                                 % (coro_name, filename, lineno))
            elif coro_frame is not None:
                lineno = coro_frame.f_lineno
                coro_repr = ('%s running at %s:%s'
                             % (coro_name, filename, lineno))
            else:
                lineno = coro_code.co_firstlineno
                coro_repr = ('%s done, defined at %s:%s'
                             % (coro_name, filename, lineno))
    
        # end of coroutines._format_coroutine(task._coro)
        # return coro_repr
        
        info.insert(1, 'coro=%s(<%s>)' % (type(coro), coro_repr))
    
        if self._fut_waiter is not None:
            info.insert(2, 'wait_for=%r' % self._fut_waiter)

        return info
    except:
        return [type(self), id(self)]

Here is the result of running that unit test on the patched Task object. Note that there is no AssertionError from repr(), and the diagnostics about the failing unit tests are now clear:

% python -m unittest bug_task_repr_3.py
.
----------------------------------------------------------------------
Ran 1 test in 0.000s
 
OK
Task exception was never retrieved
future: <Task finished coro=<class 'async_generator'>(<Test.test_wrongly_scheduled.<locals>.ag()>) exception=AttributeError("'async_generator' object has no attribute 'send'",)>
AttributeError: 'async_generator' object has no attribute 'send'
History
Date User Action Args
2018-07-08 23:32:51JDLHsetrecipients: + JDLH, asvetlov, yselivanov
2018-07-08 23:32:51JDLHsetmessageid: <1531092771.84.0.56676864532.issue34071@psf.upfronthosting.co.za>
2018-07-08 23:32:51JDLHlinkissue34071 messages
2018-07-08 23:32:51JDLHcreate