Class ShortReadStream

python/test_jsonrpc.py:31–73

Mock stream that simulates short reads from a pipe. This simulates the behavior of Unix pipes when reading data larger than the pipe buffer (typically 64KB). The read() method will return fewer bytes than requested, requiring multiple read calls.

class ShortReadStream:
    """
    Mock stream that simulates short reads from a pipe.

    This simulates the behavior of Unix pipes when reading data larger than
    the pipe buffer (typically 64KB). The read() method will return fewer
    bytes than requested, requiring multiple read calls.
    """

    def __init__(self, data: bytes, chunk_size: int = 32768):
        """
        Args:
            data: Complete data to be read
            chunk_size: Maximum bytes to return per read() call (simulates pipe buffer)
        """
        self.data = data
        self.chunk_size = chunk_size
        self.pos = 0

    def readline(self):
        """Read until newline"""
        end = self.data.find(b"\n", self.pos) + 1
        if end == 0:  # Not found
            result = self.data[self.pos :]
            self.pos = len(self.data)
        else:
            result = self.data[self.pos : end]
            self.pos = end
        return result

    def read(self, n: int) -> bytes:
        """
        Read at most n bytes, but may return fewer (short read).

        This simulates the behavior of pipes when data exceeds buffer size.
        """
        # Calculate how much we can return (limited by chunk_size)
        available = len(self.data) - self.pos
        to_read = min(n, available, self.chunk_size)

        result = self.data[self.pos : self.pos + to_read]
        self.pos += to_read
        return result


class TestReadExact: