Function _wait_for_signal

Lib/test/test_external_inspection.py:84–123

Wait for expected signal(s) from a socket with proper timeout and EOF handling. Args: sock: Connected socket to read from expected_signals: Single bytes object or list of bytes objects to wait for timeout: Socket timeout in seconds Returns: bytes: Compl

(sock, expected_signals, timeout=SHORT_TIMEOUT)

def _wait_for_signal(sock, expected_signals, timeout=SHORT_TIMEOUT):
    """
    Wait for expected signal(s) from a socket with proper timeout and EOF handling.

    Args:
        sock: Connected socket to read from
        expected_signals: Single bytes object or list of bytes objects to wait for
        timeout: Socket timeout in seconds

    Returns:
        bytes: Complete accumulated response buffer

    Raises:
        RuntimeError: If connection closed before signal received or timeout
    """
    if isinstance(expected_signals, bytes):
        expected_signals = [expected_signals]

    sock.settimeout(timeout)
    buffer = b""

    while True:
        # Check if all expected signals are in buffer
        if all(sig in buffer for sig in expected_signals):
            return buffer

        try:
            chunk = sock.recv(4096)
            if not chunk:
                # EOF - connection closed
                raise RuntimeError(
                    f"Connection closed before receiving expected signals. "
                    f"Expected: {expected_signals}, Got: {buffer[-200:]!r}"
                )
            buffer += chunk
        except socket.timeout:
            raise RuntimeError(
                f"Timeout waiting for signals. "
                f"Expected: {expected_signals}, Got: {buffer[-200:]!r}"
            )


def _wait_for_n_signals(sock, signal_pattern, count, timeout=SHORT_TIMEOUT):