[general] Write overwriting files in terms of .exists()

storages/backends/azure_storage.py

@@ -19,7 +19,6 @@
 
 from storages.base import BaseStorage
 from storages.utils import clean_name
-from storages.utils import get_available_overwrite_name
 from storages.utils import safe_join
 from storages.utils import setting
 from storages.utils import to_bytes
@@ -232,20 +231,13 @@ def _get_valid_path(self, name):
     def _open(self, name, mode="rb"):
         return AzureStorageFile(name, mode, self)
 
-    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 = clean_name(name)
-        if self.overwrite_files:
-            return get_available_overwrite_name(name, max_length)
-        return super().get_available_name(name, max_length)
-
     def exists(self, name):
         if not name:
             return True
 
+        if self.overwrite_files:
+            return False
+
         blob_client = self.client.get_blob_client(self._get_valid_path(name))
         return blob_client.exists()
 

storages/backends/dropbox.py

@@ -19,7 +19,6 @@
 from dropbox.files import WriteMode
 
 from storages.base import BaseStorage
-from storages.utils import get_available_overwrite_name
 from storages.utils import setting
 
 _DEFAULT_TIMEOUT = 100
@@ -130,6 +129,9 @@ def delete(self, name):
         self.client.files_delete(self._full_path(name))
 
     def exists(self, name):
+        if self.write_mode == "overwrite":
+            return False
+
         try:
             return bool(self.client.files_get_metadata(self._full_path(name)))
         except ApiError:
@@ -199,12 +201,5 @@ def _chunked_upload(self, content, dest_path):
                 )
                 cursor.offset = content.tell()
 
-    def get_available_name(self, name, max_length=None):
-        """Overwrite existing file with the same name."""
-        name = self._full_path(name)
-        if self.write_mode == "overwrite":
-            return get_available_overwrite_name(name, max_length)
-        return super().get_available_name(name, max_length)
-
 
 DropBoxStorage = DropboxStorage

storages/backends/gcloud.py

@@ -14,7 +14,6 @@
 from storages.compress import CompressedFileMixin
 from storages.utils import check_location
 from storages.utils import clean_name
-from storages.utils import get_available_overwrite_name
 from storages.utils import safe_join
 from storages.utils import setting
 from storages.utils import to_bytes
@@ -243,6 +242,9 @@ def exists(self, name):
             except NotFound:
                 return False
 
+        if self.file_overwrite:
+            return False
+
         name = self._normalize_name(clean_name(name))
         return bool(self.bucket.get_blob(name))
 
@@ -333,9 +335,3 @@ def url(self, name, parameters=None):
                     params[key] = value
 
             return blob.generate_signed_url(**params)
-
-    def get_available_name(self, name, max_length=None):
-        name = clean_name(name)
-        if self.file_overwrite:
-            return get_available_overwrite_name(name, max_length)
-        return super().get_available_name(name, max_length)

storages/backends/s3.py

@@ -23,7 +23,6 @@
 from storages.utils import ReadBytesWrapper
 from storages.utils import check_location
 from storages.utils import clean_name
-from storages.utils import get_available_overwrite_name
 from storages.utils import is_seekable
 from storages.utils import lookup_env
 from storages.utils import safe_join
@@ -580,6 +579,9 @@ def delete(self, name):
             raise
 
     def exists(self, name):
+        if self.file_overwrite:
+            return False
+
         name = self._normalize_name(clean_name(name))
         try:
             self.connection.meta.client.head_object(Bucket=self.bucket_name, Key=name)
@@ -696,13 +698,6 @@ def url(self, name, parameters=None, expire=None, http_method=None):
         )
         return url
 
-    def get_available_name(self, name, max_length=None):
-        """Overwrite existing file with the same name."""
-        name = clean_name(name)
-        if self.file_overwrite:
-            return get_available_overwrite_name(name, max_length)
-        return super().get_available_name(name, max_length)
-
 
 class S3StaticStorage(S3Storage):
     """Querystring auth must be disabled so that url() returns a consistent output."""

storages/utils.py

@@ -4,7 +4,6 @@
 
 from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
-from django.core.exceptions import SuspiciousFileOperation
 from django.core.files.utils import FileProxyMixin
 from django.utils.encoding import force_bytes
 
@@ -120,25 +119,6 @@ def lookup_env(names):
             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, "{}{}".format(file_root, file_ext))
-
-
 def is_seekable(file_object):
     return not hasattr(file_object, "seekable") or file_object.seekable()
 

tests/test_azure.py

@@ -2,9 +2,7 @@
 from datetime import timedelta
 from unittest import mock
 
-import django
 from azure.storage.blob import BlobProperties
-from django.core.exceptions import SuspiciousOperation
 from django.core.files.base import ContentFile
 from django.test import TestCase
 from django.test import override_settings
@@ -70,65 +68,6 @@ def test_get_valid_path_idempotency(self):
             self.storage._get_valid_path(some_path),
         )
 
-    def test_get_available_name(self):
-        self.storage.overwrite_files = False
-        client_mock = mock.MagicMock()
-        client_mock.exists.side_effect = [True, False]
-        self.storage._client.get_blob_client.return_value = client_mock
-        name = self.storage.get_available_name("foo.txt")
-        self.assertTrue(name.startswith("foo_"))
-        self.assertTrue(name.endswith(".txt"))
-        self.assertTrue(len(name) > len("foo.txt"))
-        self.assertEqual(client_mock.exists.call_count, 2)
-
-    def test_get_available_name_first(self):
-        self.storage.overwrite_files = False
-        client_mock = mock.MagicMock()
-        client_mock.exists.return_value = False
-        self.storage._client.get_blob_client.return_value = client_mock
-        self.assertEqual(
-            self.storage.get_available_name("foo bar baz.txt"), "foo bar baz.txt"
-        )
-        self.assertEqual(client_mock.exists.call_count, 1)
-
-    def test_get_available_name_max_len(self):
-        self.storage.overwrite_files = False
-        # if you wonder why this is, file-system
-        # storage will raise when file name is too long as well,
-        # the form should validate this
-        client_mock = mock.MagicMock()
-        client_mock.exists.side_effect = [True, False]
-        self.storage._client.get_blob_client.return_value = client_mock
-        self.assertRaises(ValueError, self.storage.get_available_name, "a" * 1025)
-        name = self.storage.get_available_name(
-            "a" * 1000, max_length=100
-        )  # max_len == 1024
-        self.assertEqual(len(name), 100)
-        self.assertTrue("_" in name)
-        self.assertEqual(client_mock.exists.call_count, 2)
-
-    def test_get_available_invalid(self):
-        self.storage.overwrite_files = False
-        self.storage._client.exists.return_value = False
-        if django.VERSION[:2] == (3, 0):
-            # Django 2.2.21 added this security fix:
-            # https://docs.djangoproject.com/en/3.2/releases/2.2.21/#cve-2021-31542-potential-directory-traversal-via-uploaded-files
-            # It raises SuspiciousOperation before we get to our ValueError.
-            # The fix wasn't applied to 3.0 (no longer in support), but was applied to
-            # 3.1 & 3.2.
-            self.assertRaises(ValueError, self.storage.get_available_name, "")
-            self.assertRaises(ValueError, self.storage.get_available_name, "/")
-            self.assertRaises(ValueError, self.storage.get_available_name, ".")
-            self.assertRaises(ValueError, self.storage.get_available_name, "///")
-        else:
-            self.assertRaises(SuspiciousOperation, self.storage.get_available_name, "")
-            self.assertRaises(SuspiciousOperation, self.storage.get_available_name, "/")
-            self.assertRaises(SuspiciousOperation, self.storage.get_available_name, ".")
-            self.assertRaises(
-                SuspiciousOperation, self.storage.get_available_name, "///"
-            )
-        self.assertRaises(ValueError, self.storage.get_available_name, "...")
-
     def test_url(self):
         blob_mock = mock.MagicMock()
         blob_mock.url = "https://ret_foo.blob.core.windows.net/test/some%20blob"
@@ -357,11 +296,15 @@ def test_storage_open_write(self):
         )
 
     def test_storage_exists(self):
-        blob_name = "blob"
-        client_mock = mock.MagicMock()
-        self.storage._client.get_blob_client.return_value = client_mock
-        self.assertTrue(self.storage.exists(blob_name))
-        self.assertEqual(client_mock.exists.call_count, 1)
+        overwrite_files = [True, False]
+        for owf in overwrite_files:
+            self.storage.overwrite_files = owf
+            client_mock = mock.MagicMock()
+            self.storage._client.get_blob_client.return_value = client_mock
+            assert_ = self.assertFalse if owf else self.assertTrue
+            call_count = 0 if owf else 1
+            assert_(self.storage.exists("blob"))
+            self.assertEqual(client_mock.exists.call_count, call_count)
 
     def test_delete_blob(self):
         self.storage.delete("name")

tests/test_dropbox.py

@@ -92,6 +92,11 @@ def test_not_exists(self, *args):
         exists = self.storage.exists("bar")
         self.assertFalse(exists)
 
+    @mock.patch("dropbox.Dropbox.files_get_metadata", return_value=[FILE_METADATA_MOCK])
+    def test_exists_overwrite_mode(self, *args):
+        self.storage.write_mode = "overwrite"
+        self.assertFalse(self.storage.exists("foo"))
+
     @mock.patch("dropbox.Dropbox.files_list_folder", return_value=FILES_MOCK)
     def test_listdir(self, *args):
         dirs, files = self.storage.listdir("/")

tests/test_gcloud.py

@@ -167,6 +167,7 @@ def test_delete(self):
         )
 
     def test_exists(self):
+        self.storage.file_overwrite = False
         self.storage._bucket = mock.MagicMock()
         self.assertTrue(self.storage.exists(self.filename))
         self.storage._bucket.get_blob.assert_called_with(self.filename)
@@ -186,6 +187,10 @@ def test_exists_bucket(self):
         # exists('') should return True if the bucket exists
         self.assertTrue(self.storage.exists(""))
 
+    def test_exists_file_overwrite(self):
+        self.storage.file_overwrite = True
+        self.assertFalse(self.storage.exists(self.filename))
+
     def test_listdir(self):
         file_names = ["some/path/1.txt", "2.txt", "other/path/3.txt", "4.txt"]
         subdir = ""

tests/test_s3.py

@@ -496,13 +496,15 @@ def test_storage_write_beyond_buffer_size(self):
         )
 
     def test_storage_exists(self):
+        self.storage.file_overwrite = False
         self.assertTrue(self.storage.exists("file.txt"))
         self.storage.connection.meta.client.head_object.assert_called_with(
             Bucket=self.storage.bucket_name,
             Key="file.txt",
         )
 
     def test_storage_exists_false(self):
+        self.storage.file_overwrite = False
         self.storage.connection.meta.client.head_object.side_effect = ClientError(
             {"Error": {}, "ResponseMetadata": {"HTTPStatusCode": 404}},
             "HeadObject",
@@ -514,6 +516,7 @@ def test_storage_exists_false(self):
         )
 
     def test_storage_exists_other_error_reraise(self):
+        self.storage.file_overwrite = False
         self.storage.connection.meta.client.head_object.side_effect = ClientError(
             {"Error": {}, "ResponseMetadata": {"HTTPStatusCode": 403}},
             "HeadObject",
@@ -525,6 +528,10 @@ def test_storage_exists_other_error_reraise(self):
             cm.exception.response["ResponseMetadata"]["HTTPStatusCode"], 403
         )
 
+    def test_storage_exists_overwrite(self):
+        self.storage.file_overwrite = True
+        self.assertFalse(self.storage.exists("foo"))
+
     def test_storage_delete(self):
         self.storage.delete("path/to/file.txt")
         self.storage.bucket.Object.assert_called_with("path/to/file.txt")

tests/test_utils.py

@@ -4,11 +4,9 @@
 import pathlib
 
 from django.conf import settings
-from django.core.exceptions import SuspiciousFileOperation
 from django.test import TestCase
 
 from storages import utils
-from storages.utils import get_available_overwrite_name as gaon
 
 
 class SettingTest(TestCase):
@@ -118,29 +116,6 @@ def test_with_base_url_join_nothing(self):
         self.assertEqual(path, "base_url/")
 
 
-class TestGetAvailableOverwriteName(TestCase):
-    def test_maxlength_is_none(self):
-        name = "superlong/file/with/path.txt"
-        self.assertEqual(gaon(name, None), name)
-
-    def test_maxlength_equals_name(self):
-        name = "parent/child.txt"
-        self.assertEqual(gaon(name, len(name)), name)
-
-    def test_maxlength_is_greater_than_name(self):
-        name = "parent/child.txt"
-        self.assertEqual(gaon(name, len(name) + 1), name)
-
-    def test_maxlength_less_than_name(self):
-        name = "parent/child.txt"
-        self.assertEqual(gaon(name, len(name) - 1), "parent/chil.txt")
-
-    def test_truncates_away_filename_raises(self):
-        name = "parent/child.txt"
-        with self.assertRaises(SuspiciousFileOperation):
-            gaon(name, len(name) - 5)
-
-
 class TestReadBytesWrapper(TestCase):
     def test_with_bytes_file(self):
         file = io.BytesIO(b"abcd")