Skip to content
This repository

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse code

[1.0.X] Fixed #8817: get_image_dimensions correctly closes the files …

…it opens, and leaves open the ones it doesn't. Thanks, mitsuhiko.

While I was at it, I converted the file_storage doctests to unit tests.

Backport of [10707] from trunk.

git-svn-id: http://code.djangoproject.com/svn/django/branches/releases/1.0.X@10708 bcc190cf-cafb-0310-a4f2-bffc1f526a37
  • Loading branch information...
commit 7935231ef086b5649136dfbb2e49e3e413ecb344 1 parent a109a22
Jacob Kaplan-Moss authored May 08, 2009
22  django/core/files/images.py
@@ -28,15 +28,21 @@ def get_image_dimensions(file_or_path):
28 28
     """Returns the (width, height) of an image, given an open file or a path."""
29 29
     from PIL import ImageFile as PILImageFile
30 30
     p = PILImageFile.Parser()
  31
+    close = False
31 32
     if hasattr(file_or_path, 'read'):
32 33
         file = file_or_path
33 34
     else:
34 35
         file = open(file_or_path, 'rb')
35  
-    while 1:
36  
-        data = file.read(1024)
37  
-        if not data:
38  
-            break
39  
-        p.feed(data)
40  
-        if p.image:
41  
-            return p.image.size
42  
-    return None
  36
+        close = True
  37
+    try:
  38
+        while 1:
  39
+            data = file.read(1024)
  40
+            if not data:
  41
+                break
  42
+            p.feed(data)
  43
+            if p.image:
  44
+                return p.image.size
  45
+        return None
  46
+    finally:
  47
+        if close:
  48
+            file.close()
226  tests/regressiontests/file_storage/tests.py
... ...
@@ -1,105 +1,105 @@
1  
-# coding: utf-8
2  
-"""
3  
-Tests for the file storage mechanism
4  
-
5  
->>> import tempfile
6  
->>> from django.core.files.storage import FileSystemStorage
7  
->>> from django.core.files.base import ContentFile
8  
-
9  
-# Set up a unique temporary directory
10  
->>> import os
11  
->>> temp_dir = tempfile.mktemp()
12  
->>> os.makedirs(temp_dir)
13  
-
14  
->>> temp_storage = FileSystemStorage(location=temp_dir)
15  
-
16  
-# Standard file access options are available, and work as expected.
17  
-
18  
->>> temp_storage.exists('storage_test')
19  
-False
20  
->>> file = temp_storage.open('storage_test', 'w')
21  
->>> file.write('storage contents')
22  
->>> file.close()
23  
-
24  
->>> temp_storage.exists('storage_test')
25  
-True
26  
->>> file = temp_storage.open('storage_test', 'r')
27  
->>> file.read()
28  
-'storage contents'
29  
->>> file.close()
30  
-
31  
->>> temp_storage.delete('storage_test')
32  
->>> temp_storage.exists('storage_test')
33  
-False
34  
-
35  
-# Files can only be accessed if they're below the specified location.
36  
-
37  
->>> temp_storage.exists('..')
38  
-Traceback (most recent call last):
39  
-...
40  
-SuspiciousOperation: Attempted access to '..' denied.
41  
->>> temp_storage.open('/etc/passwd')
42  
-Traceback (most recent call last):
43  
-  ...
44  
-SuspiciousOperation: Attempted access to '/etc/passwd' denied.
45  
-
46  
-# Custom storage systems can be created to customize behavior
47  
-
48  
->>> class CustomStorage(FileSystemStorage):
49  
-...     def get_available_name(self, name):
50  
-...         # Append numbers to duplicate files rather than underscores, like Trac
51  
-...
52  
-...         parts = name.split('.')
53  
-...         basename, ext = parts[0], parts[1:]
54  
-...         number = 2
55  
-...
56  
-...         while self.exists(name):
57  
-...             name = '.'.join([basename, str(number)] + ext)
58  
-...             number += 1
59  
-...
60  
-...         return name
61  
->>> custom_storage = CustomStorage(temp_dir)
62  
-
63  
->>> first = custom_storage.save('custom_storage', ContentFile('custom contents'))
64  
->>> first
65  
-u'custom_storage'
66  
->>> second = custom_storage.save('custom_storage', ContentFile('more contents'))
67  
->>> second
68  
-u'custom_storage.2'
69  
-
70  
->>> custom_storage.delete(first)
71  
->>> custom_storage.delete(second)
72  
-
73  
-# Cleanup the temp dir
74  
->>> os.rmdir(temp_dir)
75  
-
76  
-
77  
-# Regression test for #8156: files with unicode names I can't quite figure out the
78  
-# encoding situation between doctest and this file, but the actual repr doesn't
79  
-# matter; it just shouldn't return a unicode object.
80  
->>> from django.core.files.uploadedfile import UploadedFile
81  
->>> uf = UploadedFile(name=u'¿Cómo?',content_type='text')
82  
->>> uf.__repr__()
83  
-'<UploadedFile: ... (text)>'
84  
-"""
85  
-
86  
-# Tests for a race condition on file saving (#4948).
87  
-# This is written in such a way that it'll always pass on platforms
88  
-# without threading.
  1
+# -*- coding: utf-8 -*-
89 2
 import os
90  
-import time
91 3
 import shutil
92 4
 import sys
93 5
 import tempfile
94  
-from unittest import TestCase
  6
+import time
  7
+import unittest
  8
+from cStringIO import StringIO
95 9
 from django.conf import settings
  10
+from django.core.exceptions import SuspiciousOperation
96 11
 from django.core.files.base import ContentFile
  12
+from django.core.files.images import get_image_dimensions
97 13
 from django.core.files.storage import FileSystemStorage
  14
+from django.core.files.uploadedfile import UploadedFile
  15
+from unittest import TestCase
98 16
 try:
99 17
     import threading
100 18
 except ImportError:
101 19
     import dummy_threading as threading
102 20
 
  21
+try:
  22
+    # Checking for the existence of Image is enough for CPython, but
  23
+    # for PyPy, you need to check for the underlying modules
  24
+    from PIL import Image, _imaging
  25
+except ImportError:
  26
+    Image = None
  27
+
  28
+class FileStorageTests(unittest.TestCase):
  29
+    storage_class = FileSystemStorage
  30
+    
  31
+    def setUp(self):
  32
+        self.temp_dir = tempfile.mktemp()
  33
+        os.makedirs(self.temp_dir)
  34
+        self.storage = self.storage_class(location=self.temp_dir)
  35
+    
  36
+    def tearDown(self):
  37
+        os.rmdir(self.temp_dir)
  38
+        
  39
+    def test_file_access_options(self):
  40
+        """
  41
+        Standard file access options are available, and work as expected.
  42
+        """
  43
+        self.failIf(self.storage.exists('storage_test'))
  44
+        f = self.storage.open('storage_test', 'w')
  45
+        f.write('storage contents')
  46
+        f.close()
  47
+        self.assert_(self.storage.exists('storage_test'))
  48
+
  49
+        f = self.storage.open('storage_test', 'r')
  50
+        self.assertEqual(f.read(), 'storage contents')
  51
+        f.close()
  52
+        
  53
+        self.storage.delete('storage_test')
  54
+        self.failIf(self.storage.exists('storage_test'))
  55
+
  56
+    def test_file_storage_prevents_directory_traversal(self):
  57
+        """
  58
+        File storage prevents directory traversal (files can only be accessed if
  59
+        they're below the storage location).
  60
+        """
  61
+        self.assertRaises(SuspiciousOperation, self.storage.exists, '..')
  62
+        self.assertRaises(SuspiciousOperation, self.storage.exists, '/etc/passwd')
  63
+
  64
+class CustomStorage(FileSystemStorage):
  65
+    def get_available_name(self, name):
  66
+        """
  67
+        Append numbers to duplicate files rather than underscores, like Trac.
  68
+        """
  69
+        parts = name.split('.')
  70
+        basename, ext = parts[0], parts[1:]
  71
+        number = 2
  72
+        while self.exists(name):
  73
+            name = '.'.join([basename, str(number)] + ext)
  74
+            number += 1
  75
+
  76
+        return name
  77
+
  78
+class CustomStorageTests(FileStorageTests):
  79
+    storage_class = CustomStorage
  80
+    
  81
+    def test_custom_get_available_name(self):
  82
+        first = self.storage.save('custom_storage', ContentFile('custom contents'))
  83
+        self.assertEqual(first, 'custom_storage')
  84
+        second = self.storage.save('custom_storage', ContentFile('more contents'))
  85
+        self.assertEqual(second, 'custom_storage.2')
  86
+        self.storage.delete(first)
  87
+        self.storage.delete(second)
  88
+
  89
+class UnicodeFileNameTests(unittest.TestCase):
  90
+    def test_unicode_file_names(self):
  91
+        """
  92
+        Regression test for #8156: files with unicode names I can't quite figure
  93
+        out the encoding situation between doctest and this file, but the actual
  94
+        repr doesn't matter; it just shouldn't return a unicode object.
  95
+        """
  96
+        uf = UploadedFile(name=u'¿Cómo?',content_type='text')
  97
+        self.assertEqual(type(uf.__repr__()), str)
  98
+
  99
+# Tests for a race condition on file saving (#4948).
  100
+# This is written in such a way that it'll always pass on platforms
  101
+# without threading.
  102
+
103 103
 class SlowFile(ContentFile):
104 104
     def chunks(self):
105 105
         time.sleep(1)
@@ -180,3 +180,49 @@ def test_first_character_dot(self):
180 180
             self.assertTrue(os.path.exists(os.path.join(self.storage_dir, 'dotted.path/_.test')))
181 181
         else:
182 182
             self.assertTrue(os.path.exists(os.path.join(self.storage_dir, 'dotted.path/.test_')))
  183
+
  184
+if Image is not None:
  185
+    class DimensionClosingBug(TestCase):
  186
+        """
  187
+        Test that get_image_dimensions() properly closes files (#8817)
  188
+        """
  189
+        def test_not_closing_of_files(self):
  190
+            """
  191
+            Open files passed into get_image_dimensions() should stay opened.
  192
+            """
  193
+            empty_io = StringIO()
  194
+            try:
  195
+                get_image_dimensions(empty_io)
  196
+            finally:
  197
+                self.assert_(not empty_io.closed)
  198
+
  199
+        def test_closing_of_filenames(self):
  200
+            """
  201
+            get_image_dimensions() called with a filename should closed the file.
  202
+            """
  203
+            # We need to inject a modified open() builtin into the images module
  204
+            # that checks if the file was closed properly if the function is
  205
+            # called with a filename instead of an file object.
  206
+            # get_image_dimensions will call our catching_open instead of the
  207
+            # regular builtin one.
  208
+
  209
+            class FileWrapper(object):
  210
+                _closed = []
  211
+                def __init__(self, f):
  212
+                    self.f = f
  213
+                def __getattr__(self, name):
  214
+                    return getattr(self.f, name)
  215
+                def close(self):
  216
+                    self._closed.append(True)
  217
+                    self.f.close()
  218
+
  219
+            def catching_open(*args):
  220
+                return FileWrapper(open(*args))
  221
+
  222
+            from django.core.files import images
  223
+            images.open = catching_open
  224
+            try:
  225
+                get_image_dimensions(os.path.join(os.path.dirname(__file__), "test1.png"))
  226
+            finally:
  227
+                del images.open
  228
+            self.assert_(FileWrapper._closed)

0 notes on commit 7935231

Please sign in to comment.
Something went wrong with that request. Please try again.