Lib/test/test_fileio.py

   1 # Adapted from test_file.py by Daniel Stutzbach
   2
   3 from __future__ import unicode_literals
   4
   5 import sys
   6 import os
   7 import errno
   8 import unittest
   9 from array import array
  10 from weakref import proxy
  11 from functools import wraps
  12
  13 from test.test_support import (TESTFN, findfile, check_warnings, run_unittest,
  14                                make_bad_fd)
  15 from test.test_support import py3k_bytes as bytes
  16
  17 from _io import FileIO as _FileIO
  18
  19 class AutoFileTests(unittest.TestCase):
  20     # file tests for which a test file is automatically set up
  21
  22     def setUp(self):
  23         self.f = _FileIO(TESTFN, 'w')
  24
  25     def tearDown(self):
  26         if self.f:
  27             self.f.close()
  28         os.remove(TESTFN)
  29
  30     def testWeakRefs(self):
  31         # verify weak references
  32         p = proxy(self.f)
  33         p.write(bytes(range(10)))
  34         self.assertEquals(self.f.tell(), p.tell())
  35         self.f.close()
  36         self.f = None
  37         self.assertRaises(ReferenceError, getattr, p, 'tell')
  38
  39     def testSeekTell(self):
  40         self.f.write(bytes(range(20)))
  41         self.assertEquals(self.f.tell(), 20)
  42         self.f.seek(0)
  43         self.assertEquals(self.f.tell(), 0)
  44         self.f.seek(10)
  45         self.assertEquals(self.f.tell(), 10)
  46         self.f.seek(5, 1)
  47         self.assertEquals(self.f.tell(), 15)
  48         self.f.seek(-5, 1)
  49         self.assertEquals(self.f.tell(), 10)
  50         self.f.seek(-5, 2)
  51         self.assertEquals(self.f.tell(), 15)
  52
  53     def testAttributes(self):
  54         # verify expected attributes exist
  55         f = self.f
  56
  57         self.assertEquals(f.mode, "wb")
  58         self.assertEquals(f.closed, False)
  59
  60         # verify the attributes are readonly
  61         for attr in 'mode', 'closed':
  62             self.assertRaises((AttributeError, TypeError),
  63                               setattr, f, attr, 'oops')
  64
  65     def testReadinto(self):
  66         # verify readinto
  67         self.f.write(b"\x01\x02")
  68         self.f.close()
  69         a = array(b'b', b'x'*10)
  70         self.f = _FileIO(TESTFN, 'r')
  71         n = self.f.readinto(a)
  72         self.assertEquals(array(b'b', [1, 2]), a[:n])
  73
  74     def testRepr(self):
  75         self.assertEquals(repr(self.f), "<_io.FileIO name=%r mode='%s'>"
  76                                         % (self.f.name, self.f.mode))
  77         del self.f.name
  78         self.assertEquals(repr(self.f), "<_io.FileIO fd=%r mode='%s'>"
  79                                         % (self.f.fileno(), self.f.mode))
  80         self.f.close()
  81         self.assertEquals(repr(self.f), "<_io.FileIO [closed]>")
  82
  83     def testErrors(self):
  84         f = self.f
  85         self.assertTrue(not f.isatty())
  86         self.assertTrue(not f.closed)
  87         #self.assertEquals(f.name, TESTFN)
  88         self.assertRaises(ValueError, f.read, 10) # Open for reading
  89         f.close()
  90         self.assertTrue(f.closed)
  91         f = _FileIO(TESTFN, 'r')
  92         self.assertRaises(TypeError, f.readinto, "")
  93         self.assertTrue(not f.closed)
  94         f.close()
  95         self.assertTrue(f.closed)
  96
  97     def testMethods(self):
  98         methods = ['fileno', 'isatty', 'read', 'readinto',
  99                    'seek', 'tell', 'truncate', 'write', 'seekable',
 100                    'readable', 'writable']
 101         if sys.platform.startswith('atheos'):
 102             methods.remove('truncate')
 103
 104         self.f.close()
 105         self.assertTrue(self.f.closed)
 106
 107         for methodname in methods:
 108             method = getattr(self.f, methodname)
 109             # should raise on closed file
 110             self.assertRaises(ValueError, method)
 111
 112     def testOpendir(self):
 113         # Issue 3703: opening a directory should fill the errno
 114         # Windows always returns "[Errno 13]: Permission denied
 115         # Unix calls dircheck() and returns "[Errno 21]: Is a directory"
 116         try:
 117             _FileIO('.', 'r')
 118         except IOError as e:
 119             self.assertNotEqual(e.errno, 0)
 120             self.assertEqual(e.filename, ".")
 121         else:
 122             self.fail("Should have raised IOError")
 123
 124     #A set of functions testing that we get expected behaviour if someone has
 125     #manually closed the internal file descriptor.  First, a decorator:
 126     def ClosedFD(func):
 127         @wraps(func)
 128         def wrapper(self):
 129             #forcibly close the fd before invoking the problem function
 130             f = self.f
 131             os.close(f.fileno())
 132             try:
 133                 func(self, f)
 134             finally:
 135                 try:
 136                     self.f.close()
 137                 except IOError:
 138                     pass
 139         return wrapper
 140
 141     def ClosedFDRaises(func):
 142         @wraps(func)
 143         def wrapper(self):
 144             #forcibly close the fd before invoking the problem function
 145             f = self.f
 146             os.close(f.fileno())
 147             try:
 148                 func(self, f)
 149             except IOError as e:
 150                 self.assertEqual(e.errno, errno.EBADF)
 151             else:
 152                 self.fail("Should have raised IOError")
 153             finally:
 154                 try:
 155                     self.f.close()
 156                 except IOError:
 157                     pass
 158         return wrapper
 159
 160     @ClosedFDRaises
 161     def testErrnoOnClose(self, f):
 162         f.close()
 163
 164     @ClosedFDRaises
 165     def testErrnoOnClosedWrite(self, f):
 166         f.write('a')
 167
 168     @ClosedFDRaises
 169     def testErrnoOnClosedSeek(self, f):
 170         f.seek(0)
 171
 172     @ClosedFDRaises
 173     def testErrnoOnClosedTell(self, f):
 174         f.tell()
 175
 176     @ClosedFDRaises
 177     def testErrnoOnClosedTruncate(self, f):
 178         f.truncate(0)
 179
 180     @ClosedFD
 181     def testErrnoOnClosedSeekable(self, f):
 182         f.seekable()
 183
 184     @ClosedFD
 185     def testErrnoOnClosedReadable(self, f):
 186         f.readable()
 187
 188     @ClosedFD
 189     def testErrnoOnClosedWritable(self, f):
 190         f.writable()
 191
 192     @ClosedFD
 193     def testErrnoOnClosedFileno(self, f):
 194         f.fileno()
 195
 196     @ClosedFD
 197     def testErrnoOnClosedIsatty(self, f):
 198         self.assertEqual(f.isatty(), False)
 199
 200     def ReopenForRead(self):
 201         try:
 202             self.f.close()
 203         except IOError:
 204             pass
 205         self.f = _FileIO(TESTFN, 'r')
 206         os.close(self.f.fileno())
 207         return self.f
 208
 209     @ClosedFDRaises
 210     def testErrnoOnClosedRead(self, f):
 211         f = self.ReopenForRead()
 212         f.read(1)
 213
 214     @ClosedFDRaises
 215     def testErrnoOnClosedReadall(self, f):
 216         f = self.ReopenForRead()
 217         f.readall()
 218
 219     @ClosedFDRaises
 220     def testErrnoOnClosedReadinto(self, f):
 221         f = self.ReopenForRead()
 222         a = array(b'b', b'x'*10)
 223         f.readinto(a)
 224
 225 class OtherFileTests(unittest.TestCase):
 226
 227     def testAbles(self):
 228         try:
 229             f = _FileIO(TESTFN, "w")
 230             self.assertEquals(f.readable(), False)
 231             self.assertEquals(f.writable(), True)
 232             self.assertEquals(f.seekable(), True)
 233             f.close()
 234
 235             f = _FileIO(TESTFN, "r")
 236             self.assertEquals(f.readable(), True)
 237             self.assertEquals(f.writable(), False)
 238             self.assertEquals(f.seekable(), True)
 239             f.close()
 240
 241             f = _FileIO(TESTFN, "a+")
 242             self.assertEquals(f.readable(), True)
 243             self.assertEquals(f.writable(), True)
 244             self.assertEquals(f.seekable(), True)
 245             self.assertEquals(f.isatty(), False)
 246             f.close()
 247
 248             if sys.platform != "win32":
 249                 try:
 250                     f = _FileIO("/dev/tty", "a")
 251                 except EnvironmentError:
 252                     # When run in a cron job there just aren't any
 253                     # ttys, so skip the test.  This also handles other
 254                     # OS'es that don't support /dev/tty.
 255                     pass
 256                 else:
 257                     f = _FileIO("/dev/tty", "a")
 258                     self.assertEquals(f.readable(), False)
 259                     self.assertEquals(f.writable(), True)
 260                     if sys.platform != "darwin" and \
 261                        'bsd' not in sys.platform and \
 262                        not sys.platform.startswith('sunos'):
 263                         # Somehow /dev/tty appears seekable on some BSDs
 264                         self.assertEquals(f.seekable(), False)
 265                     self.assertEquals(f.isatty(), True)
 266                     f.close()
 267         finally:
 268             os.unlink(TESTFN)
 269
 270     def testModeStrings(self):
 271         # check invalid mode strings
 272         for mode in ("", "aU", "wU+", "rw", "rt"):
 273             try:
 274                 f = _FileIO(TESTFN, mode)
 275             except ValueError:
 276                 pass
 277             else:
 278                 f.close()
 279                 self.fail('%r is an invalid file mode' % mode)
 280
 281     def testUnicodeOpen(self):
 282         # verify repr works for unicode too
 283         f = _FileIO(str(TESTFN), "w")
 284         f.close()
 285         os.unlink(TESTFN)
 286
 287     def testBytesOpen(self):
 288         # Opening a bytes filename
 289         try:
 290             fn = TESTFN.encode("ascii")
 291         except UnicodeEncodeError:
 292             # Skip test
 293             return
 294         f = _FileIO(fn, "w")
 295         try:
 296             f.write(b"abc")
 297             f.close()
 298             with open(TESTFN, "rb") as f:
 299                 self.assertEquals(f.read(), b"abc")
 300         finally:
 301             os.unlink(TESTFN)
 302
 303     def testInvalidFd(self):
 304         self.assertRaises(ValueError, _FileIO, -10)
 305         self.assertRaises(OSError, _FileIO, make_bad_fd())
 306
 307     def testBadModeArgument(self):
 308         # verify that we get a sensible error message for bad mode argument
 309         bad_mode = "qwerty"
 310         try:
 311             f = _FileIO(TESTFN, bad_mode)
 312         except ValueError as msg:
 313             if msg.args[0] != 0:
 314                 s = str(msg)
 315                 if s.find(TESTFN) != -1 or s.find(bad_mode) == -1:
 316                     self.fail("bad error message for invalid mode: %s" % s)
 317             # if msg.args[0] == 0, we're probably on Windows where there may be
 318             # no obvious way to discover why open() failed.
 319         else:
 320             f.close()
 321             self.fail("no error for invalid mode: %s" % bad_mode)
 322
 323     def testTruncateOnWindows(self):
 324         def bug801631():
 325             # SF bug <http://www.python.org/sf/801631>
 326             # "file.truncate fault on windows"
 327             f = _FileIO(TESTFN, 'w')
 328             f.write(bytes(range(11)))
 329             f.close()
 330
 331             f = _FileIO(TESTFN,'r+')
 332             data = f.read(5)
 333             if data != bytes(range(5)):
 334                 self.fail("Read on file opened for update failed %r" % data)
 335             if f.tell() != 5:
 336                 self.fail("File pos after read wrong %d" % f.tell())
 337
 338             f.truncate()
 339             if f.tell() != 5:
 340                 self.fail("File pos after ftruncate wrong %d" % f.tell())
 341
 342             f.close()
 343             size = os.path.getsize(TESTFN)
 344             if size != 5:
 345                 self.fail("File size after ftruncate wrong %d" % size)
 346
 347         try:
 348             bug801631()
 349         finally:
 350             os.unlink(TESTFN)
 351
 352     def testAppend(self):
 353         try:
 354             f = open(TESTFN, 'wb')
 355             f.write(b'spam')
 356             f.close()
 357             f = open(TESTFN, 'ab')
 358             f.write(b'eggs')
 359             f.close()
 360             f = open(TESTFN, 'rb')
 361             d = f.read()
 362             f.close()
 363             self.assertEqual(d, b'spameggs')
 364         finally:
 365             try:
 366                 os.unlink(TESTFN)
 367             except:
 368                 pass
 369
 370     def testInvalidInit(self):
 371         self.assertRaises(TypeError, _FileIO, "1", 0, 0)
 372
 373     def testWarnings(self):
 374         with check_warnings() as w:
 375             self.assertEqual(w.warnings, [])
 376             self.assertRaises(TypeError, _FileIO, [])
 377             self.assertEqual(w.warnings, [])
 378             self.assertRaises(ValueError, _FileIO, "/some/invalid/name", "rt")
 379             self.assertEqual(w.warnings, [])
 380
 381
 382 def test_main():
 383     # Historically, these tests have been sloppy about removing TESTFN.
 384     # So get rid of it no matter what.
 385     try:
 386         run_unittest(AutoFileTests, OtherFileTests)
 387     finally:
 388         if os.path.exists(TESTFN):
 389             os.unlink(TESTFN)
 390
 391 if __name__ == '__main__':
 392     test_main()