"""Create a bytes-like object having no string or sequence methods"""

data = bytes(*pos, **kw)

obj = EmptyStruct()

ctypes.resize(obj, len(data))

memoryview(obj).cast("B")[:] = data

"""A RawIO implementation without read(), so as to exercise the default

def __init__(self, read_stack=()):

self._read_stack = list(read_stack)

self._write_stack = []

self._reads = 0

self._extraneous_reads = 0

def write(self, b):

self._write_stack.append(bytes(b))

return len(b)

def writable(self):

return True

def fileno(self):

return 42

def readable(self):

return True

def seekable(self):

return True

def seek(self, pos, whence):

return 0 # wrong but we gotta return something

def tell(self):

return 0 # same comment as above

def readinto(self, buf):

self._reads += 1

max_len = len(buf)

try:

data = self._read_stack[0]

except IndexError:

self._extraneous_reads += 1

return 0

if data is None:

del self._read_stack[0]

return None

n = len(data)

if len(data) <= max_len:

del self._read_stack[0]

buf[:n] = data

return n

else:

buf[:] = data[:max_len]

self._read_stack[0] = data[max_len:]

return max_len

def truncate(self, pos=None):

def read(self, n=None):

self._reads += 1

try:

return self._read_stack.pop(0)

except:

self._extraneous_reads += 1

def write(self, b):

return super().write(b) * 2

def read(self, n=None):

return super().read(n) * 2

def seek(self, pos, whence):

return -123

def tell(self):

return -456

def readinto(self, buf):

super().readinto(buf)

closed = 0

def close(self):

if not self.closed:

self.closed = 1

def __init__(self, data):

self.read_history = []

super().__init__(data)

def read(self, n=None):

res = super().read(n)

self.read_history.append(None if res is None else len(res))

return res

def readinto(self, b):

res = super().readinto(b)

self.read_history.append(res)

def seekable(self):

return False

def seek(self, *args):

raise self.UnsupportedOperation("not seekable")

def tell(self, *args):

raise self.UnsupportedOperation("not seekable")

def truncate(self, *args):

def __init__(self):

self._write_stack = []

self._blocker_char = None

def pop_written(self):

s = b"".join(self._write_stack)

self._write_stack[:] = []

return s

def block_on(self, char):

"""Block when a given char is encountered."""

self._blocker_char = char

def readable(self):

return True

def seekable(self):

return True

def writable(self):

return True

def write(self, b):

b = bytes(b)

n = -1

if self._blocker_char:

try:

n = b.index(self._blocker_char)

except ValueError:

pass

else:

if n > 0:

# write data up to the first blocker

self._write_stack.append(b[:n])

return n

else:

# cancel blocker and indicate would block

self._blocker_char = None

return None

self._write_stack.append(b)

def setUp(self):

support.unlink(support.TESTFN)

def tearDown(self):

support.unlink(support.TESTFN)

def write_ops(self, f):

self.assertEqual(f.write(b"blah."), 5)

f.truncate(0)

self.assertEqual(f.tell(), 5)

f.seek(0)

self.assertEqual(f.write(b"blah."), 5)

self.assertEqual(f.seek(0), 0)

self.assertEqual(f.write(b"Hello."), 6)

self.assertEqual(f.tell(), 6)

self.assertEqual(f.seek(-1, 1), 5)

self.assertEqual(f.tell(), 5)

buffer = bytearray(b" world\n\n\n")

self.assertEqual(f.write(buffer), 9)

buffer[:] = b"*" * 9 # Overwrite our copy of the data

self.assertEqual(f.seek(0), 0)

self.assertEqual(f.write(b"h"), 1)

self.assertEqual(f.seek(-1, 2), 13)

self.assertEqual(f.tell(), 13)

self.assertEqual(f.truncate(12), 12)

self.assertEqual(f.tell(), 13)

self.assertRaises(TypeError, f.seek, 0.0)

def read_ops(self, f, buffered=False):

data = f.read(5)

self.assertEqual(data, b"hello")

data = byteslike(data)

self.assertEqual(f.readinto(data), 5)

self.assertEqual(bytes(data), b" worl")

data = bytearray(5)

self.assertEqual(f.readinto(data), 2)

self.assertEqual(len(data), 5)

self.assertEqual(data[:2], b"d\n")

self.assertEqual(f.seek(0), 0)

self.assertEqual(f.read(20), b"hello world\n")

self.assertEqual(f.read(1), b"")

self.assertEqual(f.readinto(byteslike(b"x")), 0)

self.assertEqual(f.seek(-6, 2), 6)

self.assertEqual(f.read(5), b"world")

self.assertEqual(f.read(0), b"")

self.assertEqual(f.readinto(byteslike()), 0)

self.assertEqual(f.seek(-6, 1), 5)

self.assertEqual(f.read(5), b" worl")

self.assertEqual(f.tell(), 10)

self.assertRaises(TypeError, f.seek, 0.0)

if buffered:

f.seek(0)

self.assertEqual(f.read(), b"hello world\n")

f.seek(6)

self.assertEqual(f.read(), b"world\n")

self.assertEqual(f.read(), b"")

f.seek(0)

data = byteslike(5)

self.assertEqual(f.readinto1(data), 5)

self.assertEqual(bytes(data), b"hello")

LARGE = 2**31

def large_file_ops(self, f):

assert f.readable()

assert f.writable()

try:

self.assertEqual(f.seek(self.LARGE), self.LARGE)

except (OverflowError, ValueError):

self.skipTest("no largefile support")

self.assertEqual(f.tell(), self.LARGE)

self.assertEqual(f.write(b"xxx"), 3)

self.assertEqual(f.tell(), self.LARGE + 3)

self.assertEqual(f.seek(-1, 1), self.LARGE + 2)

self.assertEqual(f.truncate(), self.LARGE + 2)

self.assertEqual(f.tell(), self.LARGE + 2)

self.assertEqual(f.seek(0, 2), self.LARGE + 2)

self.assertEqual(f.truncate(self.LARGE + 1), self.LARGE + 1)

self.assertEqual(f.tell(), self.LARGE + 2)

self.assertEqual(f.seek(0, 2), self.LARGE + 1)

self.assertEqual(f.seek(-1, 2), self.LARGE)

self.assertEqual(f.read(2), b"x")

def test_invalid_operations(self):

# Try writing on a file opened in read mode and vice-versa.

exc = self.UnsupportedOperation

for mode in ("w", "wb"):

with self.open(support.TESTFN, mode) as fp:

self.assertRaises(exc, fp.read)

self.assertRaises(exc, fp.readline)

with self.open(support.TESTFN, "wb", buffering=0) as fp:

self.assertRaises(exc, fp.read)

self.assertRaises(exc, fp.readline)

with self.open(support.TESTFN, "rb", buffering=0) as fp:

self.assertRaises(exc, fp.write, b"blah")

self.assertRaises(exc, fp.writelines, [b"blah\n"])

with self.open(support.TESTFN, "rb") as fp:

self.assertRaises(exc, fp.write, b"blah")

self.assertRaises(exc, fp.writelines, [b"blah\n"])

with self.open(support.TESTFN, "r") as fp:

self.assertRaises(exc, fp.write, "blah")

self.assertRaises(exc, fp.writelines, ["blah\n"])

# Non-zero seeking from current or end pos

self.assertRaises(exc, fp.seek, 1, self.SEEK_CUR)

self.assertRaises(exc, fp.seek, -1, self.SEEK_END)

def test_optional_abilities(self):

# Test for OSError when optional APIs are not supported

# The purpose of this test is to try fileno(), reading, writing and

# seeking operations with various objects that indicate they do not

# support these operations.

def pipe_reader():

[r, w] = os.pipe()

os.close(w) # So that read() is harmless

return self.FileIO(r, "r")

def pipe_writer():

[r, w] = os.pipe()

self.addCleanup(os.close, r)

# Guarantee that we can write into the pipe without blocking

thread = threading.Thread(target=os.read, args=(r, 100))

thread.start()

self.addCleanup(thread.join)

return self.FileIO(w, "w")

def buffered_reader():

return self.BufferedReader(self.MockUnseekableIO())

def buffered_writer():

return self.BufferedWriter(self.MockUnseekableIO())

def buffered_random():

return self.BufferedRandom(self.BytesIO())

def buffered_rw_pair():

return self.BufferedRWPair(self.MockUnseekableIO(),

self.MockUnseekableIO())

def text_reader():

class UnseekableReader(self.MockUnseekableIO):

writable = self.BufferedIOBase.writable

write = self.BufferedIOBase.write

return self.TextIOWrapper(UnseekableReader(), "ascii")

def text_writer():

class UnseekableWriter(self.MockUnseekableIO):

readable = self.BufferedIOBase.readable

read = self.BufferedIOBase.read

return self.TextIOWrapper(UnseekableWriter(), "ascii")

tests = (

(pipe_reader, "fr"), (pipe_writer, "fw"),

(buffered_reader, "r"), (buffered_writer, "w"),

(buffered_random, "rws"), (buffered_rw_pair, "rw"),

(text_reader, "r"), (text_writer, "w"),

(self.BytesIO, "rws"), (self.StringIO, "rws"),

)

for [test, abilities] in tests:

if test is pipe_writer and not threading:

continue # Skip subtest that uses a background thread

with self.subTest(test), test() as obj:

readable = "r" in abilities

self.assertEqual(obj.readable(), readable)

writable = "w" in abilities

self.assertEqual(obj.writable(), writable)

if isinstance(obj, self.TextIOBase):

data = "3"

elif isinstance(obj, (self.BufferedIOBase, self.RawIOBase)):

data = b"3"

else:

self.fail("Unknown base class")

if "f" in abilities:

obj.fileno()

else:

self.assertRaises(OSError, obj.fileno)

if readable:

obj.read(1)

obj.read()

else:

self.assertRaises(OSError, obj.read, 1)

self.assertRaises(OSError, obj.read)

if writable:

obj.write(data)

else:

self.assertRaises(OSError, obj.write, data)

if sys.platform.startswith("win") and test in (

pipe_reader, pipe_writer):

# Pipes seem to appear as seekable on Windows

continue

seekable = "s" in abilities

self.assertEqual(obj.seekable(), seekable)

if seekable:

obj.tell()

obj.seek(0)

else:

self.assertRaises(OSError, obj.tell)

self.assertRaises(OSError, obj.seek, 0)

if writable and seekable:

obj.truncate()

obj.truncate(0)

else:

self.assertRaises(OSError, obj.truncate)

self.assertRaises(OSError, obj.truncate, 0)

def test_open_handles_NUL_chars(self):

fn_with_NUL = 'foo\0bar'

self.assertRaises(ValueError, self.open, fn_with_NUL, 'w')

bytes_fn = bytes(fn_with_NUL, 'ascii')

with warnings.catch_warnings():

warnings.simplefilter("ignore", DeprecationWarning)

self.assertRaises(ValueError, self.open, bytes_fn, 'w')

def test_raw_file_io(self):

with self.open(support.TESTFN, "wb", buffering=0) as f:

self.assertEqual(f.readable(), False)

self.assertEqual(f.writable(), True)

self.assertEqual(f.seekable(), True)

self.write_ops(f)

with self.open(support.TESTFN, "rb", buffering=0) as f:

self.assertEqual(f.readable(), True)

self.assertEqual(f.writable(), False)

self.assertEqual(f.seekable(), True)

self.read_ops(f)

def test_buffered_file_io(self):

with self.open(support.TESTFN, "wb") as f:

self.assertEqual(f.readable(), False)

self.assertEqual(f.writable(), True)

self.assertEqual(f.seekable(), True)

self.write_ops(f)

with self.open(support.TESTFN, "rb") as f:

self.assertEqual(f.readable(), True)

self.assertEqual(f.writable(), False)

self.assertEqual(f.seekable(), True)

self.read_ops(f, True)

def test_readline(self):

with self.open(support.TESTFN, "wb") as f:

f.write(b"abc\ndef\nxyzzy\nfoo\x00bar\nanother line")

with self.open(support.TESTFN, "rb") as f:

self.assertEqual(f.readline(), b"abc\n")

self.assertEqual(f.readline(10), b"def\n")

self.assertEqual(f.readline(2), b"xy")

self.assertEqual(f.readline(4), b"zzy\n")

self.assertEqual(f.readline(), b"foo\x00bar\n")

self.assertEqual(f.readline(None), b"another line")

self.assertRaises(TypeError, f.readline, 5.3)

with self.open(support.TESTFN, "r") as f:

self.assertRaises(TypeError, f.readline, 5.3)

def test_raw_bytes_io(self):

f = self.BytesIO()

self.write_ops(f)

data = f.getvalue()

self.assertEqual(data, b"hello world\n")

f = self.BytesIO(data)

self.read_ops(f, True)

def test_large_file_ops(self):

# On Windows and Mac OSX this test comsumes large resources; It takes

# a long time to build the >2GB file and takes >2GB of disk space

# therefore the resource must be enabled to run this test.

if sys.platform[:3] == 'win' or sys.platform == 'darwin':

support.requires(

'largefile',

'test requires %s bytes and a long time to run' % self.LARGE)

with self.open(support.TESTFN, "w+b", 0) as f:

self.large_file_ops(f)

with self.open(support.TESTFN, "w+b") as f:

self.large_file_ops(f)

def test_with_open(self):

for bufsize in (0, 1, 100):

f = None

with self.open(support.TESTFN, "wb", bufsize) as f:

f.write(b"xxx")

self.assertEqual(f.closed, True)

f = None

try:

with self.open(support.TESTFN, "wb", bufsize) as f:

1/0

except ZeroDivisionError:

self.assertEqual(f.closed, True)

else:

self.fail("1/0 didn't raise an exception")

# issue 5008

def test_append_mode_tell(self):

with self.open(support.TESTFN, "wb") as f:

f.write(b"xxx")

with self.open(support.TESTFN, "ab", buffering=0) as f:

self.assertEqual(f.tell(), 3)

with self.open(support.TESTFN, "ab") as f:

self.assertEqual(f.tell(), 3)

with self.open(support.TESTFN, "a") as f:

self.assertGreater(f.tell(), 0)

def test_destructor(self):

record = []

class MyFileIO(self.FileIO):

def __del__(self):

record.append(1)

try:

f = super().__del__

except AttributeError:

pass

else:

f()

def close(self):

record.append(2)

super().close()

def flush(self):

record.append(3)

super().flush()

with support.check_warnings(('', ResourceWarning)):

f = MyFileIO(support.TESTFN, "wb")

f.write(b"xxx")

del f

support.gc_collect()

self.assertEqual(record, [1, 2, 3])

with self.open(support.TESTFN, "rb") as f:

self.assertEqual(f.read(), b"xxx")

def _check_base_destructor(self, base):

record = []

class MyIO(base):

def __init__(self):

# This exercises the availability of attributes on object

# destruction.

# (in the C version, close() is called by the tp_dealloc

# function, not by __del__)

self.on_del = 1

self.on_close = 2

self.on_flush = 3

def __del__(self):

record.append(self.on_del)

try:

f = super().__del__

except AttributeError:

pass

else:

f()

def close(self):

record.append(self.on_close)

super().close()

def flush(self):

record.append(self.on_flush)

super().flush()

f = MyIO()

del f

support.gc_collect()

self.assertEqual(record, [1, 2, 3])

def test_IOBase_destructor(self):

self._check_base_destructor(self.IOBase)

def test_RawIOBase_destructor(self):

self._check_base_destructor(self.RawIOBase)

def test_BufferedIOBase_destructor(self):

self._check_base_destructor(self.BufferedIOBase)

def test_TextIOBase_destructor(self):

self._check_base_destructor(self.TextIOBase)

def test_close_flushes(self):

with self.open(support.TESTFN, "wb") as f:

f.write(b"xxx")

with self.open(support.TESTFN, "rb") as f:

self.assertEqual(f.read(), b"xxx")

def test_array_writes(self):

a = array.array('i', range(10))

n = len(a.tobytes())

def check(f):

with f:

self.assertEqual(f.write(a), n)

f.writelines((a,))

check(self.BytesIO())

check(self.FileIO(support.TESTFN, "w"))

check(self.BufferedWriter(self.MockRawIO()))

check(self.BufferedRandom(self.MockRawIO()))

check(self.BufferedRWPair(self.MockRawIO(), self.MockRawIO()))

def test_closefd(self):

self.assertRaises(ValueError, self.open, support.TESTFN, 'w',

closefd=False)

def test_read_closed(self):

with self.open(support.TESTFN, "w") as f:

f.write("egg\n")

with self.open(support.TESTFN, "r") as f:

file = self.open(f.fileno(), "r", closefd=False)

self.assertEqual(file.read(), "egg\n")

file.seek(0)

file.close()

self.assertRaises(ValueError, file.read)

def test_no_closefd_with_filename(self):

# can't use closefd in combination with a file name

self.assertRaises(ValueError, self.open, support.TESTFN, "r", closefd=False)

def test_closefd_attr(self):

with self.open(support.TESTFN, "wb") as f:

f.write(b"egg\n")

with self.open(support.TESTFN, "r") as f:

self.assertEqual(f.buffer.raw.closefd, True)

file = self.open(f.fileno(), "r", closefd=False)

self.assertEqual(file.buffer.raw.closefd, False)

def test_garbage_collection(self):

# FileIO objects are collected, and collecting them flushes

# all data to disk.

with support.check_warnings(('', ResourceWarning)):

f = self.FileIO(support.TESTFN, "wb")

f.write(b"abcxxx")

f.f = f

wr = weakref.ref(f)

del f

support.gc_collect()

self.assertIsNone(wr(), wr)

with self.open(support.TESTFN, "rb") as f:

self.assertEqual(f.read(), b"abcxxx")

def test_unbounded_file(self):

# Issue #1174606: reading from an unbounded stream such as /dev/zero.

zero = "/dev/zero"

if not os.path.exists(zero):

self.skipTest("{0} does not exist".format(zero))

if sys.maxsize > 0x7FFFFFFF:

self.skipTest("test can only run in a 32-bit address space")

if support.real_max_memuse < support._2G:

self.skipTest("test requires at least 2GB of memory")

with self.open(zero, "rb", buffering=0) as f:

self.assertRaises(OverflowError, f.read)

with self.open(zero, "rb") as f:

self.assertRaises(OverflowError, f.read)

with self.open(zero, "r") as f:

self.assertRaises(OverflowError, f.read)

def check_flush_error_on_close(self, *args, **kwargs):

# Test that the file is closed despite failed flush

# and that flush() is called before file closed.

f = self.open(*args, **kwargs)

closed = []

def bad_flush():

closed[:] = [f.closed]

raise OSError()

f.flush = bad_flush

self.assertRaises(OSError, f.close) # exception not swallowed

self.assertTrue(f.closed)

self.assertTrue(closed) # flush() called

self.assertFalse(closed[0]) # flush() called before file closed

f.flush = lambda: None # break reference loop

def test_flush_error_on_close(self):

# raw file

# Issue #5700: io.FileIO calls flush() after file closed

self.check_flush_error_on_close(support.TESTFN, 'wb', buffering=0)

fd = os.open(support.TESTFN, os.O_WRONLY|os.O_CREAT)

self.check_flush_error_on_close(fd, 'wb', buffering=0)

fd = os.open(support.TESTFN, os.O_WRONLY|os.O_CREAT)

self.check_flush_error_on_close(fd, 'wb', buffering=0, closefd=False)

os.close(fd)

# buffered io

self.check_flush_error_on_close(support.TESTFN, 'wb')

fd = os.open(support.TESTFN, os.O_WRONLY|os.O_CREAT)

self.check_flush_error_on_close(fd, 'wb')

fd = os.open(support.TESTFN, os.O_WRONLY|os.O_CREAT)

self.check_flush_error_on_close(fd, 'wb', closefd=False)

os.close(fd)

# text io

self.check_flush_error_on_close(support.TESTFN, 'w')

fd = os.open(support.TESTFN, os.O_WRONLY|os.O_CREAT)

self.check_flush_error_on_close(fd, 'w')

fd = os.open(support.TESTFN, os.O_WRONLY|os.O_CREAT)

self.check_flush_error_on_close(fd, 'w', closefd=False)

os.close(fd)

def test_multi_close(self):

f = self.open(support.TESTFN, "wb", buffering=0)

f.close()

f.close()

f.close()

self.assertRaises(ValueError, f.flush)

def test_RawIOBase_read(self):

# Exercise the default RawIOBase.read() implementation (which calls

# readinto() internally).

rawio = self.MockRawIOWithoutRead((b"abc", b"d", None, b"efg", None))

self.assertEqual(rawio.read(2), b"ab")

self.assertEqual(rawio.read(2), b"c")

self.assertEqual(rawio.read(2), b"d")

self.assertEqual(rawio.read(2), None)

self.assertEqual(rawio.read(2), b"ef")

self.assertEqual(rawio.read(2), b"g")

self.assertEqual(rawio.read(2), None)

self.assertEqual(rawio.read(2), b"")

def test_types_have_dict(self):

test = (

self.IOBase(),

self.RawIOBase(),

self.TextIOBase(),

self.StringIO(),

self.BytesIO()

)

for obj in test:

self.assertTrue(hasattr(obj, "__dict__"))

def test_opener(self):

with self.open(support.TESTFN, "w") as f:

f.write("egg\n")

fd = os.open(support.TESTFN, os.O_RDONLY)

def opener(path, flags):

return fd

with self.open("non-existent", "r", opener=opener) as f:

self.assertEqual(f.read(), "egg\n")

def test_bad_opener_negative_1(self):

# Issue #27066.

def badopener(fname, flags):

return -1

with self.assertRaises(ValueError) as cm:

open('non-existent', 'r', opener=badopener)

self.assertEqual(str(cm.exception), 'opener returned -1')

def test_bad_opener_other_negative(self):

# Issue #27066.

def badopener(fname, flags):

return -2

with self.assertRaises(ValueError) as cm:

open('non-existent', 'r', opener=badopener)

self.assertEqual(str(cm.exception), 'opener returned -2')

def test_fileio_closefd(self):

# Issue #4841

with self.open(__file__, 'rb') as f1, \

self.open(__file__, 'rb') as f2:

fileio = self.FileIO(f1.fileno(), closefd=False)

# .__init__() must not close f1

fileio.__init__(f2.fileno(), closefd=False)

f1.readline()

# .close() must not close f2

fileio.close()

f2.readline()

def test_nonbuffered_textio(self):

with support.check_no_resource_warning(self):

with self.assertRaises(ValueError):

self.open(support.TESTFN, 'w', buffering=0)

def test_invalid_newline(self):

with support.check_no_resource_warning(self):

with self.assertRaises(ValueError):

self.open(support.TESTFN, 'w', newline='invalid')

def test_buffered_readinto_mixin(self):

# Test the implementation provided by BufferedIOBase

class Stream(self.BufferedIOBase):

def read(self, size):

return b"12345"

read1 = read

stream = Stream()

for method in ("readinto", "readinto1"):

with self.subTest(method):

buffer = byteslike(5)

self.assertEqual(getattr(stream, method)(buffer), 5)

self.assertEqual(bytes(buffer), b"12345")

def test_fspath_support(self):

class PathLike:

def __init__(self, path):

self.path = path

def __fspath__(self):

return self.path

def check_path_succeeds(path):

with self.open(path, "w") as f:

f.write("egg\n")

with self.open(path, "r") as f:

self.assertEqual(f.read(), "egg\n")

check_path_succeeds(PathLike(support.TESTFN))

check_path_succeeds(PathLike(support.TESTFN.encode('utf-8')))

bad_path = PathLike(TypeError)

with self.assertRaises(TypeError):

self.open(bad_path, 'w')

# ensure that refcounting is correct with some error conditions

with self.assertRaisesRegex(ValueError, 'read/write/append mode'):

def test_IOBase_finalize(self):

# Issue #12149: segmentation fault on _PyIOBase_finalize when both a

# class which inherits IOBase and an object of this class are caught

# in a reference cycle and close() is already in the method cache.

class MyIO(self.IOBase):

def close(self):

pass

# create an instance to populate the method cache

MyIO()

obj = MyIO()

obj.obj = obj

wr = weakref.ref(obj)

del MyIO

del obj

support.gc_collect()

def test_RawIOBase_io_in_pyio_match(self):

"""Test that pyio RawIOBase class has all c RawIOBase methods"""

mismatch = support.detect_api_mismatch(pyio.RawIOBase, io.RawIOBase,

ignore=('__weakref__',))

self.assertEqual(mismatch, set(), msg='Python RawIOBase does not have all C RawIOBase methods')

def test_RawIOBase_pyio_in_io_match(self):

"""Test that c RawIOBase class has all pyio RawIOBase methods"""

mismatch = support.detect_api_mismatch(io.RawIOBase, pyio.RawIOBase)

# Tests common to BufferedReader, BufferedWriter and BufferedRandom

def test_detach(self):

raw = self.MockRawIO()

buf = self.tp(raw)

self.assertIs(buf.detach(), raw)

self.assertRaises(ValueError, buf.detach)

repr(buf) # Should still work

def test_fileno(self):

rawio = self.MockRawIO()

bufio = self.tp(rawio)

self.assertEqual(42, bufio.fileno())

def test_invalid_args(self):

rawio = self.MockRawIO()

bufio = self.tp(rawio)

# Invalid whence

self.assertRaises(ValueError, bufio.seek, 0, -1)

self.assertRaises(ValueError, bufio.seek, 0, 9)

def test_override_destructor(self):

tp = self.tp

record = []

class MyBufferedIO(tp):

def __del__(self):

record.append(1)

try:

f = super().__del__

except AttributeError:

pass

else:

f()

def close(self):

record.append(2)

super().close()

def flush(self):

record.append(3)

super().flush()

rawio = self.MockRawIO()

bufio = MyBufferedIO(rawio)

del bufio

support.gc_collect()

self.assertEqual(record, [1, 2, 3])

def test_context_manager(self):

# Test usability as a context manager

rawio = self.MockRawIO()

bufio = self.tp(rawio)

def _with():

with bufio:

pass

_with()

# bufio should now be closed, and using it a second time should raise

# a ValueError.

self.assertRaises(ValueError, _with)

def test_error_through_destructor(self):

# Test that the exception state is not modified by a destructor,

# even if close() fails.

rawio = self.CloseFailureIO()

def f():

self.tp(rawio).xyzzy

with support.captured_output("stderr") as s:

self.assertRaises(AttributeError, f)

s = s.getvalue().strip()