Function _wait_for_n_signals

Lib/test/test_external_inspection.py:126–163

Wait for N occurrences of a signal pattern. Args: sock: Connected socket to read from signal_pattern: bytes pattern to count (e.g., b"ready") count: Number of occurrences expected timeout: Socket timeout in seconds Returns: bytes: Complete accum

(sock, signal_pattern, count, timeout=SHORT_TIMEOUT)

def _wait_for_n_signals(sock, signal_pattern, count, timeout=SHORT_TIMEOUT):
    """
    Wait for N occurrences of a signal pattern.

    Args:
        sock: Connected socket to read from
        signal_pattern: bytes pattern to count (e.g., b"ready")
        count: Number of occurrences expected
        timeout: Socket timeout in seconds

    Returns:
        bytes: Complete accumulated response buffer

    Raises:
        RuntimeError: If connection closed or timeout before receiving all signals
    """
    sock.settimeout(timeout)
    buffer = b""
    found_count = 0

    while found_count < count:
        try:
            chunk = sock.recv(4096)
            if not chunk:
                raise RuntimeError(
                    f"Connection closed after {found_count}/{count} signals. "
                    f"Last 200 bytes: {buffer[-200:]!r}"
                )
            buffer += chunk
            # Count occurrences in entire buffer
            found_count = buffer.count(signal_pattern)
        except socket.timeout:
            raise RuntimeError(
                f"Timeout waiting for {count} signals (found {found_count}). "
                f"Last 200 bytes: {buffer[-200:]!r}"
            )

    return buffer


@contextmanager