Python io 模块，BufferedRandom() 实例源码

def open(self, path, mode='r', errors=None):
        stream = self.opened.get(path)
        if not stream:
            if mode[0] == 'r':
                raise FileNotFoundError(path)
            stream = self.opened[path] = BytesIO()
        stream.seek(0)
        stream = BufferedRandom(stream) if mode[-1] == 'b' else TextIOWrapper(stream)
        stream.close = stream.flush

        return stream

def test_unpickling_buffering_readline(self):
        # Issue #12687: the unpickler's buffering logic could fail with
        # text mode opcodes.
        data = list(range(10))
        for proto in protocols:
            for buf_size in range(1, 11):
                f = io.BufferedRandom(io.BytesIO(), buffer_size=buf_size)
                pickler = self.pickler_class(f, protocol=proto)
                pickler.dump(data)
                f.seek(0)
                unpickler = self.unpickler_class(f)
                self.assertEqual(unpickler.load(), data)

def test_unpickling_buffering_readline(self):
        # Issue #12687: the unpickler's buffering logic could fail with
        # text mode opcodes.
        import io
        data = list(xrange(10))
        for proto in protocols:
            for buf_size in xrange(1, 11):
                f = io.BufferedRandom(io.BytesIO(), buffer_size=buf_size)
                pickler = self.pickler_class(f, protocol=proto)
                pickler.dump(data)
                f.seek(0)
                unpickler = self.unpickler_class(f)
                self.assertEqual(unpickler.load(), data)

def test_unpickling_buffering_readline(self):
        # Issue #12687: the unpickler's buffering logic could fail with
        # text mode opcodes.
        import io
        data = list(xrange(10))
        for proto in protocols:
            for buf_size in xrange(1, 11):
                f = io.BufferedRandom(io.BytesIO(), buffer_size=buf_size)
                pickler = self.pickler_class(f, protocol=proto)
                pickler.dump(data)
                f.seek(0)
                unpickler = self.unpickler_class(f)
                self.assertEqual(unpickler.load(), data)

def open(self, mode = "r"):
        """
        Open the file.

        :param mode: The mode to open the file in. See :py:func:`io.open` for
                     documentation.
        """

        mode = ModeParser(mode)
        mode.parse()
        f = FileStream(self, mode)
        #f = io.BufferedRandom(f)
        if (mode.text):
            f = io.TextIOWrapper(f)
        return f

def test_unpickling_buffering_readline(self):
        # Issue #12687: the unpickler's buffering logic could fail with
        # text mode opcodes.
        data = list(range(10))
        for proto in protocols:
            for buf_size in range(1, 11):
                f = io.BufferedRandom(io.BytesIO(), buffer_size=buf_size)
                pickler = self.pickler_class(f, protocol=proto)
                pickler.dump(data)
                f.seek(0)
                unpickler = self.unpickler_class(f)
                self.assertEqual(unpickler.load(), data)


# Tests for dispatch_table attribute

def test_make_stream_reader_writer(self):
        f = io.BytesIO(b'Hello')
        s = iotools.make_stream('foo', f, '+b', buffering=1)
        self.assertIsInstance(s, io.BufferedRandom)
        self.assertEqual(s.read(), b'Hello')
        s.write(b' World')
        self.assertEqual(f.getvalue(), b'Hello World')

def test_unpickling_buffering_readline(self):
        # Issue #12687: the unpickler's buffering logic could fail with
        # text mode opcodes.
        data = list(range(10))
        for proto in protocols:
            for buf_size in range(1, 11):
                f = io.BufferedRandom(io.BytesIO(), buffer_size=buf_size)
                pickler = self.pickler_class(f, protocol=proto)
                pickler.dump(data)
                f.seek(0)
                unpickler = self.unpickler_class(f)
                self.assertEqual(unpickler.load(), data)


# Tests for dispatch_table attribute

def test_unpickling_buffering_readline(self):
        # Issue #12687: the unpickler's buffering logic could fail with
        # text mode opcodes.
        data = list(range(10))
        for proto in protocols:
            for buf_size in range(1, 11):
                f = io.BufferedRandom(io.BytesIO(), buffer_size=buf_size)
                pickler = self.pickler_class(f, protocol=proto)
                pickler.dump(data)
                f.seek(0)
                unpickler = self.unpickler_class(f)
                self.assertEqual(unpickler.load(), data)


# Tests for dispatch_table attribute

def make_stream(name,
                bin_file,
                mode='r',
                buffering=-1,
                encoding=None,
                errors=None,
                newline='',
                line_buffering=False,
                **kwargs):
    """Take a Python 2.x binary file and return an IO Stream.
    """
    reading = 'r' in mode
    writing = 'w' in mode
    appending = 'a' in mode
    binary = 'b' in mode
    if '+' in mode:
        reading = True
        writing = True

    encoding = None if binary else (encoding or 'utf-8')

    io_object = RawWrapper(bin_file, mode=mode, name=name)
    if buffering >= 0:
        if reading and writing:
            io_object = io.BufferedRandom(
                io_object,
                buffering or io.DEFAULT_BUFFER_SIZE
            )
        elif reading:
            io_object = io.BufferedReader(
                io_object,
                buffering or io.DEFAULT_BUFFER_SIZE
            )
        elif writing or appending:
            io_object = io.BufferedWriter(
                io_object,
                buffering or io.DEFAULT_BUFFER_SIZE
            )

    if not binary:
        io_object = io.TextIOWrapper(
            io_object,
            encoding=encoding,
            errors=errors,
            newline=newline,
            line_buffering=line_buffering,
        )

    return io_object