Merge pull request #47 from unicef/feature/storage

chg ! use media storage for files instead of default

Author: domdinicola

src/country_workspace/migrations/0005_alter_asyncjob_file.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.1.3 on 2025-03-13 09:04
+
+from django.db import migrations, models
+from country_workspace.storages import MEDIA_STORAGE
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("country_workspace", "0004_kobosubmission_delete_koboasset"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="asyncjob",
+            name="file",
+            field=models.FileField(blank=True, null=True, storage=MEDIA_STORAGE, upload_to="updates"),
+        ),
+    ]

src/country_workspace/models/jobs.py

@@ -6,6 +6,8 @@
 from django.utils.module_loading import import_string
 from django_celery_boost.models import CeleryTaskModel
 
+from country_workspace.storages import MEDIA_STORAGE
+
 
 class AsyncJob(CeleryTaskModel, models.Model):
     class JobType(models.TextChoices):
@@ -16,7 +18,7 @@ class JobType(models.TextChoices):
     type = models.CharField(max_length=50, choices=JobType.choices)
     program = models.ForeignKey("Program", related_name="jobs", on_delete=models.CASCADE, null=True, blank=True)
     batch = models.ForeignKey("Batch", related_name="jobs", on_delete=models.CASCADE, null=True, blank=True)
-    file = models.FileField(upload_to="updates", null=True, blank=True)
+    file = models.FileField(storage=MEDIA_STORAGE, upload_to="updates", null=True, blank=True)
     config = models.JSONField(default=dict, blank=True)
     action = models.CharField(max_length=500, blank=True, null=True)
     description = models.CharField(max_length=255, blank=True, null=True)

src/country_workspace/storages.py

@@ -0,0 +1,4 @@
+from storages.backends.azure_storage import AzureStorage
+from django.conf import settings
+
+MEDIA_STORAGE: AzureStorage = AzureStorage(**settings.STORAGES.get("media").get("OPTIONS"))

src/country_workspace/workspaces/admin/cleaners/bulk_update.py

@@ -5,7 +5,6 @@
 from django import forms
 from django.apps import apps
 from django.core.exceptions import ObjectDoesNotExist
-from django.core.files.storage import default_storage
 from xlsxwriter import Workbook
 
 from constance import config as constance_config
@@ -15,6 +14,7 @@
 from hope_smart_import.readers import open_xls
 
 from country_workspace.models import AsyncJob, Program
+from country_workspace.storages import MEDIA_STORAGE
 
 if TYPE_CHECKING:
     from django.db.models import QuerySet
@@ -174,7 +174,7 @@ def bulk_update_export_template(job: AsyncJob) -> bytes:
     queryset = model.objects.filter(pk__in=job.config["pks"])
     filename = "bulk_update_export_template/%s/%s/%s.xlsx" % (job.program.pk, job.owner.pk, job.config["model_name"])
     out, __ = create_xls_importer(queryset, job.program, job.config["columns"])
-    path = default_storage.save(filename, out)
+    path = MEDIA_STORAGE.save(filename, out)
     job.file = path
     job.save()
     return path

tests/workspace/actions/test_ws_bulk.py

@@ -8,7 +8,7 @@
 from django.urls import reverse
 from testutils.factories import FlexFieldFactory
 from testutils.utils import select_office
-from webtest import Checkbox, Upload
+from unittest import mock
 
 from country_workspace.state import state
 from country_workspace.workspaces.admin.cleaners.bulk_update import TYPES, create_xls_importer
@@ -20,6 +20,7 @@
     from django_webtest import DjangoTestApp
     from django_webtest.pytest_plugin import MixinWithInstanceVariables
     from pytest_django.fixtures import SettingsWrapper
+    from webtest import Checkbox
 
     from country_workspace.models import AsyncJob
     from country_workspace.workspaces.models import CountryHousehold
@@ -38,6 +39,18 @@ def app(django_app_factory: "MixinWithInstanceVariables") -> "DjangoTestApp":
     return django_app
 
 
+@pytest.fixture
+def mock_media_storage():
+    return mock.MagicMock(
+        **{
+            "save.return_value": "mocked/path/to/file",
+            "exists.return_value": False,
+            "get_available_name.return_value": "mocked/path/to/file",
+            "open.return_value": io.BytesIO(b"mocked file content"),
+        }
+    )
+
+
 @pytest.fixture(scope="session")
 def celery_config():
     return {"broker_url": os.environ["CELERY_BROKER_URL"], "result_backend": os.environ["CELERY_BROKER_URL"]}
@@ -142,56 +155,30 @@ def test_create_xls_importer(household: "CountryHousehold", force_migrated_recor
 
 
 def test_bulk_update_export(
-    app: "DjangoTestApp", force_migrated_records, settings: "SettingsWrapper", household: "CountryHousehold"
-) -> None:
-    url = reverse("workspace:workspaces_countryindividual_changelist")
-    settings.CELERY_TASK_ALWAYS_EAGER = True
-    selected_fields = stub.header_add["ind"]
-    with select_office(app, household.country_office, household.program):
-        res = app.get(url)
-        form = res.forms["changelist-form"]
-        form["action"] = "bulk_update_export"
-        form.set("_selected_action", True, index=0)
-        res = form.submit()
-
-        form = res.forms["bulk-update-form"]
-        for i in range(len(form.fields.get("fields"))):
-            target: Checkbox = form.fields.get("fields")[i]
-            if target._value in selected_fields:
-                target.checked = True
-        res = form.submit("_export")
-
-        assert res.status_code == 302
-        job: AsyncJob = household.program.jobs.first()
-        job.queue()
-
-
-def test_bulk_update_import(
     app: "DjangoTestApp",
     force_migrated_records,
     settings: "SettingsWrapper",
-    data: tuple[io.BytesIO, "CountryHousehold", str],
+    household: "CountryHousehold",
+    mock_media_storage,
 ) -> None:
-    buff, household, target = data
-    url = reverse("workspace:workspaces_countryprogram_change", args=[household.program.pk])
-    settings.CELERY_TASK_ALWAYS_EAGER = True
-
-    with select_office(app, household.country_office, household.program):
-        res = app.get(url)
-        res = res.click("Update Records")
-        res.forms["bulk-update-form"]["description"] = f"Bulk update from {target}"
-        res.forms["bulk-update-form"]["target"] = target
-        res.forms["bulk-update-form"]["file"] = Upload(f"{target}.xlsx", buff.read())
-        res = res.forms["bulk-update-form"].submit("_import")
-        household.refresh_from_db()
-        job: AsyncJob = household.program.jobs.first()
-
-        assert res.status_code == 302
-        assert job
-
-        if target == "hh":
-            admin1_v = f"admin1_{stub.header_add['hh'].index('admin1') + 2}"
-            assert household.flex_fields.get("admin1") == admin1_v
-        elif target == "ind":
-            given_name = f"given_name_{stub.header_add['ind'].index('given_name') + 2}"
-            assert household.members.filter(flex_fields__given_name=given_name).exists()
+    with mock.patch("country_workspace.workspaces.admin.cleaners.bulk_update.MEDIA_STORAGE", mock_media_storage):
+        url = reverse("workspace:workspaces_countryindividual_changelist")
+        settings.CELERY_TASK_ALWAYS_EAGER = True
+        selected_fields = stub.header_add["ind"]
+        with select_office(app, household.country_office, household.program):
+            res = app.get(url)
+            form = res.forms["changelist-form"]
+            form["action"] = "bulk_update_export"
+            form.set("_selected_action", True, index=0)
+            res = form.submit()
+
+            form = res.forms["bulk-update-form"]
+            for i in range(len(form.fields.get("fields"))):
+                target: Checkbox = form.fields.get("fields")[i]
+                if target._value in selected_fields:
+                    target.checked = True
+            res = form.submit("_export")
+
+            assert res.status_code == 302
+            job: AsyncJob = household.program.jobs.first()
+            job.queue()

tests/workspace/test_ws_import.py

@@ -1,10 +1,8 @@
-from pathlib import Path
 from typing import TYPE_CHECKING, Any
 import responses
 import pytest
 import re
 from django.urls import reverse
-from webtest import Upload
 
 from country_workspace.state import state
 from constance import config
@@ -56,31 +54,6 @@ def app(django_app_factory: "MixinWithInstanceVariables") -> "DjangoTestApp":
     return django_app
 
 
-def test_import_data_rdi(force_migrated_records, app, program):
-    # NOTE: This test is linked to the content of `data/rdi_one.xlsx`
-    res = app.get("/").follow()
-    res.forms["select-tenant"]["tenant"] = program.country_office.pk
-    res.forms["select-tenant"].submit()
-
-    url = reverse("workspace:workspaces_countryprogram_import_data", args=[program.pk])
-    data = (Path(__file__).parent.parent / "data/rdi_one.xlsx").read_bytes()
-
-    res = app.get(url)
-
-    res.forms["import-file"]["_selected_tab"] = "rdi"
-    res.forms["import-file"]["rdi-file"] = Upload("rdi_one.xlsx", data)
-    res.forms["import-file"]["rdi-detail_column_label"] = "full_name"
-    res = res.forms["import-file"].submit()
-    assert res.status_code == 302
-    assert program.households.count() == 1
-    assert program.individuals.count() == 5
-
-    hh: "CountryHousehold" = program.households.first()
-    assert hh.members.count() == 5
-    assert (head := hh.heads().first())
-    assert head.name == "Edward Jeffrey Rogers"
-
-
 @pytest.mark.django_db(transaction=True)
 @pytest.mark.parametrize(
     ("stub_data", "error_expected", "error_message"),

tests/workspace/test_ws_individual.py

@@ -56,17 +56,6 @@ def app(django_app_factory: "MixinWithInstanceVariables") -> "DjangoTestApp":
     return django_app
 
 
-def test_ind_changelist(app: "DjangoTestApp", individual: "CountryIndividual") -> None:
-    url = reverse("workspace:workspaces_countryindividual_changelist")
-    with select_office(app, individual.country_office, individual.program):
-        res = app.get(url)
-        assert res.status_code == 200, res.location
-        assert f"Add {individual._meta.verbose_name}" not in res.text
-        # filter by program
-        res = app.get(url)
-        assert res.status_code == 200, res.location
-
-
 def test_ind_change(app: "DjangoTestApp", individual: "CountryIndividual") -> None:
     url = reverse("workspace:workspaces_countryindividual_changelist")
     with select_office(app, individual.country_office, individual.program):