| # Tests StringIO and cStringIO | |
| import unittest | |
| import StringIO | |
| import cStringIO | |
| import types | |
| import array | |
| import sys | |
| from test import test_support | |
| class TestGenericStringIO(unittest.TestCase): | |
| # use a class variable MODULE to define which module is being tested | |
| # Line of data to test as string | |
| _line = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ!' | |
| # Constructor to use for the test data (._line is passed to this | |
| # constructor) | |
| constructor = str | |
| def setUp(self): | |
| self._lines = self.constructor((self._line + '\n') * 5) | |
| self._fp = self.MODULE.StringIO(self._lines) | |
| def test_reads(self): | |
| eq = self.assertEqual | |
| self.assertRaises(TypeError, self._fp.seek) | |
| eq(self._fp.read(10), self._line[:10]) | |
| eq(self._fp.read(0), '') | |
| eq(self._fp.readline(0), '') | |
| eq(self._fp.readline(), self._line[10:] + '\n') | |
| eq(len(self._fp.readlines(60)), 2) | |
| self._fp.seek(0) | |
| eq(self._fp.readline(-1), self._line + '\n') | |
| def test_writes(self): | |
| f = self.MODULE.StringIO() | |
| self.assertRaises(TypeError, f.seek) | |
| f.write(self._line[:6]) | |
| f.seek(3) | |
| f.write(self._line[20:26]) | |
| f.write(self._line[52]) | |
| self.assertEqual(f.getvalue(), 'abcuvwxyz!') | |
| def test_writelines(self): | |
| f = self.MODULE.StringIO() | |
| f.writelines([self._line[0], self._line[1], self._line[2]]) | |
| f.seek(0) | |
| self.assertEqual(f.getvalue(), 'abc') | |
| def test_writelines_error(self): | |
| def errorGen(): | |
| yield 'a' | |
| raise KeyboardInterrupt() | |
| f = self.MODULE.StringIO() | |
| self.assertRaises(KeyboardInterrupt, f.writelines, errorGen()) | |
| def test_truncate(self): | |
| eq = self.assertEqual | |
| f = self.MODULE.StringIO() | |
| f.write(self._lines) | |
| f.seek(10) | |
| f.truncate() | |
| eq(f.getvalue(), 'abcdefghij') | |
| f.truncate(5) | |
| eq(f.getvalue(), 'abcde') | |
| f.write('xyz') | |
| eq(f.getvalue(), 'abcdexyz') | |
| self.assertRaises(IOError, f.truncate, -1) | |
| f.close() | |
| self.assertRaises(ValueError, f.write, 'frobnitz') | |
| def test_closed_flag(self): | |
| f = self.MODULE.StringIO() | |
| self.assertEqual(f.closed, False) | |
| f.close() | |
| self.assertEqual(f.closed, True) | |
| f = self.MODULE.StringIO("abc") | |
| self.assertEqual(f.closed, False) | |
| f.close() | |
| self.assertEqual(f.closed, True) | |
| def test_isatty(self): | |
| f = self.MODULE.StringIO() | |
| self.assertRaises(TypeError, f.isatty, None) | |
| self.assertEqual(f.isatty(), False) | |
| f.close() | |
| self.assertRaises(ValueError, f.isatty) | |
| def test_iterator(self): | |
| eq = self.assertEqual | |
| unless = self.assertTrue | |
| eq(iter(self._fp), self._fp) | |
| # Does this object support the iteration protocol? | |
| unless(hasattr(self._fp, '__iter__')) | |
| unless(hasattr(self._fp, 'next')) | |
| i = 0 | |
| for line in self._fp: | |
| eq(line, self._line + '\n') | |
| i += 1 | |
| eq(i, 5) | |
| self._fp.close() | |
| self.assertRaises(ValueError, self._fp.next) | |
| def test_getvalue(self): | |
| self._fp.close() | |
| self.assertRaises(ValueError, self._fp.getvalue) | |
| @test_support.bigmemtest(test_support._2G + 2**26, memuse=2.001) | |
| def test_reads_from_large_stream(self, size): | |
| linesize = 2**26 # 64 MiB | |
| lines = ['x' * (linesize - 1) + '\n'] * (size // linesize) + \ | |
| ['y' * (size % linesize)] | |
| f = self.MODULE.StringIO(''.join(lines)) | |
| for i, expected in enumerate(lines): | |
| line = f.read(len(expected)) | |
| self.assertEqual(len(line), len(expected)) | |
| self.assertEqual(line, expected) | |
| self.assertEqual(f.read(), '') | |
| f.seek(0) | |
| for i, expected in enumerate(lines): | |
| line = f.readline() | |
| self.assertEqual(len(line), len(expected)) | |
| self.assertEqual(line, expected) | |
| self.assertEqual(f.readline(), '') | |
| f.seek(0) | |
| self.assertEqual(f.readlines(), lines) | |
| self.assertEqual(f.readlines(), []) | |
| f.seek(0) | |
| self.assertEqual(f.readlines(size), lines) | |
| self.assertEqual(f.readlines(), []) | |
| # In worst case cStringIO requires 2 + 1 + 1/2 + 1/2**2 + ... = 4 | |
| # bytes per input character. | |
| @test_support.bigmemtest(test_support._2G, memuse=4) | |
| def test_writes_to_large_stream(self, size): | |
| s = 'x' * 2**26 # 64 MiB | |
| f = self.MODULE.StringIO() | |
| n = size | |
| while n > len(s): | |
| f.write(s) | |
| n -= len(s) | |
| s = None | |
| f.write('x' * n) | |
| self.assertEqual(len(f.getvalue()), size) | |
| class TestStringIO(TestGenericStringIO): | |
| MODULE = StringIO | |
| def test_unicode(self): | |
| if not test_support.have_unicode: return | |
| # The StringIO module also supports concatenating Unicode | |
| # snippets to larger Unicode strings. This is tested by this | |
| # method. Note that cStringIO does not support this extension. | |
| f = self.MODULE.StringIO() | |
| f.write(self._line[:6]) | |
| f.seek(3) | |
| f.write(unicode(self._line[20:26])) | |
| f.write(unicode(self._line[52])) | |
| s = f.getvalue() | |
| self.assertEqual(s, unicode('abcuvwxyz!')) | |
| self.assertEqual(type(s), types.UnicodeType) | |
| class TestcStringIO(TestGenericStringIO): | |
| MODULE = cStringIO | |
| def test_array_support(self): | |
| # Issue #1730114: cStringIO should accept array objects | |
| a = array.array('B', [0,1,2]) | |
| f = self.MODULE.StringIO(a) | |
| self.assertEqual(f.getvalue(), '\x00\x01\x02') | |
| def test_unicode(self): | |
| if not test_support.have_unicode: return | |
| # The cStringIO module converts Unicode strings to character | |
| # strings when writing them to cStringIO objects. | |
| # Check that this works. | |
| f = self.MODULE.StringIO() | |
| f.write(u'abcde') | |
| s = f.getvalue() | |
| self.assertEqual(s, 'abcde') | |
| self.assertEqual(type(s), str) | |
| f = self.MODULE.StringIO(u'abcde') | |
| s = f.getvalue() | |
| self.assertEqual(s, 'abcde') | |
| self.assertEqual(type(s), str) | |
| self.assertRaises(UnicodeEncodeError, self.MODULE.StringIO, u'\xf4') | |
| import sys | |
| if sys.platform.startswith('java'): | |
| # Jython doesn't have a buffer object, so we just do a useless | |
| # fake of the buffer tests. | |
| buffer = str | |
| class TestBufferStringIO(TestStringIO): | |
| constructor = buffer | |
| class TestBuffercStringIO(TestcStringIO): | |
| constructor = buffer | |
| class TestMemoryviewcStringIO(TestcStringIO): | |
| constructor = memoryview | |
| def test_main(): | |
| test_support.run_unittest(TestStringIO, TestcStringIO) | |
| with test_support.check_py3k_warnings(("buffer.. not supported", | |
| DeprecationWarning)): | |
| test_support.run_unittest(TestBufferStringIO, TestBuffercStringIO) | |
| test_support.run_unittest(TestMemoryviewcStringIO) | |
| if __name__ == '__main__': | |
| test_main() |