Refactor: Use function-scoped event loop for tests

Co-authored-by: jakub.dobry <[email protected]>

Author: cursoragent

python/pytest.ini

@@ -3,6 +3,5 @@
 markers =
     skip_debug: skip test if E2B_DEBUG is set.
 asyncio_mode=auto
-asyncio_default_fixture_loop_scope=function
 
 addopts = "--import-mode=importlib" "--numprocesses=2"

python/tests/conftest.py

@@ -1,6 +1,7 @@
 import pytest
 import pytest_asyncio
 import os
+import asyncio
 
 from logging import warning
 
@@ -10,6 +11,28 @@
 timeout = 60
 
 
+@pytest_asyncio.fixture(scope="function")
+def event_loop():
+    """Create an instance of the default event loop for each test case."""
+    policy = asyncio.get_event_loop_policy()
+    loop = policy.new_event_loop()
+    yield loop
+    # Clean up any remaining tasks
+    try:
+        pending = asyncio.all_tasks(loop)
+        for task in pending:
+            task.cancel()
+        if pending:
+            loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
+    except Exception:
+        pass
+    # Close the loop gracefully
+    try:
+        loop.close()
+    except Exception:
+        pass
+
+
 @pytest.fixture()
 def template():
     return os.getenv("E2B_TESTS_TEMPLATE") or "code-interpreter-v1"
@@ -31,7 +54,7 @@ def sandbox(template, debug):
                 )
 
 
-@pytest_asyncio.fixture
+@pytest_asyncio.fixture(loop_scope="function")
 async def async_sandbox(template, debug):
     async_sandbox = await AsyncSandbox.create(template, timeout=timeout, debug=debug)
 
@@ -40,6 +63,10 @@ async def async_sandbox(template, debug):
     finally:
         try:
             await async_sandbox.kill()
+        except RuntimeError as e:
+            # Ignore "Event loop is closed" errors during cleanup in pytest-xdist
+            if "Event loop is closed" not in str(e):
+                raise
         except:  # noqa: E722
             if not debug:
                 warning(