diff --git a/docs/reference/changelog.rst b/docs/reference/changelog.rst
index aca0ea0f..dfd3ee9a 100644
--- a/docs/reference/changelog.rst
+++ b/docs/reference/changelog.rst
@@ -2,6 +2,11 @@
 Changelog
 =========
 
+0.25.2 (UNRELEASED)
+===================
+
+- Call ``loop.shutdown_asyncgens()`` before closing the event loop to ensure async generators are closed in the same manner as ``asyncio.run`` does `#1034 <https://github.com/pytest-dev/pytest-asyncio/pull/1034>`_
+
 0.25.1 (2025-01-02)
 ===================
 - Fixes an issue that caused a broken event loop when a function-scoped test was executed in between two tests with wider loop scope `#950 <https://github.com/pytest-dev/pytest-asyncio/issues/950>`_
diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py
index 8ad17c0a..2f028ae1 100644
--- a/pytest_asyncio/plugin.py
+++ b/pytest_asyncio/plugin.py
@@ -708,11 +708,12 @@ def scoped_event_loop(
         event_loop_policy,
     ) -> Iterator[asyncio.AbstractEventLoop]:
         new_loop_policy = event_loop_policy
-        with _temporary_event_loop_policy(new_loop_policy):
-            loop = _make_pytest_asyncio_loop(asyncio.new_event_loop())
+        with (
+            _temporary_event_loop_policy(new_loop_policy),
+            _provide_event_loop() as loop,
+        ):
             asyncio.set_event_loop(loop)
             yield loop
-            loop.close()
 
     # @pytest.fixture does not register the fixture anywhere, so pytest doesn't
     # know it exists. We work around this by attaching the fixture function to the
@@ -1147,16 +1148,26 @@ def _retrieve_scope_root(item: Collector | Item, scope: str) -> Collector:
 def event_loop(request: FixtureRequest) -> Iterator[asyncio.AbstractEventLoop]:
     """Create an instance of the default event loop for each test case."""
     new_loop_policy = request.getfixturevalue(event_loop_policy.__name__)
-    with _temporary_event_loop_policy(new_loop_policy):
-        loop = asyncio.get_event_loop_policy().new_event_loop()
-        # Add a magic value to the event loop, so pytest-asyncio can determine if the
-        # event_loop fixture was overridden. Other implementations of event_loop don't
-        # set this value.
-        # The magic value must be set as part of the function definition, because pytest
-        # seems to have multiple instances of the same FixtureDef or fixture function
-        loop = _make_pytest_asyncio_loop(loop)
+    with _temporary_event_loop_policy(new_loop_policy), _provide_event_loop() as loop:
         yield loop
-        loop.close()
+
+
+@contextlib.contextmanager
+def _provide_event_loop() -> Iterator[asyncio.AbstractEventLoop]:
+    loop = asyncio.get_event_loop_policy().new_event_loop()
+    # Add a magic value to the event loop, so pytest-asyncio can determine if the
+    # event_loop fixture was overridden. Other implementations of event_loop don't
+    # set this value.
+    # The magic value must be set as part of the function definition, because pytest
+    # seems to have multiple instances of the same FixtureDef or fixture function
+    loop = _make_pytest_asyncio_loop(loop)
+    try:
+        yield loop
+    finally:
+        try:
+            loop.run_until_complete(loop.shutdown_asyncgens())
+        finally:
+            loop.close()
 
 
 @pytest.fixture(scope="session")
@@ -1164,11 +1175,9 @@ def _session_event_loop(
     request: FixtureRequest, event_loop_policy: AbstractEventLoopPolicy
 ) -> Iterator[asyncio.AbstractEventLoop]:
     new_loop_policy = event_loop_policy
-    with _temporary_event_loop_policy(new_loop_policy):
-        loop = _make_pytest_asyncio_loop(asyncio.new_event_loop())
+    with _temporary_event_loop_policy(new_loop_policy), _provide_event_loop() as loop:
         asyncio.set_event_loop(loop)
         yield loop
-        loop.close()
 
 
 @pytest.fixture(scope="session", autouse=True)
diff --git a/tests/test_event_loop_fixture.py b/tests/test_event_loop_fixture.py
index 21785075..447d15d5 100644
--- a/tests/test_event_loop_fixture.py
+++ b/tests/test_event_loop_fixture.py
@@ -53,3 +53,30 @@ async def test_custom_policy_is_not_overwritten():
     )
     result = pytester.runpytest_subprocess("--asyncio-mode=strict")
     result.assert_outcomes(passed=2)
+
+
+def test_event_loop_fixture_handles_unclosed_async_gen(
+    pytester: Pytester,
+):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest
+
+            pytest_plugins = 'pytest_asyncio'
+
+            @pytest.mark.asyncio
+            async def test_something():
+                async def generator_fn():
+                    yield
+                    yield
+
+                gen = generator_fn()
+                await gen.__anext__()
+            """
+        )
+    )
+    result = pytester.runpytest_subprocess("--asyncio-mode=strict", "-W", "default")
+    result.assert_outcomes(passed=1, warnings=0)