Antoine Pitrou | c5d2b41 | 2009-06-12 20:36:25 +0000 | [diff] [blame^] | 1 | from __future__ import print_function |
| 2 | |
| 3 | import sys |
| 4 | import os |
| 5 | import unittest |
| 6 | from array import array |
| 7 | from weakref import proxy |
| 8 | |
| 9 | import io |
| 10 | import _pyio as pyio |
| 11 | |
| 12 | from test.test_support import TESTFN, findfile, run_unittest |
| 13 | from UserList import UserList |
| 14 | |
| 15 | class AutoFileTests(unittest.TestCase): |
| 16 | # file tests for which a test file is automatically set up |
| 17 | |
| 18 | def setUp(self): |
| 19 | self.f = self.open(TESTFN, 'wb') |
| 20 | |
| 21 | def tearDown(self): |
| 22 | if self.f: |
| 23 | self.f.close() |
| 24 | os.remove(TESTFN) |
| 25 | |
| 26 | def testWeakRefs(self): |
| 27 | # verify weak references |
| 28 | p = proxy(self.f) |
| 29 | p.write(b'teststring') |
| 30 | self.assertEquals(self.f.tell(), p.tell()) |
| 31 | self.f.close() |
| 32 | self.f = None |
| 33 | self.assertRaises(ReferenceError, getattr, p, 'tell') |
| 34 | |
| 35 | def testAttributes(self): |
| 36 | # verify expected attributes exist |
| 37 | f = self.f |
| 38 | f.name # merely shouldn't blow up |
| 39 | f.mode # ditto |
| 40 | f.closed # ditto |
| 41 | |
| 42 | def testReadinto(self): |
| 43 | # verify readinto |
| 44 | self.f.write(b'12') |
| 45 | self.f.close() |
| 46 | a = array('b', b'x'*10) |
| 47 | self.f = self.open(TESTFN, 'rb') |
| 48 | n = self.f.readinto(a) |
| 49 | self.assertEquals(b'12', a.tostring()[:n]) |
| 50 | |
| 51 | def testReadinto_text(self): |
| 52 | # verify readinto refuses text files |
| 53 | a = array('b', b'x'*10) |
| 54 | self.f.close() |
| 55 | self.f = self.open(TESTFN, 'r') |
| 56 | if hasattr(self.f, "readinto"): |
| 57 | self.assertRaises(TypeError, self.f.readinto, a) |
| 58 | |
| 59 | def testWritelinesUserList(self): |
| 60 | # verify writelines with instance sequence |
| 61 | l = UserList([b'1', b'2']) |
| 62 | self.f.writelines(l) |
| 63 | self.f.close() |
| 64 | self.f = self.open(TESTFN, 'rb') |
| 65 | buf = self.f.read() |
| 66 | self.assertEquals(buf, b'12') |
| 67 | |
| 68 | def testWritelinesIntegers(self): |
| 69 | # verify writelines with integers |
| 70 | self.assertRaises(TypeError, self.f.writelines, [1, 2, 3]) |
| 71 | |
| 72 | def testWritelinesIntegersUserList(self): |
| 73 | # verify writelines with integers in UserList |
| 74 | l = UserList([1,2,3]) |
| 75 | self.assertRaises(TypeError, self.f.writelines, l) |
| 76 | |
| 77 | def testWritelinesNonString(self): |
| 78 | # verify writelines with non-string object |
| 79 | class NonString: |
| 80 | pass |
| 81 | |
| 82 | self.assertRaises(TypeError, self.f.writelines, |
| 83 | [NonString(), NonString()]) |
| 84 | |
| 85 | def testErrors(self): |
| 86 | f = self.f |
| 87 | self.assertEquals(f.name, TESTFN) |
| 88 | self.assert_(not f.isatty()) |
| 89 | self.assert_(not f.closed) |
| 90 | |
| 91 | if hasattr(f, "readinto"): |
| 92 | self.assertRaises((IOError, TypeError), f.readinto, "") |
| 93 | f.close() |
| 94 | self.assert_(f.closed) |
| 95 | |
| 96 | def testMethods(self): |
| 97 | methods = [('fileno', ()), |
| 98 | ('flush', ()), |
| 99 | ('isatty', ()), |
| 100 | ('next', ()), |
| 101 | ('read', ()), |
| 102 | ('write', (b"",)), |
| 103 | ('readline', ()), |
| 104 | ('readlines', ()), |
| 105 | ('seek', (0,)), |
| 106 | ('tell', ()), |
| 107 | ('write', (b"",)), |
| 108 | ('writelines', ([],)), |
| 109 | ('__iter__', ()), |
| 110 | ] |
| 111 | if not sys.platform.startswith('atheos'): |
| 112 | methods.append(('truncate', ())) |
| 113 | |
| 114 | # __exit__ should close the file |
| 115 | self.f.__exit__(None, None, None) |
| 116 | self.assert_(self.f.closed) |
| 117 | |
| 118 | for methodname, args in methods: |
| 119 | method = getattr(self.f, methodname) |
| 120 | # should raise on closed file |
| 121 | self.assertRaises(ValueError, method, *args) |
| 122 | |
| 123 | # file is closed, __exit__ shouldn't do anything |
| 124 | self.assertEquals(self.f.__exit__(None, None, None), None) |
| 125 | # it must also return None if an exception was given |
| 126 | try: |
| 127 | 1/0 |
| 128 | except: |
| 129 | self.assertEquals(self.f.__exit__(*sys.exc_info()), None) |
| 130 | |
| 131 | def testReadWhenWriting(self): |
| 132 | self.assertRaises(IOError, self.f.read) |
| 133 | |
| 134 | class CAutoFileTests(AutoFileTests): |
| 135 | open = io.open |
| 136 | |
| 137 | class PyAutoFileTests(AutoFileTests): |
| 138 | open = staticmethod(pyio.open) |
| 139 | |
| 140 | |
| 141 | class OtherFileTests(unittest.TestCase): |
| 142 | |
| 143 | def testModeStrings(self): |
| 144 | # check invalid mode strings |
| 145 | for mode in ("", "aU", "wU+"): |
| 146 | try: |
| 147 | f = self.open(TESTFN, mode) |
| 148 | except ValueError: |
| 149 | pass |
| 150 | else: |
| 151 | f.close() |
| 152 | self.fail('%r is an invalid file mode' % mode) |
| 153 | |
| 154 | def testStdin(self): |
| 155 | # This causes the interpreter to exit on OSF1 v5.1. |
| 156 | if sys.platform != 'osf1V5': |
| 157 | self.assertRaises((IOError, ValueError), sys.stdin.seek, -1) |
| 158 | else: |
| 159 | print(( |
| 160 | ' Skipping sys.stdin.seek(-1), it may crash the interpreter.' |
| 161 | ' Test manually.'), file=sys.__stdout__) |
| 162 | self.assertRaises((IOError, ValueError), sys.stdin.truncate) |
| 163 | |
| 164 | def testBadModeArgument(self): |
| 165 | # verify that we get a sensible error message for bad mode argument |
| 166 | bad_mode = "qwerty" |
| 167 | try: |
| 168 | f = self.open(TESTFN, bad_mode) |
| 169 | except ValueError as msg: |
| 170 | if msg.args[0] != 0: |
| 171 | s = str(msg) |
| 172 | if s.find(TESTFN) != -1 or s.find(bad_mode) == -1: |
| 173 | self.fail("bad error message for invalid mode: %s" % s) |
| 174 | # if msg.args[0] == 0, we're probably on Windows where there may be |
| 175 | # no obvious way to discover why open() failed. |
| 176 | else: |
| 177 | f.close() |
| 178 | self.fail("no error for invalid mode: %s" % bad_mode) |
| 179 | |
| 180 | def testSetBufferSize(self): |
| 181 | # make sure that explicitly setting the buffer size doesn't cause |
| 182 | # misbehaviour especially with repeated close() calls |
| 183 | for s in (-1, 0, 1, 512): |
| 184 | try: |
| 185 | f = self.open(TESTFN, 'wb', s) |
| 186 | f.write(str(s).encode("ascii")) |
| 187 | f.close() |
| 188 | f.close() |
| 189 | f = self.open(TESTFN, 'rb', s) |
| 190 | d = int(f.read().decode("ascii")) |
| 191 | f.close() |
| 192 | f.close() |
| 193 | except IOError as msg: |
| 194 | self.fail('error setting buffer size %d: %s' % (s, str(msg))) |
| 195 | self.assertEquals(d, s) |
| 196 | |
| 197 | def testTruncateOnWindows(self): |
| 198 | # SF bug <http://www.python.org/sf/801631> |
| 199 | # "file.truncate fault on windows" |
| 200 | |
| 201 | os.unlink(TESTFN) |
| 202 | f = self.open(TESTFN, 'wb') |
| 203 | |
| 204 | try: |
| 205 | f.write(b'12345678901') # 11 bytes |
| 206 | f.close() |
| 207 | |
| 208 | f = self.open(TESTFN,'rb+') |
| 209 | data = f.read(5) |
| 210 | if data != b'12345': |
| 211 | self.fail("Read on file opened for update failed %r" % data) |
| 212 | if f.tell() != 5: |
| 213 | self.fail("File pos after read wrong %d" % f.tell()) |
| 214 | |
| 215 | f.truncate() |
| 216 | if f.tell() != 5: |
| 217 | self.fail("File pos after ftruncate wrong %d" % f.tell()) |
| 218 | |
| 219 | f.close() |
| 220 | size = os.path.getsize(TESTFN) |
| 221 | if size != 5: |
| 222 | self.fail("File size after ftruncate wrong %d" % size) |
| 223 | finally: |
| 224 | f.close() |
| 225 | os.unlink(TESTFN) |
| 226 | |
| 227 | def testIteration(self): |
| 228 | # Test the complex interaction when mixing file-iteration and the |
| 229 | # various read* methods. |
| 230 | dataoffset = 16384 |
| 231 | filler = b"ham\n" |
| 232 | assert not dataoffset % len(filler), \ |
| 233 | "dataoffset must be multiple of len(filler)" |
| 234 | nchunks = dataoffset // len(filler) |
| 235 | testlines = [ |
| 236 | b"spam, spam and eggs\n", |
| 237 | b"eggs, spam, ham and spam\n", |
| 238 | b"saussages, spam, spam and eggs\n", |
| 239 | b"spam, ham, spam and eggs\n", |
| 240 | b"spam, spam, spam, spam, spam, ham, spam\n", |
| 241 | b"wonderful spaaaaaam.\n" |
| 242 | ] |
| 243 | methods = [("readline", ()), ("read", ()), ("readlines", ()), |
| 244 | ("readinto", (array("b", b" "*100),))] |
| 245 | |
| 246 | try: |
| 247 | # Prepare the testfile |
| 248 | bag = self.open(TESTFN, "wb") |
| 249 | bag.write(filler * nchunks) |
| 250 | bag.writelines(testlines) |
| 251 | bag.close() |
| 252 | # Test for appropriate errors mixing read* and iteration |
| 253 | for methodname, args in methods: |
| 254 | f = self.open(TESTFN, 'rb') |
| 255 | if next(f) != filler: |
| 256 | self.fail, "Broken testfile" |
| 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.tostring() |
| 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 | # Reading after iteration hit EOF shouldn't hurt either |
| 310 | f = self.open(TESTFN, 'rb') |
| 311 | try: |
| 312 | for line in f: |
| 313 | pass |
| 314 | try: |
| 315 | f.readline() |
| 316 | f.readinto(buf) |
| 317 | f.read() |
| 318 | f.readlines() |
| 319 | except ValueError: |
| 320 | self.fail("read* failed after next() consumed file") |
| 321 | finally: |
| 322 | f.close() |
| 323 | finally: |
| 324 | os.unlink(TESTFN) |
| 325 | |
| 326 | class COtherFileTests(OtherFileTests): |
| 327 | open = io.open |
| 328 | |
| 329 | class PyOtherFileTests(OtherFileTests): |
| 330 | open = staticmethod(pyio.open) |
| 331 | |
| 332 | |
| 333 | def test_main(): |
| 334 | # Historically, these tests have been sloppy about removing TESTFN. |
| 335 | # So get rid of it no matter what. |
| 336 | try: |
| 337 | run_unittest(CAutoFileTests, PyAutoFileTests, |
| 338 | COtherFileTests, PyOtherFileTests) |
| 339 | finally: |
| 340 | if os.path.exists(TESTFN): |
| 341 | os.unlink(TESTFN) |
| 342 | |
| 343 | if __name__ == '__main__': |
| 344 | test_main() |