Switch to ThreadedChildWatcher and test (#5877)

Co-authored-by: Sviatoslav Sydorenko <[email protected]>

CHANGES/5877.bugfix

@@ -0,0 +1 @@
+Uses :py:class:`~asyncio.ThreadedChildWatcher` under POSIX to allow setting up test loop in non-main thread.

aiohttp/test_utils.py

@@ -497,7 +497,7 @@ def setup_test_loop(
             # * https://stackoverflow.com/a/58614689/595220
             # * https://bugs.python.org/issue35621
             # * https://github.com/python/cpython/pull/14344
-            watcher = asyncio.MultiLoopChildWatcher()
+            watcher = asyncio.ThreadedChildWatcher()
         except AttributeError:  # Python < 3.8
             watcher = asyncio.SafeChildWatcher()
         watcher.attach_loop(loop)

tests/test_loop.py

@@ -6,14 +6,15 @@
 import pytest
 
 from aiohttp import web
-from aiohttp.test_utils import AioHTTPTestCase
+from aiohttp.helpers import PY_38
+from aiohttp.test_utils import AioHTTPTestCase, loop_context
 
 
 @pytest.mark.skipif(
     platform.system() == "Windows", reason="the test is not valid for Windows"
 )
 async def test_subprocess_co(loop: Any) -> None:
-    assert threading.current_thread() is threading.main_thread()
+    assert PY_38 or threading.current_thread() is threading.main_thread()
     proc = await asyncio.create_subprocess_shell(
         "exit 0",
         stdin=asyncio.subprocess.DEVNULL,
@@ -43,3 +44,24 @@ def test_default_loop(self) -> None:
 
 def test_default_loop(loop: Any) -> None:
     assert asyncio.get_event_loop() is loop
+
+
+@pytest.mark.xfail(not PY_38, reason="ThreadedChildWatcher is only available in 3.8+")
+def test_setup_loop_non_main_thread() -> None:
+    child_exc = None
+
+    def target() -> None:
+        try:
+            with loop_context() as loop:
+                assert asyncio.get_event_loop() is loop
+                loop.run_until_complete(test_subprocess_co(loop))
+        except Exception as exc:
+            nonlocal child_exc
+            child_exc = exc
+
+    # Ensures setup_test_loop can be called by pytest-xdist in non-main thread.
+    t = threading.Thread(target=target)
+    t.start()
+    t.join()
+
+    assert child_exc is None