Made more extensive usage of context managers with open.

This commit is contained in:
Claude Paroz 2012-05-05 14:01:38 +02:00
parent ec5423df05
commit 865cd35c9b
33 changed files with 178 additions and 233 deletions

View File

@ -22,7 +22,8 @@ def unique_messages():
cmd = 'msguniq "%s.po"' % pf
stdout = os.popen(cmd)
msg = stdout.read()
open('%s.po' % pf, 'w').write(msg)
with open('%s.po' % pf, 'w') as fp:
fp.write(msg)
if __name__ == "__main__":
unique_messages()

View File

@ -120,12 +120,9 @@ class SpatiaLiteCreation(DatabaseCreation):
# Opening up the SpatiaLite SQL initialization file and executing
# as a script.
sql_fh = open(spatialite_sql, 'r')
try:
with open(spatialite_sql, 'r') as sql_fh:
cur = self.connection._cursor()
cur.executescript(sql_fh.read())
finally:
sql_fh.close()
def spatialite_init_file(self):
# SPATIALITE_SQL may be placed in settings to tell GeoDjango

View File

@ -7,10 +7,10 @@ def fromfile(file_h):
"""
# If given a file name, get a real handle.
if isinstance(file_h, basestring):
file_h = open(file_h, 'rb')
# Reading in the file's contents,
buf = file_h.read()
with open(file_h, 'rb') as file_h:
buf = file_h.read()
else:
buf = file_h.read()
# If we get WKB need to wrap in buffer(), so run through regexes.
if wkt_regex.match(buf) or hex_regex.match(buf):

View File

@ -47,18 +47,15 @@ class SessionStore(SessionBase):
def load(self):
session_data = {}
try:
session_file = open(self._key_to_file(), "rb")
try:
with open(self._key_to_file(), "rb") as session_file:
file_data = session_file.read()
# Don't fail if there is no data in the session file.
# We may have opened the empty placeholder file.
if file_data:
try:
session_data = self.decode(file_data)
except (EOFError, SuspiciousOperation):
self.create()
finally:
session_file.close()
# Don't fail if there is no data in the session file.
# We may have opened the empty placeholder file.
if file_data:
try:
session_data = self.decode(file_data)
except (EOFError, SuspiciousOperation):
self.create()
except IOError:
self.create()
return session_data

View File

@ -31,16 +31,13 @@ class FileBasedCache(BaseCache):
fname = self._key_to_file(key)
try:
f = open(fname, 'rb')
try:
with open(fname, 'rb') as f:
exp = pickle.load(f)
now = time.time()
if exp < now:
self._delete(fname)
else:
return pickle.load(f)
finally:
f.close()
except (IOError, OSError, EOFError, pickle.PickleError):
pass
return default
@ -61,13 +58,10 @@ class FileBasedCache(BaseCache):
if not os.path.exists(dirname):
os.makedirs(dirname)
f = open(fname, 'wb')
try:
with open(fname, 'wb') as f:
now = time.time()
pickle.dump(now + timeout, f, pickle.HIGHEST_PROTOCOL)
pickle.dump(value, f, pickle.HIGHEST_PROTOCOL)
finally:
f.close()
except (IOError, OSError):
pass
@ -94,17 +88,14 @@ class FileBasedCache(BaseCache):
self.validate_key(key)
fname = self._key_to_file(key)
try:
f = open(fname, 'rb')
try:
with open(fname, 'rb') as f:
exp = pickle.load(f)
now = time.time()
if exp < now:
self._delete(fname)
return False
else:
return True
finally:
f.close()
now = time.time()
if exp < now:
self._delete(fname)
return False
else:
return True
except (IOError, OSError, EOFError, pickle.PickleError):
return False

View File

@ -9,10 +9,9 @@ Cookbook, licensed under the Python Software License.
Example Usage::
>>> from django.core.files import locks
>>> f = open('./file', 'wb')
>>> locks.lock(f, locks.LOCK_EX)
>>> f.write('Django')
>>> f.close()
>>> with open('./file', 'wb') as f:
>>> locks.lock(f, locks.LOCK_EX)
>>> f.write('Django')
"""
__all__ = ('LOCK_EX','LOCK_SH','LOCK_NB','lock','unlock')

View File

@ -59,8 +59,7 @@ def file_move_safe(old_file_name, new_file_name, chunk_size = 1024*64, allow_ove
pass
# first open the old file, so that it won't go away
old_file = open(old_file_name, 'rb')
try:
with open(old_file_name, 'rb') as old_file:
# now open the new file, not forgetting allow_overwrite
fd = os.open(new_file_name, os.O_WRONLY | os.O_CREAT | getattr(os, 'O_BINARY', 0) |
(not allow_overwrite and os.O_EXCL or 0))
@ -73,8 +72,6 @@ def file_move_safe(old_file_name, new_file_name, chunk_size = 1024*64, allow_ove
finally:
locks.unlock(fd)
os.close(fd)
finally:
old_file.close()
copystat(old_file_name, new_file_name)
try:

View File

@ -265,7 +265,8 @@ class EmailMessage(object):
def attach_file(self, path, mimetype=None):
"""Attaches a file from the filesystem."""
filename = os.path.basename(path)
content = open(path, 'rb').read()
with open(path, 'rb') as f:
content = f.read()
self.attach(filename, content, mimetype)
def _create_message(self, msg):

View File

@ -5,8 +5,8 @@ from optparse import make_option
from django.core.management.base import BaseCommand, CommandError
def has_bom(fn):
f = open(fn, 'r')
sample = f.read(4)
with open(fn, 'r') as f:
sample = f.read(4)
return sample[:3] == '\xef\xbb\xbf' or \
sample.startswith(codecs.BOM_UTF16_LE) or \
sample.startswith(codecs.BOM_UTF16_BE)

View File

@ -112,7 +112,8 @@ def copy_plural_forms(msgs, locale, domain, verbosity, stdout=sys.stdout):
for domain in domains:
django_po = os.path.join(django_dir, 'conf', 'locale', locale, 'LC_MESSAGES', '%s.po' % domain)
if os.path.exists(django_po):
m = plural_forms_re.search(open(django_po, 'rU').read())
with open(django_po, 'rU') as fp:
m = plural_forms_re.search(fp.read())
if m:
if verbosity > 1:
stdout.write("copying plural forms: %s\n" % m.group('value'))
@ -141,11 +142,8 @@ def write_pot_file(potfile, msgs, file, work_file, is_templatized):
msgs = '\n'.join(dropwhile(len, msgs.split('\n')))
else:
msgs = msgs.replace('charset=CHARSET', 'charset=UTF-8')
f = open(potfile, 'ab')
try:
f.write(msgs)
finally:
f.close()
with open(potfile, 'ab') as fp:
fp.write(msgs)
def process_file(file, dirpath, potfile, domain, verbosity,
extensions, wrap, location, stdout=sys.stdout):
@ -164,15 +162,13 @@ def process_file(file, dirpath, potfile, domain, verbosity,
if domain == 'djangojs' and file_ext in extensions:
is_templatized = True
orig_file = os.path.join(dirpath, file)
src_data = open(orig_file).read()
with open(orig_file) as fp:
src_data = fp.read()
src_data = prepare_js_for_gettext(src_data)
thefile = '%s.c' % file
work_file = os.path.join(dirpath, thefile)
f = open(work_file, "w")
try:
f.write(src_data)
finally:
f.close()
with open(work_file, "w") as fp:
fp.write(src_data)
cmd = (
'xgettext -d %s -L C %s %s --keyword=gettext_noop '
'--keyword=gettext_lazy --keyword=ngettext_lazy:1,2 '
@ -184,14 +180,12 @@ def process_file(file, dirpath, potfile, domain, verbosity,
orig_file = os.path.join(dirpath, file)
is_templatized = file_ext in extensions
if is_templatized:
src_data = open(orig_file, "rU").read()
with open(orig_file, "rU") as fp:
src_data = fp.read()
thefile = '%s.py' % file
content = templatize(src_data, orig_file[2:])
f = open(os.path.join(dirpath, thefile), "w")
try:
f.write(content)
finally:
f.close()
with open(os.path.join(dirpath, thefile), "w") as fp:
fp.write(content)
work_file = os.path.join(dirpath, thefile)
cmd = (
'xgettext -d %s -L Python %s %s --keyword=gettext_noop '
@ -232,11 +226,8 @@ def write_po_file(pofile, potfile, domain, locale, verbosity, stdout,
os.unlink(potfile)
raise CommandError("errors happened while running msguniq\n%s" % errors)
if os.path.exists(pofile):
f = open(potfile, 'w')
try:
f.write(msgs)
finally:
f.close()
with open(potfile, 'w') as fp:
fp.write(msgs)
msgs, errors = _popen('msgmerge %s %s -q "%s" "%s"' %
(wrap, location, pofile, potfile))
if errors:
@ -247,11 +238,8 @@ def write_po_file(pofile, potfile, domain, locale, verbosity, stdout,
msgs = copy_plural_forms(msgs, locale, domain, verbosity, stdout)
msgs = msgs.replace(
"#. #-#-#-#-# %s.pot (PACKAGE VERSION) #-#-#-#-#\n" % domain, "")
f = open(pofile, 'wb')
try:
f.write(msgs)
finally:
f.close()
with open(pofile, 'wb') as fp:
fp.write(msgs)
os.unlink(potfile)
if no_obsolete:
msgs, errors = _popen('msgattrib %s %s -o "%s" --no-obsolete "%s"' %

View File

@ -157,13 +157,12 @@ def custom_sql_for_model(model, style, connection):
os.path.join(app_dir, "%s.sql" % opts.object_name.lower())]
for sql_file in sql_files:
if os.path.exists(sql_file):
fp = open(sql_file, 'U')
for statement in statements.split(fp.read().decode(settings.FILE_CHARSET)):
# Remove any comments from the file
statement = re.sub(ur"--.*([\n\Z]|$)", "", statement)
if statement.strip():
output.append(statement + u";")
fp.close()
with open(sql_file, 'U') as fp:
for statement in statements.split(fp.read().decode(settings.FILE_CHARSET)):
# Remove any comments from the file
statement = re.sub(ur"--.*([\n\Z]|$)", "", statement)
if statement.strip():
output.append(statement + u";")
return output

View File

@ -176,9 +176,8 @@ def runfastcgi(argset=[], **kwargs):
become_daemon(our_home_dir=options["workdir"], **daemon_kwargs)
if options["pidfile"]:
fp = open(options["pidfile"], "w")
fp.write("%d\n" % os.getpid())
fp.close()
with open(options["pidfile"], "w") as fp:
fp.write("%d\n" % os.getpid())
WSGIServer(get_internal_wsgi_application(), **wsgi_opts).run()

View File

@ -328,9 +328,8 @@ class SsiNode(Node):
else:
return '' # Fail silently for invalid includes.
try:
fp = open(filepath, 'r')
output = fp.read()
fp.close()
with open(filepath, 'r') as fp:
output = fp.read()
except IOError:
output = ''
if self.parsed:

View File

@ -52,11 +52,8 @@ class Loader(BaseLoader):
def load_template_source(self, template_name, template_dirs=None):
for filepath in self.get_template_sources(template_name, template_dirs):
try:
file = open(filepath)
try:
return (file.read().decode(settings.FILE_CHARSET), filepath)
finally:
file.close()
with open(filepath) as fp:
return (fp.read().decode(settings.FILE_CHARSET), filepath)
except IOError:
pass
raise TemplateDoesNotExist(template_name)

View File

@ -34,11 +34,8 @@ class Loader(BaseLoader):
tried = []
for filepath in self.get_template_sources(template_name, template_dirs):
try:
file = open(filepath)
try:
return (file.read().decode(settings.FILE_CHARSET), filepath)
finally:
file.close()
with open(filepath) as fp:
return (fp.read().decode(settings.FILE_CHARSET), filepath)
except IOError:
tried.append(filepath)
if tried:

View File

@ -226,7 +226,8 @@ def _load_testfile(filename, package, module_relative):
# get_data() opens files as 'rb', so one must do the equivalent
# conversion as universal newlines would do.
return file_contents.replace(os.linesep, '\n'), filename
return open(filename).read(), filename
with open(filename) as fp:
return fp.read(), filename
def _indent(s, indent=4):
"""
@ -2519,9 +2520,8 @@ def debug_script(src, pm=False, globs=None):
# docs say, a file so created cannot be opened by name a second time
# on modern Windows boxes, and execfile() needs to open it.
srcfilename = tempfile.mktemp(".py", "doctestdebug")
f = open(srcfilename, 'w')
f.write(src)
f.close()
with open(srcfilename, 'w') as fp:
fp.write(src)
try:
if globs:

View File

@ -15,9 +15,8 @@ Sample usage:
... link=u"http://www.holovaty.com/test/",
... description="Testing."
... )
>>> fp = open('test.rss', 'w')
>>> feed.write(fp, 'utf-8')
>>> fp.close()
>>> with open('test.rss', 'w') as fp:
>>> feed.write(fp, 'utf-8')
For definitions of the different versions of RSS, see:
http://diveintomark.org/archives/2004/02/04/incompatible-rss

View File

@ -333,11 +333,8 @@ class ExceptionReporter(object):
source = source.splitlines()
if source is None:
try:
f = open(filename)
try:
source = f.readlines()
finally:
f.close()
with open(filename) as fp:
source = fp.readlines()
except (OSError, IOError):
pass
if source is None:

View File

@ -18,33 +18,18 @@ def process_file(fn, lines):
lines.insert(0, '\n')
lines.insert(0, '.. %s:\n' % target_name(fn))
try:
f = open(fn, 'w')
with open(fn, 'w') as fp:
fp.writelines(lines)
except IOError:
print("Can't open %s for writing. Not touching it." % fn)
return
try:
f.writelines(lines)
except IOError:
print("Can't write to %s. Not touching it." % fn)
finally:
f.close()
def has_target(fn):
try:
f = open(fn, 'r')
with open(fn, 'r') as fp:
lines = fp.readlines()
except IOError:
print("Can't open %s. Not touching it." % fn)
print("Can't open or read %s. Not touching it." % fn)
return (True, None)
readok = True
try:
lines = f.readlines()
except IOError:
print("Can't read %s. Not touching it." % fn)
readok = False
finally:
f.close()
if not readok:
return (True, None)
#print fn, len(lines)
if len(lines) < 1:
@ -82,7 +67,7 @@ def main(argv=None):
print("Adding xref to %s" % fn)
process_file(fn, lines)
else:
print "Skipping %s: already has a xref" % fn
print("Skipping %s: already has a xref" % fn)
if __name__ == '__main__':
sys.exit(main())

View File

@ -210,8 +210,7 @@ class DjangoStandaloneHTMLBuilder(StandaloneHTMLBuilder):
if t == "templatefilter" and l == "ref/templates/builtins"],
}
outfilename = os.path.join(self.outdir, "templatebuiltins.js")
f = open(outfilename, 'wb')
f.write('var django_template_builtins = ')
json.dump(templatebuiltins, f)
f.write(';\n')
f.close();
with open(outfilename, 'wb') as fp:
fp.write('var django_template_builtins = ')
json.dump(templatebuiltins, fp)
fp.write(';\n')

View File

@ -38,7 +38,8 @@ ALWAYS_SKIP = [
]
def fixliterals(fname):
data = open(fname).read()
with open(fname) as fp:
data = fp.read()
last = 0
new = []
@ -101,7 +102,8 @@ def fixliterals(fname):
lastvalues[m.group(1)] = replace_value
new.append(data[last:])
open(fname, "w").write("".join(new))
with open(fname, "w") as fp:
fp.write("".join(new))
storage["lastvalues"] = lastvalues
storage.close()

View File

@ -239,9 +239,8 @@ Sample usage::
... link=u"http://www.holovaty.com/test/",
... description="Testing."
... )
>>> fp = open('test.rss', 'w')
>>> feed.write(fp, 'utf-8')
>>> fp.close()
>>> with open('test.rss', 'w') as fp:
>>> feed.write(fp, 'utf-8')
For simplifying the selection of a generator use ``feedgenerator.DefaultFeed``
which is currently ``Rss201rev2Feed``

View File

@ -101,10 +101,9 @@ objects; see `UploadedFile objects`_ for a complete reference.
Putting it all together, here's a common way you might handle an uploaded file::
def handle_uploaded_file(f):
destination = open('some/file/name.txt', 'wb+')
for chunk in f.chunks():
destination.write(chunk)
destination.close()
with open('some/file/name.txt', 'wb+') as destination:
for chunk in f.chunks():
destination.write(chunk)
Looping over ``UploadedFile.chunks()`` instead of using ``read()`` ensures that
large files don't overwhelm your system's memory.

View File

@ -36,8 +36,8 @@ You can also use a serializer object directly::
This is useful if you want to serialize data directly to a file-like object
(which includes an :class:`~django.http.HttpResponse`)::
out = open("file.xml", "w")
xml_serializer.serialize(SomeModel.objects.all(), stream=out)
with open("file.xml", "w") as out:
xml_serializer.serialize(SomeModel.objects.all(), stream=out)
.. note::

View File

@ -770,9 +770,8 @@ arguments at time of construction:
wish to upload as a value. For example::
>>> c = Client()
>>> f = open('wishlist.doc')
>>> c.post('/customers/wishes/', {'name': 'fred', 'attachment': f})
>>> f.close()
>>> with open('wishlist.doc') as fp:
>>> c.post('/customers/wishes/', {'name': 'fred', 'attachment': fp})
(The name ``attachment`` here is not relevant; use whatever name your
file-processing code expects.)

View File

@ -170,14 +170,13 @@ class Template(object):
try:
return self._content
except AttributeError:
fd = open(self.absolute_filename)
try:
content = fd.read().decode(TEMPLATE_ENCODING)
except UnicodeDecodeError as e:
message = '%s in %s' % (
e[4], self.absolute_filename.encode('UTF-8', 'ignore'))
raise UnicodeDecodeError(*(e.args[:4] + (message,)))
fd.close()
with open(self.absolute_filename) as fd:
try:
content = fd.read().decode(TEMPLATE_ENCODING)
except UnicodeDecodeError as e:
message = '%s in %s' % (
e[4], self.absolute_filename.encode('UTF-8', 'ignore'))
raise UnicodeDecodeError(*(e.args[:4] + (message,)))
self._content = content
return content
content = property(content)
@ -271,9 +270,8 @@ def get_python_code(paths):
for f in filenames:
if len([True for e in PYTHON_SOURCE_EXTENSIONS if f.endswith(e)]) > 0:
fn = os.path.join(dirpath, f)
fd = open(fn)
content = [l.decode(PYTHON_ENCODING) for l in fd.readlines()]
fd.close()
with open(fn) as fd:
content = [l.decode(PYTHON_ENCODING) for l in fd.readlines()]
retval.append((fn, content))
return retval

View File

@ -1242,8 +1242,10 @@ class OldFormForXTests(TestCase):
# it comes to validation. This specifically tests that #6302 is fixed for
# both file fields and image fields.
image_data = open(os.path.join(os.path.dirname(__file__), "test.png"), 'rb').read()
image_data2 = open(os.path.join(os.path.dirname(__file__), "test2.png"), 'rb').read()
with open(os.path.join(os.path.dirname(__file__), "test.png"), 'rb') as fp:
image_data = fp.read()
with open(os.path.join(os.path.dirname(__file__), "test2.png"), 'rb') as fp:
image_data2 = fp.read()
f = ImageFileForm(
data={'description': u'An image'},

View File

@ -27,32 +27,32 @@ class AdminScriptTestCase(unittest.TestCase):
if is_dir:
settings_dir = os.path.join(test_dir, filename)
os.mkdir(settings_dir)
settings_file = open(os.path.join(settings_dir, '__init__.py'), 'w')
settings_file_path = os.path.join(settings_dir, '__init__.py')
else:
settings_file = open(os.path.join(test_dir, filename), 'w')
settings_file.write('# Settings file automatically generated by regressiontests.admin_scripts test case\n')
exports = [
'DATABASES',
'ROOT_URLCONF',
'SECRET_KEY',
]
for s in exports:
if hasattr(settings, s):
o = getattr(settings, s)
if not isinstance(o, dict):
o = "'%s'" % o
settings_file.write("%s = %s\n" % (s, o))
settings_file_path = os.path.join(test_dir, filename)
if apps is None:
apps = ['django.contrib.auth', 'django.contrib.contenttypes', 'regressiontests.admin_scripts']
with open(settings_file_path, 'w') as settings_file:
settings_file.write('# Settings file automatically generated by regressiontests.admin_scripts test case\n')
exports = [
'DATABASES',
'ROOT_URLCONF',
'SECRET_KEY',
]
for s in exports:
if hasattr(settings, s):
o = getattr(settings, s)
if not isinstance(o, dict):
o = "'%s'" % o
settings_file.write("%s = %s\n" % (s, o))
settings_file.write("INSTALLED_APPS = %s\n" % apps)
if apps is None:
apps = ['django.contrib.auth', 'django.contrib.contenttypes', 'regressiontests.admin_scripts']
if sdict:
for k, v in sdict.items():
settings_file.write("%s = %s\n" % (k, v))
settings_file.write("INSTALLED_APPS = %s\n" % apps)
settings_file.close()
if sdict:
for k, v in sdict.items():
settings_file.write("%s = %s\n" % (k, v))
def remove_settings(self, filename, is_dir=False):
full_name = os.path.join(test_dir, filename)
@ -989,13 +989,12 @@ class ManageSettingsWithImportError(AdminScriptTestCase):
if is_dir:
settings_dir = os.path.join(test_dir, filename)
os.mkdir(settings_dir)
settings_file = open(os.path.join(settings_dir, '__init__.py'), 'w')
settings_file_path = os.path.join(settings_dir, '__init__.py')
else:
settings_file = open(os.path.join(test_dir, filename), 'w')
settings_file.write('# Settings file automatically generated by regressiontests.admin_scripts test case\n')
settings_file.write('# The next line will cause an import error:\nimport foo42bar\n')
settings_file.close()
settings_file_path = os.path.join(test_dir, filename)
with open(settings_file_path, 'w') as settings_file:
settings_file.write('# Settings file automatically generated by regressiontests.admin_scripts test case\n')
settings_file.write('# The next line will cause an import error:\nimport foo42bar\n')
def test_builtin_command(self):
"import error: manage.py builtin commands shows useful diagnostic info when settings with import errors is provided"

View File

@ -24,7 +24,8 @@ class Bug639Test(unittest.TestCase):
"""
# Grab an image for testing.
filename = os.path.join(os.path.dirname(__file__), "test.jpg")
img = open(filename, "rb").read()
with open(filename, "rb") as fp:
img = fp.read()
# Fake a POST QueryDict and FILES MultiValueDict.
data = {'title': 'Testing'}

View File

@ -24,11 +24,12 @@ UNICODE_FILENAME = u'test-0123456789_中文_Orléans.jpg'
class FileUploadTests(TestCase):
def test_simple_upload(self):
post_data = {
'name': 'Ringo',
'file_field': open(__file__),
}
response = self.client.post('/file_uploads/upload/', post_data)
with open(__file__) as fp:
post_data = {
'name': 'Ringo',
'file_field': fp,
}
response = self.client.post('/file_uploads/upload/', post_data)
self.assertEqual(response.status_code, 200)
def test_large_upload(self):
@ -87,17 +88,16 @@ class FileUploadTests(TestCase):
tdir = tempfile.gettempdir()
# This file contains chinese symbols and an accented char in the name.
file1 = open(os.path.join(tdir, UNICODE_FILENAME.encode('utf-8')), 'w+b')
file1.write('b' * (2 ** 10))
file1.seek(0)
with open(os.path.join(tdir, UNICODE_FILENAME.encode('utf-8')), 'w+b') as file1:
file1.write('b' * (2 ** 10))
file1.seek(0)
post_data = {
'file_unicode': file1,
}
post_data = {
'file_unicode': file1,
}
response = self.client.post('/file_uploads/unicode_name/', post_data)
response = self.client.post('/file_uploads/unicode_name/', post_data)
file1.close()
try:
os.unlink(file1.name)
except:
@ -294,10 +294,6 @@ class FileUploadTests(TestCase):
p = request.POST
return ret
post_data = {
'name': 'Ringo',
'file_field': open(__file__),
}
# Maybe this is a little more complicated that it needs to be; but if
# the django.test.client.FakePayload.read() implementation changes then
# this test would fail. So we need to know exactly what kind of error
@ -310,16 +306,21 @@ class FileUploadTests(TestCase):
# install the custom handler that tries to access request.POST
self.client.handler = POSTAccessingHandler()
try:
response = self.client.post('/file_uploads/upload_errors/', post_data)
except reference_error.__class__ as err:
self.assertFalse(
str(err) == str(reference_error),
"Caught a repeated exception that'll cause an infinite loop in file uploads."
)
except Exception as err:
# CustomUploadError is the error that should have been raised
self.assertEqual(err.__class__, uploadhandler.CustomUploadError)
with open(__file__) as fp:
post_data = {
'name': 'Ringo',
'file_field': fp,
}
try:
response = self.client.post('/file_uploads/upload_errors/', post_data)
except reference_error.__class__ as err:
self.assertFalse(
str(err) == str(reference_error),
"Caught a repeated exception that'll cause an infinite loop in file uploads."
)
except Exception as err:
# CustomUploadError is the error that should have been raised
self.assertEqual(err.__class__, uploadhandler.CustomUploadError)
def test_filename_case_preservation(self):
"""
@ -382,8 +383,7 @@ class DirectoryCreationTests(unittest.TestCase):
def test_not_a_directory(self):
"""The correct IOError is raised when the upload directory name exists but isn't a directory"""
# Create a file with the upload directory name
fd = open(UPLOAD_TO, 'w')
fd.close()
open(UPLOAD_TO, 'w').close()
try:
self.obj.testfile.save('foo.txt', SimpleUploadedFile('foo.txt', 'x'))
except IOError as err:

View File

@ -517,7 +517,8 @@ class FileBackendTests(BaseEmailBackendTests, TestCase):
def get_mailbox_content(self):
messages = []
for filename in os.listdir(self.tmp_dir):
session = open(os.path.join(self.tmp_dir, filename)).read().split('\n' + ('-' * 79) + '\n')
with open(os.path.join(self.tmp_dir, filename)) as fp:
session = fp.read().split('\n' + ('-' * 79) + '\n')
messages.extend(email.message_from_string(m) for m in session if m)
return messages
@ -528,7 +529,8 @@ class FileBackendTests(BaseEmailBackendTests, TestCase):
connection.send_messages([msg])
self.assertEqual(len(os.listdir(self.tmp_dir)), 1)
message = email.message_from_file(open(os.path.join(self.tmp_dir, os.listdir(self.tmp_dir)[0])))
with open(os.path.join(self.tmp_dir, os.listdir(self.tmp_dir)[0])) as fp:
message = email.message_from_file(fp)
self.assertEqual(message.get_content_type(), 'text/plain')
self.assertEqual(message.get('subject'), 'Subject')
self.assertEqual(message.get('from'), 'from@example.com')

View File

@ -29,7 +29,8 @@ class StaticTests(TestCase):
for filename in media_files:
response = self.client.get('/views/%s/%s' % (self.prefix, filename))
file_path = path.join(media_dir, filename)
self.assertEqual(open(file_path).read(), response.content)
with open(file_path) as fp:
self.assertEqual(fp.read(), response.content)
self.assertEqual(len(response.content), int(response['Content-Length']))
self.assertEqual(mimetypes.guess_type(file_path)[1], response.get('Content-Encoding', None))
@ -40,15 +41,15 @@ class StaticTests(TestCase):
def test_copes_with_empty_path_component(self):
file_name = 'file.txt'
response = self.client.get('/views/%s//%s' % (self.prefix, file_name))
file = open(path.join(media_dir, file_name))
self.assertEqual(file.read(), response.content)
with open(path.join(media_dir, file_name)) as fp:
self.assertEqual(fp.read(), response.content)
def test_is_modified_since(self):
file_name = 'file.txt'
response = self.client.get('/views/%s/%s' % (self.prefix, file_name),
HTTP_IF_MODIFIED_SINCE='Thu, 1 Jan 1970 00:00:00 GMT')
file = open(path.join(media_dir, file_name))
self.assertEqual(file.read(), response.content)
with open(path.join(media_dir, file_name)) as fp:
self.assertEqual(fp.read(), response.content)
def test_not_modified_since(self):
file_name = 'file.txt'
@ -70,8 +71,8 @@ class StaticTests(TestCase):
invalid_date = 'Mon, 28 May 999999999999 28:25:26 GMT'
response = self.client.get('/views/%s/%s' % (self.prefix, file_name),
HTTP_IF_MODIFIED_SINCE=invalid_date)
file = open(path.join(media_dir, file_name))
self.assertEqual(file.read(), response.content)
with open(path.join(media_dir, file_name)) as fp:
self.assertEqual(fp.read(), response.content)
self.assertEqual(len(response.content),
int(response['Content-Length']))
@ -85,8 +86,8 @@ class StaticTests(TestCase):
invalid_date = ': 1291108438, Wed, 20 Oct 2010 14:05:00 GMT'
response = self.client.get('/views/%s/%s' % (self.prefix, file_name),
HTTP_IF_MODIFIED_SINCE=invalid_date)
file = open(path.join(media_dir, file_name))
self.assertEqual(file.read(), response.content)
with open(path.join(media_dir, file_name)) as fp:
self.assertEqual(fp.read(), response.content)
self.assertEqual(len(response.content),
int(response['Content-Length']))

View File

@ -61,6 +61,7 @@ def get_test_modules():
for f in os.listdir(dirpath):
if (f.startswith('__init__') or
f.startswith('.') or
f == '__pycache__' or
f.startswith('sql') or
os.path.basename(f) in REGRESSION_SUBDIRS_TO_SKIP):
continue