Cleanup of tests, and removal of _assert_raises

src/trio/_core/_tests/test_run.py

@@ -178,11 +178,10 @@ async def main() -> None:
             nursery.start_soon(looper)
             nursery.start_soon(crasher)
 
-    with pytest.raises(ValueError) as excinfo:
+    with pytest.raises(ValueError, match="argh"):
         _core.run(main)
 
     assert looper_record == ["cancelled"]
-    assert excinfo.value.args == ("argh",)
 
 
 def test_main_and_task_both_crash() -> None:
@@ -433,7 +432,10 @@ async def test_cancel_scope_multierror_filtering() -> None:
     async def crasher() -> NoReturn:
         raise KeyError
 
-    try:
+    # This is outside the outer scope, so all the Cancelled
+    # exceptions should have been absorbed, leaving just a regular
+    # KeyError from crasher()
+    with pytest.raises(KeyError):
         with _core.CancelScope() as outer:
             try:
                 async with _core.open_nursery() as nursery:
@@ -461,15 +463,8 @@ async def crasher() -> NoReturn:
                     summary[type(exc)] += 1
                 assert summary == {_core.Cancelled: 3, KeyError: 1}
                 raise
-    except AssertionError:  # pragma: no cover
-        raise
-    except BaseException as exc:
-        # This is outside the outer scope, so all the Cancelled
-        # exceptions should have been absorbed, leaving just a regular
-        # KeyError from crasher()
-        assert type(exc) is KeyError
-    else:  # pragma: no cover
-        raise AssertionError()
+            else:
+                raise AssertionError("No ExceptionGroup")
 
 
 async def test_precancelled_task() -> None:
@@ -785,9 +780,10 @@ async def task2() -> None:
         await wait_all_tasks_blocked()
         nursery.cancel_scope.__exit__(None, None, None)
     finally:
-        with pytest.raises(RuntimeError) as exc_info:
+        with pytest.raises(
+            RuntimeError, match="which had already been exited"
+        ) as exc_info:
             await nursery_mgr.__aexit__(*sys.exc_info())
-        assert "which had already been exited" in str(exc_info.value)
         assert type(exc_info.value.__context__) is NonBaseMultiError
         assert len(exc_info.value.__context__.exceptions) == 3
         cancelled_in_context = False
@@ -1606,10 +1602,9 @@ async def child_xyzzy() -> None:
     async def misguided() -> None:
         await child_xyzzy()
 
-    with pytest.raises(TypeError) as excinfo:
+    with pytest.raises(TypeError, match="asyncio") as excinfo:
         _core.run(misguided)
 
-    assert "asyncio" in str(excinfo.value)
     # The traceback should point to the location of the foreign await
     assert any(  # pragma: no branch
         entry.name == "child_xyzzy" for entry in excinfo.traceback
@@ -1618,11 +1613,10 @@ async def misguided() -> None:
 
 async def test_asyncio_function_inside_nursery_does_not_explode() -> None:
     # Regression test for https://github.com/python-trio/trio/issues/552
-    with pytest.raises(TypeError) as excinfo:
+    with pytest.raises(TypeError, match="asyncio"):
         async with _core.open_nursery() as nursery:
             nursery.start_soon(sleep_forever)
             await create_asyncio_future_in_new_loop()
-    assert "asyncio" in str(excinfo.value)
 
 
 async def test_trivial_yields() -> None:
@@ -1890,12 +1884,11 @@ async def test_nursery_stop_iteration() -> None:
     async def fail() -> NoReturn:
         raise ValueError
 
-    try:
+    with pytest.raises(ExceptionGroup) as excinfo:
         async with _core.open_nursery() as nursery:
             nursery.start_soon(fail)
             raise StopIteration
-    except MultiError as e:
-        assert tuple(map(type, e.exceptions)) == (StopIteration, ValueError)
+    assert tuple(map(type, excinfo.value.exceptions)) == (StopIteration, ValueError)
 
 
 async def test_nursery_stop_async_iteration() -> None:
@@ -1944,7 +1937,7 @@ async def test_traceback_frame_removal() -> None:
     async def my_child_task() -> NoReturn:
         raise KeyError()
 
-    try:
+    with pytest.raises(ExceptionGroup) as excinfo:
         # Trick: For now cancel/nursery scopes still leave a bunch of tb gunk
         # behind. But if there's a MultiError, they leave it on the MultiError,
         # which lets us get a clean look at the KeyError itself. Someday I
@@ -1953,16 +1946,15 @@ async def my_child_task() -> NoReturn:
         async with _core.open_nursery() as nursery:
             nursery.start_soon(my_child_task)
             nursery.start_soon(my_child_task)
-    except MultiError as exc:
-        first_exc = exc.exceptions[0]
-        assert isinstance(first_exc, KeyError)
-        # The top frame in the exception traceback should be inside the child
-        # task, not trio/contextvars internals. And there's only one frame
-        # inside the child task, so this will also detect if our frame-removal
-        # is too eager.
-        tb = first_exc.__traceback__
-        assert tb is not None
-        assert tb.tb_frame.f_code is my_child_task.__code__
+    first_exc = excinfo.value.exceptions[0]
+    assert isinstance(first_exc, KeyError)
+    # The top frame in the exception traceback should be inside the child
+    # task, not trio/contextvars internals. And there's only one frame
+    # inside the child task, so this will also detect if our frame-removal
+    # is too eager.
+    tb = first_exc.__traceback__
+    assert tb is not None
+    assert tb.tb_frame.f_code is my_child_task.__code__
 
 
 def test_contextvar_support() -> None:

src/trio/_tests/test_ssl.py

@@ -8,7 +8,15 @@
 from contextlib import asynccontextmanager, contextmanager, suppress
 from functools import partial
 from ssl import SSLContext
-from typing import TYPE_CHECKING, Any, AsyncIterator, Iterator, NoReturn
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    AsyncIterator,
+    Awaitable,
+    Callable,
+    Iterator,
+    NoReturn,
+)
 
 import pytest
 
@@ -344,33 +352,21 @@ async def test_PyOpenSSLEchoStream_gives_resource_busy_errors() -> None:
     # PyOpenSSLEchoStream, so this makes sure that if we do have a bug then
     # PyOpenSSLEchoStream will notice and complain.
 
-    s = PyOpenSSLEchoStream()
-    with pytest.raises(_core.BusyResourceError) as excinfo:
-        async with _core.open_nursery() as nursery:
-            nursery.start_soon(s.send_all, b"x")
-            nursery.start_soon(s.send_all, b"x")
-    assert "simultaneous" in str(excinfo.value)
-
-    s = PyOpenSSLEchoStream()
-    with pytest.raises(_core.BusyResourceError) as excinfo:
-        async with _core.open_nursery() as nursery:
-            nursery.start_soon(s.send_all, b"x")
-            nursery.start_soon(s.wait_send_all_might_not_block)
-    assert "simultaneous" in str(excinfo.value)
-
-    s = PyOpenSSLEchoStream()
-    with pytest.raises(_core.BusyResourceError) as excinfo:
-        async with _core.open_nursery() as nursery:
-            nursery.start_soon(s.wait_send_all_might_not_block)
-            nursery.start_soon(s.wait_send_all_might_not_block)
-    assert "simultaneous" in str(excinfo.value)
+    async def do_test(
+        func1: str, args1: tuple[object, ...], func2: str, args2: tuple[object, ...]
+    ) -> None:
+        s = PyOpenSSLEchoStream()
+        with pytest.raises(_core.BusyResourceError, match="simultaneous"):
+            async with _core.open_nursery() as nursery:
+                nursery.start_soon(getattr(s, func1), *args1)
+                nursery.start_soon(getattr(s, func2), *args2)
 
-    s = PyOpenSSLEchoStream()
-    with pytest.raises(_core.BusyResourceError) as excinfo:
-        async with _core.open_nursery() as nursery:
-            nursery.start_soon(s.receive_some, 1)
-            nursery.start_soon(s.receive_some, 1)
-    assert "simultaneous" in str(excinfo.value)
+    await do_test("send_all", (b"x",), "send_all", (b"x",))
+    await do_test("send_all", (b"x",), "wait_send_all_might_not_block", ())
+    await do_test(
+        "wait_send_all_might_not_block", (), "wait_send_all_might_not_block", ()
+    )
+    await do_test("receive_some", (1,), "receive_some", (1,))
 
 
 @contextmanager  # type: ignore[misc]  # decorated contains Any
@@ -727,45 +723,35 @@ async def sleeper_with_slow_wait_writable_and_expect(method: str) -> None:
 
 
 async def test_resource_busy_errors(client_ctx: SSLContext) -> None:
-    async def do_send_all() -> None:
+    S: TypeAlias = trio.SSLStream[
+        trio.StapledStream[trio.abc.SendStream, trio.abc.ReceiveStream]
+    ]
+
+    async def do_send_all(s: S) -> None:
         with assert_checkpoints():
             await s.send_all(b"x")
 
-    async def do_receive_some() -> None:
+    async def do_receive_some(s: S) -> None:
         with assert_checkpoints():
             await s.receive_some(1)
 
-    async def do_wait_send_all_might_not_block() -> None:
+    async def do_wait_send_all_might_not_block(s: S) -> None:
         with assert_checkpoints():
             await s.wait_send_all_might_not_block()
 
-    s, _ = ssl_lockstep_stream_pair(client_ctx)
-    with pytest.raises(_core.BusyResourceError) as excinfo:
-        async with _core.open_nursery() as nursery:
-            nursery.start_soon(do_send_all)
-            nursery.start_soon(do_send_all)
-    assert "another task" in str(excinfo.value)
-
-    s, _ = ssl_lockstep_stream_pair(client_ctx)
-    with pytest.raises(_core.BusyResourceError) as excinfo:
-        async with _core.open_nursery() as nursery:
-            nursery.start_soon(do_receive_some)
-            nursery.start_soon(do_receive_some)
-    assert "another task" in str(excinfo.value)
-
-    s, _ = ssl_lockstep_stream_pair(client_ctx)
-    with pytest.raises(_core.BusyResourceError) as excinfo:
-        async with _core.open_nursery() as nursery:
-            nursery.start_soon(do_send_all)
-            nursery.start_soon(do_wait_send_all_might_not_block)
-    assert "another task" in str(excinfo.value)
+    async def do_test(
+        func1: Callable[[S], Awaitable[None]], func2: Callable[[S], Awaitable[None]]
+    ) -> None:
+        s, _ = ssl_lockstep_stream_pair(client_ctx)
+        with pytest.raises(_core.BusyResourceError, match="another task"):
+            async with _core.open_nursery() as nursery:
+                nursery.start_soon(func1, s)
+                nursery.start_soon(func2, s)
 
-    s, _ = ssl_lockstep_stream_pair(client_ctx)
-    with pytest.raises(_core.BusyResourceError) as excinfo:
-        async with _core.open_nursery() as nursery:
-            nursery.start_soon(do_wait_send_all_might_not_block)
-            nursery.start_soon(do_wait_send_all_might_not_block)
-    assert "another task" in str(excinfo.value)
+    await do_test(do_send_all, do_send_all)
+    await do_test(do_receive_some, do_receive_some)
+    await do_test(do_send_all, do_wait_send_all_might_not_block)
+    await do_test(do_wait_send_all_might_not_block, do_wait_send_all_might_not_block)
 
 
 async def test_wait_writable_calls_underlying_wait_writable() -> None:

src/trio/_tests/test_util.py

@@ -49,21 +49,19 @@ async def test_ConflictDetector() -> None:
         with ul2:
             print("ok")
 
-    with pytest.raises(_core.BusyResourceError) as excinfo:
+    with pytest.raises(_core.BusyResourceError, match="ul1"):
         with ul1:
             with ul1:
                 pass  # pragma: no cover
-    assert "ul1" in str(excinfo.value)
 
     async def wait_with_ul1() -> None:
         with ul1:
             await wait_all_tasks_blocked()
 
-    with pytest.raises(_core.BusyResourceError) as excinfo:
+    with pytest.raises(_core.BusyResourceError, match="ul1"):
         async with _core.open_nursery() as nursery:
             nursery.start_soon(wait_with_ul1)
             nursery.start_soon(wait_with_ul1)
-    assert "ul1" in str(excinfo.value)
 
 
 def test_module_metadata_is_fixed_up() -> None:

src/trio/testing/_check_streams.py

@@ -42,6 +42,8 @@ async def __aexit__(
             await aclose_forcefully(self._second)
 
 
+# This is used in this file instead of pytest.raises in order to avoid a dependency
+# on pytest, as the check_* functions are publicly exported.
 @contextmanager
 def _assert_raises(exc: type[BaseException]) -> Generator[None, None, None]:
     __tracebackhide__ = True