--- /dev/null
+test_largefile
-s: single -- run only a single test (see below)
-r: random -- randomize test execution order
-l: leakdebug -- if cycle garbage collection is enabled, run with DEBUG_LEAK
+--have-resources -- run tests that require large resources (time/space)
If non-option arguments are present, they are names for tests to run,
unless -x is given, in which case they are names for tests not to run.
import test_support
def main(tests=None, testdir=None, verbose=0, quiet=0, generate=0,
- exclude=0, single=0, randomize=0, leakdebug=0):
+ exclude=0, single=0, randomize=0, leakdebug=0,
+ use_large_resources=0):
"""Execute a test suite.
This also parses command-line options and modifies its behavior
"""
try:
- opts, args = getopt.getopt(sys.argv[1:], 'vgqxsrl')
+ opts, args = getopt.getopt(sys.argv[1:], 'vgqxsrl', ['have-resources'])
except getopt.error, msg:
print msg
print __doc__
if o == '-s': single = 1
if o == '-r': randomize = 1
if o == '-l': leakdebug = 1
+ if o == '--have-resources': use_large_resources = 1
if generate and verbose:
print "-g and -v don't go together!"
return 2
if randomize:
random.shuffle(tests)
test_support.verbose = verbose # Tell tests to be moderately quiet
+ test_support.use_large_resources = use_large_resources
save_modules = sys.modules.keys()
for test in tests:
if not quiet:
print test
- ok = runtest(test, generate, verbose, testdir)
+ ok = runtest(test, generate, verbose, quiet, testdir)
if ok > 0:
good.append(test)
elif ok == 0:
bad.append(test)
else:
- if not quiet:
- print "test", test,
- print "skipped -- an optional feature could not be imported"
skipped.append(test)
# Unload the newly imported modules (best effort finalization)
for module in sys.modules.keys():
tests.sort()
return stdtests + tests
-def runtest(test, generate, verbose, testdir = None):
+def runtest(test, generate, verbose, quiet, testdir = None):
"""Run a single test.
test -- the name of the test
generate -- if true, generate output, instead of running the test
and comparing it to a previously created output file
verbose -- if true, print more messages
+ quiet -- if true, don't print 'skipped' messages (probably redundant)
testdir -- test directory
"""
test_support.unload(test)
finally:
sys.stdout = save_stdout
except (ImportError, test_support.TestSkipped), msg:
+ if not quiet:
+ print "test", test,
+ print "skipped -- ", msg
return -1
except KeyboardInterrupt, v:
raise KeyboardInterrupt, v, sys.exc_info()[2]
--- /dev/null
+#!python
+
+#----------------------------------------------------------------------
+# test largefile support on system where this makes sense
+#
+#XXX how to only run this when support is there
+#XXX how to only optionally run this, it will take along time
+#----------------------------------------------------------------------
+
+import test_support
+import os, struct, stat, sys
+
+
+# only run if the current system support large files
+f = open(test_support.TESTFN, 'w')
+try:
+ # 2**31 == 2147483648
+ f.seek(2147483649L)
+except OverflowError:
+ raise test_support.TestSkipped, "platform does not have largefile support"
+else:
+ f.close()
+
+
+# create >2GB file (2GB = 2147483648 bytes)
+size = 2500000000L
+name = test_support.TESTFN
+
+
+# on Windows this test comsumes large resources:
+# it takes a long time to build the >2GB file and takes >2GB of disk space
+# therefore test_support.use_large_resources must be defined to run this test
+if sys.platform[:3] == 'win' and not test_support.use_large_resources:
+ raise test_support.TestSkipped, \
+ "test requires %s bytes and a long time to run" % str(size)
+
+
+
+def expect(got_this, expect_this):
+ if test_support.verbose:
+ print '%s =?= %s ...' % (`got_this`, `expect_this`),
+ if got_this != expect_this:
+ if test_support.verbose:
+ print 'no'
+ raise test_support.TestFailed, 'got %s, but expected %s' %\
+ (str(got_this), str(expect_this))
+ else:
+ if test_support.verbose:
+ print 'yes'
+
+
+# test that each file function works as expected for a large (i.e. >2GB, do
+# we have to check >4GB) files
+
+if test_support.verbose:
+ print 'create large file via seek (may be sparse file) ...'
+f = open(name, 'w')
+f.seek(size)
+f.write('a')
+f.flush()
+expect(os.fstat(f.fileno())[stat.ST_SIZE], size+1)
+if test_support.verbose:
+ print 'check file size with os.fstat'
+f.close()
+if test_support.verbose:
+ print 'check file size with os.stat'
+expect(os.stat(name)[stat.ST_SIZE], size+1)
+
+if test_support.verbose:
+ print 'play around with seek() and read() with the built largefile'
+f = open(name, 'r')
+expect(f.tell(), 0)
+expect(f.read(1), '\000')
+expect(f.tell(), 1)
+f.seek(0)
+expect(f.tell(), 0)
+f.seek(0, 0)
+expect(f.tell(), 0)
+f.seek(42)
+expect(f.tell(), 42)
+f.seek(42, 0)
+expect(f.tell(), 42)
+f.seek(42, 1)
+expect(f.tell(), 84)
+f.seek(0, 1)
+expect(f.tell(), 84)
+f.seek(0, 2) # seek from the end
+expect(f.tell(), size + 1 + 0)
+f.seek(-10, 2)
+expect(f.tell(), size + 1 - 10)
+f.seek(-size-1, 2)
+expect(f.tell(), 0)
+f.seek(size)
+expect(f.tell(), size)
+expect(f.read(1), 'a') # the 'a' that was written at the end of the file above
+f.close()
+
+if test_support.verbose:
+ print 'play around with os.lseek() with the built largefile'
+f = open(name, 'r')
+expect(os.lseek(f.fileno(), 0, 0), 0)
+expect(os.lseek(f.fileno(), 42, 0), 42)
+expect(os.lseek(f.fileno(), 42, 1), 84)
+expect(os.lseek(f.fileno(), 0, 1), 84)
+expect(os.lseek(f.fileno(), 0, 2), size+1+0)
+expect(os.lseek(f.fileno(), -10, 2), size+1-10)
+expect(os.lseek(f.fileno(), -size-1, 2), 0)
+expect(os.lseek(f.fileno(), size, 0), size)
+expect(f.read(1), 'a') # the 'a' that was written at the end of the file above
+f.close()
+
+
+# XXX add tests for truncate if it exists
+# XXX has truncate ever worked on Windows? specifically on WinNT I get:
+# "IOError: [Errno 13] Permission denied"
+##try:
+## newsize = size - 10
+## f.seek(newsize)
+## f.truncate()
+## expect(f.tell(), newsize)
+## newsize = newsize - 1
+## f.seek(0)
+## f.truncate(newsize)
+## expect(f.tell(), newsize)
+##except AttributeError:
+## pass
+
+os.unlink(name)
+
#endif
#ifdef MS_WIN32
-#define ftruncate _chsize
#define fileno _fileno
+/* can (almost fully) duplicate with _chsize, see file_truncate */
#define HAVE_FTRUNCATE
#endif
#include <errno.h>
#endif
+/* define the appropriate 64-bit capable tell() function */
+#ifdef MS_WIN64
+# define TELL64 _telli64
+#endif
+
+
typedef struct {
PyObject_HEAD
FILE *f_fp;
return Py_None;
}
+
+/* a portable fseek() function
+ return 0 on success, non-zero on failure (with errno set) */
+int
+_portable_fseek(fp, offset, whence)
+ FILE* fp;
+#if defined(HAVE_LARGEFILE_SUPPORT) && SIZEOF_OFF_T < 8 && SIZEOF_FPOS_T >= 8
+ fpos_t offset;
+#else
+ off_t offset;
+#endif
+ int whence;
+{
+#if defined(HAVE_FSEEKO)
+ return fseeko(fp, offset, whence);
+#elif defined(HAVE_FSEEK64)
+ return fseek64(fp, offset, whence);
+#elif defined(HAVE_LARGEFILE_SUPPORT) && SIZEOF_FPOS_T >= 8
+ /* lacking a 64-bit capable fseek() (as Win64 does) use a 64-bit capable
+ fsetpos() and tell() to implement fseek()*/
+ fpos_t pos;
+ switch (whence) {
+ case SEEK_CUR:
+ if (fgetpos(fp, &pos) != 0)
+ return -1;
+ offset += pos;
+ break;
+ case SEEK_END:
+ /* do a "no-op" seek first to sync the buffering so that
+ the low-level tell() can be used correctly */
+ if (fseek(fp, 0, SEEK_END) != 0)
+ return -1;
+ if ((pos = TELL64(fileno(fp))) == -1L)
+ return -1;
+ offset += pos;
+ break;
+ /* case SEEK_SET: break; */
+ }
+ return fsetpos(fp, &offset);
+#else
+ return fseek(fp, offset, whence);
+#endif
+}
+
+
+/* a portable ftell() function
+ Return -1 on failure with errno set appropriately, current file
+ position on success */
+#if defined(HAVE_LARGEFILE_SUPPORT) && SIZEOF_OFF_T < 8 && SIZEOF_FPOS_T >= 8
+fpos_t
+#else
+off_t
+#endif
+_portable_ftell(fp)
+ FILE* fp;
+{
+#if defined(HAVE_FTELLO) && defined(HAVE_LARGEFILE_SUPPORT)
+ return ftello(fp);
+#elif defined(HAVE_FTELL64) && defined(HAVE_LARGEFILE_SUPPORT)
+ return ftell64(fp);
+#elif SIZEOF_FPOS_T >= 8 && defined(HAVE_LARGEFILE_SUPPORT)
+ fpos_t pos;
+ if (fgetpos(fp, &pos) != 0)
+ return -1;
+ return pos;
+#else
+ return ftell(fp);
+#endif
+}
+
+
static PyObject *
file_seek(PyFileObject *f, PyObject *args)
{
int whence;
int ret;
+#if defined(HAVE_LARGEFILE_SUPPORT) && SIZEOF_OFF_T < 8 && SIZEOF_FPOS_T >= 8
+ fpos_t offset, pos;
+#else
off_t offset;
+#endif /* !MS_WIN64 */
PyObject *offobj;
if (f->f_fp == NULL)
#endif
if (PyErr_Occurred())
return NULL;
+
Py_BEGIN_ALLOW_THREADS
errno = 0;
-#if defined(HAVE_FSEEKO)
- ret = fseeko(f->f_fp, offset, whence);
-#elif defined(HAVE_FSEEK64)
- ret = fseek64(f->f_fp, offset, whence);
-#else
- ret = fseek(f->f_fp, offset, whence);
-#endif
+ ret = _portable_fseek(f->f_fp, offset, whence);
Py_END_ALLOW_THREADS
+
if (ret != 0) {
PyErr_SetFromErrno(PyExc_IOError);
clearerr(f->f_fp);
return Py_None;
}
+
#ifdef HAVE_FTRUNCATE
static PyObject *
file_truncate(PyFileObject *f, PyObject *args)
{
int ret;
+#if defined(HAVE_LARGEFILE_SUPPORT) && SIZEOF_OFF_T < 8 && SIZEOF_FPOS_T >= 8
+ fpos_t newsize;
+#else
off_t newsize;
+#endif
PyObject *newsizeobj;
if (f->f_fp == NULL)
/* Default to current position*/
Py_BEGIN_ALLOW_THREADS
errno = 0;
-#if defined(HAVE_FTELLO) && defined(HAVE_LARGEFILE_SUPPORT)
- newsize = ftello(f->f_fp);
-#elif defined(HAVE_FTELL64) && defined(HAVE_LARGEFILE_SUPPORT)
- newsize = ftell64(f->f_fp);
-#else
- newsize = ftell(f->f_fp);
-#endif
+ newsize = _portable_ftell(f->f_fp);
Py_END_ALLOW_THREADS
if (newsize == -1) {
PyErr_SetFromErrno(PyExc_IOError);
errno = 0;
ret = fflush(f->f_fp);
Py_END_ALLOW_THREADS
- if (ret == 0) {
- Py_BEGIN_ALLOW_THREADS
+ if (ret != 0) goto onioerror;
+
+#ifdef MS_WIN32
+ /* can use _chsize; if, however, the newsize overflows 32-bits then
+ _chsize is *not* adequate; in this case, an OverflowError is raised */
+ if (newsize > LONG_MAX) {
+ PyErr_SetString(PyExc_OverflowError,
+ "the new size is too long for _chsize (it is limited to 32-bit values)");
+ return NULL;
+ } else {
+ Py_BEGIN_ALLOW_THREADS
errno = 0;
- ret = ftruncate(fileno(f->f_fp), newsize);
+ ret = _chsize(fileno(f->f_fp), newsize);
Py_END_ALLOW_THREADS
+ if (ret != 0) goto onioerror;
}
- if (ret != 0) {
- PyErr_SetFromErrno(PyExc_IOError);
- clearerr(f->f_fp);
- return NULL;
- }
+#else
+ Py_BEGIN_ALLOW_THREADS
+ errno = 0;
+ ret = ftruncate(fileno(f->f_fp), newsize);
+ Py_END_ALLOW_THREADS
+ if (ret != 0) goto onioerror;
+#endif /* !MS_WIN32 */
+
Py_INCREF(Py_None);
return Py_None;
+
+onioerror:
+ PyErr_SetFromErrno(PyExc_IOError);
+ clearerr(f->f_fp);
+ return NULL;
}
#endif /* HAVE_FTRUNCATE */
static PyObject *
file_tell(PyFileObject *f, PyObject *args)
{
- off_t offset;
+#if defined(HAVE_LARGEFILE_SUPPORT) && SIZEOF_OFF_T < 8 && SIZEOF_FPOS_T >= 8
+ fpos_t pos;
+#else
+ off_t pos;
+#endif
+
if (f->f_fp == NULL)
return err_closed();
if (!PyArg_NoArgs(args))
return NULL;
Py_BEGIN_ALLOW_THREADS
errno = 0;
-#if defined(HAVE_FTELLO) && defined(HAVE_LARGEFILE_SUPPORT)
- offset = ftello(f->f_fp);
-#elif defined(HAVE_FTELL64) && defined(HAVE_LARGEFILE_SUPPORT)
- offset = ftell64(f->f_fp);
-#else
- offset = ftell(f->f_fp);
-#endif
+ pos = _portable_ftell(f->f_fp);
Py_END_ALLOW_THREADS
- if (offset == -1) {
+ if (pos == -1) {
PyErr_SetFromErrno(PyExc_IOError);
clearerr(f->f_fp);
return NULL;
}
#if !defined(HAVE_LARGEFILE_SUPPORT)
- return PyInt_FromLong(offset);
+ return PyInt_FromLong(pos);
#else
- return PyLong_FromLongLong(offset);
+ return PyLong_FromLongLong(pos);
#endif
}
buffersize = new_buffersize(f, (size_t)0);
else
buffersize = bytesrequested;
+ if (buffersize > INT_MAX) {
+ PyErr_SetString(PyExc_OverflowError,
+ "requested number of bytes is more than a Python string can hold");
+ return NULL;
+ }
v = PyString_FromStringAndSize((char *)NULL, buffersize);
if (v == NULL)
return NULL;
file_readinto(PyFileObject *f, PyObject *args)
{
char *ptr;
- int ntodo, ndone, nnow;
+ size_t ntodo, ndone, nnow;
if (f->f_fp == NULL)
return err_closed();
ndone += nnow;
ntodo -= nnow;
}
- return PyInt_FromLong(ndone);
+ return PyInt_FromLong((long)ndone);
}
register FILE *fp;
register int c;
register char *buf, *end;
- int n1, n2;
+ size_t n1, n2;
PyObject *v;
fp = f->f_fp;
break;
n1 = n2;
n2 += 1000;
+ if (n2 > INT_MAX) {
+ PyErr_SetString(PyExc_OverflowError,
+ "line is longer than a Python string can hold");
+ return NULL;
+ }
Py_BLOCK_THREADS
if (_PyString_Resize(&v, n2) < 0)
return NULL;
/* Need a larger buffer to fit this line */
nfilled += nread;
buffersize *= 2;
+ if (buffersize > INT_MAX) {
+ PyErr_SetString(PyExc_OverflowError,
+ "line is too long for a Python string");
+ goto error;
+ }
if (big_buffer == NULL) {
/* Create the big buffer */
big_buffer = PyString_FromStringAndSize(
projects / python / commitdiff