Switch to ThreadedChildWatcher and test (#5877)

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

@@ -528,7 +528,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

@@ -5,14 +5,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) -> None:
-    assert isinstance(threading.current_thread(), threading._MainThread)
+    assert PY_38 or isinstance(threading.current_thread(), threading._MainThread)
     proc = await asyncio.create_subprocess_shell(
         "exit 0",
         stdin=asyncio.subprocess.DEVNULL,
@@ -40,3 +41,24 @@ def test_default_loop(self) -> None:
 
 def test_default_loop(loop) -> None:
     assert asyncio.get_event_loop_policy().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_policy().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