Revert D51681518: Multisect successfully blamed "D51681518: [later] c…

…inder compatability for differences in all_tasks" for otest failure

Summary:
This diff is reverting D51681518
Depends on D51696082
D51681518: [later] cinder compatability for differences in all_tasks by fried has been identified to be causing the following test failure:

Tests affected:
- [fblearner/flow/recurring_service/tests:test_server - fblearner.flow.recurring_service.tests.test_server.TestRecurringService: test_error_handling](https://www.internalfb.com/intern/test/281475062751785/)

Here's the Multisect link:
https://www.internalfb.com/multisect/3637357
Here are the tasks that are relevant to this breakage:

We're generating a revert to back out the changes in this diff, please note the backout may land if someone accepts it.

If you believe this diff has been generated in error you may Commandeer and Abandon it.

Reviewed By: fried

Differential Revision: D51696084

fbshipit-source-id: c99379a6037af9b533a06ccfc4038f0eab05f395

later/unittest/case.py

@@ -25,7 +25,6 @@
 import asyncio.tasks
 import sys
 import unittest.mock as mock
-import weakref
 from functools import wraps
 from typing import Any, Callable, TypeVar
 
@@ -40,7 +39,6 @@
 _IGNORE_TASK_LEAKS_ATTR = "__later_testcase_ignore_tasks__"
 _IGNORE_AIO_ERRS_ATTR = "__later_testcase_ignore_asyncio__"
 atleastpy38: bool = sys.version_info[:2] >= (3, 8)
-_unmanaged_tasks: weakref.WeakSet[asyncio.Task] = weakref.WeakSet()
 
 
 class TestTask(asyncio.Task):
@@ -50,7 +48,6 @@ class TestTask(asyncio.Task):
     def __init__(self, coro, *args, **kws) -> None:
         # pyre-fixme[16]: Module `coroutines` has no attribute `_format_coroutine`.
         self._coro_repr = asyncio.coroutines._format_coroutine(coro)
-        _unmanaged_tasks.add(self)
         super().__init__(coro, *args, **kws)
 
     def __repr__(self) -> str:
@@ -63,29 +60,24 @@ def __repr__(self) -> str:
             repr_info[1] = coro  # pragma: nocover
         return f"<{self.__class__.__name__} {' '.join(repr_info)}>"
 
-    def _mark_managed(self):
-        if not self._managed:
-            self._managed = True
-            _unmanaged_tasks.remove(self)
-
     def __await__(self):
-        self._mark_managed()
+        self._managed = True
         return super().__await__()
 
     def result(self):
         if self.done():
-            self._mark_managed()
+            self._managed = True
         return super().result()
 
     def exception(self):
         if self.done():
-            self._mark_managed()
+            self._managed = True
         return super().exception()
 
     def add_done_callback(self, fn, *, context=None) -> None:
         @wraps(fn)
         def mark_managed(fut):
-            self._mark_managed()
+            self._managed = True
             return fn(fut)
 
         super().add_done_callback(mark_managed, context=context)
@@ -143,7 +135,7 @@ def all_tasks(loop):
     i = 0
     while True:
         try:
-            tasks = list(_all_tasks) + list(_unmanaged_tasks)
+            tasks = list(_all_tasks)
         except RuntimeError:  # pragma: nocover
             i += 1
             if i >= 1000: