Merge pull request #2365 from asimurka/fix/async-task-pending

LCORE-1627: fix e2e proxy teardown asyncio task warnings

Author: tisnik

tests/e2e/features/steps/proxy.py

@@ -286,7 +286,11 @@ def _stop_proxy(context: Context, attr: str, loop_attr: str) -> None:
         except Exception:
             pass
         loop.call_soon_threadsafe(loop.stop)
-        time.sleep(0.5)
+        thread = getattr(proxy, "_thread", None)
+        if thread is not None:
+            thread.join(timeout=30)
+        else:
+            time.sleep(0.5)
     if hasattr(context, attr):
         delattr(context, attr)
     if hasattr(context, loop_attr):
@@ -368,10 +372,21 @@ def start_tunnel_proxy(context: Context, port: int) -> None:
 
     def run_proxy() -> None:
         asyncio.set_event_loop(loop)
-        loop.run_until_complete(proxy.start())
-        loop.run_forever()
+        try:
+            loop.run_until_complete(proxy.start())
+            loop.run_forever()
+        finally:
+            # Cancel leftover handler tasks so the loop can close cleanly.
+            if pending := asyncio.all_tasks(loop):
+                for task in pending:
+                    task.cancel()
+                loop.run_until_complete(
+                    asyncio.gather(*pending, return_exceptions=True)
+                )
+            loop.close()
 
     thread = threading.Thread(target=run_proxy, daemon=True)
+    proxy._thread = thread
     thread.start()
     time.sleep(1)
 
@@ -463,10 +478,21 @@ def start_interception_proxy(context: Context, port: int) -> None:
 
     def run_proxy() -> None:
         asyncio.set_event_loop(loop)
-        loop.run_until_complete(proxy.start())
-        loop.run_forever()
+        try:
+            loop.run_until_complete(proxy.start())
+            loop.run_forever()
+        finally:
+            # Cancel leftover handler tasks so the loop can close cleanly.
+            if pending := asyncio.all_tasks(loop):
+                for task in pending:
+                    task.cancel()
+                loop.run_until_complete(
+                    asyncio.gather(*pending, return_exceptions=True)
+                )
+            loop.close()
 
     thread = threading.Thread(target=run_proxy, daemon=True)
+    proxy._thread = thread
     thread.start()
     time.sleep(1)
 

tests/e2e/proxy/interception_proxy.py

@@ -25,6 +25,7 @@
 import json
 import logging
 import ssl
+import threading
 from pathlib import Path
 from typing import Any, Optional
 
@@ -39,7 +40,7 @@
 IN_CLUSTER_CA_CERT_PATH = Path("/tmp/interception-proxy-ca.pem")
 
 
-class InterceptionProxy:
+class InterceptionProxy:  # pylint: disable=too-many-instance-attributes
     """Async TLS-intercepting proxy for testing.
 
     Attributes:
@@ -64,6 +65,7 @@ def __init__(
         self.connect_count = 0
         self._server: Optional[asyncio.Server] = None
         self._handler_tasks: set[asyncio.Task[Any]] = set()
+        self._thread: Optional[threading.Thread] = None
 
     def _make_server_ssl_context(self, hostname: str) -> ssl.SSLContext:
         """Create an SSL context with a certificate for the given hostname.

tests/e2e/proxy/tunnel_proxy.py

@@ -21,6 +21,7 @@
 import asyncio
 import json
 import logging
+import threading
 from typing import Any, Optional
 
 # In-cluster defaults (``python tunnel_proxy.py``).
@@ -48,6 +49,7 @@ def __init__(self, host: str = "127.0.0.1", port: int = 8888) -> None:
         self.last_connect_target: Optional[str] = None
         self._server: Optional[asyncio.Server] = None
         self._handler_tasks: set[asyncio.Task[Any]] = set()
+        self._thread: Optional[threading.Thread] = None
 
     async def _handle_client(
         self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter