diff --git a/README.md b/README.md index 34f8d84..b458766 100644 --- a/README.md +++ b/README.md @@ -213,11 +213,15 @@ and the campaign results: ## Honest limits Code that goes through the event-loop API is supported; code that -bypasses it is fenced: threads and executors, raw sockets, subprocesses, -signals, and TLS raise `SimulationFenceError` rather than silently -breaking determinism. Name resolution stays inside the simulation: -`getaddrinfo` resolves sim host names to stable synthetic addresses and -raises `socket.gaierror` for anything else — no real DNS, ever. +bypasses it is fenced: threads and executors, raw socket reads and +writes, subprocesses, signals, and loop-level TLS upgrades raise +`SimulationFenceError` rather than silently breaking determinism. +`sock_connect` on an `AF_INET` stream socket is the exception — it is +simulated, so a client that connects a socket and hands it to +`create_connection` runs, while the datagram and raw variants still fence. +Name resolution stays inside the simulation: `getaddrinfo` resolves sim +host names to stable synthetic addresses and raises `socket.gaierror` for +anything else — no real DNS, ever. Write-side flow control is not simulated. The full contract is in [docs/supported-api.md](https://github.com/dhruvl/simloop/blob/main/docs/supported-api.md). What that contract costs real libraries — what aiohttp, anyio, websockets and diff --git a/docs/compatibility.md b/docs/compatibility.md index e423514..0ec7b74 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -46,11 +46,12 @@ carries a date instead. | anyio | 4.14.2 | works: task group, memory object stream (one, two, three), anyio.sleep and move_on_after; virtual clock reached 1.75s | Asyncio backend only; nothing here touches a socket. | | redis (RESP wire protocol) | n/a | works: PING, SET and GET round trips over one connection: ['+PONG', '+OK', '0'] | Hand-rolled RESP over sim streams; no client library, no real server. | | websockets | 17.0.1 | works: handshake, one echoed frame ('HELLO') and close over ws:// | asyncio server and client on two sim hosts, ws:// only. | -| aiohttp (client) | 3.14.3 | fenced: simloop does not simulate 'sock_connect'; see docs/supported-api.md for the supported asyncio subset | ClientSession GET at a sim host answered by a raw stream server. | -| httpx | 0.28.1 | fails: AttributeError: 'NoneType' object has no attribute 'getpeername' | AsyncClient GET at a sim host answered by a raw stream server. | +| aiohttp (client) | 3.14.3 | works: ClientSession GET returned 'hello from the simulation' | ClientSession GET at a sim host answered by a raw stream server. | +| httpx | 0.28.1 | works: AsyncClient GET returned 'hello from the simulation' | AsyncClient GET at a sim host answered by a raw stream server. | -Rows are grouped: the libraries expected to run on the loop first, then the -client stacks expected to leave it. +Rows are grouped: the libraries that need nothing but the loop and its +streams first, then the client stacks that expect a socket object +underneath them. ## Reading the rows @@ -78,37 +79,67 @@ clients are built on — a length-prefixed request/response protocol on one long-lived connection — so the probe speaks RESP by hand against a small server on a second sim host, and the row claims no more than that. -**aiohttp's client** leaves the simulation at its first connection attempt. -The fence, verbatim: +**aiohttp's client** issues its GET and reads the body back. Its connector +resolves the name through `loop.getaddrinfo`, then hands the addresses to +`aiohappyeyeballs`, which creates a real `AF_INET` stream socket, calls +`loop.sock_connect` on it, and passes that socket to +`loop.create_connection(sock=...)`. The simulation answers the sequence +without letting the socket reach a network: `sock_connect` resolves the +target against the host table and records it, moving no packet and no +clock, and the `create_connection` call closes the real descriptor and +opens a simulated connection to the recorded address, paying the same +single round trip a direct `create_connection` would. The connector's +`setsockopt(TCP_NODELAY)` lands on the stand-in object +`get_extra_info("socket")` returns, which accepts option calls and does +nothing with them. + +**httpx** completes the same request through httpcore and anyio. anyio +resolves the name (the simulated resolver accepts the ASCII-encoded form +anyio sends) and connects through `loop.create_connection`, with no socket +of its own. The step that used to end this row comes next: httpcore asks +the new stream who it is connected to, and anyio answers by reading the +socket object out of `transport.get_extra_info("socket")` +(`anyio/abc/_sockets.py`, `extra_attributes`) and calling `getpeername()` +on it. A transport with no operating-system socket now answers with a +stand-in that reports the peer's synthetic address and port, so the +introspection succeeds and the response body comes back. + +Both client probes make one request against a responder that sends +`Connection: close`, so neither row says anything about connection reuse. +The piece a pool depends on is the descriptor `fileno()` returns: httpcore +polls it to decide whether a pooled connection has died, and the +simulation backs it with a parked descriptor the transport owns, which +stays unreadable while the peer is alive and becomes readable once the +peer's EOF arrives; a reset or a teardown closes it and `fileno()` returns +`-1`, which the same poll reads as dead just as well. That contract is +pinned by the test suite, not by these rows. + +Both client rows are `http://` only, and the two stacks stop differently +on `https://`. aiohttp asks for TLS through `create_connection(ssl=...)`, +which fences: ``` -simloop does not simulate 'sock_connect'; see docs/supported-api.md for the supported asyncio subset +simloop does not simulate 'create_connection(ssl=...)'; see docs/supported-api.md for the supported asyncio subset ``` -Its connector resolves the name through `loop.getaddrinfo` (which the -simulation answers), then hands the addresses to `aiohappyeyeballs`, which -opens a real socket and calls `loop.sock_connect` on it. Raw sockets are -fenced, so the simulation stops there rather than letting a real connection -out. - -**httpx** never reaches a fence, and gets further than its verdict looks. -It goes through httpcore and anyio; anyio resolves the name (the simulated -resolver accepts the ASCII-encoded form anyio sends), connects through -`loop.create_connection`, and the connection *succeeds*. The failure is -introspection after the fact: httpcore asks the new stream for its local -and remote addresses, anyio answers by reading the raw socket object out of -`transport.get_extra_info("socket")` (`anyio/abc/_sockets.py`, -`extra_attributes`), and the simulation's honest answer for a transport -with no operating-system socket is `None`. Whether a request would complete -past that line is untested here. +httpx reaches no fence. httpcore wraps the byte stream with anyio's +`TLSStream`, which drives an `ssl` memory BIO inside the process and +sends the handshake as ordinary bytes over the simulated connection, so +`loop.start_tls` is never called and nothing stops the attempt. Where it +ends is up to whatever is listening: aimed at the plaintext responder +these probes use, the handshake goes unanswered and the request dies of +httpx's own `ConnectTimeout`. That timeout is a one-off measurement +rather than a row — no probe on this page requests `https://`. ## Not tested - **asyncpg**: reaching its first fence needs a live PostgreSQL server to connect to, which no probe can provide; it is untested rather than fenced-or-not. -- **TLS anywhere**: `start_tls` is fenced, so `https://` and `wss://` are out - of scope for every probe on this page. +- **TLS anywhere**: no probe on this page requests `https://` or `wss://`. + simloop fences `start_tls` and `create_connection(ssl=...)`, but a stack + that runs its handshake in memory reaches neither — it reaches a simulated + network with nothing on it that speaks TLS unless the test puts it there. - Anything that reaches outside the loop by design — threads, executors, subprocesses, signals, real DNS. Those are fences, listed in [docs/supported-api.md](supported-api.md), not compatibility questions. diff --git a/docs/supported-api.md b/docs/supported-api.md index f5c8360..5583a04 100644 --- a/docs/supported-api.md +++ b/docs/supported-api.md @@ -44,6 +44,8 @@ a host belong to an implicit `driver` host. | `loop.getaddrinfo` | Resolves against the host table, never DNS: a registered host name, its synthetic address, or a loopback-shaped name (`None`, `""`, `localhost`, `127.0.0.1`, `0.0.0.0`) meaning the calling task's own host. Returns stdlib-shaped rows — `(AF_INET, SOCK_STREAM, IPPROTO_TCP, "", (address, port))` and the `SOCK_DGRAM` / `IPPROTO_UDP` row — filtered by `family`, `type` and `proto`. Ports are numeric (`int`, a digit string, or `None` for 0); resolver `flags` have nothing to vary | | `loop.getnameinfo` | Reverse lookup: a synthetic address maps back to its host name, and a host name (what `get_extra_info("peername")` reports) maps to itself. `NI_NUMERICHOST` returns the address instead; services are always numeric | | `loop.net.address` / `hostname` | The mapping itself: every registered host owns one synthetic IPv4 address from `10.7.0.0/16` — `10.7.0.1`, `10.7.0.2`, ... handed out in registration order, starting with the implicit `driver` host | +| `loop.sock_connect` + `create_connection(sock=...)` | The two-call connect sequence aiohttp's connector performs. On an `AF_INET` stream socket, `sock_connect` places the target in the host table and records it against the socket — no packet moves and no virtual time passes, and an address that is not a `(host, port)` pair, or a target the host table cannot place, raises `OSError` there. `create_connection(sock=...)` then claims that recorded address: it closes the real descriptor (the loop takes ownership, as the stdlib does) and opens a simulated connection, paying the same one round trip a direct connect pays, so a closed port raises `ConnectionRefusedError` from this call rather than the first. Passing a socket that no `sock_connect` on this loop parked raises `OSError`; passing `host`/`port` alongside `sock` raises `ValueError`. Binding a source address first — `TCPConnector(local_addr=...)` — is not supported: the connector binds the real socket before the simulation is consulted, and a synthetic address belongs to no real interface, so the bind fails outside the loop | +| `transport.get_extra_info("socket")` (streams) | A stand-in object, not a network socket: `family` / `type` / `proto` report `AF_INET` / `SOCK_STREAM` / `IPPROTO_TCP`, `getsockname()` and `getpeername()` return `(synthetic address, port)` tuples for the two ends, and `setsockopt`, `shutdown` and `close` are accepted and do nothing. `fileno()` returns a parked descriptor the transport owns, created on first call, that polls unreadable while the peer is alive and readable once the peer's EOF arrives — which is how a pool that checks readability sees a dead connection. A reset, a local close or the end of the run closes the descriptor instead, and `fileno()` reports `-1` from then on; a readability poll reads that as dead too. The number itself is assigned by the operating system, so it is not reproducible across runs and is outside the determinism guarantee — nothing in a trace depends on it. No bytes ever cross it: `recv`, `send` and anything else not listed above raise `AttributeError` rather than pretend. Datagram transports still report `None` | Names and their synthetic addresses are interchangeable wherever an endpoint is accepted, so a client can resolve a name and connect to what it got back. @@ -62,5 +64,17 @@ reliable by construction; and addressing is IPv4-only and entirely synthetic Anything that reaches outside the simulation raises `SimulationFenceError`: executors and threads (`run_in_executor`, `call_soon_threadsafe`), signal -handlers, subprocesses, raw sockets (`sock_*`), file-descriptor callbacks -(`add_reader` / `add_writer`), TLS upgrades, `sendfile`, and pipes. +handlers, subprocesses, file-descriptor callbacks (`add_reader` / +`add_writer`), loop-level TLS upgrades (`start_tls`, +`create_connection(ssl=...)`), `sendfile`, and pipes. TLS a library +performs in memory reaches no loop API and so reaches no fence; what that +means in practice is in [docs/compatibility.md](compatibility.md). + +The socket calls are fenced with one exception. `sock_connect` on an +`AF_INET` stream socket is simulated — it is how client stacks reach the +network, and the table above says what it does. Every other socket kind +fences there — datagram, raw and IPv6 alike — and the rest of the family +stays fenced outright: `sock_recv`, `sock_recv_into`, `sock_sendall`, +`sock_sendto`, `sock_recvfrom`, `sock_recvfrom_into`, `sock_accept` and +`sock_sendfile`. Client stacks do not need them: the socket they connect +is upgraded into a transport instead of being read and written directly. diff --git a/src/simloop/_loop.py b/src/simloop/_loop.py index 8b20d35..f2ba3e2 100644 --- a/src/simloop/_loop.py +++ b/src/simloop/_loop.py @@ -6,10 +6,11 @@ import gc import heapq import random +import socket import sys from array import array from asyncio import events -from collections.abc import Callable, Iterable +from collections.abc import Callable, Iterable, Sequence from contextvars import Context from typing import TYPE_CHECKING, Any, NoReturn, TypeVarTuple, Unpack @@ -117,6 +118,10 @@ def __init__(self, seed: int = 0) -> None: self._unhandled: list[BaseException] = [] self._exception_handler: _ExceptionHandler | None = None self._task_factory: _TaskFactory | None = None + # sock_connect(sock, addr) -> create_connection(sock=sock) is how + # aiohttp reaches the network; the address is only visible in the + # first call, so it is parked here until the upgrade claims it. + self._sock_targets: dict[Any, tuple[Any, int]] = {} self._net = SimNetwork(self) @classmethod @@ -328,6 +333,15 @@ def is_closed(self) -> bool: def close(self) -> None: if self._running: raise RuntimeError("cannot close a running event loop") + self._sock_targets.clear() + # Connections still open at the end of a run — what a pooling client + # leaves behind — hold their liveness descriptors inside the socket + # object's reference cycle with the transport, so the descriptors + # would only come back when the cycle collector ran. Nothing can poll + # them once the loop is closed, so release them here instead. + for transport in self._net._streams.values(): + if transport._extra_socket is not None: + transport._extra_socket._dispose() self._closed = True def _check_closed(self) -> None: @@ -386,7 +400,24 @@ async def create_connection( port: Any = None, **kwargs: Any, ) -> Any: + sock = kwargs.pop("sock", None) _reject_kwargs("create_connection", kwargs) + if sock is not None: + # The stdlib treats a passed-in socket as already connected and + # takes ownership of it. Here "connected" means sock_connect + # parked a target for it; the real descriptor is closed at once + # because the simulation only needed the address it carried. + if host is not None or port is not None: + raise ValueError( + "host/port and sock can not be specified at the same time" + ) + target = self._sock_targets.pop(sock, None) + if target is None: + raise OSError( + "the given socket was not connected via sock_connect on this loop" + ) + sock.close() + host, port = target return await self._net._open_connection(protocol_factory, host, port) async def create_server( @@ -585,8 +616,39 @@ def sock_recv_into(self, *args: Any, **kwargs: Any) -> Any: def sock_sendall(self, *args: Any, **kwargs: Any) -> Any: _fence("sock_sendall") - def sock_connect(self, *args: Any, **kwargs: Any) -> Any: - _fence("sock_connect") + async def sock_connect(self, sock: Any, address: Any) -> None: + # aiohttp's connector creates a real TCP socket and connects it here + # before handing it to create_connection(sock=...). The simulation + # accepts exactly that shape; every other socket kind still fences. + if ( + getattr(sock, "family", None) != socket.AF_INET + or getattr(sock, "type", None) != socket.SOCK_STREAM + ): + raise SimulationFenceError( + "simloop does not simulate 'sock_connect' for anything but " + "AF_INET stream sockets; see docs/supported-api.md for the " + "supported asyncio subset" + ) + if ( + isinstance(address, (str, bytes, bytearray)) + or not isinstance(address, Sequence) + or len(address) != 2 + or not isinstance(address[0], (str, bytes)) + or isinstance(address[1], bool) + or not isinstance(address[1], int) + ): + # Caught here rather than deeper: a bare host string would be read + # character by character, and a string port would surface much + # later as a confusing error about something else entirely. + raise OSError( + f"sock_connect needs an AF_INET (host, port) address, got {address!r}" + ) + host, port = address[0], address[1] + self._net._resolve(host) # unknown targets fail here, loudly + # No packet moves and no time passes yet: the connection handshake + # (and its one-RTT cost) happens when create_connection claims the + # socket, keeping the total cost identical to a direct connect. + self._sock_targets[sock] = (host, port) def sock_accept(self, *args: Any, **kwargs: Any) -> Any: _fence("sock_accept") diff --git a/src/simloop/_transports.py b/src/simloop/_transports.py index 0845114..8d13bef 100644 --- a/src/simloop/_transports.py +++ b/src/simloop/_transports.py @@ -9,6 +9,7 @@ from __future__ import annotations import asyncio +import socket from typing import TYPE_CHECKING, Any if TYPE_CHECKING: @@ -89,6 +90,79 @@ def get_protocol(self) -> Any: return self._protocol +class _SimSocket: + """Stand-in for the OS socket a simulated stream connection does not have. + + Client stacks introspect the transport's socket: anyio eagerly calls + getpeername()/.family on every typed-attribute lookup, aiohttp sets + TCP_NODELAY through it, and httpcore polls fileno() to decide whether a + pooled connection died. Addresses are reported as synthetic-IP tuples so + callers that feed them to ipaddress.ip_address() get a plausible + AF_INET sockaddr; option and teardown calls are accepted and do nothing. + Anything not listed here raises AttributeError, keeping the simulation + loud about surface it does not fake. + """ + + def __init__(self, transport: _SimStreamTransport) -> None: + self._transport = transport + self.family = socket.AF_INET + self.type = socket.SOCK_STREAM + self.proto = socket.IPPROTO_TCP + self._park: tuple[socket.socket, socket.socket] | None = None + self._disposed = False + + def _address(self, endpoint: _Addr) -> tuple[str, int]: + name, port = endpoint + return (self._transport._net.address(name), port) + + def getsockname(self) -> tuple[str, int]: + return self._address(self._transport._local) + + def getpeername(self) -> tuple[str, int]: + return self._address(self._transport._remote) + + def setsockopt(self, *args: Any) -> None: + return None + + def shutdown(self, how: int) -> None: + return None + + def close(self) -> None: + return None + + def fileno(self) -> int: + # httpcore's is_socket_readable() polls this fd to decide whether a + # pooled connection died. A parked socketpair end reads not-readable + # until its peer end is closed, so the poll answers exactly one + # question: has the peer's EOF arrived? A reset or a teardown closes + # both ends instead and reports -1, which the same poll reads as dead + # just as a closed real socket would. Lazy: connections nobody + # introspects cost no descriptors, which matters at campaign scale. + if self._disposed: + return -1 + if self._park is None: + self._park = socket.socketpair() + # The peer's FIN/RST can land before anyone asks for the socket, + # so a freshly parked pair inherits whatever the transport + # already saw rather than claiming a dead connection is live. + if self._transport._peer_closed: + self._park[0].close() + return self._park[1].fileno() + + def _peer_gone(self) -> None: + # The peer's EOF arrived: closing the held end makes the exposed end + # poll readable, exactly when a real kernel would report it. + if self._park is not None: + self._park[0].close() + + def _dispose(self) -> None: + self._disposed = True + if self._park is not None: + self._park[0].close() + self._park[1].close() + self._park = None + + class _SimStreamTransport(asyncio.Transport): """One end of a reliable, ordered byte-stream connection. @@ -116,6 +190,8 @@ def __init__( self._backlog: list[bytes] = [] self._eof_pending = False self._limits = (16 * 1024, 64 * 1024) # (low, high): recorded, inert + self._extra_socket: _SimSocket | None = None + self._peer_closed = False # the peer's FIN or RST has arrived def _begin(self, protocol: Any) -> None: self._protocol = protocol @@ -179,6 +255,8 @@ def _finish(self, exc: Exception | None) -> None: self._closed = True self._closing = True self._net._drop_stream(self._conn, self._local[0]) + if self._extra_socket is not None: + self._extra_socket._dispose() protocol, self._protocol = self._protocol, None if protocol is not None: protocol.connection_lost(exc) @@ -196,6 +274,9 @@ def _data_arrived(self, data: bytes) -> None: self._protocol.data_received(data) def _eof_arrived(self) -> None: + self._peer_closed = True + if self._extra_socket is not None: + self._extra_socket._peer_gone() if self._closed: return if self._read_paused: @@ -206,6 +287,11 @@ def _eof_arrived(self) -> None: self.close() def _reset_arrived(self) -> None: + # No _peer_gone() here: a reset tears the connection down on the spot, + # and _finish closes both parked ends. The descriptor goes to -1 + # rather than turning readable, which a liveness poll reads the same + # way — as a connection to discard. + self._peer_closed = True if self._closed: return self._finish(ConnectionResetError("Connection reset by peer")) @@ -244,6 +330,15 @@ def get_extra_info(self, name: str, default: Any = None) -> Any: return self._local if name == "peername": return self._remote + if name == "socket": + if self._extra_socket is None: + self._extra_socket = _SimSocket(self) + if self._closed: + # Asked for after teardown: born closed, like the socket a + # finished connection leaves behind. Nothing here may hand + # out a live descriptor for a connection that is gone. + self._extra_socket._dispose() + return self._extra_socket return default def set_protocol(self, protocol: Any) -> None: diff --git a/tests/test_client_sockets.py b/tests/test_client_sockets.py new file mode 100644 index 0000000..af5965c --- /dev/null +++ b/tests/test_client_sockets.py @@ -0,0 +1,563 @@ +"""The fake socket stream transports expose for client-stack introspection. + +anyio reads get_extra_info("socket") and eagerly calls getpeername()/.family +on it (anyio/abc/_sockets.py); httpcore polls fileno() to decide whether a +pooled connection died. These tests pin the contract both stacks rely on. +""" + +from __future__ import annotations + +import asyncio +import gc +import ipaddress +import os +import select +import socket +from collections.abc import Callable, Coroutine +from typing import Any + +import pytest + +from simloop import SimLoop, SimulationFenceError + + +def _network(seed: int = 0) -> SimLoop: + loop = SimLoop(seed=seed) + loop.net.host("server") + loop.net.host("client") + return loop + + +async def _hold_open(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + await reader.read() # stay open until the client closes + writer.close() + + +def _connected_pair(loop: SimLoop, handler: Any = _hold_open) -> tuple[Any, Any, Any]: + """One established client connection. + + Returns (client_writer, accepted_writer, server): the two ends of the same + connection plus the listening server. + """ + + accepted: list[Any] = [] + + async def accept( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + accepted.append(writer) + await handler(reader, writer) + + async def main() -> tuple[Any, Any, Any]: + server = await loop.net.host("server").create_task( + asyncio.start_server(accept, "0.0.0.0", 9000) + ) + writer_box: list[Any] = [] + + async def connect() -> None: + _, writer = await asyncio.open_connection("server", 9000) + writer_box.append(writer) + + await loop.net.host("client").create_task(connect()) + return writer_box[0], accepted[0], server + + trio: tuple[Any, Any, Any] = loop.run_until_complete(main()) + return trio + + +def _settle(loop: SimLoop) -> None: + """Let queued closes cross the simulated network before teardown.""" + loop.run_until_complete(asyncio.sleep(1.0)) + + +def test_stream_transport_exposes_a_socket_object() -> None: + loop = _network() + try: + writer, _, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + assert sock is not None + assert sock is writer.transport.get_extra_info("socket") # stable singleton + assert sock.family == socket.AF_INET + assert sock.type == socket.SOCK_STREAM + assert sock.proto == socket.IPPROTO_TCP + server.close() + writer.close() + _settle(loop) + finally: + loop.close() + + +def test_socket_addresses_are_synthetic_ip_tuples() -> None: + loop = _network() + try: + writer, accepted, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + local_ip, local_port = sock.getsockname() + peer_ip, peer_port = sock.getpeername() + # Plausible AF_INET sockaddrs: ip_address() must accept them. + ipaddress.ip_address(local_ip) + ipaddress.ip_address(peer_ip) + assert local_ip == loop.net.address("client") + assert peer_ip == loop.net.address("server") + assert peer_port == 9000 + # Same endpoints the transport itself reports, by name. + assert writer.transport.get_extra_info("peername") == ("server", 9000) + assert writer.transport.get_extra_info("sockname")[1] == local_port + # The accepting side sees the client from the other direction: the two + # fake sockets describe one connection, not two unrelated ones. + accepted_sock = accepted.transport.get_extra_info("socket") + assert accepted_sock.getpeername() == (local_ip, local_port) + assert accepted_sock.getsockname() == (peer_ip, 9000) + server.close() + writer.close() + _settle(loop) + finally: + loop.close() + + +def test_socket_options_and_teardown_calls_are_inert() -> None: + loop = _network() + try: + writer, _, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + assert sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True) is None + assert sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) is None + assert sock.shutdown(socket.SHUT_RDWR) is None + assert sock.close() is None + with pytest.raises(AttributeError): + sock.recv(1) # no pretending to be readable + server.close() + writer.close() + _settle(loop) + finally: + loop.close() + + +def test_datagram_transports_still_answer_none() -> None: + loop = _network() + + async def main() -> Any: + transport, _ = await loop.create_datagram_endpoint( + asyncio.DatagramProtocol, local_addr=("0.0.0.0", 5000) + ) + try: + return transport.get_extra_info("socket") + finally: + transport.close() + + try: + assert loop.run_until_complete(main()) is None + finally: + loop.close() + + +def _readable(fd: int) -> bool: + """The exact check httpcore's is_socket_readable performs.""" + if fd < 0: + return True + rready, _, _ = select.select([fd], [], [], 0) + return bool(rready) + + +def test_fileno_is_lazy_and_not_readable_while_live() -> None: + loop = _network() + try: + writer, _, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + assert sock._park is None # no kernel object until someone asks + fd = sock.fileno() + assert fd >= 0 + assert fd == sock.fileno() # stable + assert not _readable(fd) # live connection: httpcore keeps pooling it + server.close() + writer.close() + _settle(loop) + finally: + loop.close() + + +def test_fd_turns_readable_when_the_peer_closes() -> None: + loop = _network() + + async def close_after_a_beat( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + await asyncio.sleep(0.5) # after the client has taken its fd + writer.close() + + try: + writer, _, server = _connected_pair(loop, close_after_a_beat) + sock = writer.transport.get_extra_info("socket") + fd = sock.fileno() + assert not _readable(fd) + # Let the FIN cross the simulated network. + _settle(loop) + assert _readable(fd) # httpcore now sees the connection as expired + server.close() + writer.close() + _settle(loop) + finally: + loop.close() + + +def test_fd_born_readable_when_the_peer_left_first() -> None: + """A FIN can land before anyone asks for the socket.""" + loop = _network() + + async def close_immediately( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + writer.close() + + try: + writer, _, server = _connected_pair(loop, close_immediately) + _settle(loop) # FIN already delivered + sock = writer.transport.get_extra_info("socket") + assert sock.fileno() >= 0 # the local end is still open: a real fd + assert _readable(sock.fileno()) # never claim a dead connection is live + server.close() + writer.close() + _settle(loop) + finally: + loop.close() + + +def test_fd_lifecycle_ends_with_the_transport() -> None: + loop = _network() + try: + writer, _, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + fd = sock.fileno() + assert not _readable(fd) + writer.close() + _settle(loop) + assert sock._park is None # both ends closed, nothing leaked + assert sock.fileno() == -1 # closed-socket semantics + server.close() + finally: + loop.close() + + +def test_reset_closes_the_descriptor_instead_of_arming_it() -> None: + """An aborting peer tears the connection down; the fd goes, not readable.""" + loop = _network() + + async def abort_after_a_beat( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + await asyncio.sleep(0.5) # after the client has taken its fd + writer.transport.abort() + + try: + writer, _, server = _connected_pair(loop, abort_after_a_beat) + sock = writer.transport.get_extra_info("socket") + assert sock.fileno() >= 0 + _settle(loop) # let the RST cross the network + assert sock.fileno() == -1 # the reset finished the connection outright + assert sock._park is None + assert _readable(sock.fileno()) # a poll still reads it as dead + server.close() + writer.close() + _settle(loop) + finally: + loop.close() + + +def test_fileno_after_teardown_never_creates_a_descriptor() -> None: + loop = _network() + try: + writer, _, server = _connected_pair(loop) + sock = writer.transport.get_extra_info("socket") + writer.close() + _settle(loop) + assert sock.fileno() == -1 + assert sock._park is None + server.close() + finally: + loop.close() + + +def test_sock_connect_records_a_stream_socket_target() -> None: + loop = _network() + + async def main() -> None: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.setblocking(False) # what aiohappyeyeballs does + try: + await loop.sock_connect(sock, (loop.net.address("server"), 9000)) + assert loop._sock_targets[sock] == (loop.net.address("server"), 9000) + finally: + sock.close() + + loop.run_until_complete(main()) + loop.close() + + +def test_sock_connect_rejects_unknown_addresses() -> None: + loop = _network() + + async def main() -> None: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + with pytest.raises(OSError, match="10.9.9.9"): + await loop.sock_connect(sock, ("10.9.9.9", 9000)) + finally: + sock.close() + + loop.run_until_complete(main()) + loop.close() + + +@pytest.mark.parametrize( + "address", + [ + "server", # a bare string is a sequence of characters, not an endpoint + ("server",), # one item short + ("server", "9000"), # a port that only looks like a number + ], +) +def test_sock_connect_rejects_malformed_addresses(address: Any) -> None: + loop = _network() + + async def main() -> None: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + with pytest.raises(OSError, match="host, port"): + await loop.sock_connect(sock, address) + finally: + sock.close() + + loop.run_until_complete(main()) + loop.close() + + +def test_sock_connect_still_fences_datagram_sockets() -> None: + loop = _network() + + async def main() -> None: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + with pytest.raises(SimulationFenceError, match="sock_connect"): + await loop.sock_connect(sock, (loop.net.address("server"), 9000)) + finally: + sock.close() + + loop.run_until_complete(main()) + loop.close() + + +def test_sock_connect_consumes_no_virtual_time() -> None: + loop = _network() + + async def main() -> float: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + before = loop.time() + await loop.sock_connect(sock, ("server", 9000)) + return loop.time() - before + finally: + sock.close() + + assert loop.run_until_complete(main()) == 0.0 + loop.close() + + +def _aiohttp_style_connect( + loop: SimLoop, port: int = 9000, sockets: list[Any] | None = None +) -> Callable[[], Coroutine[Any, Any, tuple[Any, Any, Any]]]: + """The exact two-call sequence aiohttp + aiohappyeyeballs performs.""" + + async def connect() -> tuple[Any, Any, Any]: + infos = await loop.getaddrinfo("server", port, type=socket.SOCK_STREAM) + family, kind, proto, _, address = infos[0] + sock = socket.socket(family=family, type=kind, proto=proto) + sock.setblocking(False) + if sockets is not None: + sockets.append(sock) # so a refused connect can be inspected too + await loop.sock_connect(sock, address) + transport, protocol = await loop.create_connection( + asyncio.Protocol, ssl=None, server_hostname=None, sock=sock + ) + return sock, transport, protocol + + return connect + + +def test_parked_socket_upgrades_to_a_sim_connection() -> None: + loop = _network() + loop.net.set_defaults(latency=(0.001, 0.001)) # so the round trip is visible + + async def main() -> tuple[Any, float]: + server = await loop.net.host("server").create_task( + asyncio.start_server(_hold_open, "0.0.0.0", 9000) + ) + started = loop.time() + sock, transport, _ = await loop.net.host("client").create_task( + _aiohttp_style_connect(loop)() + ) + elapsed = loop.time() - started + transport.close() + server.close() + return sock, elapsed + + sock, elapsed = loop.run_until_complete(main()) + _settle(loop) # let both ends finish closing + assert sock.fileno() == -1 # the loop took ownership and closed it + assert elapsed == pytest.approx(0.002) # SYN out, accept back: one round trip + assert not loop._sock_targets # the parked entry was claimed + loop.close() + + +def test_refused_port_raises_from_the_upgrade() -> None: + loop = _network() + loop.net.set_defaults(latency=(0.001, 0.001)) + sockets: list[Any] = [] + + async def main() -> float: + started = loop.time() + with pytest.raises(ConnectionRefusedError): + await loop.net.host("client").create_task( + _aiohttp_style_connect(loop, port=9999, sockets=sockets)() + ) + return loop.time() - started + + elapsed = loop.run_until_complete(main()) + # Refusal costs what a real one does: SYN out, refusal back. + assert elapsed == pytest.approx(0.002) + assert sockets[0].fileno() == -1 # the loop owned it and closed it anyway + loop.close() + + +def test_unparked_socket_is_rejected() -> None: + loop = _network() + + async def main() -> None: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + with pytest.raises(OSError, match="sock_connect"): + await loop.create_connection(asyncio.Protocol, sock=sock) + finally: + sock.close() + + loop.run_until_complete(main()) + loop.close() + + +def test_sock_and_host_together_follow_stdlib_rules() -> None: + loop = _network() + + async def main() -> None: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + await loop.sock_connect(sock, ("server", 9000)) + with pytest.raises(ValueError): + await loop.create_connection( + asyncio.Protocol, "server", 9000, sock=sock + ) + finally: + sock.close() + + loop.run_until_complete(main()) + loop.close() + + +def test_closing_the_loop_releases_parked_descriptors() -> None: + """A pooling client leaves its connections open; the run still ends clean.""" + loop = _network() + + async def main() -> tuple[Any, Any, Any]: + # A bare protocol server, not start_server: the connection has to + # survive to the end of the run, and a stream handler parked on the + # other side of it would be a leak of its own. + server = await loop.net.host("server").create_task( + loop.create_server(asyncio.Protocol, "0.0.0.0", 9000) + ) + _, transport, _ = await loop.net.host("client").create_task( + _aiohttp_style_connect(loop)() + ) + return server, transport, transport.get_extra_info("socket") + + server, transport, sock = loop.run_until_complete(main()) + fd = sock.fileno() + assert fd >= 0 + server.close() + loop.close() # the connection is still open, exactly as a pool leaves it + assert sock._park is None + with pytest.raises(OSError): + os.fstat(fd) # the descriptor went back to the OS, not just its owner + gc.collect() # under -W error::ResourceWarning: nothing left to complain + assert transport.get_extra_info("socket") is sock + + +class _Collect(asyncio.Protocol): + """A protocol that keeps what it was sent, so replies are observable.""" + + def __init__(self) -> None: + self.data = bytearray() + + def data_received(self, data: bytes) -> None: + self.data += data + + +async def _echo_once( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter +) -> None: + writer.write(await reader.read(64)) + await writer.drain() + writer.close() + + +async def _client_stack_probe(loop: SimLoop, name: str) -> None: + """One aiohttp-shaped request: resolve, connect, introspect, write, close.""" + infos = await loop.getaddrinfo("server", 9000, type=socket.SOCK_STREAM) + family, kind, proto, _, address = infos[0] + sock = socket.socket(family=family, type=kind, proto=proto) + sock.setblocking(False) + await loop.sock_connect(sock, address) + transport, _ = await loop.create_connection(_Collect, sock=sock) + fake = transport.get_extra_info("socket") + fake.getpeername() + fake.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True) + fake.fileno() # the descriptor a pool would poll + transport.write(name.encode()) + await asyncio.sleep(0.05) + transport.close() + + +def _run_client_stack(seed: int) -> str: + """Three concurrent clients driving the whole new surface; trace hash out.""" + loop = SimLoop(seed=seed) + loop.net.host("server") + loop.net.set_defaults(latency=(0.001, 0.005)) + + async def main() -> None: + server = await loop.net.host("server").create_task( + asyncio.start_server(_echo_once, "0.0.0.0", 9000) + ) + clients = [ + loop.net.host(name).create_task(_client_stack_probe(loop, name)) + for name in ("alice", "bob", "carol") + ] + for client in clients: + await client + server.close() + + try: + loop.run_until_complete(main()) + loop.run_until_complete(asyncio.sleep(1.0)) + return loop.trace_hash() + finally: + loop.close() + + +def test_client_stack_path_is_deterministic() -> None: + for seed in range(3): + hashes = {_run_client_stack(seed) for _ in range(2)} + assert len(hashes) == 1, f"seed {seed} produced diverging traces" + + +def test_client_stack_path_still_varies_with_the_seed() -> None: + # The fake socket's descriptors are OS-assigned and deliberately absent + # from the trace, so only the schedule and the latency draws move. + hashes = {_run_client_stack(seed) for seed in range(10)} + assert len(hashes) >= 8 diff --git a/tests/test_resolution.py b/tests/test_resolution.py index 0c97c89..603e401 100644 --- a/tests/test_resolution.py +++ b/tests/test_resolution.py @@ -164,6 +164,17 @@ def test_getaddrinfo_honors_type_family_and_proto_filters() -> None: assert len(_resolve("server", 80, family=socket.AF_INET)) == 2 +def test_getaddrinfo_returns_exactly_one_row_per_socket_kind() -> None: + # aiohappyeyeballs (aiohttp) and anyio (httpx) both stagger connection + # attempts across candidate addresses when more than one comes back, + # arming extra timers and racing tasks. Both stay on their simple path + # because this resolver returns exactly one row per socket kind. Anyone + # making this return more rows must first decide what happy-eyeballs + # should mean under simulation. + assert len(_resolve("server", 80, type=socket.SOCK_STREAM)) == 1 + assert len(_resolve("server", 80, type=socket.SOCK_DGRAM)) == 1 + + def test_getaddrinfo_accepts_a_bytes_host_name() -> None: # Resolver stacks encode names before resolving them — anyio hands the # stdlib resolver ASCII bytes — so bytes must resolve like their text.