chg ! tests & rebase

Author: vitali-yanushchyk-valor

…rations/0005_alter_individual_options.py …rations/0006_alter_individual_options.pysrc/country_workspace/migrations/0005_alter_individual_options.py renamed to src/country_workspace/migrations/0006_alter_individual_options.py src/country_workspace/migrations/0005_alter_individual_options.py renamed to src/country_workspace/migrations/0006_alter_individual_options.py

@@ -5,7 +5,7 @@
 
 class Migration(migrations.Migration):
     dependencies = [
-        ("country_workspace", "0004_kobosubmission_delete_koboasset"),
+        ("country_workspace", "0005_alter_asyncjob_file"),
     ]
 
     operations = [

tests/admin/test_admin_cleaners.py

@@ -15,8 +15,8 @@ def household():
 
 
 def test_validate_program(household):
-    assert validate_program(Mock(program=household.program)) == {"invalid": 0, "total": 0, "valid": 1}
+    assert validate_program(Mock(program=household.program)) == {"invalid": 0, "valid": 1}
     with mock.patch(
         "country_workspace.workspaces.admin.cleaners.validate.Household.validate_with_checker", return_value=False
     ):
-        assert validate_program(Mock(program=household.program)) == {"invalid": 1, "total": 0, "valid": 0}
+        assert validate_program(Mock(program=household.program)) == {"invalid": 1, "valid": 0}

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,6 +37,37 @@ 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:

tests/workspace/actions/test_ws_bulk.py

@@ -13,6 +13,8 @@
 from country_workspace.state import state
 from country_workspace.workspaces.admin.cleaners.bulk_update import TYPES, create_xls_importer
 from tests.workspace.actions import stub
+from webtest import Upload
+
 
 if TYPE_CHECKING:
     from django_webtest import DjangoTestApp
@@ -157,12 +159,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"]
@@ -180,3 +182,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()