Fixed #34110 -- Added in-memory file storage.

Thanks Paolo Melchiorre, Carlton Gibson, and Mariusz Felisiak for
reviews.

1 files changed, 290 insertions, 0 deletions

diff --git a/tests/file_storage/test_inmemory_storage.py b/tests/file_storage/test_inmemory_storage.py
new file mode 100644
index 0000000000..b1d2e2e5a1
--- /dev/null
+++ b/tests/file_storage/test_inmemory_storage.py
@@ -0,0 +1,290 @@
+import os
+import sys
+import time
+import unittest
+
+from django.core.files.base import ContentFile
+from django.core.files.storage import InMemoryStorage
+from django.core.files.uploadedfile import TemporaryUploadedFile
+from django.test import SimpleTestCase, override_settings
+
+
+class MemoryStorageIOTests(unittest.TestCase):
+    def setUp(self):
+        self.storage = InMemoryStorage()
+
+    def test_write_string(self):
+        with self.storage.open("file.txt", "w") as fd:
+            fd.write("hello")
+        with self.storage.open("file.txt", "r") as fd:
+            self.assertEqual(fd.read(), "hello")
+        with self.storage.open("file.dat", "wb") as fd:
+            fd.write(b"hello")
+        with self.storage.open("file.dat", "rb") as fd:
+            self.assertEqual(fd.read(), b"hello")
+
+    def test_convert_str_to_bytes_and_back(self):
+        """InMemoryStorage handles conversion from str to bytes and back."""
+        with self.storage.open("file.txt", "w") as fd:
+            fd.write("hello")
+        with self.storage.open("file.txt", "rb") as fd:
+            self.assertEqual(fd.read(), b"hello")
+        with self.storage.open("file.dat", "wb") as fd:
+            fd.write(b"hello")
+        with self.storage.open("file.dat", "r") as fd:
+            self.assertEqual(fd.read(), "hello")
+
+    def test_open_missing_file(self):
+        self.assertRaises(FileNotFoundError, self.storage.open, "missing.txt")
+
+    def test_open_dir_as_file(self):
+        with self.storage.open("a/b/file.txt", "w") as fd:
+            fd.write("hello")
+        self.assertRaises(IsADirectoryError, self.storage.open, "a/b")
+
+    def test_file_saving(self):
+        self.storage.save("file.txt", ContentFile("test"))
+        self.assertEqual(self.storage.open("file.txt", "r").read(), "test")
+
+        self.storage.save("file.dat", ContentFile(b"test"))
+        self.assertEqual(self.storage.open("file.dat", "rb").read(), b"test")
+
+    @unittest.skipIf(
+        sys.platform == "win32", "Windows doesn't support moving open files."
+    )
+    def test_removing_temporary_file_after_save(self):
+        """A temporary file is removed when saved into storage."""
+        with TemporaryUploadedFile("test", "text/plain", 1, "utf8") as file:
+            self.storage.save("test.txt", file)
+            self.assertFalse(os.path.exists(file.temporary_file_path()))
+
+    def test_large_file_saving(self):
+        large_file = ContentFile("A" * ContentFile.DEFAULT_CHUNK_SIZE * 3)
+        self.storage.save("file.txt", large_file)
+
+    def test_file_size(self):
+        """
+        File size is equal to the size of bytes-encoded version of the saved
+        data.
+        """
+        self.storage.save("file.txt", ContentFile("test"))
+        self.assertEqual(self.storage.size("file.txt"), 4)
+
+        # A unicode char encoded to UTF-8 takes 2 bytes.
+        self.storage.save("unicode_file.txt", ContentFile("è"))
+        self.assertEqual(self.storage.size("unicode_file.txt"), 2)
+
+        self.storage.save("file.dat", ContentFile(b"\xf1\xf1"))
+        self.assertEqual(self.storage.size("file.dat"), 2)
+
+    def test_listdir(self):
+        self.assertEqual(self.storage.listdir(""), ([], []))
+
+        self.storage.save("file_a.txt", ContentFile("test"))
+        self.storage.save("file_b.txt", ContentFile("test"))
+        self.storage.save("dir/file_c.txt", ContentFile("test"))
+
+        dirs, files = self.storage.listdir("")
+        self.assertEqual(sorted(files), ["file_a.txt", "file_b.txt"])
+        self.assertEqual(dirs, ["dir"])
+
+    def test_list_relative_path(self):
+        self.storage.save("a/file.txt", ContentFile("test"))
+
+        _dirs, files = self.storage.listdir("./a/./.")
+        self.assertEqual(files, ["file.txt"])
+
+    def test_exists(self):
+        self.storage.save("dir/subdir/file.txt", ContentFile("test"))
+        self.assertTrue(self.storage.exists("dir"))
+        self.assertTrue(self.storage.exists("dir/subdir"))
+        self.assertTrue(self.storage.exists("dir/subdir/file.txt"))
+
+    def test_delete(self):
+        """Deletion handles both files and directory trees."""
+        self.storage.save("dir/subdir/file.txt", ContentFile("test"))
+        self.storage.save("dir/subdir/other_file.txt", ContentFile("test"))
+        self.assertTrue(self.storage.exists("dir/subdir/file.txt"))
+        self.assertTrue(self.storage.exists("dir/subdir/other_file.txt"))
+
+        self.storage.delete("dir/subdir/other_file.txt")
+        self.assertFalse(self.storage.exists("dir/subdir/other_file.txt"))
+
+        self.storage.delete("dir/subdir")
+        self.assertFalse(self.storage.exists("dir/subdir/file.txt"))
+        self.assertFalse(self.storage.exists("dir/subdir"))
+
+    def test_delete_missing_file(self):
+        self.storage.delete("missing_file.txt")
+        self.storage.delete("missing_dir/missing_file.txt")
+
+    def test_file_node_cannot_have_children(self):
+        """Navigate to children of a file node raises FileExistsError."""
+        self.storage.save("file.txt", ContentFile("test"))
+        self.assertRaises(FileExistsError, self.storage.listdir, "file.txt/child_dir")
+        self.assertRaises(
+            FileExistsError,
+            self.storage.save,
+            "file.txt/child_file.txt",
+            ContentFile("test"),
+        )
+
+    @override_settings(MEDIA_URL=None)
+    def test_url(self):
+        self.assertRaises(ValueError, self.storage.url, ("file.txt",))
+
+        storage = InMemoryStorage(base_url="http://www.example.com")
+        self.assertEqual(storage.url("file.txt"), "http://www.example.com/file.txt")
+
+    def test_url_with_none_filename(self):
+        storage = InMemoryStorage(base_url="/test_media_url/")
+        self.assertEqual(storage.url(None), "/test_media_url/")
+
+
+class MemoryStorageTimesTests(unittest.TestCase):
+    def setUp(self):
+        self.storage = InMemoryStorage()
+
+    def test_file_modified_time(self):
+        """
+        File modified time should change after file changing
+        """
+        self.storage.save("file.txt", ContentFile("test"))
+        modified_time = self.storage.get_modified_time("file.txt")
+
+        time.sleep(0.1)
+
+        with self.storage.open("file.txt", "w") as fd:
+            fd.write("new content")
+
+        new_modified_time = self.storage.get_modified_time("file.txt")
+        self.assertTrue(new_modified_time > modified_time)
+
+    def test_file_accessed_time(self):
+        """File accessed time should chage after consecutive opening."""
+        self.storage.save("file.txt", ContentFile("test"))
+        accessed_time = self.storage.get_accessed_time("file.txt")
+
+        time.sleep(0.1)
+
+        self.storage.open("file.txt", "r")
+        new_accessed_time = self.storage.get_accessed_time("file.txt")
+        self.assertGreater(new_accessed_time, accessed_time)
+
+    def test_file_created_time(self):
+        """File creation time should not change after I/O operations."""
+        self.storage.save("file.txt", ContentFile("test"))
+        created_time = self.storage.get_created_time("file.txt")
+
+        time.sleep(0.1)
+
+        # File opening doesn't change creation time.
+        file = self.storage.open("file.txt", "r")
+        after_open_created_time = self.storage.get_created_time("file.txt")
+        self.assertEqual(after_open_created_time, created_time)
+        # Writing to a file doesn't change its creation time.
+        file.write("New test")
+        self.storage.save("file.txt", file)
+        after_write_created_time = self.storage.get_created_time("file.txt")
+        self.assertEqual(after_write_created_time, created_time)
+
+    def test_directory_times_changing_after_file_creation(self):
+        """
+        Directory modified and accessed time should change when a new file is
+        created inside.
+        """
+        self.storage.save("dir/file1.txt", ContentFile("test"))
+        created_time = self.storage.get_created_time("dir")
+        modified_time = self.storage.get_modified_time("dir")
+        accessed_time = self.storage.get_accessed_time("dir")
+
+        time.sleep(0.1)
+
+        self.storage.save("dir/file2.txt", ContentFile("test"))
+        new_modified_time = self.storage.get_modified_time("dir")
+        new_accessed_time = self.storage.get_accessed_time("dir")
+        after_file_creation_created_time = self.storage.get_created_time("dir")
+        self.assertGreater(new_modified_time, modified_time)
+        self.assertGreater(new_accessed_time, accessed_time)
+        self.assertEqual(created_time, after_file_creation_created_time)
+
+    def test_directory_times_changing_after_file_deletion(self):
+        """
+        Directory modified and accessed time should change when a new file is
+        deleted inside.
+        """
+        self.storage.save("dir/file.txt", ContentFile("test"))
+        created_time = self.storage.get_created_time("dir")
+        modified_time = self.storage.get_modified_time("dir")
+        accessed_time = self.storage.get_accessed_time("dir")
+
+        time.sleep(0.1)
+
+        self.storage.delete("dir/file.txt")
+        new_modified_time = self.storage.get_modified_time("dir")
+        new_accessed_time = self.storage.get_accessed_time("dir")
+        after_file_deletion_created_time = self.storage.get_created_time("dir")
+        self.assertGreater(new_modified_time, modified_time)
+        self.assertGreater(new_accessed_time, accessed_time)
+        self.assertEqual(created_time, after_file_deletion_created_time)
+
+
+class InMemoryStorageTests(SimpleTestCase):
+    def test_deconstruction(self):
+        storage = InMemoryStorage()
+        path, args, kwargs = storage.deconstruct()
+        self.assertEqual(path, "django.core.files.storage.InMemoryStorage")
+        self.assertEqual(args, ())
+        self.assertEqual(kwargs, {})
+
+        kwargs_orig = {
+            "location": "/custom_path",
+            "base_url": "http://myfiles.example.com/",
+            "file_permissions_mode": "0o755",
+            "directory_permissions_mode": "0o600",
+        }
+        storage = InMemoryStorage(**kwargs_orig)
+        path, args, kwargs = storage.deconstruct()
+        self.assertEqual(kwargs, kwargs_orig)
+
+    @override_settings(
+        MEDIA_ROOT="media_root",
+        MEDIA_URL="media_url/",
+        FILE_UPLOAD_PERMISSIONS=0o777,
+        FILE_UPLOAD_DIRECTORY_PERMISSIONS=0o777,
+    )
+    def test_setting_changed(self):
+        """
+        Properties using settings values as defaults should be updated on
+        referenced settings change while specified values should be unchanged.
+        """
+        storage = InMemoryStorage(
+            location="explicit_location",
+            base_url="explicit_base_url/",
+            file_permissions_mode=0o666,
+            directory_permissions_mode=0o666,
+        )
+        defaults_storage = InMemoryStorage()
+        settings = {
+            "MEDIA_ROOT": "overridden_media_root",
+            "MEDIA_URL": "/overridden_media_url/",
+            "FILE_UPLOAD_PERMISSIONS": 0o333,
+            "FILE_UPLOAD_DIRECTORY_PERMISSIONS": 0o333,
+        }
+        with self.settings(**settings):
+            self.assertEqual(storage.base_location, "explicit_location")
+            self.assertIn("explicit_location", storage.location)
+            self.assertEqual(storage.base_url, "explicit_base_url/")
+            self.assertEqual(storage.file_permissions_mode, 0o666)
+            self.assertEqual(storage.directory_permissions_mode, 0o666)
+            self.assertEqual(defaults_storage.base_location, settings["MEDIA_ROOT"])
+            self.assertIn(settings["MEDIA_ROOT"], defaults_storage.location)
+            self.assertEqual(defaults_storage.base_url, settings["MEDIA_URL"])
+            self.assertEqual(
+                defaults_storage.file_permissions_mode,
+                settings["FILE_UPLOAD_PERMISSIONS"],
+            )
+            self.assertEqual(
+                defaults_storage.directory_permissions_mode,
+                settings["FILE_UPLOAD_DIRECTORY_PERMISSIONS"],
+            )