|
| 1 | +import sys |
| 2 | +import os |
| 3 | +import unittest |
| 4 | +from array import array |
| 5 | +from weakref import proxy |
| 6 | + |
| 7 | +import io |
| 8 | +import _pyio as pyio |
| 9 | + |
| 10 | +from test.support import TESTFN |
| 11 | +from test import support |
| 12 | +from collections import UserList |
| 13 | + |
| 14 | +class AutoFileTests: |
| 15 | + # file tests for which a test file is automatically set up |
| 16 | + |
| 17 | + def setUp(self): |
| 18 | + self.f = self.open(TESTFN, 'wb') |
| 19 | + |
| 20 | + def tearDown(self): |
| 21 | + if self.f: |
| 22 | + self.f.close() |
| 23 | + support.unlink(TESTFN) |
| 24 | + |
| 25 | + def testWeakRefs(self): |
| 26 | + # verify weak references |
| 27 | + p = proxy(self.f) |
| 28 | + p.write(b'teststring') |
| 29 | + self.assertEqual(self.f.tell(), p.tell()) |
| 30 | + self.f.close() |
| 31 | + self.f = None |
| 32 | + self.assertRaises(ReferenceError, getattr, p, 'tell') |
| 33 | + |
| 34 | + def testAttributes(self): |
| 35 | + # verify expected attributes exist |
| 36 | + f = self.f |
| 37 | + f.name # merely shouldn't blow up |
| 38 | + f.mode # ditto |
| 39 | + f.closed # ditto |
| 40 | + |
| 41 | + def testReadinto(self): |
| 42 | + # verify readinto |
| 43 | + self.f.write(b'12') |
| 44 | + self.f.close() |
| 45 | + a = array('b', b'x'*10) |
| 46 | + self.f = self.open(TESTFN, 'rb') |
| 47 | + n = self.f.readinto(a) |
| 48 | + self.assertEqual(b'12', a.tobytes()[:n]) |
| 49 | + |
| 50 | + def testReadinto_text(self): |
| 51 | + # verify readinto refuses text files |
| 52 | + a = array('b', b'x'*10) |
| 53 | + self.f.close() |
| 54 | + self.f = self.open(TESTFN, 'r') |
| 55 | + if hasattr(self.f, "readinto"): |
| 56 | + self.assertRaises(TypeError, self.f.readinto, a) |
| 57 | + |
| 58 | + def testWritelinesUserList(self): |
| 59 | + # verify writelines with instance sequence |
| 60 | + l = UserList([b'1', b'2']) |
| 61 | + self.f.writelines(l) |
| 62 | + self.f.close() |
| 63 | + self.f = self.open(TESTFN, 'rb') |
| 64 | + buf = self.f.read() |
| 65 | + self.assertEqual(buf, b'12') |
| 66 | + |
| 67 | + def testWritelinesIntegers(self): |
| 68 | + # verify writelines with integers |
| 69 | + self.assertRaises(TypeError, self.f.writelines, [1, 2, 3]) |
| 70 | + |
| 71 | + def testWritelinesIntegersUserList(self): |
| 72 | + # verify writelines with integers in UserList |
| 73 | + l = UserList([1,2,3]) |
| 74 | + self.assertRaises(TypeError, self.f.writelines, l) |
| 75 | + |
| 76 | + def testWritelinesNonString(self): |
| 77 | + # verify writelines with non-string object |
| 78 | + class NonString: |
| 79 | + pass |
| 80 | + |
| 81 | + self.assertRaises(TypeError, self.f.writelines, |
| 82 | + [NonString(), NonString()]) |
| 83 | + |
| 84 | + def testErrors(self): |
| 85 | + f = self.f |
| 86 | + self.assertEqual(f.name, TESTFN) |
| 87 | + self.assertFalse(f.isatty()) |
| 88 | + self.assertFalse(f.closed) |
| 89 | + |
| 90 | + if hasattr(f, "readinto"): |
| 91 | + self.assertRaises((OSError, TypeError), f.readinto, "") |
| 92 | + f.close() |
| 93 | + self.assertTrue(f.closed) |
| 94 | + |
| 95 | + def testMethods(self): |
| 96 | + methods = [('fileno', ()), |
| 97 | + ('flush', ()), |
| 98 | + ('isatty', ()), |
| 99 | + ('__next__', ()), |
| 100 | + ('read', ()), |
| 101 | + ('write', (b"",)), |
| 102 | + ('readline', ()), |
| 103 | + ('readlines', ()), |
| 104 | + ('seek', (0,)), |
| 105 | + ('tell', ()), |
| 106 | + ('write', (b"",)), |
| 107 | + ('writelines', ([],)), |
| 108 | + ('__iter__', ()), |
| 109 | + ] |
| 110 | + methods.append(('truncate', ())) |
| 111 | + |
| 112 | + # __exit__ should close the file |
| 113 | + self.f.__exit__(None, None, None) |
| 114 | + self.assertTrue(self.f.closed) |
| 115 | + |
| 116 | + for methodname, args in methods: |
| 117 | + method = getattr(self.f, methodname) |
| 118 | + # should raise on closed file |
| 119 | + self.assertRaises(ValueError, method, *args) |
| 120 | + |
| 121 | + # file is closed, __exit__ shouldn't do anything |
| 122 | + self.assertEqual(self.f.__exit__(None, None, None), None) |
| 123 | + # it must also return None if an exception was given |
| 124 | + try: |
| 125 | + 1/0 |
| 126 | + except: |
| 127 | + self.assertEqual(self.f.__exit__(*sys.exc_info()), None) |
| 128 | + |
| 129 | + def testReadWhenWriting(self): |
| 130 | + self.assertRaises(OSError, self.f.read) |
| 131 | + |
| 132 | +class CAutoFileTests(AutoFileTests, unittest.TestCase): |
| 133 | + open = io.open |
| 134 | + |
| 135 | +@unittest.skipIf(sys.platform == "win32", "TODO: RUSTPYTHON, test setUp errors on Windows") |
| 136 | +class PyAutoFileTests(AutoFileTests, unittest.TestCase): |
| 137 | + open = staticmethod(pyio.open) |
| 138 | + |
| 139 | + |
| 140 | +class OtherFileTests: |
| 141 | + |
| 142 | + def tearDown(self): |
| 143 | + support.unlink(TESTFN) |
| 144 | + |
| 145 | + def testModeStrings(self): |
| 146 | + # check invalid mode strings |
| 147 | + self.open(TESTFN, 'wb').close() |
| 148 | + for mode in ("", "aU", "wU+", "U+", "+U", "rU+"): |
| 149 | + try: |
| 150 | + f = self.open(TESTFN, mode) |
| 151 | + except ValueError: |
| 152 | + pass |
| 153 | + else: |
| 154 | + f.close() |
| 155 | + self.fail('%r is an invalid file mode' % mode) |
| 156 | + |
| 157 | + def testBadModeArgument(self): |
| 158 | + # verify that we get a sensible error message for bad mode argument |
| 159 | + bad_mode = "qwerty" |
| 160 | + try: |
| 161 | + f = self.open(TESTFN, bad_mode) |
| 162 | + except ValueError as msg: |
| 163 | + if msg.args[0] != 0: |
| 164 | + s = str(msg) |
| 165 | + if TESTFN in s or bad_mode not in s: |
| 166 | + self.fail("bad error message for invalid mode: %s" % s) |
| 167 | + # if msg.args[0] == 0, we're probably on Windows where there may be |
| 168 | + # no obvious way to discover why open() failed. |
| 169 | + else: |
| 170 | + f.close() |
| 171 | + self.fail("no error for invalid mode: %s" % bad_mode) |
| 172 | + |
| 173 | + def _checkBufferSize(self, s): |
| 174 | + try: |
| 175 | + f = self.open(TESTFN, 'wb', s) |
| 176 | + f.write(str(s).encode("ascii")) |
| 177 | + f.close() |
| 178 | + f.close() |
| 179 | + f = self.open(TESTFN, 'rb', s) |
| 180 | + d = int(f.read().decode("ascii")) |
| 181 | + f.close() |
| 182 | + f.close() |
| 183 | + except OSError as msg: |
| 184 | + self.fail('error setting buffer size %d: %s' % (s, str(msg))) |
| 185 | + self.assertEqual(d, s) |
| 186 | + |
| 187 | + def testSetBufferSize(self): |
| 188 | + # make sure that explicitly setting the buffer size doesn't cause |
| 189 | + # misbehaviour especially with repeated close() calls |
| 190 | + for s in (-1, 0, 512): |
| 191 | + with support.check_no_warnings(self, |
| 192 | + message='line buffering', |
| 193 | + category=RuntimeWarning): |
| 194 | + self._checkBufferSize(s) |
| 195 | + |
| 196 | + # test that attempts to use line buffering in binary mode cause |
| 197 | + # a warning |
| 198 | + with self.assertWarnsRegex(RuntimeWarning, 'line buffering'): |
| 199 | + self._checkBufferSize(1) |
| 200 | + |
| 201 | + def testTruncateOnWindows(self): |
| 202 | + # SF bug <http://www.python.org/sf/801631> |
| 203 | + # "file.truncate fault on windows" |
| 204 | + |
| 205 | + f = self.open(TESTFN, 'wb') |
| 206 | + |
| 207 | + try: |
| 208 | + f.write(b'12345678901') # 11 bytes |
| 209 | + f.close() |
| 210 | + |
| 211 | + f = self.open(TESTFN,'rb+') |
| 212 | + data = f.read(5) |
| 213 | + if data != b'12345': |
| 214 | + self.fail("Read on file opened for update failed %r" % data) |
| 215 | + if f.tell() != 5: |
| 216 | + self.fail("File pos after read wrong %d" % f.tell()) |
| 217 | + |
| 218 | + f.truncate() |
| 219 | + if f.tell() != 5: |
| 220 | + self.fail("File pos after ftruncate wrong %d" % f.tell()) |
| 221 | + |
| 222 | + f.close() |
| 223 | + size = os.path.getsize(TESTFN) |
| 224 | + if size != 5: |
| 225 | + self.fail("File size after ftruncate wrong %d" % size) |
| 226 | + finally: |
| 227 | + f.close() |
| 228 | + |
| 229 | + def testIteration(self): |
| 230 | + # Test the complex interaction when mixing file-iteration and the |
| 231 | + # various read* methods. |
| 232 | + dataoffset = 16384 |
| 233 | + filler = b"ham\n" |
| 234 | + assert not dataoffset % len(filler), \ |
| 235 | + "dataoffset must be multiple of len(filler)" |
| 236 | + nchunks = dataoffset // len(filler) |
| 237 | + testlines = [ |
| 238 | + b"spam, spam and eggs\n", |
| 239 | + b"eggs, spam, ham and spam\n", |
| 240 | + b"saussages, spam, spam and eggs\n", |
| 241 | + b"spam, ham, spam and eggs\n", |
| 242 | + b"spam, spam, spam, spam, spam, ham, spam\n", |
| 243 | + b"wonderful spaaaaaam.\n" |
| 244 | + ] |
| 245 | + methods = [("readline", ()), ("read", ()), ("readlines", ()), |
| 246 | + ("readinto", (array("b", b" "*100),))] |
| 247 | + |
| 248 | + # Prepare the testfile |
| 249 | + bag = self.open(TESTFN, "wb") |
| 250 | + bag.write(filler * nchunks) |
| 251 | + bag.writelines(testlines) |
| 252 | + bag.close() |
| 253 | + # Test for appropriate errors mixing read* and iteration |
| 254 | + for methodname, args in methods: |
| 255 | + f = self.open(TESTFN, 'rb') |
| 256 | + self.assertEqual(next(f), filler) |
| 257 | + meth = getattr(f, methodname) |
| 258 | + meth(*args) # This simply shouldn't fail |
| 259 | + f.close() |
| 260 | + |
| 261 | + # Test to see if harmless (by accident) mixing of read* and |
| 262 | + # iteration still works. This depends on the size of the internal |
| 263 | + # iteration buffer (currently 8192,) but we can test it in a |
| 264 | + # flexible manner. Each line in the bag o' ham is 4 bytes |
| 265 | + # ("h", "a", "m", "\n"), so 4096 lines of that should get us |
| 266 | + # exactly on the buffer boundary for any power-of-2 buffersize |
| 267 | + # between 4 and 16384 (inclusive). |
| 268 | + f = self.open(TESTFN, 'rb') |
| 269 | + for i in range(nchunks): |
| 270 | + next(f) |
| 271 | + testline = testlines.pop(0) |
| 272 | + try: |
| 273 | + line = f.readline() |
| 274 | + except ValueError: |
| 275 | + self.fail("readline() after next() with supposedly empty " |
| 276 | + "iteration-buffer failed anyway") |
| 277 | + if line != testline: |
| 278 | + self.fail("readline() after next() with empty buffer " |
| 279 | + "failed. Got %r, expected %r" % (line, testline)) |
| 280 | + testline = testlines.pop(0) |
| 281 | + buf = array("b", b"\x00" * len(testline)) |
| 282 | + try: |
| 283 | + f.readinto(buf) |
| 284 | + except ValueError: |
| 285 | + self.fail("readinto() after next() with supposedly empty " |
| 286 | + "iteration-buffer failed anyway") |
| 287 | + line = buf.tobytes() |
| 288 | + if line != testline: |
| 289 | + self.fail("readinto() after next() with empty buffer " |
| 290 | + "failed. Got %r, expected %r" % (line, testline)) |
| 291 | + |
| 292 | + testline = testlines.pop(0) |
| 293 | + try: |
| 294 | + line = f.read(len(testline)) |
| 295 | + except ValueError: |
| 296 | + self.fail("read() after next() with supposedly empty " |
| 297 | + "iteration-buffer failed anyway") |
| 298 | + if line != testline: |
| 299 | + self.fail("read() after next() with empty buffer " |
| 300 | + "failed. Got %r, expected %r" % (line, testline)) |
| 301 | + try: |
| 302 | + lines = f.readlines() |
| 303 | + except ValueError: |
| 304 | + self.fail("readlines() after next() with supposedly empty " |
| 305 | + "iteration-buffer failed anyway") |
| 306 | + if lines != testlines: |
| 307 | + self.fail("readlines() after next() with empty buffer " |
| 308 | + "failed. Got %r, expected %r" % (line, testline)) |
| 309 | + f.close() |
| 310 | + |
| 311 | + # Reading after iteration hit EOF shouldn't hurt either |
| 312 | + f = self.open(TESTFN, 'rb') |
| 313 | + try: |
| 314 | + for line in f: |
| 315 | + pass |
| 316 | + try: |
| 317 | + f.readline() |
| 318 | + f.readinto(buf) |
| 319 | + f.read() |
| 320 | + f.readlines() |
| 321 | + except ValueError: |
| 322 | + self.fail("read* failed after next() consumed file") |
| 323 | + finally: |
| 324 | + f.close() |
| 325 | + |
| 326 | +class COtherFileTests(OtherFileTests, unittest.TestCase): |
| 327 | + open = io.open |
| 328 | + |
| 329 | + # TODO: RUSTPYTHON |
| 330 | + @unittest.expectedFailure |
| 331 | + def testSetBufferSize(self): |
| 332 | + super().testSetBufferSize() |
| 333 | + |
| 334 | +class PyOtherFileTests(OtherFileTests, unittest.TestCase): |
| 335 | + open = staticmethod(pyio.open) |
| 336 | + |
| 337 | + # TODO: RUSTPYTHON |
| 338 | + if sys.platform == "win32": |
| 339 | + @unittest.expectedFailure |
| 340 | + def testIteration(self): |
| 341 | + super().testIteration() |
| 342 | + |
| 343 | + # TODO: RUSTPYTHON |
| 344 | + if sys.platform == "win32": |
| 345 | + @unittest.expectedFailure |
| 346 | + def testModeStrings(self): |
| 347 | + super().testModeStrings() |
| 348 | + |
| 349 | + # TODO: RUSTPYTHON |
| 350 | + if sys.platform == "win32": |
| 351 | + @unittest.expectedFailure |
| 352 | + def testSetBufferSize(self): |
| 353 | + super().testSetBufferSize() |
| 354 | + |
| 355 | + # TODO: RUSTPYTHON |
| 356 | + if sys.platform == "win32": |
| 357 | + @unittest.expectedFailure |
| 358 | + def testTruncateOnWindows(self): |
| 359 | + super().testTruncateOnWindows() |
| 360 | + |
| 361 | + |
| 362 | +if __name__ == '__main__': |
| 363 | + unittest.main() |
0 commit comments