from __future__ import annotations import array import errno import gc import io import os import platform import re import socket import sys import tempfile import threading import time from collections.abc import Generator, Iterable, Iterator from contextlib import suppress from ipaddress import IPv4Address, IPv6Address from pathlib import Path from socket import AddressFamily from ssl import SSLContext, SSLError from threading import Thread from typing import TYPE_CHECKING, Any, Literal, NoReturn, Protocol, TypeVar, cast from unittest import mock from unittest.mock import MagicMock, patch import psutil import pytest from _pytest.fixtures import SubRequest from _pytest.logging import LogCaptureFixture from _pytest.monkeypatch import MonkeyPatch from _pytest.tmpdir import TempPathFactory from pytest import FixtureRequest from pytest_mock.plugin import MockerFixture from anyio import ( BrokenResourceError, BusyResourceError, ClosedResourceError, EndOfStream, Event, TCPConnectable, TypedAttributeLookupError, UNIXConnectable, as_connectable, connect_tcp, connect_unix, create_connected_udp_socket, create_connected_unix_datagram_socket, create_task_group, create_tcp_listener, create_udp_socket, create_unix_datagram_socket, create_unix_listener, fail_after, getaddrinfo, getnameinfo, move_on_after, notify_closing, wait_all_tasks_blocked, wait_readable, wait_socket_readable, wait_socket_writable, wait_writable, ) from anyio._core._eventloop import get_async_backend from anyio.abc import ( ConnectedUDPSocket, ConnectedUNIXDatagramSocket, IPSockAddrType, Listener, SocketAttribute, SocketListener, SocketStream, UDPSocket, UNIXDatagramSocket, UNIXSocketStream, ) from anyio.lowlevel import checkpoint from anyio.streams.stapled import MultiListener from anyio.streams.tls import TLSConnectable from .conftest import asyncio_params, no_other_refs if sys.version_info < (3, 11): from exceptiongroup import ExceptionGroup if TYPE_CHECKING: from _typeshed import FileDescriptorLike AnyIPAddressFamily = Literal[ AddressFamily.AF_UNSPEC, AddressFamily.AF_INET, AddressFamily.AF_INET6 ] # If a socket can bind to ::1, the current environment has IPv6 properly configured has_ipv6 = False if socket.has_ipv6: try: s = socket.socket(AddressFamily.AF_INET6) try: s.bind(("::1", 0)) finally: s.close() del s except OSError: pass else: has_ipv6 = True skip_ipv6_mark = pytest.mark.skipif(not has_ipv6, reason="IPv6 is not available") skip_unix_abstract_mark = pytest.mark.skipif( not sys.platform.startswith("linux"), reason="Abstract namespace sockets is a Linux only feature", ) @pytest.fixture def fake_localhost_dns(monkeypatch: MonkeyPatch) -> None: def fake_getaddrinfo(*args: Any, **kwargs: Any) -> object: # Make it return IPv4 addresses first so we can test the IPv6 preference results = real_getaddrinfo(*args, **kwargs) return sorted(results, key=lambda item: item[0]) real_getaddrinfo = socket.getaddrinfo monkeypatch.setattr("socket.getaddrinfo", fake_getaddrinfo) @pytest.fixture( params=[ pytest.param(AddressFamily.AF_INET, id="ipv4"), pytest.param(AddressFamily.AF_INET6, id="ipv6", marks=[skip_ipv6_mark]), ] ) def family(request: SubRequest) -> AnyIPAddressFamily: return request.param @pytest.fixture def check_asyncio_bug(anyio_backend_name: str, family: AnyIPAddressFamily) -> None: if ( anyio_backend_name == "asyncio" and sys.platform == "win32" and family == AddressFamily.AF_INET6 ): import asyncio policy = asyncio.get_event_loop_policy() if policy.__class__.__name__ == "WindowsProactorEventLoopPolicy": pytest.skip("Does not work due to a known bug (39148)") class SockFdFactoryProtocol(Protocol): def __call__( self, family: socket.AddressFamily, kind: socket.SocketKind, *, bound: bool = False, connected: bool = False, ) -> socket.socket | int: ... @pytest.fixture( params=[pytest.param(False, id="sock"), pytest.param(True, id="fileno")] ) def sock_or_fd_factory( request: SubRequest, tmp_path_factory: TempPathFactory ) -> SockFdFactoryProtocol: def factory( family: socket.AddressFamily, kind: socket.SocketKind, *, bound: bool = False, connected: bool = False, ) -> socket.socket | int: sock = socket.socket(family, kind) if bound or connected: if family in (socket.AF_INET, socket.AF_INET6): local_addr: str | tuple[str, int] = ("localhost", 0) else: local_addr = str(tmp_path_factory.mktemp("anyio") / "socket") if bound: sock.bind(local_addr) if kind == socket.SOCK_STREAM: sock.listen() elif connected: server_sock = socket.socket(family, kind) request.addfinalizer(server_sock.close) server_sock.bind(local_addr) if kind == socket.SOCK_STREAM: server_sock.listen() sock.connect(server_sock.getsockname()) if request.param: return sock.detach() else: request.addfinalizer(sock.close) return sock return factory _T = TypeVar("_T") def _identity(v: _T) -> _T: return v def fill_socket(sock: socket.socket) -> None: try: while True: sock.send(b"x" * 65536) except BlockingIOError: pass # _ProactorBasePipeTransport.abort() after _ProactorBasePipeTransport.close() # does not cancel writes: https://bugs.python.org/issue44428 _ignore_win32_resource_warnings = ( pytest.mark.filterwarnings( "ignore:unclosed Iterator[socket.socket]: sock = socket.socket(family, socket.SOCK_STREAM) sock.settimeout(1) sock.bind(("localhost", 0)) sock.listen() yield sock sock.close() @pytest.fixture def server_addr(self, server_sock: socket.socket) -> tuple[str, int]: return server_sock.getsockname()[:2] async def test_extra_attributes( self, server_sock: socket.socket, server_addr: tuple[str, int], family: AnyIPAddressFamily, ) -> None: async with await connect_tcp(*server_addr) as stream: raw_socket = stream.extra(SocketAttribute.raw_socket) assert stream.extra(SocketAttribute.family) == family assert ( stream.extra(SocketAttribute.local_address) == raw_socket.getsockname()[:2] ) assert ( stream.extra(SocketAttribute.local_port) == raw_socket.getsockname()[1] ) assert stream.extra(SocketAttribute.remote_address) == server_addr assert stream.extra(SocketAttribute.remote_port) == server_addr[1] async def test_send_receive( self, server_sock: socket.socket, server_addr: tuple[str, int] ) -> None: async with await connect_tcp(*server_addr) as stream: client, _ = server_sock.accept() await stream.send(b"blah") request = client.recv(100) client.sendall(request[::-1]) response = await stream.receive() client.close() assert response == b"halb" async def test_send_large_buffer( self, server_sock: socket.socket, server_addr: tuple[str, int] ) -> None: def serve() -> None: client, _ = server_sock.accept() client.sendall(buffer) client.close() buffer = ( b"\xff" * 1024 * 1024 ) # should exceed the maximum kernel send buffer size async with await connect_tcp(*server_addr) as stream: thread = Thread(target=serve, daemon=True) thread.start() response = b"" while len(response) < len(buffer): chunk = await stream.receive() assert isinstance(chunk, bytes) response += chunk thread.join() assert response == buffer async def test_send_eof( self, server_sock: socket.socket, server_addr: tuple[str, int] ) -> None: def serve() -> None: client, _ = server_sock.accept() request = b"" while True: data = client.recv(100) request += data if not data: break client.sendall(request[::-1]) client.close() async with await connect_tcp(*server_addr) as stream: thread = Thread(target=serve, daemon=True) thread.start() await stream.send(b"hello, ") await stream.send(b"world\n") await stream.send_eof() response = await stream.receive() thread.join() assert response == b"\ndlrow ,olleh" async def test_iterate( self, server_sock: socket.socket, server_addr: tuple[str, int] ) -> None: def serve() -> None: client, _ = server_sock.accept() client.sendall(b"bl") event.wait(1) client.sendall(b"ah") client.close() event = threading.Event() thread = Thread(target=serve, daemon=True) thread.start() chunks = [] async with await connect_tcp(*server_addr) as stream: async for chunk in stream: chunks.append(chunk) event.set() thread.join() assert chunks == [b"bl", b"ah"] async def test_socket_options( self, family: AnyIPAddressFamily, server_addr: tuple[str, int] ) -> None: async with await connect_tcp(*server_addr) as stream: raw_socket = stream.extra(SocketAttribute.raw_socket) assert raw_socket.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY) != 0 @skip_ipv6_mark @pytest.mark.parametrize( "local_addr, expected_client_addr", [ pytest.param("", "::1", id="dualstack"), pytest.param("127.0.0.1", "127.0.0.1", id="ipv4"), pytest.param("::1", "::1", id="ipv6"), ], ) async def test_happy_eyeballs( self, local_addr: str, expected_client_addr: str, fake_localhost_dns: None ) -> None: client_addr = None, None def serve() -> None: nonlocal client_addr client, client_addr = server_sock.accept() client.close() family = ( AddressFamily.AF_INET if local_addr == "127.0.0.1" else AddressFamily.AF_INET6 ) server_sock = socket.socket(family) server_sock.bind((local_addr, 0)) server_sock.listen() port = server_sock.getsockname()[1] thread = Thread(target=serve, daemon=True) thread.start() async with await connect_tcp("localhost", port): pass thread.join() server_sock.close() assert client_addr[0] == expected_client_addr @pytest.mark.skipif( sys.implementation.name == "pypy", reason=( "gc.get_referrers is broken on PyPy (see " "https://github.com/pypy/pypy/issues/5075)" ), ) async def test_happy_eyeballs_refcycles( self, free_tcp_port: int, anyio_backend_name: str ) -> None: """ Test derived from https://github.com/python/cpython/pull/124859 """ if anyio_backend_name == "asyncio" and sys.version_info < (3, 10): pytest.skip( "asyncio.BaseEventLoop.create_connection creates refcycles on py 3.9" ) exc = None try: async with await connect_tcp("127.0.0.1", free_tcp_port): pass except OSError as e: exc = e.__cause__ assert isinstance(exc, OSError) assert gc.get_referrers(exc) == no_other_refs() @pytest.mark.parametrize( "target, exception_class", [ pytest.param( "localhost", ExceptionGroup, id="multi", marks=[skip_ipv6_mark] ), pytest.param("127.0.0.1", ConnectionRefusedError, id="single"), ], ) async def test_connection_refused( self, target: str, exception_class: type[ExceptionGroup] | type[ConnectionRefusedError], fake_localhost_dns: None, free_tcp_port: int, ) -> None: with pytest.raises(OSError) as exc: await connect_tcp(target, free_tcp_port) assert exc.match("All connection attempts failed") assert isinstance(exc.value.__cause__, exception_class) if isinstance(exc.value.__cause__, ExceptionGroup): for exception in exc.value.__cause__.exceptions: assert isinstance(exception, ConnectionRefusedError) async def test_receive_timeout( self, server_sock: socket.socket, server_addr: tuple[str, int] ) -> None: def serve() -> None: conn, _ = server_sock.accept() time.sleep(1) conn.close() thread = Thread(target=serve, daemon=True) thread.start() async with await connect_tcp(*server_addr) as stream: start_time = time.monotonic() with move_on_after(0.1): while time.monotonic() - start_time < 0.3: await stream.receive(1) pytest.fail("The timeout was not respected") async def test_concurrent_send(self, server_addr: tuple[str, int]) -> None: async def send_data() -> NoReturn: while True: await stream.send(b"\x00" * 4096) async with await connect_tcp(*server_addr) as stream: async with create_task_group() as tg: tg.start_soon(send_data) await wait_all_tasks_blocked() with pytest.raises(BusyResourceError) as exc: await stream.send(b"foo") exc.match("already writing to") tg.cancel_scope.cancel() async def test_concurrent_receive(self, server_addr: tuple[str, int]) -> None: async with await connect_tcp(*server_addr) as client: async with create_task_group() as tg: tg.start_soon(client.receive) await wait_all_tasks_blocked() try: with pytest.raises(BusyResourceError) as exc: await client.receive() exc.match("already reading from") finally: tg.cancel_scope.cancel() async def test_close_during_receive(self, server_addr: tuple[str, int]) -> None: async def interrupt() -> None: await wait_all_tasks_blocked() await stream.aclose() async with await connect_tcp(*server_addr) as stream: async with create_task_group() as tg: tg.start_soon(interrupt) with pytest.raises(ClosedResourceError): await stream.receive() async def test_receive_after_close(self, server_addr: tuple[str, int]) -> None: stream = await connect_tcp(*server_addr) await stream.aclose() with pytest.raises(ClosedResourceError): await stream.receive() async def test_send_after_close(self, server_addr: tuple[str, int]) -> None: stream = await connect_tcp(*server_addr) await stream.aclose() with pytest.raises(ClosedResourceError): await stream.send(b"foo") async def test_receive_after_peer_closed( self, family: AnyIPAddressFamily, request: FixtureRequest ) -> None: server_sock = socket.create_server(("localhost", 0), family=family) request.addfinalizer(server_sock.close) server_sock.settimeout(1) server_addr = server_sock.getsockname()[:2] async with await connect_tcp(*server_addr) as stream: client_sock, _ = server_sock.accept() client_sock.close() with pytest.raises(EndOfStream): await stream.receive(1) with pytest.raises(ClosedResourceError): await stream.receive(1) async def test_send_after_peer_closed( self, family: AnyIPAddressFamily, request: FixtureRequest ) -> None: server_sock = socket.create_server(("localhost", 0), family=family) request.addfinalizer(server_sock.close) server_sock.settimeout(1) server_addr = server_sock.getsockname()[:2] async with await connect_tcp(*server_addr) as stream: client_sock, _ = server_sock.accept() client_sock.close() with pytest.raises(BrokenResourceError): for _ in range(1000): await stream.send(b"foo") with pytest.raises(ClosedResourceError): await stream.send(b"foo") async def test_connect_tcp_with_tls( self, server_context: SSLContext, client_context: SSLContext, server_sock: socket.socket, server_addr: tuple[str, int], ) -> None: def serve() -> None: with suppress(socket.timeout): client, addr = server_sock.accept() client.settimeout(1) client = server_context.wrap_socket(client, server_side=True) data = client.recv(100) client.sendall(data[::-1]) client.unwrap() client.close() # The TLSStream tests are more comprehensive than this one! thread = Thread(target=serve, daemon=True) thread.start() async with await connect_tcp( *server_addr, tls_hostname="localhost", ssl_context=client_context ) as stream: await stream.send(b"hello") response = await stream.receive() assert response == b"olleh" thread.join() async def test_connect_tcp_with_tls_cert_check_fail( self, server_context: SSLContext, server_sock: socket.socket, server_addr: tuple[str, int], ) -> None: thread_exception = None def serve() -> None: nonlocal thread_exception client, addr = server_sock.accept() with client: client.settimeout(1) try: server_context.wrap_socket(client, server_side=True) except OSError: pass except BaseException as exc: thread_exception = exc thread = Thread(target=serve, daemon=True) thread.start() with pytest.raises(SSLError): await connect_tcp(*server_addr, tls_hostname="localhost") thread.join() assert thread_exception is None @pytest.mark.parametrize("anyio_backend", asyncio_params) async def test_unretrieved_future_exception_server_crash( self, family: AnyIPAddressFamily, caplog: LogCaptureFixture ) -> None: """ Test that there won't be any leftover Futures that don't get their exceptions retrieved. See https://github.com/encode/httpcore/issues/382 for details. """ def serve() -> None: sock, addr = server_sock.accept() event.wait(3) sock.close() del sock gc.collect() with socket.socket(family, socket.SOCK_STREAM) as server_sock: server_sock.settimeout(1) server_sock.bind(("localhost", 0)) server_sock.listen() server_addr = server_sock.getsockname()[:2] event = threading.Event() thread = Thread(target=serve) thread.start() async with await connect_tcp(*server_addr) as stream: await stream.send(b"GET") event.set() with pytest.raises(BrokenResourceError): await stream.receive() thread.join() gc.collect() caplog_text = "\n".join( msg for msg in caplog.messages if not re.search("took [0-9.]+ seconds", msg) ) assert not caplog_text async def test_from_socket( self, family: AnyIPAddressFamily, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory(family, socket.SOCK_STREAM, connected=True) async with await SocketStream.from_socket(sock_or_fd) as stream: assert isinstance(stream, SocketStream) assert stream.extra(SocketAttribute.family) == family async def test_from_socket_not_a_socket(self) -> None: with pytest.raises( TypeError, match="expected an int or socket, got str instead", ): await SocketStream.from_socket("foo") # type: ignore[arg-type] async def test_from_socket_pass_file_fd(self, tmp_path: Path) -> None: with pytest.raises( ValueError, match="the file descriptor does not refer to a socket", ): with tmp_path.joinpath("foo").open("wb") as fd: await SocketStream.from_socket(fd.fileno()) async def test_from_socket_wrong_socket_type( self, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory( socket.AF_INET, socket.SocketKind.SOCK_DGRAM, connected=True ) with pytest.raises( ValueError, match="socket type mismatch: expected SOCK_STREAM, got SOCK_DGRAM", ): await SocketStream.from_socket(sock_or_fd) async def test_from_socket_not_connected( self, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory(socket.AF_INET, socket.SOCK_STREAM, bound=True) with pytest.raises(ValueError, match="the socket must be connected"): await SocketStream.from_socket(sock_or_fd) @pytest.mark.network class TestTCPListener: async def test_extra_attributes(self, family: AnyIPAddressFamily) -> None: async with await create_tcp_listener( local_host="localhost", family=family ) as multi: assert multi.extra(SocketAttribute.family) == family for listener in multi.listeners: raw_socket = listener.extra(SocketAttribute.raw_socket) assert listener.extra(SocketAttribute.family) == family assert ( listener.extra(SocketAttribute.local_address) == raw_socket.getsockname()[:2] ) assert ( listener.extra(SocketAttribute.local_port) == raw_socket.getsockname()[1] ) pytest.raises( TypedAttributeLookupError, listener.extra, SocketAttribute.remote_address, ) pytest.raises( TypedAttributeLookupError, listener.extra, SocketAttribute.remote_port, ) @pytest.mark.parametrize( "family", [ pytest.param(AddressFamily.AF_INET, id="ipv4"), pytest.param(AddressFamily.AF_INET6, id="ipv6", marks=[skip_ipv6_mark]), pytest.param(socket.AF_UNSPEC, id="both", marks=[skip_ipv6_mark]), ], ) async def test_accept(self, family: AnyIPAddressFamily) -> None: async with await create_tcp_listener( local_host="localhost", family=family ) as multi: for listener in multi.listeners: client = socket.socket(listener.extra(SocketAttribute.family)) client.settimeout(1) addr = listener.extra(SocketAttribute.local_address) host, port = addr[0], addr[1] # On Windows, connecting to ANY (:: or 0.0.0.0) is invalid (WinError 10049). # Replace it with loopback (::1 or 127.0.0.1) to make the test portable. if sys.platform == "win32" and host in ("::", "0.0.0.0"): family = listener.extra(SocketAttribute.family) if family == socket.AF_INET6: addr = ("::1", port) elif family == socket.AF_INET: addr = ("127.0.0.1", port) client.connect(addr) assert isinstance(listener, SocketListener) stream = await listener.accept() client.sendall(b"blah") request = await stream.receive() await stream.send(request[::-1]) assert client.recv(100) == b"halb" client.close() await stream.aclose() async def test_accept_after_close(self, family: AnyIPAddressFamily) -> None: async with await create_tcp_listener( local_host="localhost", family=family ) as multi: for listener in multi.listeners: await listener.aclose() assert isinstance(listener, SocketListener) with pytest.raises(ClosedResourceError): await listener.accept() async def test_socket_options(self, family: AnyIPAddressFamily) -> None: async with await create_tcp_listener( local_host="localhost", family=family ) as multi: for listener in multi.listeners: raw_socket = listener.extra(SocketAttribute.raw_socket) if sys.platform == "win32": assert ( raw_socket.getsockopt( socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE ) != 0 ) else: assert ( raw_socket.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) != 0 ) raw_socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 80000) assert raw_socket.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF) in ( 80000, 160000, ) client = socket.socket(raw_socket.family) client.settimeout(1) client.connect(raw_socket.getsockname()) assert isinstance(listener, SocketListener) async with await listener.accept() as stream: raw_socket = stream.extra(SocketAttribute.raw_socket) assert raw_socket.gettimeout() == 0 assert raw_socket.family == listener.extra(SocketAttribute.family) assert ( raw_socket.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY) != 0 ) client.close() @pytest.mark.skipif( not hasattr(socket, "SO_REUSEPORT"), reason="SO_REUSEPORT option not supported" ) async def test_reuse_port(self, family: AnyIPAddressFamily) -> None: multi1 = await create_tcp_listener( local_host="localhost", family=family, reuse_port=True ) assert len(multi1.listeners) == 1 multi2 = await create_tcp_listener( local_host="localhost", local_port=multi1.listeners[0].extra(SocketAttribute.local_port), family=family, reuse_port=True, ) assert len(multi2.listeners) == 1 assert multi1.listeners[0].extra( SocketAttribute.local_address ) == multi2.listeners[0].extra(SocketAttribute.local_address) await multi1.aclose() await multi2.aclose() async def test_close_from_other_task(self, family: AnyIPAddressFamily) -> None: listener = await create_tcp_listener(local_host="localhost", family=family) with pytest.raises(ExceptionGroup) as exc: async with create_task_group() as tg: tg.start_soon(listener.serve, lambda stream: None) await wait_all_tasks_blocked() await listener.aclose() tg.cancel_scope.cancel() assert len(exc.value.exceptions) == 1 assert isinstance(exc.value.exceptions[0], ExceptionGroup) nested_grp = exc.value.exceptions[0] assert len(nested_grp.exceptions) == 1 assert isinstance(nested_grp.exceptions[0], ExceptionGroup) async def test_send_after_eof(self, family: AnyIPAddressFamily) -> None: async def handle(stream: SocketStream) -> None: async with stream: await stream.send(b"Hello\n") multi = await create_tcp_listener(family=family, local_host="localhost") async with multi, create_task_group() as tg: tg.start_soon(multi.serve, handle) await wait_all_tasks_blocked() with socket.socket(family) as client: client.connect(multi.extra(SocketAttribute.local_address)) client.shutdown(socket.SHUT_WR) client.setblocking(False) with fail_after(1): while True: try: message = client.recv(10) except BlockingIOError: await checkpoint() else: assert message == b"Hello\n" break tg.cancel_scope.cancel() async def test_eof_after_send(self, family: AnyIPAddressFamily) -> None: """Regression test for #701.""" received_bytes = b"" async def handle(stream: SocketStream) -> None: nonlocal received_bytes async with stream: received_bytes = await stream.receive() with pytest.raises(EndOfStream), fail_after(1): await stream.receive() tg.cancel_scope.cancel() multi = await create_tcp_listener(family=family, local_host="localhost") async with multi, create_task_group() as tg: with socket.socket(family) as client: client.connect(multi.extra(SocketAttribute.local_address)) client.send(b"Hello") client.shutdown(socket.SHUT_WR) await multi.serve(handle) assert received_bytes == b"Hello" @skip_ipv6_mark @pytest.mark.skipif( sys.platform == "win32", reason="Windows does not support interface name suffixes", ) async def test_bind_link_local(self) -> None: # Regression test for #554 link_local_ipv6_address = next( ( addr.address for addresses in psutil.net_if_addrs().values() for addr in addresses if addr.address.startswith("fe80::") and "%" in addr.address ), None, ) if link_local_ipv6_address is None: pytest.fail("Could not find a link-local IPv6 interface") async with await create_tcp_listener(local_host=link_local_ipv6_address): pass async def test_from_socket( self, family: AnyIPAddressFamily, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory(family, socket.SOCK_STREAM, bound=True) async with await SocketListener.from_socket(sock_or_fd) as listener: assert isinstance(listener, SocketListener) assert listener.extra(SocketAttribute.family) == family async def test_from_socket_not_bound( self, family: AnyIPAddressFamily, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory(family, socket.SOCK_STREAM) with pytest.raises(ValueError, match="the socket must be bound"): await SocketListener.from_socket(sock_or_fd) @pytest.mark.parametrize( "local_host,family,expected_listeners,local_port", [ (None, AddressFamily.AF_UNSPEC, 1 if socket.has_dualstack_ipv6() else 2, 0), ("localhost", AddressFamily.AF_UNSPEC, 2, 0), ("localhost", AddressFamily.AF_INET, 1, 0), ("::", AddressFamily.AF_UNSPEC, 1, 0), ("127.0.0.1", AddressFamily.AF_INET, 1, 0), ("::1", AddressFamily.AF_INET6, 1, 0), ( None, AddressFamily.AF_UNSPEC, 1 if socket.has_dualstack_ipv6() else 2, 54321, ), ("localhost", AddressFamily.AF_UNSPEC, 2, 54321), ("localhost", AddressFamily.AF_INET, 1, 54321), ], ) async def test_tcp_listener_same_port( self, local_host: str | None, family: AnyIPAddressFamily, expected_listeners: int, local_port: int, ) -> None: async with await create_tcp_listener( local_host=local_host, family=family, local_port=local_port ) as multi: ports = { listener.extra(SocketAttribute.local_port) for listener in multi.listeners } assert len(ports) == 1 if local_port != 0: assert ports == {local_port} assert len(multi.listeners) == expected_listeners @skip_ipv6_mark async def test_tcp_listener_dualstack_disabled( self, monkeypatch: MonkeyPatch ) -> None: monkeypatch.setattr(socket, "has_dualstack_ipv6", lambda: False) async with await create_tcp_listener( local_host=None, family=AddressFamily.AF_UNSPEC ) as multi: families = { listener.extra(SocketAttribute.family) for listener in multi.listeners } assert families == {socket.AF_INET, socket.AF_INET6} assert len(multi.listeners) == 2 ports = { listener.extra(SocketAttribute.local_port) for listener in multi.listeners } assert len(ports) == 1 async def test_tcp_listener_retry_after_partial_failure(self) -> None: """ Simulate a case where the first bind() succeeds with an ephemeral port, the second bind() fails with EADDRINUSE, and verify that create_tcp_listener retries and eventually succeeds with all listeners bound to the same port. """ real_bind = socket.socket.bind bind_count = 0 fail_once = True def fake_bind( self: socket.socket, addr: tuple[str, int] | tuple[str, int, int, int] ) -> None: nonlocal bind_count, fail_once port = addr[1] if isinstance(addr, tuple) and len(addr) >= 2 else None bind_count += 1 if bind_count == 1 and port == 0: return real_bind(self, addr) if fail_once and port != 0: fail_once = False raise OSError(errno.EADDRINUSE, "simulated collision on second bind") return real_bind(self, addr) with patch.object(socket.socket, "bind", new=fake_bind): async with await create_tcp_listener( local_host="localhost", family=socket.AF_UNSPEC, local_port=0 ) as multi: assert bind_count >= 2 ports = { listener.extra(SocketAttribute.local_port) for listener in multi.listeners } assert len(ports) == 1 assert all( isinstance(listener, SocketListener) for listener in multi.listeners ) async def test_tcp_listener_total_bind_failure(self) -> None: """ Test for a situation where bind() always fails when other listeners are being bound to the same port as the first listener which was randomly assigned a free port by the kernel. """ def raise_oserror(addr: tuple[str, int]) -> None: # Pretend that every explicitly requested port is already in use if addr[1] != 0: raise OSError(errno.EADDRINUSE, "bind failure") mock_socket_instance = MagicMock() mock_socket_instance.bind.side_effect = raise_oserror asynclib = get_async_backend() with ( patch("anyio._core._sockets.socket") as mock_anyio_sockets, patch.object( asynclib, "create_tcp_listener", return_value=MagicMock(SocketListener) ), pytest.raises( OSError, match="Could not create 2 listeners with a consistent port" ), ): mock_anyio_sockets.socket.configure_mock(return_value=mock_socket_instance) await create_tcp_listener(local_host="localhost") @pytest.mark.skipif( sys.platform == "win32", reason="UNIX sockets are not available on Windows" ) class TestUNIXStream: @pytest.fixture( params=[ "path", pytest.param("abstract", marks=[skip_unix_abstract_mark]), ] ) def socket_path(self, request: SubRequest) -> Generator[Path, None, None]: # Use stdlib tempdir generation # Fixes `OSError: AF_UNIX path too long` from pytest generated temp_path with tempfile.TemporaryDirectory() as path: if request.param == "path": yield Path(path) / "socket" else: yield Path(f"\0{path}") / "socket" @pytest.fixture(params=[False, True], ids=["str", "path"]) def socket_path_or_str(self, request: SubRequest, socket_path: Path) -> Path | str: return socket_path if request.param else str(socket_path) @pytest.fixture def server_sock(self, socket_path: Path) -> Iterable[socket.socket]: sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.settimeout(1) sock.bind(str(socket_path)) sock.listen() yield sock sock.close() async def test_extra_attributes( self, server_sock: socket.socket, socket_path: Path ) -> None: async with await connect_unix(socket_path) as stream: raw_socket = stream.extra(SocketAttribute.raw_socket) assert stream.extra(SocketAttribute.family) == socket.AF_UNIX assert ( stream.extra(SocketAttribute.local_address) == raw_socket.getsockname() ) remote_addr = stream.extra(SocketAttribute.remote_address) if isinstance(remote_addr, str): assert stream.extra(SocketAttribute.remote_address) == str(socket_path) else: assert isinstance(remote_addr, bytes) assert stream.extra(SocketAttribute.remote_address) == bytes( socket_path ) pytest.raises( TypedAttributeLookupError, stream.extra, SocketAttribute.local_port ) pytest.raises( TypedAttributeLookupError, stream.extra, SocketAttribute.remote_port ) async def test_send_receive( self, server_sock: socket.socket, socket_path_or_str: Path | str ) -> None: async with await connect_unix(socket_path_or_str) as stream: client, _ = server_sock.accept() await stream.send(b"blah") request = client.recv(100) client.sendall(request[::-1]) response = await stream.receive() client.close() assert response == b"halb" async def test_receive_large_buffer( self, server_sock: socket.socket, socket_path: Path ) -> None: def serve() -> None: client, _ = server_sock.accept() client.sendall(buffer) client.close() buffer = ( b"\xff" * 1024 * 512 + b"\x00" * 1024 * 512 ) # should exceed the maximum kernel send buffer size async with await connect_unix(socket_path) as stream: thread = Thread(target=serve, daemon=True) thread.start() response = b"" while len(response) < len(buffer): response += await stream.receive() thread.join() assert response == buffer async def test_send_large_buffer( self, server_sock: socket.socket, socket_path: Path ) -> None: response = b"" def serve() -> None: nonlocal response client, _ = server_sock.accept() while True: data = client.recv(1024) if not data: break response += data client.close() buffer = ( b"\xff" * 1024 * 512 + b"\x00" * 1024 * 512 ) # should exceed the maximum kernel send buffer size async with await connect_unix(socket_path) as stream: thread = Thread(target=serve, daemon=True) thread.start() await stream.send(buffer) thread.join() assert response == buffer async def test_receive_fds( self, server_sock: socket.socket, socket_path: Path, tmp_path: Path ) -> None: def serve() -> None: path1 = tmp_path / "file1" path2 = tmp_path / "file2" path1.write_text("Hello, ") path2.write_text("World!") with path1.open() as file1, path2.open() as file2: fdarray = array.array("i", [file1.fileno(), file2.fileno()]) client, _ = server_sock.accept() cmsg = (socket.SOL_SOCKET, socket.SCM_RIGHTS, fdarray) with client: client.sendmsg([b"test"], [cmsg]) async with await connect_unix(socket_path) as stream: thread = Thread(target=serve, daemon=True) thread.start() message, fds = await stream.receive_fds(10, 2) thread.join() text = "" for fd in fds: with os.fdopen(fd) as file: text += file.read() assert message == b"test" assert text == "Hello, World!" async def test_receive_fds_bad_args( self, server_sock: socket.socket, socket_path: Path ) -> None: async with await connect_unix(socket_path) as stream: for msglen in (-1, "foo"): with pytest.raises( ValueError, match="msglen must be a non-negative integer" ): await stream.receive_fds(msglen, 0) # type: ignore[arg-type] for maxfds in (0, "foo"): with pytest.raises( ValueError, match="maxfds must be a positive integer" ): await stream.receive_fds(0, maxfds) # type: ignore[arg-type] async def test_send_fds( self, server_sock: socket.socket, socket_path: Path, tmp_path: Path ) -> None: def serve() -> None: fds = array.array("i") client, _ = server_sock.accept() msg, ancdata, *_ = client.recvmsg(10, socket.CMSG_LEN(2 * fds.itemsize)) client.close() assert msg == b"test" for cmsg_level, cmsg_type, cmsg_data in ancdata: assert cmsg_level == socket.SOL_SOCKET assert cmsg_type == socket.SCM_RIGHTS fds.frombytes( cmsg_data[: len(cmsg_data) - (len(cmsg_data) % fds.itemsize)] ) text = "" for fd in fds: with os.fdopen(fd) as file: text += file.read() assert text == "Hello, World!" path1 = tmp_path / "file1" path2 = tmp_path / "file2" path1.write_text("Hello, ") path2.write_text("World!") with path1.open() as file1, path2.open() as file2, fail_after(2): assert isinstance(file1, io.TextIOWrapper) assert isinstance(file2, io.TextIOWrapper) async with await connect_unix(socket_path) as stream: thread = Thread(target=serve, daemon=True) thread.start() await stream.send_fds(b"test", [file1, file2]) thread.join() async def test_send_eof( self, server_sock: socket.socket, socket_path: Path ) -> None: def serve() -> None: client, _ = server_sock.accept() request = b"" while True: data = client.recv(100) request += data if not data: break client.sendall(request[::-1]) client.close() async with await connect_unix(socket_path) as stream: thread = Thread(target=serve, daemon=True) thread.start() await stream.send(b"hello, ") await stream.send(b"world\n") await stream.send_eof() response = await stream.receive() thread.join() assert response == b"\ndlrow ,olleh" async def test_iterate(self, server_sock: socket.socket, socket_path: Path) -> None: def serve() -> None: client, _ = server_sock.accept() client.sendall(b"bl") time.sleep(0.05) client.sendall(b"ah") client.close() thread = Thread(target=serve, daemon=True) thread.start() async with await connect_unix(socket_path) as stream: chunks = [chunk async for chunk in stream] thread.join() assert chunks == [b"bl", b"ah"] async def test_send_fds_bad_args( self, server_sock: socket.socket, socket_path: Path ) -> None: async with await connect_unix(socket_path) as stream: with pytest.raises(ValueError, match="message must not be empty"): await stream.send_fds(b"", [0]) with pytest.raises(ValueError, match="fds must not be empty"): await stream.send_fds(b"test", []) async def test_concurrent_send( self, server_sock: socket.socket, socket_path: Path ) -> None: async def send_data() -> NoReturn: while True: await client.send(b"\x00" * 4096) async with await connect_unix(socket_path) as client: async with create_task_group() as tg: tg.start_soon(send_data) await wait_all_tasks_blocked() with pytest.raises(BusyResourceError) as exc: await client.send(b"foo") exc.match("already writing to") tg.cancel_scope.cancel() async def test_concurrent_receive( self, server_sock: socket.socket, socket_path: Path ) -> None: async with await connect_unix(socket_path) as client: async with create_task_group() as tg: tg.start_soon(client.receive) await wait_all_tasks_blocked() try: with pytest.raises(BusyResourceError) as exc: await client.receive() exc.match("already reading from") finally: tg.cancel_scope.cancel() async def test_close_during_receive( self, server_sock: socket.socket, socket_path: Path ) -> None: async def interrupt() -> None: await wait_all_tasks_blocked() await stream.aclose() async with await connect_unix(socket_path) as stream: async with create_task_group() as tg: tg.start_soon(interrupt) with pytest.raises(ClosedResourceError): await stream.receive() async def test_receive_after_close( self, server_sock: socket.socket, socket_path: Path ) -> None: stream = await connect_unix(socket_path) await stream.aclose() with pytest.raises(ClosedResourceError): await stream.receive() async def test_send_after_close( self, server_sock: socket.socket, socket_path: Path ) -> None: stream = await connect_unix(socket_path) await stream.aclose() with pytest.raises(ClosedResourceError): await stream.send(b"foo") async def test_cannot_connect(self, socket_path: Path) -> None: if str(socket_path).startswith("\0"): with pytest.raises(ConnectionRefusedError): await connect_unix(socket_path) else: with pytest.raises(FileNotFoundError): await connect_unix(socket_path) async def test_connecting_using_bytes( self, server_sock: socket.socket, socket_path: Path ) -> None: async with await connect_unix(str(socket_path).encode()): pass @pytest.mark.skipif( platform.system() == "Darwin", reason="macOS requires valid UTF-8 paths" ) async def test_connecting_with_non_utf8(self, socket_path: Path) -> None: actual_path = str(socket_path).encode() + b"\xf0" with socket.socket(socket.AF_UNIX) as server: server.bind(actual_path) server.listen(1) async with await connect_unix(actual_path): pass async def test_from_socket( self, socket_path: Path, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory( socket.AF_UNIX, socket.SOCK_STREAM, connected=True ) async with await UNIXSocketStream.from_socket(sock_or_fd) as stream: assert isinstance(stream, UNIXSocketStream) async def test_from_socket_wrong_socket_type( self, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory( socket.AF_UNIX, socket.SOCK_DGRAM, connected=True ) with pytest.raises( ValueError, match="socket type mismatch: expected SOCK_STREAM, got SOCK_DGRAM", ): await UNIXSocketStream.from_socket(sock_or_fd) async def test_from_socket_wrong_address_family( self, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory( socket.AF_INET, socket.SOCK_STREAM, connected=True ) with pytest.raises( ValueError, match="address family mismatch: expected AF_UNIX, got AF_INET", ): await UNIXSocketStream.from_socket(sock_or_fd) async def test_from_socket_not_connected( self, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory(socket.AF_UNIX, socket.SOCK_STREAM) with pytest.raises(ValueError, match="the socket must be connected"): await UNIXSocketStream.from_socket(sock_or_fd) @pytest.mark.skipif( sys.platform == "win32", reason="UNIX sockets are not available on Windows" ) class TestUNIXListener: @pytest.fixture( params=[ "path", pytest.param("abstract", marks=[skip_unix_abstract_mark]), ] ) def socket_path(self, request: SubRequest) -> Generator[Path, None, None]: # Use stdlib tempdir generation # Fixes `OSError: AF_UNIX path too long` from pytest generated temp_path with tempfile.TemporaryDirectory() as path: if request.param == "path": yield Path(path) / "socket" else: yield Path(f"\0{path}") / "socket" @pytest.fixture(params=[False, True], ids=["str", "path"]) def socket_path_or_str(self, request: SubRequest, socket_path: Path) -> Path | str: return socket_path if request.param else str(socket_path) async def test_extra_attributes(self, socket_path: Path) -> None: async with await create_unix_listener(socket_path) as listener: raw_socket = listener.extra(SocketAttribute.raw_socket) assert listener.extra(SocketAttribute.family) == socket.AF_UNIX assert ( listener.extra(SocketAttribute.local_address) == raw_socket.getsockname() ) pytest.raises( TypedAttributeLookupError, listener.extra, SocketAttribute.local_port ) pytest.raises( TypedAttributeLookupError, listener.extra, SocketAttribute.remote_address, ) pytest.raises( TypedAttributeLookupError, listener.extra, SocketAttribute.remote_port ) async def test_accept(self, socket_path_or_str: Path | str) -> None: async with await create_unix_listener(socket_path_or_str) as listener: client = socket.socket(socket.AF_UNIX) client.settimeout(1) client.connect(str(socket_path_or_str)) stream = await listener.accept() client.sendall(b"blah") request = await stream.receive() await stream.send(request[::-1]) assert client.recv(100) == b"halb" client.close() await stream.aclose() async def test_socket_options(self, socket_path: Path) -> None: async with await create_unix_listener(socket_path) as listener: listener_socket = listener.extra(SocketAttribute.raw_socket) assert listener_socket.family == socket.AF_UNIX listener_socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 80000) assert listener_socket.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF) in ( 80000, 160000, ) client = socket.socket(listener_socket.family) client.settimeout(1) client.connect(listener_socket.getsockname()) async with await listener.accept() as stream: assert stream.extra(SocketAttribute.raw_socket).gettimeout() == 0 assert stream.extra(SocketAttribute.family) == listener_socket.family client.close() async def test_send_after_eof(self, socket_path: Path) -> None: async def handle(stream: SocketStream) -> None: async with stream: await stream.send(b"Hello\n") async with ( await create_unix_listener(socket_path) as listener, create_task_group() as tg, ): tg.start_soon(listener.serve, handle) await wait_all_tasks_blocked() with socket.socket(socket.AF_UNIX) as client: client.connect(str(socket_path)) client.shutdown(socket.SHUT_WR) client.setblocking(False) with fail_after(1): while True: try: message = client.recv(10) except BlockingIOError: await checkpoint() else: assert message == b"Hello\n" break tg.cancel_scope.cancel() async def test_bind_twice(self, socket_path: Path) -> None: """Test that the previous socket is removed before binding to the path.""" for _ in range(2): async with await create_unix_listener(socket_path): pass async def test_listening_bytes_path(self, socket_path: Path) -> None: async with await create_unix_listener(str(socket_path).encode()): pass @pytest.mark.skipif( platform.system() == "Darwin", reason="macOS requires valid UTF-8 paths" ) async def test_listening_invalid_ascii(self, socket_path: Path) -> None: real_path = str(socket_path).encode() + b"\xf0" async with await create_unix_listener(real_path): pass async def test_from_socket(self, sock_or_fd_factory: SockFdFactoryProtocol) -> None: sock_or_fd = sock_or_fd_factory(socket.AF_UNIX, socket.SOCK_STREAM, bound=True) async with await SocketListener.from_socket(sock_or_fd) as listener: assert isinstance(listener, SocketListener) assert listener.extra(SocketAttribute.family) == socket.AF_UNIX async def test_multi_listener(tmp_path_factory: TempPathFactory) -> None: async def handle(stream: SocketStream) -> None: client_addresses.append(stream.extra(SocketAttribute.remote_address)) event.set() await stream.aclose() client_addresses: list[str | IPSockAddrType] = [] listeners: list[Listener] = [await create_tcp_listener(local_host="localhost")] with tempfile.TemporaryDirectory() as path: if sys.platform != "win32": listeners.append(await create_unix_listener(Path(path) / "socket")) expected_addresses: list[str | IPSockAddrType] = [] async with MultiListener(listeners) as multi_listener: async with create_task_group() as tg: tg.start_soon(multi_listener.serve, handle) for listener in multi_listener.listeners: event = Event() local_address = listener.extra(SocketAttribute.local_address) if ( sys.platform != "win32" and listener.extra(SocketAttribute.family) == socket.AF_UNIX ): assert isinstance(local_address, str) stream: SocketStream = await connect_unix(local_address) else: assert isinstance(local_address, tuple) host, port = local_address host = ( "::1" if listener.extra(SocketAttribute.family) == socket.AF_INET6 else "127.0.0.1" ) stream = await connect_tcp(host, port) expected_addresses.append( stream.extra(SocketAttribute.local_address) ) await event.wait() await stream.aclose() tg.cancel_scope.cancel() assert client_addresses == expected_addresses @pytest.mark.network @pytest.mark.usefixtures("check_asyncio_bug") class TestUDPSocket: async def test_extra_attributes(self, family: AnyIPAddressFamily) -> None: async with await create_udp_socket( family=family, local_host="localhost" ) as udp: raw_socket = udp.extra(SocketAttribute.raw_socket) assert raw_socket.gettimeout() == 0 assert udp.extra(SocketAttribute.family) == family assert ( udp.extra(SocketAttribute.local_address) == raw_socket.getsockname()[:2] ) assert udp.extra(SocketAttribute.local_port) == raw_socket.getsockname()[1] pytest.raises( TypedAttributeLookupError, udp.extra, SocketAttribute.remote_address ) pytest.raises( TypedAttributeLookupError, udp.extra, SocketAttribute.remote_port ) async def test_send_receive(self, family: AnyIPAddressFamily) -> None: async with await create_udp_socket( local_host="localhost", family=family ) as sock: host, port = sock.extra(SocketAttribute.local_address) # type: ignore[misc] await sock.sendto(b"blah", host, port) request, addr = await sock.receive() assert request == b"blah" assert addr == sock.extra(SocketAttribute.local_address) await sock.sendto(b"halb", host, port) response, addr = await sock.receive() assert response == b"halb" assert addr == (host, port) async def test_iterate(self, family: AnyIPAddressFamily) -> None: async def serve() -> None: async for packet, addr in server: await server.send((packet[::-1], addr)) async with await create_udp_socket( family=family, local_host="localhost" ) as server: host, port = server.extra( # type: ignore[misc] SocketAttribute.local_address ) async with await create_udp_socket( family=family, local_host="localhost" ) as client: async with create_task_group() as tg: tg.start_soon(serve) await client.sendto(b"FOOBAR", host, port) assert await client.receive() == (b"RABOOF", (host, port)) await client.sendto(b"123456", host, port) assert await client.receive() == (b"654321", (host, port)) tg.cancel_scope.cancel() @pytest.mark.skipif( not hasattr(socket, "SO_REUSEPORT"), reason="SO_REUSEPORT option not supported" ) async def test_reuse_port(self, family: AnyIPAddressFamily) -> None: async with await create_udp_socket( family=family, local_host="localhost", reuse_port=True ) as udp: port = udp.extra(SocketAttribute.local_port) assert port != 0 async with await create_udp_socket( family=family, local_host="localhost", local_port=port, reuse_port=True ) as udp2: assert port == udp2.extra(SocketAttribute.local_port) async def test_concurrent_receive(self) -> None: async with await create_udp_socket( family=AddressFamily.AF_INET, local_host="localhost" ) as udp: async with create_task_group() as tg: tg.start_soon(udp.receive) await wait_all_tasks_blocked() try: with pytest.raises(BusyResourceError) as exc: await udp.receive() exc.match("already reading from") finally: tg.cancel_scope.cancel() async def test_close_during_receive(self) -> None: async def close_when_blocked() -> None: await wait_all_tasks_blocked() await udp.aclose() async with await create_udp_socket( family=AddressFamily.AF_INET, local_host="localhost" ) as udp: async with create_task_group() as tg: tg.start_soon(close_when_blocked) with pytest.raises(ClosedResourceError): await udp.receive() async def test_receive_after_close(self) -> None: udp = await create_udp_socket( family=AddressFamily.AF_INET, local_host="localhost" ) await udp.aclose() with pytest.raises(ClosedResourceError): await udp.receive() async def test_send_after_close(self) -> None: udp = await create_udp_socket( family=AddressFamily.AF_INET, local_host="localhost" ) host, port = udp.extra(SocketAttribute.local_address) # type: ignore[misc] await udp.aclose() with pytest.raises(ClosedResourceError): await udp.sendto(b"foo", host, port) async def test_create_unbound_socket(self, family: AnyIPAddressFamily) -> None: """Regression test for #360.""" async with await create_udp_socket(family=family) as udp: local_address = cast( IPSockAddrType, udp.extra(SocketAttribute.local_address) ) assert local_address[1] > 0 async def test_from_socket( self, family: AnyIPAddressFamily, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory(family, socket.SOCK_DGRAM, bound=True) async with await UDPSocket.from_socket(sock_or_fd) as udp_socket: assert isinstance(udp_socket, UDPSocket) assert udp_socket.extra(SocketAttribute.family) == family async def test_from_socket_wrong_socket_type( self, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory( socket.AF_INET, socket.SOCK_STREAM, connected=True ) with pytest.raises( ValueError, match="socket type mismatch: expected SOCK_DGRAM, got SOCK_STREAM", ): await UDPSocket.from_socket(sock_or_fd) @pytest.mark.network @pytest.mark.usefixtures("check_asyncio_bug") class TestConnectedUDPSocket: async def test_extra_attributes(self, family: AnyIPAddressFamily) -> None: async with await create_connected_udp_socket( "localhost", 5000, family=family ) as udp: raw_socket = udp.extra(SocketAttribute.raw_socket) assert udp.extra(SocketAttribute.family) == family assert ( udp.extra(SocketAttribute.local_address) == raw_socket.getsockname()[:2] ) assert udp.extra(SocketAttribute.local_port) == raw_socket.getsockname()[1] assert ( udp.extra(SocketAttribute.remote_address) == raw_socket.getpeername()[:2] ) assert udp.extra(SocketAttribute.remote_port) == 5000 async def test_send_receive(self, family: AnyIPAddressFamily) -> None: async with await create_udp_socket( family=family, local_host="localhost" ) as udp1: host, port = udp1.extra(SocketAttribute.local_address) # type: ignore[misc] async with await create_connected_udp_socket( host, port, local_host="localhost", family=family ) as udp2: host, port = udp2.extra( SocketAttribute.local_address # type: ignore[misc] ) await udp2.send(b"blah") request = await udp1.receive() assert request == (b"blah", (host, port)) await udp1.sendto(b"halb", host, port) response = await udp2.receive() assert response == b"halb" async def test_iterate(self, family: AnyIPAddressFamily) -> None: async def serve() -> None: async for packet in udp2: await udp2.send(packet[::-1]) async with await create_udp_socket( family=family, local_host="localhost" ) as udp1: host, port = udp1.extra(SocketAttribute.local_address) # type: ignore[misc] async with await create_connected_udp_socket(host, port) as udp2: host, port = udp2.extra( # type: ignore[misc] SocketAttribute.local_address ) async with create_task_group() as tg: tg.start_soon(serve) await udp1.sendto(b"FOOBAR", host, port) assert await udp1.receive() == (b"RABOOF", (host, port)) await udp1.sendto(b"123456", host, port) assert await udp1.receive() == (b"654321", (host, port)) tg.cancel_scope.cancel() @pytest.mark.skipif( not hasattr(socket, "SO_REUSEPORT"), reason="SO_REUSEPORT option not supported" ) async def test_reuse_port(self, family: AnyIPAddressFamily) -> None: async with await create_connected_udp_socket( "localhost", 6000, family=family, local_host="localhost", reuse_port=True ) as udp: port = udp.extra(SocketAttribute.local_port) assert port != 0 async with await create_connected_udp_socket( "localhost", 6001, family=family, local_host="localhost", local_port=port, reuse_port=True, ) as udp2: assert port == udp2.extra(SocketAttribute.local_port) async def test_concurrent_receive(self) -> None: async with await create_connected_udp_socket( "localhost", 5000, local_host="localhost", family=AddressFamily.AF_INET ) as udp: async with create_task_group() as tg: tg.start_soon(udp.receive) await wait_all_tasks_blocked() try: with pytest.raises(BusyResourceError) as exc: await udp.receive() exc.match("already reading from") finally: tg.cancel_scope.cancel() async def test_close_during_receive(self) -> None: async def close_when_blocked() -> None: await wait_all_tasks_blocked() await udp.aclose() async with await create_connected_udp_socket( "localhost", 5000, local_host="localhost", family=AddressFamily.AF_INET ) as udp: async with create_task_group() as tg: tg.start_soon(close_when_blocked) with pytest.raises(ClosedResourceError): await udp.receive() async def test_receive_after_close(self, family: AnyIPAddressFamily) -> None: udp = await create_connected_udp_socket( "localhost", 5000, local_host="localhost", family=family ) await udp.aclose() with pytest.raises(ClosedResourceError): await udp.receive() async def test_send_after_close(self, family: AnyIPAddressFamily) -> None: udp = await create_connected_udp_socket( "localhost", 5000, local_host="localhost", family=family ) await udp.aclose() with pytest.raises(ClosedResourceError): await udp.send(b"foo") async def test_from_socket( self, family: AnyIPAddressFamily, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory(family, socket.SOCK_DGRAM, connected=True) async with await ConnectedUDPSocket.from_socket(sock_or_fd) as udp_socket: assert isinstance(udp_socket, ConnectedUDPSocket) assert udp_socket.extra(SocketAttribute.family) == family async def test_from_socket_wrong_socket_type( self, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory( socket.AF_INET, socket.SOCK_STREAM, connected=True ) with pytest.raises( ValueError, match="socket type mismatch: expected SOCK_DGRAM, got SOCK_STREAM", ): await ConnectedUDPSocket.from_socket(sock_or_fd) async def test_from_socket_not_connected( self, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory(socket.AF_INET, socket.SOCK_DGRAM, bound=True) with pytest.raises(ValueError, match="the socket must be connected"): await ConnectedUDPSocket.from_socket(sock_or_fd) @pytest.mark.skipif( sys.platform == "win32", reason="UNIX sockets are not available on Windows" ) class TestUNIXDatagramSocket: @pytest.fixture( params=[ "path", pytest.param("abstract", marks=[skip_unix_abstract_mark]), ] ) def socket_path(self, request: SubRequest) -> Generator[Path, None, None]: # Use stdlib tempdir generation # Fixes `OSError: AF_UNIX path too long` from pytest generated temp_path with tempfile.TemporaryDirectory() as path: if request.param == "path": yield Path(path) / "socket" else: yield Path(f"\0{path}") / "socket" @pytest.fixture(params=[False, True], ids=["str", "path"]) def socket_path_or_str(self, request: SubRequest, socket_path: Path) -> Path | str: return socket_path if request.param else str(socket_path) @pytest.fixture def peer_socket_path(self) -> Generator[Path, None, None]: # Use stdlib tempdir generation # Fixes `OSError: AF_UNIX path too long` from pytest generated temp_path with tempfile.TemporaryDirectory() as path: yield Path(path) / "peer_socket" async def test_extra_attributes(self, socket_path: Path) -> None: async with await create_unix_datagram_socket(local_path=socket_path) as unix_dg: raw_socket = unix_dg.extra(SocketAttribute.raw_socket) assert raw_socket.gettimeout() == 0 assert unix_dg.extra(SocketAttribute.family) == socket.AF_UNIX assert ( unix_dg.extra(SocketAttribute.local_address) == raw_socket.getsockname() ) pytest.raises( TypedAttributeLookupError, unix_dg.extra, SocketAttribute.local_port ) pytest.raises( TypedAttributeLookupError, unix_dg.extra, SocketAttribute.remote_address ) pytest.raises( TypedAttributeLookupError, unix_dg.extra, SocketAttribute.remote_port ) async def test_send_receive(self, socket_path_or_str: Path | str) -> None: async with await create_unix_datagram_socket( local_path=socket_path_or_str, ) as sock: path = str(socket_path_or_str) await sock.sendto(b"blah", path) request, addr = await sock.receive() assert request == b"blah" if isinstance(addr, bytes): assert addr == path.encode() else: assert addr == path await sock.sendto(b"halb", path) response, addr = await sock.receive() assert response == b"halb" if isinstance(addr, bytes): assert addr == path.encode() else: assert addr == path async def test_iterate(self, peer_socket_path: Path, socket_path: Path) -> None: async def serve() -> None: async for packet, addr in server: await server.send((packet[::-1], addr)) async with await create_unix_datagram_socket( local_path=peer_socket_path, ) as server: peer_path = str(peer_socket_path) async with await create_unix_datagram_socket( local_path=socket_path ) as client: async with create_task_group() as tg: tg.start_soon(serve) await client.sendto(b"FOOBAR", peer_path) assert await client.receive() == (b"RABOOF", peer_path) await client.sendto(b"123456", peer_path) assert await client.receive() == (b"654321", peer_path) tg.cancel_scope.cancel() async def test_concurrent_receive(self) -> None: async with await create_unix_datagram_socket() as unix_dg: async with create_task_group() as tg: tg.start_soon(unix_dg.receive) await wait_all_tasks_blocked() try: with pytest.raises(BusyResourceError) as exc: await unix_dg.receive() exc.match("already reading from") finally: tg.cancel_scope.cancel() async def test_close_during_receive(self) -> None: async def close_when_blocked() -> None: await wait_all_tasks_blocked() await unix_dg.aclose() async with await create_unix_datagram_socket() as unix_dg: async with create_task_group() as tg: tg.start_soon(close_when_blocked) with pytest.raises(ClosedResourceError): await unix_dg.receive() async def test_receive_after_close(self) -> None: unix_dg = await create_unix_datagram_socket() await unix_dg.aclose() with pytest.raises(ClosedResourceError): await unix_dg.receive() async def test_send_after_close(self, socket_path: Path) -> None: unix_dg = await create_unix_datagram_socket(local_path=socket_path) path = str(socket_path) await unix_dg.aclose() with pytest.raises(ClosedResourceError): await unix_dg.sendto(b"foo", path) async def test_local_path_bytes(self, socket_path: Path) -> None: async with await create_unix_datagram_socket( local_path=str(socket_path).encode() ): pass @pytest.mark.skipif( platform.system() == "Darwin", reason="macOS requires valid UTF-8 paths" ) async def test_local_path_invalid_ascii(self, socket_path: Path) -> None: real_path = str(socket_path).encode() + b"\xf0" async with await create_unix_datagram_socket(local_path=real_path): pass async def test_from_socket(self, sock_or_fd_factory: SockFdFactoryProtocol) -> None: sock_or_fd = sock_or_fd_factory(socket.AF_UNIX, socket.SOCK_DGRAM) async with await UNIXDatagramSocket.from_socket(sock_or_fd) as udp_socket: assert isinstance(udp_socket, UNIXDatagramSocket) assert udp_socket.extra(SocketAttribute.family) == socket.AF_UNIX async def test_from_socket_wrong_socket_type( self, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory(socket.AF_UNIX, socket.SOCK_STREAM) with pytest.raises( ValueError, match="socket type mismatch: expected SOCK_DGRAM, got SOCK_STREAM", ): await UNIXDatagramSocket.from_socket(sock_or_fd) @pytest.mark.skipif( sys.platform == "win32", reason="UNIX sockets are not available on Windows" ) class TestConnectedUNIXDatagramSocket: @pytest.fixture( params=[ "path", pytest.param("abstract", marks=[skip_unix_abstract_mark]), ] ) def socket_path(self, request: SubRequest) -> Generator[Path, None, None]: # Use stdlib tempdir generation # Fixes `OSError: AF_UNIX path too long` from pytest generated temp_path with tempfile.TemporaryDirectory() as path: if request.param == "path": yield Path(path) / "socket" else: yield Path(f"\0{path}") / "socket" @pytest.fixture(params=[False, True], ids=["str", "path"]) def socket_path_or_str(self, request: SubRequest, socket_path: Path) -> Path | str: return socket_path if request.param else str(socket_path) @pytest.fixture( params=[ pytest.param("path", id="path-peer"), pytest.param( "abstract", marks=[skip_unix_abstract_mark], id="abstract-peer" ), ] ) def peer_socket_path(self) -> Generator[Path, None, None]: # Use stdlib tempdir generation # Fixes `OSError: AF_UNIX path too long` from pytest generated temp_path with tempfile.TemporaryDirectory() as path: yield Path(path) / "peer_socket" @pytest.fixture(params=[False, True], ids=["peer_str", "peer_path"]) def peer_socket_path_or_str( self, request: SubRequest, peer_socket_path: Path ) -> Path | str: return peer_socket_path if request.param else str(peer_socket_path) @pytest.fixture def peer_sock(self, peer_socket_path: Path) -> Iterable[socket.socket]: sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) sock.settimeout(1) sock.bind(str(peer_socket_path)) yield sock sock.close() async def test_extra_attributes( self, socket_path: Path, peer_socket_path: Path, peer_sock: socket.socket, ) -> None: async with await create_connected_unix_datagram_socket( remote_path=peer_socket_path, local_path=socket_path, ) as unix_dg: raw_socket = unix_dg.extra(SocketAttribute.raw_socket) assert raw_socket is not None assert unix_dg.extra(SocketAttribute.family) == AddressFamily.AF_UNIX assert os.fsencode( cast(os.PathLike, unix_dg.extra(SocketAttribute.local_address)) ) == os.fsencode(socket_path) assert os.fsencode( cast(os.PathLike, unix_dg.extra(SocketAttribute.remote_address)) ) == os.fsencode(peer_socket_path) pytest.raises( TypedAttributeLookupError, unix_dg.extra, SocketAttribute.local_port ) pytest.raises( TypedAttributeLookupError, unix_dg.extra, SocketAttribute.remote_port ) async def test_send_receive( self, socket_path_or_str: Path | str, peer_socket_path_or_str: Path | str, ) -> None: async with await create_unix_datagram_socket( local_path=peer_socket_path_or_str, ) as unix_dg1: async with await create_connected_unix_datagram_socket( peer_socket_path_or_str, local_path=socket_path_or_str, ) as unix_dg2: socket_path = os.fsdecode(socket_path_or_str) await unix_dg2.send(b"blah") data, remote_addr = await unix_dg1.receive() assert (data, os.fsdecode(remote_addr)) == (b"blah", socket_path) await unix_dg1.sendto(b"halb", socket_path) response = await unix_dg2.receive() assert response == b"halb" async def test_iterate( self, socket_path: Path, peer_socket_path: Path, ) -> None: async def serve() -> None: async for packet in unix_dg2: await unix_dg2.send(packet[::-1]) async with await create_unix_datagram_socket( local_path=peer_socket_path, ) as unix_dg1: async with await create_connected_unix_datagram_socket( peer_socket_path, local_path=socket_path ) as unix_dg2: path = os.fsdecode(socket_path) async with create_task_group() as tg: tg.start_soon(serve) await unix_dg1.sendto(b"FOOBAR", path) data, addr = await unix_dg1.receive() assert (data, os.fsdecode(addr)) == (b"RABOOF", path) await unix_dg1.sendto(b"123456", path) data, addr = await unix_dg1.receive() assert (data, os.fsdecode(addr)) == (b"654321", path) tg.cancel_scope.cancel() async def test_concurrent_receive( self, peer_socket_path: Path, peer_sock: socket.socket ) -> None: async with await create_connected_unix_datagram_socket( peer_socket_path ) as unix_dg: async with create_task_group() as tg: tg.start_soon(unix_dg.receive) await wait_all_tasks_blocked() try: with pytest.raises(BusyResourceError) as exc: await unix_dg.receive() exc.match("already reading from") finally: tg.cancel_scope.cancel() async def test_close_during_receive( self, peer_socket_path_or_str: Path | str, peer_sock: socket.socket ) -> None: async def close_when_blocked() -> None: await wait_all_tasks_blocked() await udp.aclose() async with await create_connected_unix_datagram_socket( peer_socket_path_or_str ) as udp: async with create_task_group() as tg: tg.start_soon(close_when_blocked) with pytest.raises(ClosedResourceError): await udp.receive() async def test_receive_after_close( self, peer_socket_path_or_str: Path | str, peer_sock: socket.socket ) -> None: udp = await create_connected_unix_datagram_socket(peer_socket_path_or_str) await udp.aclose() with pytest.raises(ClosedResourceError): await udp.receive() async def test_send_after_close( self, peer_socket_path_or_str: Path | str, peer_sock: socket.socket ) -> None: udp = await create_connected_unix_datagram_socket(peer_socket_path_or_str) await udp.aclose() with pytest.raises(ClosedResourceError): await udp.send(b"foo") async def test_from_socket(self, sock_or_fd_factory: SockFdFactoryProtocol) -> None: sock_or_fd = sock_or_fd_factory( socket.AF_UNIX, socket.SOCK_DGRAM, connected=True ) async with await ConnectedUNIXDatagramSocket.from_socket( sock_or_fd ) as udp_socket: assert isinstance(udp_socket, ConnectedUNIXDatagramSocket) assert udp_socket.extra(SocketAttribute.family) == socket.AF_UNIX async def test_from_socket_wrong_socket_type( self, socket_path: Path, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory( socket.AF_UNIX, socket.SOCK_STREAM, connected=True ) with pytest.raises( ValueError, match="socket type mismatch: expected SOCK_DGRAM, got SOCK_STREAM", ): await ConnectedUNIXDatagramSocket.from_socket(sock_or_fd) async def test_from_socket_not_connected( self, sock_or_fd_factory: SockFdFactoryProtocol ) -> None: sock_or_fd = sock_or_fd_factory(socket.AF_UNIX, socket.SOCK_DGRAM, bound=True) with pytest.raises(ValueError, match="the socket must be connected"): await ConnectedUNIXDatagramSocket.from_socket(sock_or_fd) async def test_tcp_connectable(mocker: MockerFixture) -> None: mock_connect_tcp = mocker.patch("anyio._core._sockets.connect_tcp") connectable = TCPConnectable("localhost", 1234) await connectable.connect() mock_connect_tcp.assert_called_once_with("localhost", 1234) async def test_unix_connectable(mocker: MockerFixture) -> None: mock_connect_tcp = mocker.patch("anyio._core._sockets.connect_unix") connectable = UNIXConnectable("/foo/bar") await connectable.connect() mock_connect_tcp.assert_called_once_with("/foo/bar") class TestAsConnectable: def test_pass_connectable(self) -> None: connectable = TCPConnectable("localhost", 1234) assert as_connectable(connectable) is connectable @pytest.mark.parametrize( "addr", [ pytest.param("localhost", id="string"), pytest.param(IPv4Address("127.0.0.1"), id="ipv4addr"), pytest.param(IPv6Address("::1"), id="ipv6addr"), ], ) def test_tcp(self, addr: str | IPv4Address | IPv6Address) -> None: connectable = as_connectable((addr, 1234)) assert isinstance(connectable, TCPConnectable) assert connectable.host == addr assert connectable.port == 1234 @pytest.mark.parametrize( "path", [ pytest.param("/foo/bar", id="str"), pytest.param(b"/foo/bar", id="bytes"), pytest.param(Path("/foo/bar"), id="strpath"), ], ) def test_unix(self, path: str | bytes | os.PathLike[str]) -> None: connectable = as_connectable(path) assert isinstance(connectable, UNIXConnectable) assert connectable.path == path def test_bad_type(self) -> None: with pytest.raises(TypeError, match="cannot convert 1234 to a connectable"): as_connectable(1234) # type: ignore[arg-type] def test_tls_true(self) -> None: connectable = as_connectable( "/foo/bar", tls=True, tls_hostname="example.com", tls_standard_compatible=False, ) assert isinstance(connectable, TLSConnectable) assert connectable.hostname == "example.com" assert not connectable.standard_compatible assert isinstance(connectable, TLSConnectable) def test_tls_explicit_context(self, client_context: SSLContext) -> None: connectable = as_connectable( "/foo/bar", tls=True, ssl_context=client_context, tls_hostname="example.com" ) assert isinstance(connectable, TLSConnectable) assert connectable.ssl_context is client_context assert connectable.hostname == "example.com" def test_tls_tcp_implicit_hostname(self, client_context: SSLContext) -> None: connectable = as_connectable(("localhost", 1234), tls=True) assert isinstance(connectable, TLSConnectable) assert connectable.hostname == "localhost" @pytest.mark.network async def test_getaddrinfo() -> None: # IDNA 2003 gets this wrong correct = await getaddrinfo("faß.de", 0) wrong = await getaddrinfo("fass.de", 0) assert correct != wrong @pytest.mark.parametrize("sock_type", [socket.SOCK_STREAM, socket.SOCK_STREAM]) async def test_getaddrinfo_ipv6addr( sock_type: Literal[socket.SocketKind.SOCK_STREAM], ) -> None: # IDNA trips up over raw IPv6 addresses proto = 0 if platform.system() == "Windows" else 6 assert await getaddrinfo("::1", 0, type=sock_type) == [ ( socket.AF_INET6, socket.SOCK_STREAM, proto, "", ("::1", 0), ) ] async def test_getaddrinfo_ipv6_disabled() -> None: gai_result = [(AddressFamily.AF_INET6, socket.SOCK_STREAM, 6, "", (1, b""))] with mock.patch.object(get_async_backend(), "getaddrinfo", return_value=gai_result): assert await getaddrinfo("::1", 0) == [] async def test_getnameinfo() -> None: expected_result = socket.getnameinfo(("127.0.0.1", 6666), 0) result = await getnameinfo(("127.0.0.1", 6666)) assert result == expected_result async def test_connect_tcp_getaddrinfo_context() -> None: """ See https://github.com/agronholm/anyio/issues/815 """ with pytest.raises(socket.gaierror) as exc_info: async with await connect_tcp("anyio.invalid", 6666): pass assert exc_info.value.__context__ is None @pytest.mark.parametrize("socket_type", ["socket", "fd"]) @pytest.mark.parametrize("event", ["readable", "writable"]) async def test_wait_socket(event: str, socket_type: str) -> None: wait = wait_readable if event == "readable" else wait_writable with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_sock: server_sock.bind(("127.0.0.1", 0)) port = server_sock.getsockname()[1] server_sock.listen() with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client_sock: client_sock.connect(("127.0.0.1", port)) client_sock.sendall(b"Hello, world") conn, addr = server_sock.accept() with conn: sock_or_fd: FileDescriptorLike = ( conn.fileno() if socket_type == "fd" else conn ) with fail_after(3): await wait(sock_or_fd) assert conn.recv(1024) == b"Hello, world" async def test_deprecated_wait_socket(anyio_backend_name: str) -> None: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: with pytest.warns( DeprecationWarning, match="This function is deprecated; use `wait_readable` instead", ): with move_on_after(0.1): await wait_socket_readable(sock) with pytest.warns( DeprecationWarning, match="This function is deprecated; use `wait_writable` instead", ): with move_on_after(0.1): await wait_socket_writable(sock) @pytest.mark.parametrize("socket_type", ["socket", "fd"]) async def test_interrupted_by_close(socket_type: str) -> None: a_sock, b = socket.socketpair() with a_sock, b: a_sock.setblocking(False) b.setblocking(False) a: FileDescriptorLike = a_sock.fileno() if socket_type == "fd" else a_sock async def reader() -> None: with pytest.raises(ClosedResourceError): await wait_readable(a) async def writer() -> None: with pytest.raises(ClosedResourceError): await wait_writable(a) fill_socket(a_sock) async with create_task_group() as tg: tg.start_soon(reader) tg.start_soon(writer) await wait_all_tasks_blocked() notify_closing(a_sock) a_sock.close()