_enter_task = None
_leave_task = None
all_tasks = None
+ Task = None
+
+ def test_register_task_resurrection(self):
+ register_task = self._register_task
+ class EvilLoop:
+ def get_debug(self):
+ return False
+
+ def call_exception_handler(self, context):
+ register_task(context["task"])
+
+ async def coro_fn ():
+ pass
+
+ coro = coro_fn()
+ self.addCleanup(coro.close)
+ loop = EvilLoop()
+ with self.assertRaises(AttributeError):
+ self.Task(coro, loop=loop)
+
def test__register_task_1(self):
class TaskLike:
_leave_task = staticmethod(tasks._py_leave_task)
all_tasks = staticmethod(tasks._py_all_tasks)
current_task = staticmethod(tasks._py_current_task)
+ Task = tasks._PyTask
@unittest.skipUnless(hasattr(tasks, '_c_register_task'),
_leave_task = staticmethod(tasks._c_leave_task)
all_tasks = staticmethod(tasks._c_all_tasks)
current_task = staticmethod(tasks._c_current_task)
+ Task = tasks._CTask
else:
_register_task = _unregister_task = _enter_task = _leave_task = None
+
class BaseCurrentLoopTests:
current_task = None
static void
TaskObj_dealloc(PyObject *self)
{
- _PyObject_ResurrectStart(self);
- // Unregister the task here so that even if any subclass of Task
- // which doesn't end up calling TaskObj_finalize not crashes.
- unregister_task((TaskObj *)self);
-
- PyObject_CallFinalizer(self);
-
- if (_PyObject_ResurrectEnd(self)) {
- return;
+ if (PyObject_CallFinalizerFromDealloc(self) < 0) {
+ return; // resurrected
}
+ // unregister the task after finalization so that
+ // if the task gets resurrected, it remains registered
+ unregister_task((TaskObj *)self);
PyTypeObject *tp = Py_TYPE(self);
PyObject_GC_UnTrack(self);