Tests for network backends.

[tomchristie]

Although we have test coverage for our network backends, we're not currently testing against them directly.
Since #699 added the network backends to the public API we should also implement some tests against them.

There's three classes of test case that we'd want covered here...

• Tests for the backends, without TLS.
• Tests for the backends, with TLS.
• Tests for the backends, with TLS-in-TLS.

I'd suggest that we start just by implementing some test cases for the first of those, with tests for each of...

• httpcore.SyncBackend
• httpcore.AnyIOBackend
• httpcore.TrioBackend

So, just testing connecting, reading, writing.

Punching this one directly into an issue, since I think it's clear enough to not require a discussion phase.

[karpetrosyan]

I'd suggest that we start just by implementing some test cases for the first of those, with tests for each of...

Should we use real connections or mock the underlying socket?

[tomchristie]

No sure. I'd assumed that we'd need to work with real sockets for these tests, I don't think we can mock it out without monkey-patching right?

[karpetrosyan]

That's a little complicated, but it solves a lot of problems.
This is very similar to the "network_backend" argument.

_Address: typing.TypeAlias = typing.Tuple[typing.Optional[str], int]

class SocketInterface(Protocol):

    def send(self, data: bytes) -> int: ...

    def sendall(self, data: bytes) -> None: ...

    def recv(self, bufsize: int) -> bytes: ...

    @typing.overload
    def setsockopt(self, level: int, optname: int, value: typing.Union[int, bytes]) -> None: ...
    @typing.overload
    def setsockopt(self, level: int, optname: int, value: None, optlen: int) -> None: ...

    def settimeout(self, value: typing.Union[float, None]) -> None: ...

    def close(self) -> None: ...

class SocketFactory:
    def create_connection(
            self,
            addr: _Address,
            timeout: typing.Optional[float] = None,
            source_address: typing.Optional[_Address] = None
    ) -> SocketInterface:
        return socket.create_connection(addr, timeout, source_address)

class SyncBackend(NetworkBackend):
    def connect_tcp(
        self,
        host: str,
        port: int,
        timeout: typing.Optional[float] = None,
        local_address: typing.Optional[str] = None,
        socket_options: typing.Optional[typing.Iterable[SOCKET_OPTION]] = None,
        socket_factory: typing.Optional[SocketFactory] = None
    ) -> NetworkStream:
        # Note that we automatically include `TCP_NODELAY`
        # in addition to any other custom socket options.
        if socket_options is None:
            socket_options = []  # pragma: no cover
        address = (host, port)
        source_address = None if local_address is None else (local_address, 0)
        exc_map: ExceptionMapping = {
            socket.timeout: ConnectTimeout,
            OSError: ConnectError,
        }

        if socket_factory is None:
            socket_factory = SocketFactory()

        with map_exceptions(exc_map):
            sock = socket_factory.create_connection(
                address,
                timeout,
                source_address=source_address,
            )
            for option in socket_options:
                sock.setsockopt(*option)  # pragma: no cover
            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
        return SyncStream(sock)

[karpetrosyan]

But it'll be too complicated because we'll need different interfaces for the socket and trio/anyio bytestreams, as well as different factories for these streams, so I think it'd be better to just test with real connections.

[tomchristie]

I don't think we've sufficient motivation to treat this as a hard issue right now. This is maybe worth us covering, but I'm going to bump it down to a discussion. If it's valuable we ought to have a "what is this resolving for us, and is it worth the cost it brings".