add support.set_event_loop_policy for setting the event loop policy in tests

Author: graingert

Lib/test/libregrtest/save_env.py

@@ -97,7 +97,7 @@ def get_asyncio_events__event_loop_policy(self):
         return support.maybe_get_event_loop_policy()
     def restore_asyncio_events__event_loop_policy(self, policy):
         asyncio = self.get_module('asyncio')
-        asyncio.set_event_loop_policy(policy)
+        support.set_event_loop_policy(policy)
 
     def get_sys_argv(self):
         return id(sys.argv), sys.argv, sys.argv[:]

Lib/test/support/__init__.py

@@ -2089,6 +2089,13 @@ def maybe_get_event_loop_policy():
     import asyncio.events
     return asyncio.events._event_loop_policy
 
+def set_event_loop_policy(policy):
+    """Set the global event loop policy ignoring deprecation warnings"""
+    import asyncio
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", DeprecationWarning)
+        asyncio.set_event_loop_policy(policy)
+
 # Helpers for testing hashing.
 NHASHBITS = sys.hash_info.width # number of bits in hash() result
 assert NHASHBITS in (32, 64)

Lib/test/test_asyncgen.py

@@ -429,7 +429,7 @@ def setUp(self):
     def tearDown(self):
         self.loop.close()
         self.loop = None
-        asyncio.set_event_loop_policy(None)
+        support.set_event_loop_policy(None)
 
     def check_async_iterator_anext(self, ait_class):
         with self.subTest(anext="pure-Python"):

Lib/test/test_asyncio/test_base_events.py

@@ -24,7 +24,7 @@
 
 
 def tearDownModule():
-    asyncio.set_event_loop_policy(None)
+    support.set_event_loop_policy(None)
 
 
 def mock_socket_module():

Lib/test/test_asyncio/test_buffered_proto.py

@@ -2,10 +2,11 @@
 import unittest
 
 from test.test_asyncio import functional as func_tests
+from test import support
 
 
 def tearDownModule():
-    asyncio.set_event_loop_policy(None)
+    support.set_event_loop_policy(None)
 
 
 class ReceiveStuffProto(asyncio.BufferedProtocol):

Lib/test/test_asyncio/test_context.py

@@ -1,10 +1,11 @@
 import asyncio
 import decimal
 import unittest
+from test import support
 
 
 def tearDownModule():
-    asyncio.set_event_loop_policy(None)
+    support.set_event_loop_policy(None)
 
 
 @unittest.skipUnless(decimal.HAVE_CONTEXTVAR, "decimal is built with a thread-local context")

Lib/test/test_asyncio/test_eager_task_factory.py

@@ -8,12 +8,13 @@
 from asyncio import tasks
 from test.test_asyncio import utils as test_utils
 from test.support.script_helper import assert_python_ok
+from test import support
 
 MOCK_ANY = mock.ANY
 
 
 def tearDownModule():
-    asyncio.set_event_loop_policy(None)
+    support.set_event_loop_policy(None)
 
 
 class EagerTaskFactoryLoopTests:

Lib/test/test_asyncio/test_events.py

@@ -39,7 +39,7 @@
 
 
 def tearDownModule():
-    asyncio.set_event_loop_policy(None)
+    support.set_event_loop_policy(None)
 
 
 def broken_unix_getsockname():
@@ -2684,8 +2684,9 @@ def test_get_event_loop_policy(self):
         self.assertIs(policy, asyncio.get_event_loop_policy())
 
     def test_set_event_loop_policy(self):
-        self.assertRaises(
-            TypeError, asyncio.set_event_loop_policy, object())
+        with self.assertWarns(DeprecationWarning):
+            self.assertRaises(
+                TypeError, asyncio.set_event_loop_policy, object())
 
         old_policy = asyncio.get_event_loop_policy()
 
@@ -2790,7 +2791,8 @@ def get_event_loop(self):
 
         old_policy = asyncio.get_event_loop_policy()
         try:
-            asyncio.set_event_loop_policy(Policy())
+            with self.assertWarns(DeprecationWarning):
+                asyncio.set_event_loop_policy(Policy())
             loop = asyncio.new_event_loop()
 
             with self.assertRaises(TestError):
@@ -2818,7 +2820,7 @@ async def func():
                 asyncio.get_event_loop()
 
         finally:
-            asyncio.set_event_loop_policy(old_policy)
+            support.set_event_loop_policy(old_policy)
             if loop is not None:
                 loop.close()
 
@@ -2830,7 +2832,8 @@ async def func():
     def test_get_event_loop_returns_running_loop2(self):
         old_policy = asyncio.get_event_loop_policy()
         try:
-            asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
+            with self.assertWarns(DeprecationWarning):
+                asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
             loop = asyncio.new_event_loop()
             self.addCleanup(loop.close)
 
@@ -2861,7 +2864,7 @@ async def func():
                 asyncio.get_event_loop()
 
         finally:
-            asyncio.set_event_loop_policy(old_policy)
+            support.set_event_loop_policy(old_policy)
             if loop is not None:
                 loop.close()
 

Lib/test/test_asyncio/test_futures.py

@@ -16,7 +16,7 @@
 
 
 def tearDownModule():
-    asyncio.set_event_loop_policy(None)
+    support.set_event_loop_policy(None)
 
 
 def _fakefunc(f):

Lib/test/test_asyncio/test_futures2.py

@@ -4,10 +4,11 @@
 import traceback
 import unittest
 from asyncio import tasks
+from test import support
 
 
 def tearDownModule():
-    asyncio.set_event_loop_policy(None)
+    support.set_event_loop_policy(None)
 
 
 class FutureTests: