Google,S3,Azure: Respect max_length when overwriting files

Closes #513

storages/backends/azure_storage.py

@@ -15,7 +15,9 @@
 from django.utils.deconstruct import deconstructible
 from django.utils.encoding import force_bytes, force_text
 
-from storages.utils import clean_name, safe_join, setting
+from storages.utils import (
+    clean_name, get_available_overwrite_name, safe_join, setting,
+)
 
 
 @deconstructible
@@ -190,10 +192,10 @@ def get_available_name(self, name, max_length=_AZURE_NAME_MAX_LEN):
         Returns a filename that's free on the target storage system, and
         available for new content to be written to.
         """
+        name = self.get_valid_name(name)
         if self.overwrite_files:
-            return self.get_valid_name(name)
-        return super(AzureStorage, self).get_available_name(
-            self.get_valid_name(name), max_length)
+            return get_available_overwrite_name(name, max_length)
+        return super(AzureStorage, self).get_available_name(name, max_length)
 
     def exists(self, name):
         return self.service.exists(

storages/backends/gcloud.py

@@ -8,7 +8,10 @@
 from django.utils.deconstruct import deconstructible
 from django.utils.encoding import force_bytes, smart_str
 
-from storages.utils import check_location, clean_name, safe_join, setting
+from storages.utils import (
+    check_location, clean_name, get_available_overwrite_name, safe_join,
+    setting,
+)
 
 try:
     from google.cloud.storage.client import Client
@@ -255,7 +258,7 @@ def url(self, name):
         return blob.public_url
 
     def get_available_name(self, name, max_length=None):
+        name = clean_name(name)
         if self.file_overwrite:
-            name = clean_name(name)
-            return name
+            return get_available_overwrite_name(name, max_length)
         return super(GoogleCloudStorage, self).get_available_name(name, max_length)

storages/backends/s3boto.py

@@ -15,7 +15,8 @@
 from django.utils.six import BytesIO
 
 from storages.utils import (
-    check_location, clean_name, lookup_env, safe_join, setting,
+    check_location, clean_name, get_available_overwrite_name, lookup_env,
+    safe_join, setting,
 )
 
 try:
@@ -489,7 +490,7 @@ def url(self, name, headers=None, response_headers=None, expire=None):
 
     def get_available_name(self, name, max_length=None):
         """ Overwrite existing file with the same name. """
+        name = self._clean_name(name)
         if self.file_overwrite:
-            name = self._clean_name(name)
-            return name
+            return get_available_overwrite_name(name, max_length)
         return super(S3BotoStorage, self).get_available_name(name, max_length)

storages/backends/s3boto3.py

@@ -18,7 +18,10 @@
 from django.utils.six.moves.urllib import parse as urlparse
 from django.utils.timezone import is_naive, localtime
 
-from storages.utils import check_location, lookup_env, safe_join, setting
+from storages.utils import (
+    check_location, get_available_overwrite_name, lookup_env, safe_join,
+    setting,
+)
 
 try:
     import boto3.session
@@ -615,7 +618,7 @@ def url(self, name, parameters=None, expire=None):
 
     def get_available_name(self, name, max_length=None):
         """Overwrite existing file with the same name."""
+        name = self._clean_name(name)
         if self.file_overwrite:
-            name = self._clean_name(name)
-            return name
+            return get_available_overwrite_name(name, max_length)
         return super(S3Boto3Storage, self).get_available_name(name, max_length)

storages/utils.py

@@ -2,7 +2,9 @@
 import posixpath
 
 from django.conf import settings
-from django.core.exceptions import ImproperlyConfigured
+from django.core.exceptions import (
+    ImproperlyConfigured, SuspiciousFileOperation,
+)
 from django.utils.encoding import force_text
 
 
@@ -97,3 +99,22 @@ def lookup_env(names):
         value = os.environ.get(name)
         if value:
             return value
+
+
+def get_available_overwrite_name(name, max_length):
+    if max_length is None or len(name) < max_length:
+        return name
+
+    # Adapted from Django
+    dir_name, file_name = os.path.split(name)
+    file_root, file_ext = os.path.splitext(file_name)
+    truncation = len(name) - max_length
+
+    file_root = file_root[:-truncation]
+    if not file_root:
+        raise SuspiciousFileOperation(
+            'Storage tried to truncate away entire filename "%s". '
+            'Please make sure that the corresponding file field '
+            'allows sufficient "max_length".' % name
+        )
+    return os.path.join(dir_name, "%s%s" % (file_root, file_ext))

tests/test_gcloud.py

@@ -8,7 +8,9 @@
 import datetime
 import mimetypes
 
-from django.core.exceptions import ImproperlyConfigured
+from django.core.exceptions import (
+    ImproperlyConfigured, SuspiciousFileOperation,
+)
 from django.core.files.base import ContentFile
 from django.test import TestCase
 from django.utils import timezone
@@ -348,6 +350,14 @@ def test_get_available_name_unicode(self):
         filename = 'ủⓝï℅ⅆℇ.txt'
         self.assertEqual(self.storage.get_available_name(filename), filename)
 
+    def test_get_available_name_overwrite_maxlength(self):
+        self.storage.file_overwrite = True
+
+        self.assertEqual(self.storage.get_available_name('test/foo.txt', 11), 'test/fo.txt')
+        self.assertEqual(self.storage.get_available_name('test_a/foobar.txt', None), 'test_a/foobar.txt')
+        with self.assertRaises(SuspiciousFileOperation):
+            self.storage.get_available_name('test_a/foobar.txt', 10)
+
     def test_cache_control(self):
         data = 'This is some test content.'
         filename = 'cache_control_file.txt'