chg ! tests

Author: vitali-yanushchyk-valor

tests/conftest.py

@@ -7,6 +7,9 @@
 import pytest
 import responses
 
+from django.core.files.storage.base import Storage
+from django.core.files.base import ContentFile
+
 if TYPE_CHECKING:
     from country_workspace.models import User
 
@@ -34,12 +37,44 @@ def pytest_addoption(parser):
     )
 
 
+class MockStorage(Storage):
+    def __init__(self):
+        self._file_content = None
+
+    def save(self, name, content, max_length=None):
+        self._file_content = content.read() if hasattr(content, "read") else content
+        return "mocked/path/to/file"
+
+    def get_available_name(self, name, max_length=None):
+        return "mocked/path/to/file"
+
+    def open(self, name, mode="rb"):
+        if self._file_content is not None:
+            return ContentFile(self._file_content)
+        return ContentFile(b"id,name\n1,Test")
+
+
+@pytest.fixture
+def mock_storage():
+    return MockStorage()
+
+
+@pytest.fixture(autouse=True)
+def patch_asyncjob(mock_storage):
+    from django.db import models
+    from country_workspace.models.jobs import AsyncJob
+
+    AsyncJob._meta.get_field("file").storage = mock_storage
+    AsyncJob._meta.get_field("file").__class__ = models.FileField
+
+
 def pytest_configure(config):
     if not config.option.enable_selenium and ("selenium" not in getattr(config.option, "markexpr", None)):
         if config.option.markexpr:
             config.option.markexpr += " and not selenium"
         else:
             config.option.markexpr = "not selenium"
+
     os.environ["DJANGO_SETTINGS_MODULE"] = "country_workspace.config.settings"
     os.environ.setdefault("STATIC_URL", "/static/")
     os.environ.setdefault("MEDIA_ROOT", "/tmp/static/")

tests/workspace/actions/test_ws_bulk.py

@@ -9,6 +9,8 @@
 from testutils.factories import FlexFieldFactory
 from testutils.utils import select_office
 from unittest import mock
+from webtest import Upload
+
 
 from country_workspace.state import state
 from country_workspace.workspaces.admin.cleaners.bulk_update import TYPES, create_xls_importer
@@ -39,18 +41,6 @@ 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"]}
@@ -159,12 +149,12 @@ def test_bulk_update_export(
     force_migrated_records,
     settings: "SettingsWrapper",
     household: "CountryHousehold",
-    mock_media_storage,
+    mock_storage,
 ) -> None:
-    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"]
+    url = reverse("workspace:workspaces_countryindividual_changelist")
+    settings.CELERY_TASK_ALWAYS_EAGER = True
+    selected_fields = stub.header_add["ind"]
+    with mock.patch("country_workspace.workspaces.admin.cleaners.bulk_update.MEDIA_STORAGE", mock_storage):
         with select_office(app, household.country_office, household.program):
             res = app.get(url)
             form = res.forms["changelist-form"]
@@ -182,3 +172,35 @@ def test_bulk_update_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",
+    household: "CountryHousehold",
+    data: tuple[io.BytesIO, "CountryHousehold", str],
+) -> 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()