Use Django storage for Wagtail deletion archive

.env_SAMPLE

@@ -177,7 +177,6 @@ export HUD_API_ENDPOINT=http://localhost:8000/hud-api-replace/
 # export OIDC_OP_JWKS_ENDPOINT=[Not used for test OIDC provider]
 # export OIDC_OP_ADMIN_ROLE=[Not supported by test OIDC provider]
 
-
 ###############################
 # Deletion archival storage
 ###############################

cfgov/cfgov/settings/base.py

@@ -271,7 +271,23 @@
     "django.contrib.staticfiles.finders.FileSystemFinder",
 ]
 
-STATICFILES_STORAGE = "django.contrib.staticfiles.storage.StaticFilesStorage"
+STORAGES = {
+    "default": {
+        "BACKEND": "django.core.files.storage.FileSystemStorage",
+    },
+    "staticfiles": {
+        "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage",
+    },
+}
+
+if WAGTAIL_DELETION_ARCHIVE_PATH := os.getenv("WAGTAIL_DELETION_ARCHIVE_PATH"):
+    STORAGES["wagtail_deletion_archival"] = {
+        "BACKEND": "django.core.files.storage.FileSystemStorage",
+        "OPTIONS": {
+            "location": WAGTAIL_DELETION_ARCHIVE_PATH,
+        }
+    }
+
 
 # Add the frontend build output to static files.
 STATICFILES_DIRS = [
@@ -363,12 +379,15 @@
 if os.environ.get("S3_ENABLED", "False") == "True":
     AWS_ACCESS_KEY_ID = os.environ["AWS_ACCESS_KEY_ID"]
     AWS_SECRET_ACCESS_KEY = os.environ["AWS_SECRET_ACCESS_KEY"]
-    DEFAULT_FILE_STORAGE = "storages.backends.s3boto3.S3Boto3Storage"
     AWS_S3_CUSTOM_DOMAIN = os.environ.get("AWS_S3_CUSTOM_DOMAIN")
     MEDIA_URL = os.path.join(
         AWS_STORAGE_BUCKET_NAME + ".s3.amazonaws.com", AWS_LOCATION, ""
     )
 
+    STORAGES["default"] = {
+        "BACKEND": "storages.backends.s3boto3.S3Boto3Storage",
+    }
+
 
 # GovDelivery
 GOVDELIVERY_ACCOUNT_CODE = os.environ.get("GOVDELIVERY_ACCOUNT_CODE")

cfgov/cfgov/settings/production.py

@@ -85,9 +85,9 @@
 EMAIL_HOST = os.getenv("EMAIL_HOST")
 DEFAULT_FROM_EMAIL = "[email protected]"
 
-STATICFILES_STORAGE = (
-    "django.contrib.staticfiles.storage.ManifestStaticFilesStorage"
-)
+STORAGES["staticfiles"] = {
+    "BACKEND": "django.contrib.staticfiles.storage.ManifestStaticFilesStorage"
+}
 
 STATIC_ROOT = os.environ["DJANGO_STATIC_ROOT"]
 

cfgov/cfgov/settings/test.py

@@ -41,6 +41,9 @@
 
 GOVDELIVERY_API = "core.govdelivery.MockGovDelivery"
 
+# Disable Wagtail deletion archive storage during testing.
+STORAGES.pop("wagtail_deletion_archival", None)
+
 STATICFILES_FINDERS += [
     "core.testutils.mock_staticfiles.MockStaticfilesFinder",
 ]

cfgov/wagtail_deletion_archival/apps.py

@@ -1,15 +1,6 @@
 from django.apps import AppConfig
-from django.conf import settings
-
-import fs
 
 
 class WagtailDeletionArchivalConfig(AppConfig):
     name = "wagtail_deletion_archival"
     label = "wagtail_deletion_archival"
-    filesystem_name = getattr(
-        settings, "WAGTAIL_DELETION_ARCHIVE_FILESYSTEM", None
-    )
-    filesystem = (
-        fs.open_fs(filesystem_name) if filesystem_name is not None else None
-    )

cfgov/wagtail_deletion_archival/templates/wagtail_deletion_archival/import_page.html

@@ -1,14 +1,14 @@
 {% extends "wagtailadmin/base.html" %}
 {% load i18n wagtailadmin_tags %}
-{% block titletag %}Import page into {{ title }}{% endblock %}
+{% block titletag %}Import page into {{ parent_page.get_admin_display_title }}{% endblock %}
 
 {% block content %}
-    {% include "wagtailadmin/shared/header.html" with title="Import page" subtitle=page.get_admin_display_title icon="doc-empty-inverse" %}
+    {% include "wagtailadmin/shared/header.html" with title="Import page" subtitle=parent_page.get_admin_display_title icon="doc-empty-inverse" %}
 
     <div class="nice-padding">
         {% include "wagtailadmin/shared/non_field_errors.html" %}
 
-        <form enctype="multipart/form-data" action="{% url 'import_page' parent_page.id %}" method="POST">
+        <form enctype="multipart/form-data" action="{% url 'wagtail_deletion_archive_import' parent_page.id %}" method="POST">
             {% csrf_token %}
             <ul class="fields">
                 {% for field in form %}

cfgov/wagtail_deletion_archival/tests/temp_storage.py

@@ -0,0 +1,29 @@
+from tempfile import TemporaryDirectory
+
+from django.test import override_settings
+from django.test.utils import TestContextDecorator
+
+
+class uses_temp_archive_storage(TestContextDecorator):
+    def __init__(self):
+        super().__init__(attr_name="temp_storage", kwarg_name="temp_storage")
+
+    def enable(self):
+        self.temp_storage = TemporaryDirectory()
+
+        self.settings = override_settings(
+            STORAGES={
+                "wagtail_deletion_archival": {
+                    "BACKEND": "django.core.files.storage.FileSystemStorage",
+                    "OPTIONS": {
+                        "location": self.temp_storage.name,
+                    },
+                }
+            }
+        )
+        self.settings.enable()
+        return self.temp_storage.name
+
+    def disable(self):
+        self.settings.disable()
+        self.temp_storage.cleanup()

cfgov/wagtail_deletion_archival/tests/test_utils.py

@@ -1,18 +1,46 @@
 import json
-import logging
+from glob import glob
+from unittest import skipIf
 
-from django.test import TestCase
+from django.conf import settings
+from django.test import SimpleTestCase, TestCase, override_settings
+from django.urls import reverse
 
 from wagtail.models import Site
+from wagtail.test.utils import WagtailTestUtils
+
+from freezegun import freeze_time
 
 from v1.models import BlogPage
+from wagtail_deletion_archival.tests.temp_storage import (
+    uses_temp_archive_storage,
+)
 from wagtail_deletion_archival.utils import (
-    export_page,
+    convert_page_to_json,
+    get_archive_storage,
     get_last_migration,
     import_page,
+    make_archive_filename,
 )
 
 
+class GetArchivalStorageTests(SimpleTestCase):
+    @override_settings(STORAGES={})
+    def test_no_archival_storage(self):
+        self.assertIsNone(get_archive_storage())
+
+    @override_settings(
+        STORAGES={
+            "wagtail_deletion_archival": {
+                "BACKEND": "django.core.files.storage.FileSystemStorage"
+            }
+        }
+    )
+    def test_archive_storage(self):
+        self.assertIsNotNone(get_archive_storage())
+
+
+@skipIf(settings.SKIP_DJANGO_MIGRATIONS, "Requires Django migrations")
 class LastMigrationTestCase(TestCase):
     def test_get_last_migration_has_migrations(self):
         app_label = BlogPage._meta.app_label
@@ -24,7 +52,8 @@ def test_get_last_migration_no_migrations(self):
         self.assertEqual(last_migration, "")
 
 
-class ExportPageTestCase(TestCase):
+@skipIf(settings.SKIP_DJANGO_MIGRATIONS, "Requires Django migrations")
+class ConvertPageToJsonTestCase(TestCase):
     def setUp(self):
         self.page = BlogPage(
             title="Test page",
@@ -46,7 +75,7 @@ def setUp(self):
         )
 
     def test_export_page(self):
-        page_json = export_page(self.page)
+        page_json = convert_page_to_json(self.page)
         page_data = json.loads(page_json)
 
         self.assertEqual(page_data["app_label"], "v1")
@@ -60,6 +89,7 @@ def test_export_page(self):
         )
 
 
+@skipIf(settings.SKIP_DJANGO_MIGRATIONS, "Requires Django migrations")
 class ImportPageTestCase(TestCase):
     def setUp(self):
         self.root_page = Site.objects.get(is_default_site=True).root_page
@@ -69,11 +99,7 @@ def setUp(self):
             content=[("text", "Hello, world!")],
             live=True,
         )
-        self.page_json = export_page(self.original_page)
-        logging.disable(logging.NOTSET)
-
-    def tearDown(self):
-        logging.disable(logging.CRITICAL)
+        self.page_json = convert_page_to_json(self.original_page)
 
     def test_import_page_model_does_not_exist(self):
         page_json = json.dumps(
@@ -105,3 +131,107 @@ def test_import_page(self):
         self.assertEqual(page.title, self.original_page.title)
         self.assertEqual(page.slug, self.original_page.slug)
         self.assertEqual(page.content, self.original_page.content)
+
+
+class MakeArchiveFilenameTests(SimpleTestCase):
+    @freeze_time("2020-01-01")
+    def test_make_archive_filename(self):
+        self.assertEqual(
+            make_archive_filename("foo"), "foo-2020-01-01T00:00:00+00:00.json"
+        )
+
+
+@skipIf(settings.SKIP_DJANGO_MIGRATIONS, "Requires Django migrations")
+class ArchivePageOnDeletionTestCase(TestCase, WagtailTestUtils):
+    def setUp(self):
+        self.login()
+
+        root_page = Site.objects.get(is_default_site=True).root_page
+
+        self.test_page1 = BlogPage(title="test page 1", slug="test-page1")
+        root_page.add_child(instance=self.test_page1)
+
+        self.test_child_page = BlogPage(title="child page", slug="test-child")
+        self.test_page1.add_child(instance=self.test_child_page)
+
+        self.test_page2 = BlogPage(title="test page 2", slug="test-page2")
+        root_page.add_child(instance=self.test_page2)
+
+        self.test_page3 = BlogPage(title="test page 3", slug="test-page3")
+        root_page.add_child(instance=self.test_page3)
+
+    @uses_temp_archive_storage()
+    def test_delete_page(self, temp_storage):
+        # Before the test no JSON files exist in the archive dir.
+        self.assertFalse(glob(f"{temp_storage}/**/*.json", recursive=True))
+
+        self.client.post(
+            reverse("wagtailadmin_pages:delete", args=(self.test_page1.pk,))
+        )
+
+        # The page slug exists in the archive dir.
+        self.assertTrue(
+            glob(
+                f"{temp_storage}/{self.test_page1.slug}/*.json",
+                recursive=True,
+            )
+        )
+
+        # The child page slug exists in the archive dir.
+        self.assertTrue(
+            glob(
+                f"{temp_storage}/{self.test_page1.slug}/{self.test_child_page.slug}/*.json",
+                recursive=True,
+            )
+        )
+
+    @uses_temp_archive_storage()
+    def test_bulk_delete_page(self, temp_storage):
+        self.client.post(
+            reverse(
+                "wagtail_bulk_action",
+                args=(
+                    "wagtailcore",
+                    "page",
+                    "delete",
+                ),
+            )
+            + f"?id={self.test_page1.pk}&id={self.test_page2.pk}"
+        )
+
+        # Page 1, its child page, and Page 2 should all be archived.
+        self.assertEqual(
+            len(
+                glob(
+                    f"{temp_storage}/**/*.json",
+                    recursive=True,
+                )
+            ),
+            3,
+        )
+
+    @uses_temp_archive_storage()
+    def test_delete_page_confirmation(self, temp_storage):
+        self.client.get(
+            reverse("wagtailadmin_pages:delete", args=(self.test_page3.pk,))
+        )
+        self.assertFalse(glob(f"{temp_storage}/**/*.json", recursive=True))
+
+
+class ArchivePageOnDeletionWhenArchivingIsDisabledTestCase(
+    TestCase, WagtailTestUtils
+):
+    def test_delete_works_when_archiving_is_disabled(self):
+        self.assertIsNone(get_archive_storage())
+
+        root_page = Site.objects.get(is_default_site=True).root_page
+        self.test_page1 = BlogPage(title="test page 1", slug="test-page1")
+        root_page.add_child(instance=self.test_page1)
+
+        self.login()
+        self.client.post(
+            reverse("wagtailadmin_pages:delete", args=(self.test_page1.pk,))
+        )
+
+        with self.assertRaises(BlogPage.DoesNotExist):
+            BlogPage.objects.get(slug="test-page1")