Python issue #22926: In debug mode, call_soon(), call_at() and call_later()

methods of BaseEventLoop now use the identifier of the current thread to ensure
that they are called from the thread running the event loop.

Before, the get_event_loop() method was used to check the thread, and no
exception was raised when the thread had no event loop. Now the methods always
raise an exception in debug mode when called from the wrong thread. It should
help to notice misusage of the API.

3 files changed, 66 insertions, 39 deletions

diff --git a/tests/test_base_events.py b/tests/test_base_events.py
index afb65b2..5906fb7 100644
--- a/tests/test_base_events.py
+++ b/tests/test_base_events.py
@@ -5,6 +5,7 @@ import logging
 import math
 import socket
 import sys
+import threading
 import time
 import unittest
 from unittest import mock
@@ -143,28 +144,71 @@ class BaseEventLoopTests(test_utils.TestCase):
         # are really slow
         self.assertLessEqual(dt, 0.9, dt)
 
-    def test_assert_is_current_event_loop(self):
+    def check_thread(self, loop, debug):
         def cb():
             pass
 
-        other_loop = base_events.BaseEventLoop()
-        other_loop._selector = mock.Mock()
-        asyncio.set_event_loop(other_loop)
+        loop.set_debug(debug)
+        if debug:
+            msg = ("Non-thread-safe operation invoked on an event loop other "
+                  "than the current one")
+            with self.assertRaisesRegex(RuntimeError, msg):
+                loop.call_soon(cb)
+            with self.assertRaisesRegex(RuntimeError, msg):
+                loop.call_later(60, cb)
+            with self.assertRaisesRegex(RuntimeError, msg):
+                loop.call_at(loop.time() + 60, cb)
+        else:
+            loop.call_soon(cb)
+            loop.call_later(60, cb)
+            loop.call_at(loop.time() + 60, cb)
+
+    def test_check_thread(self):
+        def check_in_thread(loop, event, debug, create_loop, fut):
+            # wait until the event loop is running
+            event.wait()
+
+            try:
+                if create_loop:
+                    loop2 = base_events.BaseEventLoop()
+                    try:
+                        asyncio.set_event_loop(loop2)
+                        self.check_thread(loop, debug)
+                    finally:
+                        asyncio.set_event_loop(None)
+                        loop2.close()
+                else:
+                    self.check_thread(loop, debug)
+            except Exception as exc:
+                loop.call_soon_threadsafe(fut.set_exception, exc)
+            else:
+                loop.call_soon_threadsafe(fut.set_result, None)
+
+        def test_thread(loop, debug, create_loop=False):
+            event = threading.Event()
+            fut = asyncio.Future(loop=loop)
+            loop.call_soon(event.set)
+            args = (loop, event, debug, create_loop, fut)
+            thread = threading.Thread(target=check_in_thread, args=args)
+            thread.start()
+            loop.run_until_complete(fut)
+            thread.join()
 
-        # raise RuntimeError if the event loop is different in debug mode
-        self.loop.set_debug(True)
-        with self.assertRaises(RuntimeError):
-            self.loop.call_soon(cb)
-        with self.assertRaises(RuntimeError):
-            self.loop.call_later(60, cb)
-        with self.assertRaises(RuntimeError):
-            self.loop.call_at(self.loop.time() + 60, cb)
+        self.loop._process_events = mock.Mock()
+        self.loop._write_to_self = mock.Mock()
+
+        # raise RuntimeError if the thread has no event loop
+        test_thread(self.loop, True)
 
         # check disabled if debug mode is disabled
-        self.loop.set_debug(False)
-        self.loop.call_soon(cb)
-        self.loop.call_later(60, cb)
-        self.loop.call_at(self.loop.time() + 60, cb)
+        test_thread(self.loop, False)
+
+        # raise RuntimeError if the event loop of the thread is not the called
+        # event loop
+        test_thread(self.loop, True, create_loop=True)
+
+        # check disabled if debug mode is disabled
+        test_thread(self.loop, False, create_loop=True)
 
     def test_run_once_in_executor_handle(self):
         def cb():
diff --git a/tests/test_proactor_events.py b/tests/test_proactor_events.py
index 9e9b41a..8258238 100644
--- a/tests/test_proactor_events.py
+++ b/tests/test_proactor_events.py
@@ -440,17 +440,16 @@ class BaseProactorEventLoopTests(test_utils.TestCase):
         self.loop = EventLoop(self.proactor)
         self.set_event_loop(self.loop, cleanup=False)
 
-    @mock.patch.object(BaseProactorEventLoop, '_call_soon')
+    @mock.patch.object(BaseProactorEventLoop, 'call_soon')
     @mock.patch.object(BaseProactorEventLoop, '_socketpair')
-    def test_ctor(self, socketpair, _call_soon):
+    def test_ctor(self, socketpair, call_soon):
         ssock, csock = socketpair.return_value = (
             mock.Mock(), mock.Mock())
         loop = BaseProactorEventLoop(self.proactor)
         self.assertIs(loop._ssock, ssock)
         self.assertIs(loop._csock, csock)
         self.assertEqual(loop._internal_fds, 1)
-        _call_soon.assert_called_with(loop._loop_self_reading, (),
-                                      check_loop=False)
+        call_soon.assert_called_with(loop._loop_self_reading)
 
     def test_close_self_pipe(self):
         self.loop._close_self_pipe()
diff --git a/tests/test_subprocess.py b/tests/test_subprocess.py
index 5cb0f03..b284e6b 100644
--- a/tests/test_subprocess.py
+++ b/tests/test_subprocess.py
@@ -230,19 +230,12 @@ if sys.platform != 'win32':
         def setUp(self):
             policy = asyncio.get_event_loop_policy()
             self.loop = policy.new_event_loop()
-
-            # ensure that the event loop is passed explicitly in asyncio
-            policy.set_event_loop(None)
+            self.set_event_loop(self.loop)
 
             watcher = self.Watcher()
             watcher.attach_loop(self.loop)
             policy.set_child_watcher(watcher)
-
-        def tearDown(self):
-            policy = asyncio.get_event_loop_policy()
-            policy.set_child_watcher(None)
-            self.loop.close()
-            super().tearDown()
+            self.addCleanup(policy.set_child_watcher, None)
 
     class SubprocessSafeWatcherTests(SubprocessWatcherMixin,
                                      test_utils.TestCase):
@@ -259,17 +252,8 @@ else:
     class SubprocessProactorTests(SubprocessMixin, test_utils.TestCase):
 
         def setUp(self):
-            policy = asyncio.get_event_loop_policy()
             self.loop = asyncio.ProactorEventLoop()
-
-            # ensure that the event loop is passed explicitly in asyncio
-            policy.set_event_loop(None)
-
-        def tearDown(self):
-            policy = asyncio.get_event_loop_policy()
-            self.loop.close()
-            policy.set_event_loop(None)
-            super().tearDown()
+            self.set_event_loop(self.loop)
 
 
 if __name__ == '__main__':