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 cmd = 'msguniq "%s.po"' % pf
stdout = os.popen(cmd) stdout = os.popen(cmd)
msg = stdout.read() msg = stdout.read()
open('%s.po' % pf, 'w').write(msg) with open('%s.po' % pf, 'w') as fp:
fp.write(msg)
if __name__ == "__main__": if __name__ == "__main__":
unique_messages() unique_messages()

View File

@ -120,12 +120,9 @@ class SpatiaLiteCreation(DatabaseCreation):
# Opening up the SpatiaLite SQL initialization file and executing # Opening up the SpatiaLite SQL initialization file and executing
# as a script. # as a script.
sql_fh = open(spatialite_sql, 'r') with open(spatialite_sql, 'r') as sql_fh:
try:
cur = self.connection._cursor() cur = self.connection._cursor()
cur.executescript(sql_fh.read()) cur.executescript(sql_fh.read())
finally:
sql_fh.close()
def spatialite_init_file(self): def spatialite_init_file(self):
# SPATIALITE_SQL may be placed in settings to tell GeoDjango # 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 given a file name, get a real handle.
if isinstance(file_h, basestring): if isinstance(file_h, basestring):
file_h = open(file_h, 'rb') with open(file_h, 'rb') as file_h:
buf = file_h.read()
# Reading in the file's contents, else:
buf = file_h.read() buf = file_h.read()
# If we get WKB need to wrap in buffer(), so run through regexes. # If we get WKB need to wrap in buffer(), so run through regexes.
if wkt_regex.match(buf) or hex_regex.match(buf): if wkt_regex.match(buf) or hex_regex.match(buf):

View File

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

View File

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

View File

@ -9,10 +9,9 @@ Cookbook, licensed under the Python Software License.
Example Usage:: Example Usage::
>>> from django.core.files import locks >>> from django.core.files import locks
>>> f = open('./file', 'wb') >>> with open('./file', 'wb') as f:
>>> locks.lock(f, locks.LOCK_EX) >>> locks.lock(f, locks.LOCK_EX)
>>> f.write('Django') >>> f.write('Django')
>>> f.close()
""" """
__all__ = ('LOCK_EX','LOCK_SH','LOCK_NB','lock','unlock') __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 pass
# first open the old file, so that it won't go away # first open the old file, so that it won't go away
old_file = open(old_file_name, 'rb') with open(old_file_name, 'rb') as old_file:
try:
# now open the new file, not forgetting allow_overwrite # 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) | 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)) (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: finally:
locks.unlock(fd) locks.unlock(fd)
os.close(fd) os.close(fd)
finally:
old_file.close()
copystat(old_file_name, new_file_name) copystat(old_file_name, new_file_name)
try: try:

View File

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

View File

@ -5,8 +5,8 @@ from optparse import make_option
from django.core.management.base import BaseCommand, CommandError from django.core.management.base import BaseCommand, CommandError
def has_bom(fn): def has_bom(fn):
f = open(fn, 'r') with open(fn, 'r') as f:
sample = f.read(4) sample = f.read(4)
return sample[:3] == '\xef\xbb\xbf' or \ return sample[:3] == '\xef\xbb\xbf' or \
sample.startswith(codecs.BOM_UTF16_LE) or \ sample.startswith(codecs.BOM_UTF16_LE) or \
sample.startswith(codecs.BOM_UTF16_BE) 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: for domain in domains:
django_po = os.path.join(django_dir, 'conf', 'locale', locale, 'LC_MESSAGES', '%s.po' % domain) django_po = os.path.join(django_dir, 'conf', 'locale', locale, 'LC_MESSAGES', '%s.po' % domain)
if os.path.exists(django_po): 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 m:
if verbosity > 1: if verbosity > 1:
stdout.write("copying plural forms: %s\n" % m.group('value')) 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'))) msgs = '\n'.join(dropwhile(len, msgs.split('\n')))
else: else:
msgs = msgs.replace('charset=CHARSET', 'charset=UTF-8') msgs = msgs.replace('charset=CHARSET', 'charset=UTF-8')
f = open(potfile, 'ab') with open(potfile, 'ab') as fp:
try: fp.write(msgs)
f.write(msgs)
finally:
f.close()
def process_file(file, dirpath, potfile, domain, verbosity, def process_file(file, dirpath, potfile, domain, verbosity,
extensions, wrap, location, stdout=sys.stdout): 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: if domain == 'djangojs' and file_ext in extensions:
is_templatized = True is_templatized = True
orig_file = os.path.join(dirpath, file) 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) src_data = prepare_js_for_gettext(src_data)
thefile = '%s.c' % file thefile = '%s.c' % file
work_file = os.path.join(dirpath, thefile) work_file = os.path.join(dirpath, thefile)
f = open(work_file, "w") with open(work_file, "w") as fp:
try: fp.write(src_data)
f.write(src_data)
finally:
f.close()
cmd = ( cmd = (
'xgettext -d %s -L C %s %s --keyword=gettext_noop ' 'xgettext -d %s -L C %s %s --keyword=gettext_noop '
'--keyword=gettext_lazy --keyword=ngettext_lazy:1,2 ' '--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) orig_file = os.path.join(dirpath, file)
is_templatized = file_ext in extensions is_templatized = file_ext in extensions
if is_templatized: 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 thefile = '%s.py' % file
content = templatize(src_data, orig_file[2:]) content = templatize(src_data, orig_file[2:])
f = open(os.path.join(dirpath, thefile), "w") with open(os.path.join(dirpath, thefile), "w") as fp:
try: fp.write(content)
f.write(content)
finally:
f.close()
work_file = os.path.join(dirpath, thefile) work_file = os.path.join(dirpath, thefile)
cmd = ( cmd = (
'xgettext -d %s -L Python %s %s --keyword=gettext_noop ' '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) os.unlink(potfile)
raise CommandError("errors happened while running msguniq\n%s" % errors) raise CommandError("errors happened while running msguniq\n%s" % errors)
if os.path.exists(pofile): if os.path.exists(pofile):
f = open(potfile, 'w') with open(potfile, 'w') as fp:
try: fp.write(msgs)
f.write(msgs)
finally:
f.close()
msgs, errors = _popen('msgmerge %s %s -q "%s" "%s"' % msgs, errors = _popen('msgmerge %s %s -q "%s" "%s"' %
(wrap, location, pofile, potfile)) (wrap, location, pofile, potfile))
if errors: 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 = copy_plural_forms(msgs, locale, domain, verbosity, stdout)
msgs = msgs.replace( msgs = msgs.replace(
"#. #-#-#-#-# %s.pot (PACKAGE VERSION) #-#-#-#-#\n" % domain, "") "#. #-#-#-#-# %s.pot (PACKAGE VERSION) #-#-#-#-#\n" % domain, "")
f = open(pofile, 'wb') with open(pofile, 'wb') as fp:
try: fp.write(msgs)
f.write(msgs)
finally:
f.close()
os.unlink(potfile) os.unlink(potfile)
if no_obsolete: if no_obsolete:
msgs, errors = _popen('msgattrib %s %s -o "%s" --no-obsolete "%s"' % 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())] os.path.join(app_dir, "%s.sql" % opts.object_name.lower())]
for sql_file in sql_files: for sql_file in sql_files:
if os.path.exists(sql_file): if os.path.exists(sql_file):
fp = open(sql_file, 'U') with open(sql_file, 'U') as fp:
for statement in statements.split(fp.read().decode(settings.FILE_CHARSET)): for statement in statements.split(fp.read().decode(settings.FILE_CHARSET)):
# Remove any comments from the file # Remove any comments from the file
statement = re.sub(ur"--.*([\n\Z]|$)", "", statement) statement = re.sub(ur"--.*([\n\Z]|$)", "", statement)
if statement.strip(): if statement.strip():
output.append(statement + u";") output.append(statement + u";")
fp.close()
return output return output

View File

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

View File

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

View File

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

View File

@ -34,11 +34,8 @@ class Loader(BaseLoader):
tried = [] tried = []
for filepath in self.get_template_sources(template_name, template_dirs): for filepath in self.get_template_sources(template_name, template_dirs):
try: try:
file = open(filepath) with open(filepath) as fp:
try: return (fp.read().decode(settings.FILE_CHARSET), filepath)
return (file.read().decode(settings.FILE_CHARSET), filepath)
finally:
file.close()
except IOError: except IOError:
tried.append(filepath) tried.append(filepath)
if tried: 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 # get_data() opens files as 'rb', so one must do the equivalent
# conversion as universal newlines would do. # conversion as universal newlines would do.
return file_contents.replace(os.linesep, '\n'), filename 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): 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 # docs say, a file so created cannot be opened by name a second time
# on modern Windows boxes, and execfile() needs to open it. # on modern Windows boxes, and execfile() needs to open it.
srcfilename = tempfile.mktemp(".py", "doctestdebug") srcfilename = tempfile.mktemp(".py", "doctestdebug")
f = open(srcfilename, 'w') with open(srcfilename, 'w') as fp:
f.write(src) fp.write(src)
f.close()
try: try:
if globs: if globs:

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@ -239,9 +239,8 @@ Sample usage::
... link=u"http://www.holovaty.com/test/", ... link=u"http://www.holovaty.com/test/",
... description="Testing." ... description="Testing."
... ) ... )
>>> fp = open('test.rss', 'w') >>> with open('test.rss', 'w') as fp:
>>> feed.write(fp, 'utf-8') >>> feed.write(fp, 'utf-8')
>>> fp.close()
For simplifying the selection of a generator use ``feedgenerator.DefaultFeed`` For simplifying the selection of a generator use ``feedgenerator.DefaultFeed``
which is currently ``Rss201rev2Feed`` 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:: Putting it all together, here's a common way you might handle an uploaded file::
def handle_uploaded_file(f): def handle_uploaded_file(f):
destination = open('some/file/name.txt', 'wb+') with open('some/file/name.txt', 'wb+') as destination:
for chunk in f.chunks(): for chunk in f.chunks():
destination.write(chunk) destination.write(chunk)
destination.close()
Looping over ``UploadedFile.chunks()`` instead of using ``read()`` ensures that Looping over ``UploadedFile.chunks()`` instead of using ``read()`` ensures that
large files don't overwhelm your system's memory. 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 This is useful if you want to serialize data directly to a file-like object
(which includes an :class:`~django.http.HttpResponse`):: (which includes an :class:`~django.http.HttpResponse`)::
out = open("file.xml", "w") with open("file.xml", "w") as out:
xml_serializer.serialize(SomeModel.objects.all(), stream=out) xml_serializer.serialize(SomeModel.objects.all(), stream=out)
.. note:: .. note::

View File

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

View File

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

View File

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

View File

@ -27,32 +27,32 @@ class AdminScriptTestCase(unittest.TestCase):
if is_dir: if is_dir:
settings_dir = os.path.join(test_dir, filename) settings_dir = os.path.join(test_dir, filename)
os.mkdir(settings_dir) 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: else:
settings_file = open(os.path.join(test_dir, filename), 'w') settings_file_path = os.path.join(test_dir, filename)
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))
if apps is None: with open(settings_file_path, 'w') as settings_file:
apps = ['django.contrib.auth', 'django.contrib.contenttypes', 'regressiontests.admin_scripts'] 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: settings_file.write("INSTALLED_APPS = %s\n" % apps)
for k, v in sdict.items():
settings_file.write("%s = %s\n" % (k, v))
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): def remove_settings(self, filename, is_dir=False):
full_name = os.path.join(test_dir, filename) full_name = os.path.join(test_dir, filename)
@ -989,13 +989,12 @@ class ManageSettingsWithImportError(AdminScriptTestCase):
if is_dir: if is_dir:
settings_dir = os.path.join(test_dir, filename) settings_dir = os.path.join(test_dir, filename)
os.mkdir(settings_dir) 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: else:
settings_file = open(os.path.join(test_dir, filename), 'w') settings_file_path = os.path.join(test_dir, filename)
settings_file.write('# Settings file automatically generated by regressiontests.admin_scripts test case\n') with open(settings_file_path, 'w') as settings_file:
settings_file.write('# The next line will cause an import error:\nimport foo42bar\n') 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()
def test_builtin_command(self): def test_builtin_command(self):
"import error: manage.py builtin commands shows useful diagnostic info when settings with import errors is provided" "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. # Grab an image for testing.
filename = os.path.join(os.path.dirname(__file__), "test.jpg") 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. # Fake a POST QueryDict and FILES MultiValueDict.
data = {'title': 'Testing'} data = {'title': 'Testing'}

View File

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

View File

@ -517,7 +517,8 @@ class FileBackendTests(BaseEmailBackendTests, TestCase):
def get_mailbox_content(self): def get_mailbox_content(self):
messages = [] messages = []
for filename in os.listdir(self.tmp_dir): 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) messages.extend(email.message_from_string(m) for m in session if m)
return messages return messages
@ -528,7 +529,8 @@ class FileBackendTests(BaseEmailBackendTests, TestCase):
connection.send_messages([msg]) connection.send_messages([msg])
self.assertEqual(len(os.listdir(self.tmp_dir)), 1) 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_content_type(), 'text/plain')
self.assertEqual(message.get('subject'), 'Subject') self.assertEqual(message.get('subject'), 'Subject')
self.assertEqual(message.get('from'), 'from@example.com') self.assertEqual(message.get('from'), 'from@example.com')

View File

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

View File

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