Merge pull request #998 from nolar/managed-event-loops

Manage asyncio event loops explicitly

kopf/_core/actions/loggers.py

@@ -8,7 +8,6 @@
 the operators' code, and can lead to information loss or mismatch
 (e.g. when logging call is added, but posting is forgotten).
 """
-import asyncio
 import copy
 import enum
 import logging
@@ -193,10 +192,6 @@ def configure(
         if not debug:
             logger.handlers[:] = [logging.NullHandler()]
 
-    # Since Python 3.10, get_event_loop() is deprecated, issues a warning. Here is a way around:
-    loop = asyncio.get_event_loop_policy().get_event_loop()
-    loop.set_debug(bool(debug))
-
 
 def make_formatter(
         log_format: LogFormat = LogFormat.FULL,

kopf/_core/reactor/running.py

@@ -53,29 +53,32 @@ def run(
     It will take care of a new event loop's creation and finalization for this
     call. See: :func:`asyncio.run`.
     """
-    loop = loop if loop is not None else asyncio.get_event_loop_policy().get_event_loop()
+    coro = operator(
+        lifecycle=lifecycle,
+        indexers=indexers,
+        registry=registry,
+        settings=settings,
+        memories=memories,
+        insights=insights,
+        identity=identity,
+        standalone=standalone,
+        clusterwide=clusterwide,
+        namespaces=namespaces,
+        namespace=namespace,
+        priority=priority,
+        peering_name=peering_name,
+        liveness_endpoint=liveness_endpoint,
+        stop_flag=stop_flag,
+        ready_flag=ready_flag,
+        vault=vault,
+        memo=memo,
+        _command=_command,
+    )
     try:
-        loop.run_until_complete(operator(
-            lifecycle=lifecycle,
-            indexers=indexers,
-            registry=registry,
-            settings=settings,
-            memories=memories,
-            insights=insights,
-            identity=identity,
-            standalone=standalone,
-            clusterwide=clusterwide,
-            namespaces=namespaces,
-            namespace=namespace,
-            priority=priority,
-            peering_name=peering_name,
-            liveness_endpoint=liveness_endpoint,
-            stop_flag=stop_flag,
-            ready_flag=ready_flag,
-            vault=vault,
-            memo=memo,
-            _command=_command,
-        ))
+        if loop is not None:
+            loop.run_until_complete(coro)
+        else:
+            asyncio.run(coro)
     except asyncio.CancelledError:
         pass
 

tests/cli/test_logging.py

@@ -1,4 +1,3 @@
-import asyncio
 import logging
 
 import pytest
@@ -59,7 +58,6 @@ def test_no_lowlevel_dumps_in_nondebug(invoke, caplog, options, preload, real_ru
 
     alien_records = [m for m in caplog.records if not m.name.startswith('kopf')]
     assert len(alien_records) == 0
-    assert not asyncio.get_event_loop_policy().get_event_loop().get_debug()
 
 
 @pytest.mark.parametrize('options', [
@@ -74,4 +72,3 @@ def test_lowlevel_dumps_in_debug_mode(invoke, caplog, options, preload, real_run
 
     alien_records = [m for m in caplog.records if not m.name.startswith('kopf')]
     assert len(alien_records) >= 1
-    assert asyncio.get_event_loop_policy().get_event_loop().get_debug()

tests/conftest.py

@@ -35,6 +35,9 @@ def pytest_configure(config):
     config.addinivalue_line('filterwarnings', 'ignore:The loop argument:DeprecationWarning:asyncio')
     config.addinivalue_line('filterwarnings', 'ignore:is deprecated, use current_thread:DeprecationWarning:threading')
 
+    # TODO: Remove when fixed in https://github.com/pytest-dev/pytest-asyncio/issues/460:
+    config.addinivalue_line('filterwarnings', 'ignore:There is no current event loop:DeprecationWarning:pytest_asyncio')
+
 
 def pytest_addoption(parser):
     parser.addoption("--only-e2e", action="store_true", help="Execute end-to-end tests only.")

tests/primitives/test_flags.py

@@ -136,7 +136,7 @@ async def test_waiting_of_none_does_nothing():
 async def test_waiting_for_unraised_times_out(flag, timer):
     with pytest.raises(asyncio.TimeoutError), timer:
         await asyncio.wait_for(wait_flag(flag), timeout=0.1)
-    assert timer.seconds >= 0.1
+    assert timer.seconds >= 0.099  # uvloop finishes it earlier than 0.1
 
 
 async def test_waiting_for_preraised_is_instant(flag, timer):