Function run_scenario

src/trio/_tests/test_highlevel_open_tcp_stream.py:344–388

(
    # The port to connect to
    port: int,
    # A list of
    #  (ip, delay, result)
    # tuples, where delay is in seconds and result is "success" or "error"
    # The ip's will be returned from getaddrinfo in this order, and then
    # connect() calls to them will have the given result.
    ip_list: Sequence[tuple[str, float, str]],
    *,
    # If False, AF_INET4/6 sockets error out on creation, before connect is
    # even called.
    ipv4_supported: bool = True,
    ipv6_supported: bool = True,
    # Normally, we return (winning_sock, scenario object)
    # If this is True, we require there to be an exception, and return
    #   (exception, scenario object)
    expect_error: tuple[type[BaseException], ...] | type[BaseException] = (),
    happy_eyeballs_delay: float | None = 0.25,
    local_address: str | None = None,
)

async def run_scenario(
    # The port to connect to
    port: int,
    # A list of
    #  (ip, delay, result)
    # tuples, where delay is in seconds and result is "success" or "error"
    # The ip's will be returned from getaddrinfo in this order, and then
    # connect() calls to them will have the given result.
    ip_list: Sequence[tuple[str, float, str]],
    *,
    # If False, AF_INET4/6 sockets error out on creation, before connect is
    # even called.
    ipv4_supported: bool = True,
    ipv6_supported: bool = True,
    # Normally, we return (winning_sock, scenario object)
    # If this is True, we require there to be an exception, and return
    #   (exception, scenario object)
    expect_error: tuple[type[BaseException], ...] | type[BaseException] = (),
    happy_eyeballs_delay: float | None = 0.25,
    local_address: str | None = None,
) -> tuple[SocketType, Scenario] | tuple[BaseException, Scenario]:
    supported_families = set()
    if ipv4_supported:
        supported_families.add(trio.socket.AF_INET)
    if ipv6_supported:
        supported_families.add(trio.socket.AF_INET6)
    scenario = Scenario(port, ip_list, supported_families)
    trio.socket.set_custom_hostname_resolver(scenario)
    trio.socket.set_custom_socket_factory(scenario)

    try:
        stream = await open_tcp_stream(
            "test.example.com",
            port,
            happy_eyeballs_delay=happy_eyeballs_delay,
            local_address=local_address,
        )
        assert expect_error == ()
        scenario.check(stream.socket)
        return (stream.socket, scenario)
    except AssertionError:  # pragma: no cover
        raise
    except expect_error as exc:
        scenario.check(None)
        return (exc, scenario)


async def test_one_host_quick_success(autojump_clock: MockClock) -> None: