Function wait

tests/pq/test_pgconn.py:26–46

(
    conn: PGconn | PGcancelConn,
    poll_method: str = "connect_poll",
    return_on: pq.PollingStatus = pq.PollingStatus.OK,
    timeout: int | None = None,
)

def wait(
    conn: PGconn | PGcancelConn,
    poll_method: str = class="st">"connect_poll",
    return_on: pq.PollingStatus = pq.PollingStatus.OK,
    timeout: int | None = None,
) -> None:
    poll = getattr(conn, poll_method)
    while True:
        assert conn.status != pq.ConnStatus.BAD, conn.error_message

        if (rv := poll()) == return_on:
            return
        elif rv == pq.PollingStatus.READING:
            select([conn.socket], [], [], timeout)
        elif rv == pq.PollingStatus.WRITING:
            select([], [conn.socket], [], timeout)
        else:
            pytest.fail(fclass="st">"unexpected poll result: {rv}")
    assert (
        conn.status == pq.ConnStatus.OK
    ), fclass="st">"unexpected connection status: {conn.error_message}"


def test_connectdb(dsn):