Repository: agronholm/anyio

Issue:
during asyncio tests set a default loop exception_handler that warns (similar to -p unraisablehook)
consider the following test:
```python
import pytest
import asyncio

@pytest.mark.anyio
@pytest.mark.parameterize("anyio_backend", ["asyncio"])
async def test_foo():
    def raise_():
        raise Exception

    asyncio.get_running_loop().call_soon(raise_)
    await asyncio.sleep(0)
```

it passes cleanly.

However if you explicitly collect any errors:

```python
import pytest
import asyncio

@pytest.mark.anyio
@pytest.mark.parameterize("anyio_backend", ["asyncio"])
async def test_foo():
    loop = asyncio.get_running_loop()
    old_exception_handler = loop.get_exception_handler()
    exceptions = []

    def exception_handler(*args, **kwargs):
        exceptions.append((args, kwargs))

    loop.set_exception_handler(exception_handler)
    try:
        def raise_():
            raise Exception

        asyncio.get_running_loop().call_soon(raise_)
        await asyncio.sleep(0)
        assert exceptions == []
    finally:
        loop.set_exception_handler(old_exception_handler)
```

```
=================================== test session starts ====================================
platform linux -- Python 3.9.2, pytest-6.2.2, py-1.10.0, pluggy-0.13.1
rootdir: /home/graingert/projects/anyio, configfile: pyproject.toml
plugins: hypothesis-6.2.0, asyncio-0.14.0, anyio-2.1.0.post21+dirty
collected 1 item                                                                           

ham.py F                                                                             [100%]

========================================= FAILURES =========================================
_________________________________________ test_foo _________________________________________
ham.py:20: in test_foo
    assert exceptions == []
E   AssertionError: assert [((<_UnixSele...py:15'}), {})] == []
E     Left contains one more item: ((<_UnixSelectorEventLoop running=True closed=False debug=False>, {'exception': Exception(), 'handle': <Handle test_fo...y:15>, 'message': 'Exception in callback test_foo.<locals>.raise_() at /home/graingert/projects/anyio/ham.py:17'}), {})
E     Use -v to get the full diff
==================================== 1 failed in 0.04s ====================================
```


pytest-dev/pytest-asyncio#205
https://github.com/pytest-dev/pytest/blob/master/src/_pytest/unraisableexception.py#L62-L78

Discussion:
agronholm: I'll merge this tomorrow unless you have some concerns left.

Patch:
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index bff403b..1b836f3 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -61,6 +61,9 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 - Changed ``CancelScope.deadline`` to be writable
 - ``Lock`` and ``Condition`` can now only be released by the task that acquired them. This behavior
   is now consistent on all backends whereas previously only Trio enforced this.
+- Changed the asyncio test runner to capture unhandled exceptions from asynchronous callbacks and
+  unbound native tasks which are then raised after the test function (or async fixture setup or
+  teardown) completes
 
 **2.2.0**
 
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index b730f02..cb7b7e0 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -837,7 +837,9 @@ class SocketStream(abc.SocketStream):
             try:
                 self._transport.write(item)
             except RuntimeError as exc:
-                if self._closed:
+                if self._protocol.write_future.exception():
+                    await self._protocol.write_future
+                elif self._closed:
                     raise ClosedResourceError from None
                 elif self._transport.is_closing():
                     raise BrokenResourceError from exc
@@ -1021,6 +1023,9 @@ class UNIXSocketStream(abc.SocketStream):
 
 
 class TCPSocketListener(abc.SocketListener):
+    _accept_scope: Optional[CancelScope] = None
+    _closed = False
+
     def __init__(self, raw_socket: socket.SocketType):
         self.__raw_socket = raw_socket
         self._loop = cast(asyncio.BaseEventLoop, get_running_loop())
@@ -1031,30 +1036,40 @@ class TCPSocketListener(abc.SocketListener):
         return self.__raw_socket
 
     async def accept(self) -> abc.SocketStream:
+        if self._closed:
+            raise ClosedResourceError
+
         with self._accept_guard:
             await checkpoint()
-            try:
-                client_sock, _addr = await self._loop.sock_accept(self._raw_socket)
-            except asyncio.CancelledError:
-                # Workaround for https://bugs.python.org/issue41317
+            with CancelScope() as self._accept_scope:
                 try:
-                    self._loop.remove_reader(self._raw_socket)
-                except (ValueError, NotImplementedError):
-                    if self._raw_socket.fileno() == -1:
+                    client_sock, _addr = await self._loop.sock_accept(self._raw_socket)
+                except asyncio.CancelledError:
+                    # Workaround for https://bugs.python.org/issue41317
+                    try:
+                        self._loop.remove_reader(self._raw_socket)
+                    except (ValueError, NotImplementedError):
+                        pass
+
+                    if self._closed:
                         raise ClosedResourceError from None
 
-                raise
+                    raise
+                finally:
+                    self._accept_scope = None
 
         client_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
         transport, protocol = await self._loop.connect_accepted_socket(StreamProtocol, client_sock)
         return SocketStream(cast(asyncio.Transport, transport), cast(StreamProtocol, protocol))
 
     async def aclose(self) -> None:
-        # Needed on Windows + Python < 3.8
-        try:
-            self._loop.remove_reader(self._raw_socket)
-        except NotImplementedError:
-            pass
+        if self._closed:
+            return
+
+        self._closed = True
+        if self._accept_scope:
+            self._accept_scope.cancel()
+            await sleep(0)
 
         self._raw_socket.close()
 
@@ -1432,7 +1447,8 @@ class _SignalReceiver:
 
     def _deliver(self, signum: int) -> None:
         self._signal_queue.append(signum)
-        self._future.set_result(None)
+        if not self._future.done():
+            self._future.set_result(None)
 
     def __aiter__(self):
         return self
@@ -1561,4 +1577,22 @@ class TestRunner(abc.TestRunner):
             self._loop.close()
 
     def call(self, func: Callable[..., Awaitable], *args, **kwargs):
-        return self._loop.run_until_complete(func(*args, **kwargs))
+        def exception_handler(loop: asyncio.AbstractEventLoop, context: Dict[str, Any]) -> None:
+            exceptions.append(context['exception'])
+
+        exceptions: List[Exception] = []
+        self._loop.set_exception_handler(exception_handler)
+        try:
+            retval = self._loop.run_until_complete(func(*args, **kwargs))
+        except Exception as exc:
+            retval = None
+            exceptions.append(exc)
+        finally:
+            self._loop.set_exception_handler(None)
+
+        if len(exceptions) == 1:
+            raise exceptions[0]
+        elif exceptions:
+            raise ExceptionGroup(exceptions)
+
+        return retval


Test patch:
diff --git a/tests/test_pytest_plugin.py b/tests/test_pytest_plugin.py
index 928916a..ad8098a 100644
--- a/tests/test_pytest_plugin.py
+++ b/tests/test_pytest_plugin.py
@@ -9,31 +9,35 @@ pytestmark = pytest.mark.filterwarnings(
 def test_plugin(testdir):
     testdir.makeconftest(
         """
-        import anyio
         import sniffio
         import pytest
 
+        from anyio import sleep
+
 
         @pytest.fixture
         async def async_fixture():
-            await anyio.sleep(0)
+            await sleep(0)
             return sniffio.current_async_library()
+
+
+        @pytest.fixture
+        async def some_feature():
+            yield None
+            await sleep(0)
         """
     )
 
     testdir.makepyfile(
         """
-        import pytest
+        import asyncio
 
+        import pytest
         import sniffio
-        from anyio import get_all_backends, sleep
         from hypothesis import strategies, given
 
+        from anyio import get_all_backends, sleep
 
-        @pytest.fixture
-        async def some_feature():
-            yield None
-            await sleep(0)
 
         @pytest.mark.anyio
         async def test_marked_test():
@@ -49,26 +53,88 @@ def test_plugin(testdir):
             # Test that regular functions can use async fixtures too
             assert async_fixture == anyio_backend_name
 
-        @pytest.mark.parametrize('anyio_backend', ['asyncio'], scope='class')
+        @pytest.mark.anyio
+        async def test_skip_inline(some_feature):
+            # Test for github #214
+            pytest.skip("Test that skipping works")
+        """
+    )
+
+    result = testdir.runpytest('-v')
+    result.assert_outcomes(passed=3 * len(get_all_backends()), skipped=len(get_all_backends()))
+
+
+def test_asyncio(testdir):
+    testdir.makeconftest(
+        """
+        import asyncio
+        import pytest
+
+
+        @pytest.fixture(scope='class')
+        def anyio_backend():
+            return 'asyncio'
+
+        @pytest.fixture
+        async def setup_fail_fixture():
+            def callback():
+                raise RuntimeError('failing fixture setup')
+
+            asyncio.get_event_loop().call_soon(callback)
+            await asyncio.sleep(0)
+            yield None
+
+        @pytest.fixture
+        async def teardown_fail_fixture():
+            def callback():
+                raise RuntimeError('failing fixture teardown')
+
+            yield None
+            asyncio.get_event_loop().call_soon(callback)
+            await asyncio.sleep(0)
+        """
+    )
+
+    testdir.makepyfile(
+        """
+        import asyncio
+
+        import pytest
+
+        pytestmark = pytest.mark.anyio
+
+
         class TestClassFixtures:
             @pytest.fixture(scope='class')
             async def async_class_fixture(self, anyio_backend):
-                await sleep(0)
+                await asyncio.sleep(0)
                 return anyio_backend
 
             def test_class_fixture_in_test_method(self, async_class_fixture, anyio_backend_name):
                 assert anyio_backend_name == 'asyncio'
                 assert async_class_fixture == 'asyncio'
 
-        @pytest.mark.anyio
-        async def test_skip_inline(some_feature):
-            # Test for github #214
-            pytest.skip("Test that skipping works")
+        async def test_callback_exception_during_test():
+            def callback():
+                nonlocal started
+                started = True
+                raise Exception('foo')
+
+            started = False
+            asyncio.get_event_loop().call_soon(callback)
+            await asyncio.sleep(0)
+            assert started
+
+        async def test_callback_exception_during_setup(setup_fail_fixture):
+            pass
+
+        async def test_callback_exception_during_teardown(teardown_fail_fixture):
+            pass
         """
     )
 
     result = testdir.runpytest('-v')
-    result.assert_outcomes(passed=3 * len(get_all_backends()) + 1, skipped=len(get_all_backends()))
+    result.assert_outcomes(passed=2, failed=1, errors=2)
 
 
 def test_autouse_async_fixture(testdir):
diff --git a/tests/test_sockets.py b/tests/test_sockets.py
index 746f9c7..7c38b83 100644
--- a/tests/test_sockets.py
+++ b/tests/test_sockets.py
@@ -372,6 +372,13 @@ class TestTCPListener:
                 client.close()
                 await stream.aclose()
 
+    async def test_accept_after_close(self, family):
+        async with await create_tcp_listener(local_host='localhost', family=family) as multi:
+            for listener in multi.listeners:
+                await listener.aclose()
+                with pytest.raises(ClosedResourceError):
+                    await listener.accept()
+
     async def test_socket_options(self, family):
         async with await create_tcp_listener(local_host='localhost', family=family) as multi:
             for listener in multi.listeners:
diff --git a/tests/test_threads.py b/tests/test_threads.py
index 54df8a0..9d07ed7 100644
--- a/tests/test_threads.py
+++ b/tests/test_threads.py
@@ -9,11 +9,10 @@ import pytest
 
 from anyio import (
     create_blocking_portal, create_capacity_limiter, create_event, create_task_group,
-    get_cancelled_exc_class, get_current_task, run_async_from_thread, run_sync_in_worker_thread,
-    sleep, start_blocking_portal, wait_all_tasks_blocked)
-from anyio._core._threads import run_sync_from_thread
+    get_cancelled_exc_class, get_current_task, run_async_from_thread, run_sync_from_thread,
+    run_sync_in_worker_thread, sleep, start_blocking_portal, wait_all_tasks_blocked)
 
-if sys.version_info < (3, 9):
+if sys.version_info < (3, 7):
     current_task = asyncio.Task.current_task
 else:
     current_task = asyncio.current_task