[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

src/trio/_core/_concat_tb.py

@@ -107,7 +107,7 @@ def controller(  # type: ignore[no-any-unimported]
             return operation.delegate()  # type: ignore[no-any-return]
 
         return cast(
-            TracebackType,
+            "TracebackType",
             tputil.make_proxy(controller, type(base_tb), base_tb),
         )  # Returns proxy to traceback
 

src/trio/_core/_io_windows.py

@@ -700,7 +700,7 @@ def _refresh_afd(self, base_handle: Handle) -> None:
 
             lpOverlapped = ffi.new("LPOVERLAPPED")
 
-            poll_info = cast(_AFDPollInfo, ffi.new("AFD_POLL_INFO *"))
+            poll_info = cast("_AFDPollInfo", ffi.new("AFD_POLL_INFO *"))
             poll_info.Timeout = 2**63 - 1  # INT64_MAX
             poll_info.NumberOfHandles = 1
             poll_info.Exclusive = 0
@@ -713,9 +713,9 @@ def _refresh_afd(self, base_handle: Handle) -> None:
                     kernel32.DeviceIoControl(
                         afd_group.handle,
                         IoControlCodes.IOCTL_AFD_POLL,
-                        cast(CType, poll_info),
+                        cast("CType", poll_info),
                         ffi.sizeof("AFD_POLL_INFO"),
-                        cast(CType, poll_info),
+                        cast("CType", poll_info),
                         ffi.sizeof("AFD_POLL_INFO"),
                         ffi.NULL,
                         lpOverlapped,
@@ -937,13 +937,13 @@ async def _perform_overlapped(
         # operation will not be cancellable, depending on how Windows is
         # feeling today. So we need to check for cancellation manually.
         await _core.checkpoint_if_cancelled()
-        lpOverlapped = cast(_Overlapped, ffi.new("LPOVERLAPPED"))
+        lpOverlapped = cast("_Overlapped", ffi.new("LPOVERLAPPED"))
         try:
             submit_fn(lpOverlapped)
         except OSError as exc:
             if exc.winerror != ErrorCodes.ERROR_IO_PENDING:
                 raise
-        await self.wait_overlapped(handle, cast(CData, lpOverlapped))
+        await self.wait_overlapped(handle, cast("CData", lpOverlapped))
         return lpOverlapped
 
     @_public

src/trio/_core/_local.py

@@ -44,7 +44,7 @@ class RunVar(Generic[T]):
     def get(self, default: T | type[_NoValue] = _NoValue) -> T:
         """Gets the value of this :class:`RunVar` for the current run call."""
         try:
-            return cast(T, _run.GLOBAL_RUN_CONTEXT.runner._locals[self])
+            return cast("T", _run.GLOBAL_RUN_CONTEXT.runner._locals[self])
         except AttributeError:
             raise RuntimeError("Cannot be used outside of a run context") from None
         except KeyError:

src/trio/_core/_run.py

@@ -941,7 +941,7 @@ def started(self: _TaskStatus[StatusT], value: StatusT) -> None: ...
     def started(self, value: StatusT | None = None) -> None:
         if self._value is not _NoStatus:
             raise RuntimeError("called 'started' twice on the same task status")
-        self._value = cast(StatusT, value)  # If None, StatusT == None
+        self._value = cast("StatusT", value)  # If None, StatusT == None
 
         # If the old nursery is cancelled, then quietly quit now; the child
         # will eventually exit on its own, and we don't want to risk moving
@@ -2407,7 +2407,7 @@ def run(
     # Inlined copy of runner.main_task_outcome.unwrap() to avoid
     # cluttering every single Trio traceback with an extra frame.
     if isinstance(runner.main_task_outcome, Value):
-        return cast(RetT, runner.main_task_outcome.value)
+        return cast("RetT", runner.main_task_outcome.value)
     elif isinstance(runner.main_task_outcome, Error):
         raise runner.main_task_outcome.error
     else:  # pragma: no cover
@@ -2531,7 +2531,7 @@ def my_done_callback(run_outcome):
     # this time, so it shouldn't be possible to get an exception here,
     # except for a TrioInternalError.
     next_send = cast(
-        EventResult,
+        "EventResult",
         None,
     )  # First iteration must be `None`, every iteration after that is EventResult
     for _tick in range(5):  # expected need is 2 iterations + leave some wiggle room

src/trio/_core/_tests/test_guest_mode.py

@@ -467,7 +467,7 @@ def trio_done_callback(main_outcome: Outcome[T]) -> None:
             trio_done_fut.set_result(main_outcome)
 
         if pass_not_threadsafe:
-            run_sync_soon_not_threadsafe = cast(InHost, loop.call_soon)
+            run_sync_soon_not_threadsafe = cast("InHost", loop.call_soon)
 
         trio.lowlevel.start_guest_run(
             trio_fn,

src/trio/_core/_traps.py

@@ -67,7 +67,7 @@ def _real_async_yield(
 # Real yield value is from trio's main loop, but type checkers can't
 # understand that, so we cast it to make type checkers understand.
 _async_yield = cast(
-    Callable[[MessageType], Awaitable[outcome.Outcome[object]]],
+    "Callable[[MessageType], Awaitable[outcome.Outcome[object]]]",
     _real_async_yield,
 )
 

src/trio/_core/_windows_cffi.py

@@ -395,9 +395,9 @@ class _Overlapped(Protocol):
     hEvent: Handle
 
 
-kernel32 = cast(_Kernel32, ffi.dlopen("kernel32.dll"))
-ntdll = cast(_Nt, ffi.dlopen("ntdll.dll"))
-ws2_32 = cast(_Ws2, ffi.dlopen("ws2_32.dll"))
+kernel32 = cast("_Kernel32", ffi.dlopen("kernel32.dll"))
+ntdll = cast("_Nt", ffi.dlopen("ntdll.dll"))
+ws2_32 = cast("_Ws2", ffi.dlopen("ws2_32.dll"))
 
 ################################################################
 # Magic numbers

src/trio/_tests/test_highlevel_open_tcp_listeners.py

@@ -314,7 +314,7 @@ async def handler(stream: SendStream) -> None:
         # nursery.start is incorrectly typed, awaiting #2773
         value = await nursery.start(serve_tcp, handler, 0)
         assert isinstance(value, list)
-        listeners = cast(list[SocketListener], value)
+        listeners = cast("list[SocketListener]", value)
         stream = await open_stream_to_socket_listener(listeners[0])
         async with stream:
             assert await stream.receive_some(1) == b"x"

src/trio/_tests/test_highlevel_serve_listeners.py

@@ -102,7 +102,7 @@ async def do_tests(parent_nursery: Nursery) -> None:
             listeners,
         )
         assert isinstance(value, list)
-        l2 = cast(list[MemoryListener], value)
+        l2 = cast("list[MemoryListener]", value)
         assert l2 == listeners
         # This is just split into another function because gh-136 isn't
         # implemented yet

src/trio/_tests/test_highlevel_ssl_helpers.py

@@ -92,7 +92,7 @@ async def test_open_ssl_over_tcp_stream_and_everything_else(
             ),
         )
         assert isinstance(value, list)
-        res = cast(list[SSLListener[SocketListener]], value)  # type: ignore[type-var]
+        res = cast("list[SSLListener[SocketListener]]", value)  # type: ignore[type-var]
         (listener,) = res
         async with listener:
             # listener.transport_listener is of type Listener[Stream]

src/trio/_tests/test_socket.py

@@ -655,7 +655,7 @@ async def res(
                     local=local,  # noqa: B023  # local is not bound in function definition
                 )
                 assert isinstance(value, tuple)
-                return cast(tuple[Union[str, int], ...], value)
+                return cast("tuple[Union[str, int], ...]", value)
 
             assert_eq(await res((addrs.arbitrary, "http")), (addrs.arbitrary, 80))
             if v6:

src/trio/testing/_raises_group.py

@@ -242,7 +242,7 @@ def matches(self, exception: BaseException) -> TypeGuard[MatchE]:
             return False
         # If exception_type is None check() accepts BaseException.
         # If non-none, we have done an isinstance check above.
-        return self.check is None or self.check(cast(MatchE, exception))
+        return self.check is None or self.check(cast("MatchE", exception))
 
     def __str__(self) -> str:
         reqs = []