Function consume_socket

dummyserver/testcase.py:20–38

(
    sock: SSLTransport | socket.socket,
    chunks: int = 65536,
    quit_event: threading.Event | None = None,
)

def consume_socket(
    sock: SSLTransport | socket.socket,
    chunks: int = 65536,
    quit_event: threading.Event | None = None,
) -> bytearray:
    consumed = bytearray()
    sock.settimeout(LONG_TIMEOUT)
    while True:
        if quit_event and quit_event.is_set():
            break
        try:
            b = sock.recv(chunks)
        except TimeoutError:
            continue
        assert isinstance(b, bytes)
        consumed += b
        if b.endswith(b"\r\n\r\n"):
            break
    return consumed


class SocketDummyServerTestCase: