Function test_connect_reuse_proxy_headers

tests/test_connector.py:3257–3331

(  # type: ignore[misc]
    loop: asyncio.AbstractEventLoop,
    test_case: str,
    wait_for_con: bool,
    expect_proxy_auth_header: bool,
)

    ],
)
async def test_connect_reuse_proxy_headers(  # type: ignore[misc]
    loop: asyncio.AbstractEventLoop,
    test_case: str,
    wait_for_con: bool,
    expect_proxy_auth_header: bool,
) -> None:
    proto = create_mocked_conn(loop)
    proto.is_connected.return_value = True

    if test_case != "dont_use_proxy":
        proxy = (
            URL("http://user:password@example.com")
            if test_case == "use_proxy_with_embedded_auth"
            else URL("http://example.com")
        )
        proxy_headers = (
            CIMultiDict({hdrs.AUTHORIZATION: "Basic dXNlcjpwYXNzd29yZA=="})
            if test_case == "use_proxy_with_auth_headers"
            else None
        )
    else:
        proxy = None
        proxy_headers = None
    key = ConnectionKey(
        "localhost",
        80,
        False,
        True,
        proxy,
        None,
        hash(tuple(proxy_headers.items())) if proxy_headers else None,
    )
    req = ClientRequest(
        "GET",
        URL("http://localhost:80"),
        loop=loop,
        response_class=mock.Mock(),
        proxy=proxy,
        proxy_headers=proxy_headers,
    )

    conn = aiohttp.BaseConnector(limit=1)

    async def _create_con(*args: Any, **kwargs: Any) -> None:
        conn._conns[key] = deque([(proto, loop.time())])

    with contextlib.ExitStack() as stack:
        if wait_for_con:
            # Simulate no available connections
            stack.enter_context(
                mock.patch.object(
                    conn, "_available_connections", autospec=True, return_value=0
                )
            )
            # Upon waiting for a connection, populate _conns with our proto,
            # mocking a connection becoming immediately available
            stack.enter_context(
                mock.patch.object(