Fix ensure_versioned_database_exists to not open a file for writing

Author: Jakoma02

contentcuration/contentcuration/tests/utils/restricted_filesystemstorage.py

@@ -0,0 +1,32 @@
+from django.core.files.storage import FileSystemStorage
+
+
+class RestrictedFileSystemStorage:
+    """
+    A wrapper around FileSystemStorage that is restricted to more closely match
+    the behavior of S3Storage which is used in production. In particuler,
+    it does not expose the `path` method, and opening files for writing
+    is not allowed.
+    This cannot be solved by just mocking the `path` method, because
+    it is used by the `FileSystemStorage` class internally.
+    """
+
+    def __init__(self, *args, **kwargs):
+        self._inner = FileSystemStorage(*args, **kwargs)
+
+    def __getattr__(self, name):
+        if name == "path":
+            raise NotImplementedError(
+                "The 'path' method is intentionally not available."
+            )
+        return getattr(self._inner, name)
+
+    def open(self, name, mode="rb"):
+        if "w" in mode:
+            raise ValueError(
+                "Opening files for writing will not be available in production."
+            )
+        return self._inner.open(name, mode)
+
+    def __dir__(self):
+        return [x for x in dir(self._inner) if x != "path"]

contentcuration/contentcuration/tests/utils/test_publish.py

@@ -3,10 +3,12 @@
 from unittest import mock
 
 from django.conf import settings
-from django.core.files.storage import FileSystemStorage
 
 from contentcuration.tests import testdata
 from contentcuration.tests.base import StudioTestCase
+from contentcuration.tests.utils.restricted_filesystemstorage import (
+    RestrictedFileSystemStorage,
+)
 from contentcuration.utils.publish import ensure_versioned_database_exists
 
 
@@ -17,7 +19,7 @@ def setUp(self):
         self._temp_directory_ctx = tempfile.TemporaryDirectory()
         self.test_db_root_dir = self._temp_directory_ctx.__enter__()
 
-        storage = FileSystemStorage(location=self.test_db_root_dir)
+        storage = RestrictedFileSystemStorage(location=self.test_db_root_dir)
 
         self._storage_patch_ctx = mock.patch(
             "contentcuration.utils.publish.storage",

contentcuration/contentcuration/tests/utils/test_restricted_filesystemstorage.py

@@ -0,0 +1,44 @@
+import tempfile
+
+from django.core.files.base import ContentFile
+from django.test import TestCase
+
+from contentcuration.tests.utils.restricted_filesystemstorage import (
+    RestrictedFileSystemStorage,
+)
+
+
+class RestrictedFileSystemStorageTestCase(TestCase):
+    # Sanity-checks that the RestrictedFileSystemStorage wrapper used in tests
+    # works as expected, not actually testing application code
+
+    def setUp(self):
+        super().setUp()
+
+        self._temp_directory_ctx = tempfile.TemporaryDirectory()
+        self.temp_dir = self._temp_directory_ctx.__enter__()
+
+        self.storage = RestrictedFileSystemStorage(location=self.temp_dir)
+
+    def tearDown(self):
+        self._temp_directory_ctx.__exit__(None, None, None)
+        super().tearDown()
+
+    def test_opening_for_read_works(self):
+        test_content = "test content"
+
+        self.storage.save("filename", ContentFile(test_content))
+        with self.storage.open("filename", "r") as f:
+            content = f.read()
+            self.assertEqual(content, test_content)
+
+    def test_opening_for_write_does_not_work(self):
+        test_content = "test content"
+
+        with self.assertRaises(ValueError):
+            with self.storage.open("filename", "w") as f:
+                f.write(test_content)
+
+    def test_path_does_not_work(self):
+        with self.assertRaises(NotImplementedError):
+            self.storage.path("filename")

contentcuration/contentcuration/utils/publish.py

@@ -2,7 +2,6 @@
 import json
 import logging as logmodule
 import os
-import shutil
 import tempfile
 import time
 import uuid
@@ -1119,8 +1118,7 @@ def ensure_versioned_database_exists(channel):
             )
 
         with storage.open(unversioned_db_storage_path, "rb") as unversioned_db_file:
-            with storage.open(versioned_db_storage_path, "wb") as versioned_db_file:
-                shutil.copyfileobj(unversioned_db_file, versioned_db_file)
+            storage.save(versioned_db_storage_path, unversioned_db_file)
 
         logging.info(
             f"Versioned database for channel {channel.id} did not exist, copied the unversioned database to {versioned_db_storage_path}."

contentcuration/kolibri_public/tests/test_export_channels_to_kolibri_public.py

@@ -5,7 +5,6 @@
 from unittest import mock
 
 from django.conf import settings
-from django.core.files.storage import FileSystemStorage
 from django.core.management import call_command
 from django.test import TestCase
 from kolibri_content.apps import KolibriContentConfig
@@ -18,59 +17,9 @@
 )
 
 from contentcuration.models import Country
-
-
-class FileSystemStorageWithoutPath:
-    """
-    A wrapper around FileSystemStorage that does not expose the `path` method,
-    as this will not be available in production where S3Storage is used.
-    This cannot be solved by just mocking the `path` method, because
-    it is used by the `FileSystemStorage` class internally.
-    """
-
-    def __init__(self, *args, **kwargs):
-        self._inner = FileSystemStorage(*args, **kwargs)
-
-    def __getattr__(self, name):
-        if name == "path":
-            raise NotImplementedError(
-                "The 'path' method is intentionally not available."
-            )
-        return getattr(self._inner, name)
-
-    def __dir__(self):
-        return [x for x in dir(self._inner) if x != "path"]
-
-
-class FileSystemStorageWithoutPathTestCase(TestCase):
-    # Sanity-checks that the wrapper above works as expected,
-    # not actually testing application code
-
-    def setUp(self):
-        super().setUp()
-
-        self._temp_directory_ctx = tempfile.TemporaryDirectory()
-        self.temp_dir = self._temp_directory_ctx.__enter__()
-
-        self.storage = FileSystemStorageWithoutPath(location=self.temp_dir)
-
-    def tearDown(self):
-        self._temp_directory_ctx.__exit__(None, None, None)
-        super().tearDown()
-
-    def test_open_works(self):
-        test_content = "test content"
-
-        with self.storage.open("filename", "w") as f:
-            f.write(test_content)
-
-        with open(os.path.join(self.temp_dir, "filename"), "r") as f:
-            content = f.read()
-            self.assertEqual(content, test_content)
-
-    def test_path_does_not_work(self):
-        with self.assertRaises(NotImplementedError):
-            self.storage.path("filename")
+from contentcuration.tests.utils.restricted_filesystemstorage import (
+    RestrictedFileSystemStorage,
+)
 
 
 class ExportTestCase(TestCase):
@@ -80,7 +29,7 @@ def setUp(self):
         self._temp_directory_ctx = tempfile.TemporaryDirectory()
         test_db_root_dir = self._temp_directory_ctx.__enter__()
 
-        self.storage = FileSystemStorageWithoutPath(location=test_db_root_dir)
+        self.storage = RestrictedFileSystemStorage(location=test_db_root_dir)
 
         self._storage_patch_ctx = mock.patch(
             "kolibri_public.utils.export_channel_to_kolibri_public.storage",