From 28d068427a8864a90cb491d42f59a00187d2e32d Mon Sep 17 00:00:00 2001 From: Arsen P Date: Thu, 28 May 2026 12:29:56 +0300 Subject: [PATCH 1/7] Add functionality to run formulas on existing records - Introduced a new form, `RunTransformerForm`, to allow users to select a batch and specify the record type for formula application. - Implemented a new admin action, `run_on_existing_records`, to handle the execution of formulas directly in the Country Workspace. - Created a corresponding HTML template for the form, enhancing the user interface for running formulas without rule commits. - Added permission checks to ensure users can only run formulas on batches they are authorized to process. --- .../workspaces/admin/transformer.py | 125 +++++++++++++++++- .../run_transformer_form.html | 69 ++++++++++ 2 files changed, 192 insertions(+), 2 deletions(-) create mode 100644 src/country_workspace/workspaces/templates/workspace/admin_extra_buttons/run_transformer_form.html diff --git a/src/country_workspace/workspaces/admin/transformer.py b/src/country_workspace/workspaces/admin/transformer.py index f4412454d..da5aa5753 100644 --- a/src/country_workspace/workspaces/admin/transformer.py +++ b/src/country_workspace/workspaces/admin/transformer.py @@ -1,15 +1,70 @@ -from django.contrib import admin +from admin_extra_buttons.decorators import button +from django import forms +from django.contrib import admin, messages from django.core.cache import cache from django.db.models import QuerySet from django.forms import ModelForm -from django.http import HttpRequest +from django.http import HttpRequest, HttpResponse, HttpResponseRedirect +from django.shortcuts import render +from django.urls import reverse +from django.utils.translation import gettext_lazy as _ +from strategy_field.utils import fqn +from country_workspace.models import AsyncJob, Batch from country_workspace.state import state from country_workspace.workspaces.models import CountryTransformer from country_workspace.workspaces.options import WorkspaceModelAdmin from country_workspace.workspaces.sites import workspace +class RunTransformerForm(forms.Form): + class ApplyToOptions(forms.TextChoices): + HOUSEHOLDS = "households", _("Households only") + INDIVIDUALS = "individuals", _("Individuals only") + BOTH = "both", _("Households and Individuals") + + batch = forms.ModelChoiceField( + queryset=Batch.objects.none(), + label=_("Batch"), + help_text=_("Select an existing batch to update records before pushing to HOPE."), + ) + apply_to = forms.ChoiceField( + label=_("Apply formula to"), + choices=ApplyToOptions.choices, + help_text=_("Choose which record type should be updated by this formula."), + ) + + def __init__(self, *args: object, **kwargs: object) -> None: + office = kwargs.pop("office", None) + program = kwargs.pop("program", None) + super().__init__(*args, **kwargs) + + qs = Batch.objects.order_by("-import_date") + if office: + qs = qs.filter(country_office=office) + if program: + qs = qs.filter(program=program) + self.fields["batch"].queryset = qs.select_related("program") + + if not program: + self.fields["apply_to"].choices = [ + (self.ApplyToOptions.INDIVIDUALS, self.ApplyToOptions.INDIVIDUALS.label), + (self.ApplyToOptions.BOTH, self.ApplyToOptions.BOTH.label), + ] + return + + if program.is_master_detail: + self.fields["apply_to"].choices = [ + (self.ApplyToOptions.HOUSEHOLDS, self.ApplyToOptions.HOUSEHOLDS.label), + (self.ApplyToOptions.INDIVIDUALS, self.ApplyToOptions.INDIVIDUALS.label), + (self.ApplyToOptions.BOTH, self.ApplyToOptions.BOTH.label), + ] + else: + self.fields["apply_to"].choices = [ + (self.ApplyToOptions.INDIVIDUALS, self.ApplyToOptions.INDIVIDUALS.label), + ] + + @admin.register(CountryTransformer, site=workspace) class CountryTransformerAdmin(WorkspaceModelAdmin): list_display = ("name", "description", "created_by", "created_at") @@ -63,6 +118,72 @@ def delete_queryset(self, request: HttpRequest, queryset: QuerySet[CountryTransf super().delete_queryset(request, queryset) self._invalidate_transformer_cache() + @button( + label="Run Formula on Existing Records", + change_form=True, + html_attrs={"title": "Run this formula in Country Workspace without rule commits"}, + ) + def run_on_existing_records(self, request: HttpRequest, pk: str) -> HttpResponse: + obj = self.get_object(request, pk) + if not obj: + return HttpResponse("Transformer not found", status=404) + + if request.method == "POST" and "apply" in request.POST: + form = RunTransformerForm(request.POST, office=state.tenant, program=state.program) + if form.is_valid(): + batch = form.cleaned_data["batch"] + apply_to = form.cleaned_data["apply_to"] + if not request.user.has_perm("country_workspace.reprocess_batch", batch.program): # type: ignore[attr-defined] + self.message_user( + request, + _("You do not have permission to run formulas on this batch."), + messages.ERROR, + ) + return HttpResponseRedirect(self.get_change_url(request, obj)) + + config: dict[str, int] = {"batch_id": batch.pk} + if batch.program.is_master_detail and apply_to in ( + RunTransformerForm.ApplyToOptions.HOUSEHOLDS, + RunTransformerForm.ApplyToOptions.BOTH, + ): + config["household_transformer_id"] = obj.pk + if apply_to in ( + RunTransformerForm.ApplyToOptions.INDIVIDUALS, + RunTransformerForm.ApplyToOptions.BOTH, + ): + config["individual_transformer_id"] = obj.pk + + job = AsyncJob.objects.create( + description=f"Run formula '{obj.name}' on batch {batch.name}", + type=AsyncJob.JobType.TASK, + owner=request.user, + action=fqn("country_workspace.workspaces.admin.batch.reprocessing.reprocess_batch"), + program=batch.program, + batch=batch, + config=config, + ) + job.queue() + + self.message_user( + request, + _("Formula execution has been scheduled for the selected batch."), + messages.SUCCESS, + ) + return HttpResponseRedirect(reverse("workspace:workspaces_countrybatch_changelist")) + + self.message_user(request, _("Please correct the errors below."), messages.ERROR) + else: + form = RunTransformerForm(office=state.tenant, program=state.program) + + context = self.get_common_context( + request, + pk=pk, + title=_("Run Formula on Existing Records"), + form=form, + transformer=obj, + ) + return render(request, "workspace/admin_extra_buttons/run_transformer_form.html", context) + def _invalidate_transformer_cache(self) -> None: """Invalidate cache keys related to transformers.""" if state.tenant: diff --git a/src/country_workspace/workspaces/templates/workspace/admin_extra_buttons/run_transformer_form.html b/src/country_workspace/workspaces/templates/workspace/admin_extra_buttons/run_transformer_form.html new file mode 100644 index 000000000..c5439be26 --- /dev/null +++ b/src/country_workspace/workspaces/templates/workspace/admin_extra_buttons/run_transformer_form.html @@ -0,0 +1,69 @@ +{% extends 'workspace/_base.html' %} +{% load i18n %} +{% block breadcrumbs %} +{% endblock breadcrumbs %} +{% block content %} +
+
+
+

+ {% translate 'Run Formula on Existing Records' %} +

+
+

+ {% blocktranslate with transformer_name=transformer.name %} + Formula: {{ transformer_name }} + {% endblocktranslate %} +

+

+ {% translate "This executes the selected formula directly in Country Workspace and updates existing records before they are pushed to HOPE." %} +

+

+ {% translate "No admin rule creation, rule commits, or custom code execution is required." %} +

+
+ {% csrf_token %} + {% if form.errors %} +
+

+ {% translate "Please correct the errors below." %} +

+
+ {% endif %} +
+ {% for field in form %} +
+
+ + {{ field }} + {% if field.help_text %} +

+ {{ field.help_text }} +

+ {% endif %} + {% if field.errors %} +
    + {% for error in field.errors %} +
  • + {{ error }} +
    • + {% endfor %} +
+ {% endif %} +
+
+ {% endfor %} +
+
+ + + +
+
+
+
+
+
+{% endblock content %} From 3bbe33fd5ef3dacf917c4fa286dbf2d9cb1b31cd Mon Sep 17 00:00:00 2001 From: Arsen P Date: Thu, 28 May 2026 12:37:22 +0300 Subject: [PATCH 2/7] Add unit tests for RunTransformerForm and run_on_existing_records functionality - Introduced tests for the `RunTransformerForm` to validate choices based on program type. - Added tests for the `run_on_existing_records` method in the `CountryTransformerAdmin`, covering scenarios for rendering the form, handling permissions, and processing valid and invalid submissions. - Utilized mocking to simulate requests and responses, ensuring isolated and reliable test outcomes. --- tests/workspace/admin/test_transformer.py | 170 +++++++++++++++++++++- 1 file changed, 169 insertions(+), 1 deletion(-) diff --git a/tests/workspace/admin/test_transformer.py b/tests/workspace/admin/test_transformer.py index 047500210..5c932d7ef 100644 --- a/tests/workspace/admin/test_transformer.py +++ b/tests/workspace/admin/test_transformer.py @@ -3,9 +3,10 @@ import pytest from django.contrib.auth.models import User from django.http import HttpRequest +from django.http.response import HttpResponse from country_workspace.state import state -from country_workspace.workspaces.admin.transformer import CountryTransformerAdmin +from country_workspace.workspaces.admin.transformer import CountryTransformerAdmin, RunTransformerForm from country_workspace.workspaces.models import CountryTransformer @@ -111,3 +112,170 @@ def test_invalidate_transformer_cache_no_tenant(self, mock_cache, transformer_ad state.tenant = None transformer_admin._invalidate_transformer_cache() mock_cache.delete.assert_not_called() + + +class TestRunTransformerForm: + def _build_mock_queryset(self) -> MagicMock: + qs = MagicMock() + qs.filter.return_value = qs + qs.select_related.return_value = qs + qs.all.return_value = qs + return qs + + @patch("country_workspace.workspaces.admin.transformer.Batch.objects") + def test_choices_without_program(self, mock_batch_objects): + qs = self._build_mock_queryset() + mock_batch_objects.order_by.return_value = qs + + form = RunTransformerForm() + + choices = [choice[0] for choice in form.fields["apply_to"].choices] + assert choices == [ + RunTransformerForm.ApplyToOptions.INDIVIDUALS, + RunTransformerForm.ApplyToOptions.BOTH, + ] + + @patch("country_workspace.workspaces.admin.transformer.Batch.objects") + def test_choices_master_detail_program(self, mock_batch_objects): + qs = self._build_mock_queryset() + mock_batch_objects.order_by.return_value = qs + program = MagicMock(is_master_detail=True) + + form = RunTransformerForm(program=program) + + choices = [choice[0] for choice in form.fields["apply_to"].choices] + assert choices == [ + RunTransformerForm.ApplyToOptions.HOUSEHOLDS, + RunTransformerForm.ApplyToOptions.INDIVIDUALS, + RunTransformerForm.ApplyToOptions.BOTH, + ] + + @patch("country_workspace.workspaces.admin.transformer.Batch.objects") + def test_choices_non_master_detail_program(self, mock_batch_objects): + qs = self._build_mock_queryset() + mock_batch_objects.order_by.return_value = qs + program = MagicMock(is_master_detail=False) + + form = RunTransformerForm(program=program) + + choices = [choice[0] for choice in form.fields["apply_to"].choices] + assert choices == [RunTransformerForm.ApplyToOptions.INDIVIDUALS] + + +class TestRunOnExistingRecords: + def _build_request(self, method: str = "GET", post_data: dict | None = None, has_perm: bool = True) -> MagicMock: + request = MagicMock(spec=HttpRequest) + request.method = method + request.POST = post_data or {} + request.user = MagicMock(spec=User) + request.user.has_perm.return_value = has_perm + return request + + def test_returns_404_when_transformer_not_found(self, transformer_admin): + request = self._build_request() + with patch.object(transformer_admin, "get_object", return_value=None): + response = transformer_admin.run_on_existing_records(request, "123") + assert response.status_code == 404 + + def test_get_renders_form(self, transformer_admin): + request = self._build_request("GET") + transformer = MagicMock(pk=1, name="T1") + state.tenant = MagicMock() + state.program = MagicMock() + + with ( + patch.object(transformer_admin, "get_object", return_value=transformer), + patch( + "country_workspace.workspaces.admin.transformer.render", return_value=HttpResponse("ok") + ) as mock_render, + ): + response = transformer_admin.run_on_existing_records(request, "1") + + assert response.status_code == 200 + mock_render.assert_called_once() + + def test_post_invalid_form_shows_error(self, transformer_admin): + request = self._build_request("POST", {"apply": "yes"}) + transformer = MagicMock(pk=1, name="T1") + state.tenant = MagicMock() + state.program = MagicMock() + + mock_form = MagicMock() + mock_form.is_valid.return_value = False + + with ( + patch.object(transformer_admin, "get_object", return_value=transformer), + patch("country_workspace.workspaces.admin.transformer.RunTransformerForm", return_value=mock_form), + patch.object(transformer_admin, "message_user") as mock_message_user, + patch("country_workspace.workspaces.admin.transformer.render", return_value=HttpResponse("ok")), + ): + transformer_admin.run_on_existing_records(request, "1") + + mock_message_user.assert_called() + assert "Please correct the errors below." in mock_message_user.call_args.args[1] + + def test_post_valid_without_permission_redirects_to_change(self, transformer_admin): + request = self._build_request("POST", {"apply": "yes"}, has_perm=False) + transformer = MagicMock(pk=1, name="T1") + program = MagicMock(is_master_detail=True) + batch = MagicMock(pk=10, name="Batch 1", program=program) + state.tenant = MagicMock() + state.program = MagicMock() + + mock_form = MagicMock() + mock_form.is_valid.return_value = True + mock_form.cleaned_data = { + "batch": batch, + "apply_to": RunTransformerForm.ApplyToOptions.BOTH, + } + + with ( + patch.object(transformer_admin, "get_object", return_value=transformer), + patch("country_workspace.workspaces.admin.transformer.RunTransformerForm", return_value=mock_form), + patch.object(transformer_admin, "get_change_url", return_value="/change/") as mock_get_change_url, + patch.object(transformer_admin, "message_user") as mock_message_user, + ): + response = transformer_admin.run_on_existing_records(request, "1") + + assert response.status_code == 302 + assert response.url == "/change/" + mock_get_change_url.assert_called_once_with(request, transformer) + assert "do not have permission" in mock_message_user.call_args.args[1] + + def test_post_valid_schedules_job_for_both_in_master_detail(self, transformer_admin): + request = self._build_request("POST", {"apply": "yes"}, has_perm=True) + transformer = MagicMock(pk=99, name="Eligibility Rule") + program = MagicMock(is_master_detail=True) + batch = MagicMock(pk=10, name="Batch 1", program=program) + state.tenant = MagicMock() + state.program = MagicMock() + + mock_form = MagicMock() + mock_form.is_valid.return_value = True + mock_form.cleaned_data = { + "batch": batch, + "apply_to": RunTransformerForm.ApplyToOptions.BOTH, + } + job = MagicMock() + + with ( + patch.object(transformer_admin, "get_object", return_value=transformer), + patch("country_workspace.workspaces.admin.transformer.RunTransformerForm", return_value=mock_form), + patch( + "country_workspace.workspaces.admin.transformer.AsyncJob.objects.create", return_value=job + ) as mock_create, + patch("country_workspace.workspaces.admin.transformer.reverse", return_value="/batch/"), + patch.object(transformer_admin, "message_user") as mock_message_user, + ): + response = transformer_admin.run_on_existing_records(request, "99") + + assert response.status_code == 302 + assert response.url == "/batch/" + mock_create.assert_called_once() + assert mock_create.call_args.kwargs["config"] == { + "batch_id": 10, + "household_transformer_id": 99, + "individual_transformer_id": 99, + } + job.queue.assert_called_once() + assert "scheduled" in mock_message_user.call_args.args[1].lower() From 06f9b23931fe0cedfcaa30f814e13c343ed3d974 Mon Sep 17 00:00:00 2001 From: Arsen P Date: Thu, 28 May 2026 12:50:45 +0300 Subject: [PATCH 3/7] Refactor RunTransformerForm to use models.TextChoices - Updated the ApplyToOptions class in RunTransformerForm to inherit from models.TextChoices instead of forms.TextChoices, aligning with Django best practices for defining choice fields. - This change enhances the integration with Django's model layer, improving maintainability and consistency across the application. --- src/country_workspace/workspaces/admin/transformer.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/country_workspace/workspaces/admin/transformer.py b/src/country_workspace/workspaces/admin/transformer.py index da5aa5753..b7754c5ce 100644 --- a/src/country_workspace/workspaces/admin/transformer.py +++ b/src/country_workspace/workspaces/admin/transformer.py @@ -2,6 +2,7 @@ from django import forms from django.contrib import admin, messages from django.core.cache import cache +from django.db import models from django.db.models import QuerySet from django.forms import ModelForm from django.http import HttpRequest, HttpResponse, HttpResponseRedirect @@ -18,7 +19,7 @@ class RunTransformerForm(forms.Form): - class ApplyToOptions(forms.TextChoices): + class ApplyToOptions(models.TextChoices): HOUSEHOLDS = "households", _("Households only") INDIVIDUALS = "individuals", _("Individuals only") BOTH = "both", _("Households and Individuals") From c77861b1ac99b08996cef30e7ca367e7c8ec0621 Mon Sep 17 00:00:00 2001 From: Arsen P Date: Fri, 29 May 2026 13:11:37 +0300 Subject: [PATCH 4/7] Refactor test cases in test_transformer.py to use explicit patching - Updated test cases in `test_transformer.py` to use explicit patching for the `get_object` and `get_change_url` methods of `CountryTransformerAdmin`, improving clarity and maintainability of the tests. - This change enhances the readability of the test setup by clearly indicating the target of the mocks. --- tests/workspace/admin/test_transformer.py | 30 ++++++++++++++++++----- 1 file changed, 24 insertions(+), 6 deletions(-) diff --git a/tests/workspace/admin/test_transformer.py b/tests/workspace/admin/test_transformer.py index 5c932d7ef..b18a5af9a 100644 --- a/tests/workspace/admin/test_transformer.py +++ b/tests/workspace/admin/test_transformer.py @@ -173,7 +173,10 @@ def _build_request(self, method: str = "GET", post_data: dict | None = None, has def test_returns_404_when_transformer_not_found(self, transformer_admin): request = self._build_request() - with patch.object(transformer_admin, "get_object", return_value=None): + with patch( + "country_workspace.workspaces.admin.transformer.CountryTransformerAdmin.get_object", + return_value=None, + ): response = transformer_admin.run_on_existing_records(request, "123") assert response.status_code == 404 @@ -184,7 +187,10 @@ def test_get_renders_form(self, transformer_admin): state.program = MagicMock() with ( - patch.object(transformer_admin, "get_object", return_value=transformer), + patch( + "country_workspace.workspaces.admin.transformer.CountryTransformerAdmin.get_object", + return_value=transformer, + ), patch( "country_workspace.workspaces.admin.transformer.render", return_value=HttpResponse("ok") ) as mock_render, @@ -204,7 +210,10 @@ def test_post_invalid_form_shows_error(self, transformer_admin): mock_form.is_valid.return_value = False with ( - patch.object(transformer_admin, "get_object", return_value=transformer), + patch( + "country_workspace.workspaces.admin.transformer.CountryTransformerAdmin.get_object", + return_value=transformer, + ), patch("country_workspace.workspaces.admin.transformer.RunTransformerForm", return_value=mock_form), patch.object(transformer_admin, "message_user") as mock_message_user, patch("country_workspace.workspaces.admin.transformer.render", return_value=HttpResponse("ok")), @@ -230,9 +239,15 @@ def test_post_valid_without_permission_redirects_to_change(self, transformer_adm } with ( - patch.object(transformer_admin, "get_object", return_value=transformer), + patch( + "country_workspace.workspaces.admin.transformer.CountryTransformerAdmin.get_object", + return_value=transformer, + ), patch("country_workspace.workspaces.admin.transformer.RunTransformerForm", return_value=mock_form), - patch.object(transformer_admin, "get_change_url", return_value="/change/") as mock_get_change_url, + patch( + "country_workspace.workspaces.admin.transformer.CountryTransformerAdmin.get_change_url", + return_value="/change/", + ) as mock_get_change_url, patch.object(transformer_admin, "message_user") as mock_message_user, ): response = transformer_admin.run_on_existing_records(request, "1") @@ -259,7 +274,10 @@ def test_post_valid_schedules_job_for_both_in_master_detail(self, transformer_ad job = MagicMock() with ( - patch.object(transformer_admin, "get_object", return_value=transformer), + patch( + "country_workspace.workspaces.admin.transformer.CountryTransformerAdmin.get_object", + return_value=transformer, + ), patch("country_workspace.workspaces.admin.transformer.RunTransformerForm", return_value=mock_form), patch( "country_workspace.workspaces.admin.transformer.AsyncJob.objects.create", return_value=job From 8f71bd2ea992c8ee780fb7f077bd1dc9479916fe Mon Sep 17 00:00:00 2001 From: Arsen P Date: Fri, 5 Jun 2026 10:46:35 +0300 Subject: [PATCH 5/7] Update test cases in test_transformer.py to pass transformer_admin instance - Modified the `run_on_existing_records` method calls in the test cases to explicitly pass the `transformer_admin` instance, enhancing clarity and ensuring proper context is maintained during testing. - This change improves the accuracy of the tests by aligning with the method's expected parameters. --- tests/workspace/admin/test_transformer.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/tests/workspace/admin/test_transformer.py b/tests/workspace/admin/test_transformer.py index b18a5af9a..dd5ef6bf2 100644 --- a/tests/workspace/admin/test_transformer.py +++ b/tests/workspace/admin/test_transformer.py @@ -177,7 +177,7 @@ def test_returns_404_when_transformer_not_found(self, transformer_admin): "country_workspace.workspaces.admin.transformer.CountryTransformerAdmin.get_object", return_value=None, ): - response = transformer_admin.run_on_existing_records(request, "123") + response = transformer_admin.run_on_existing_records(transformer_admin, request, "123") assert response.status_code == 404 def test_get_renders_form(self, transformer_admin): @@ -195,7 +195,7 @@ def test_get_renders_form(self, transformer_admin): "country_workspace.workspaces.admin.transformer.render", return_value=HttpResponse("ok") ) as mock_render, ): - response = transformer_admin.run_on_existing_records(request, "1") + response = transformer_admin.run_on_existing_records(transformer_admin, request, "1") assert response.status_code == 200 mock_render.assert_called_once() @@ -218,7 +218,7 @@ def test_post_invalid_form_shows_error(self, transformer_admin): patch.object(transformer_admin, "message_user") as mock_message_user, patch("country_workspace.workspaces.admin.transformer.render", return_value=HttpResponse("ok")), ): - transformer_admin.run_on_existing_records(request, "1") + transformer_admin.run_on_existing_records(transformer_admin, request, "1") mock_message_user.assert_called() assert "Please correct the errors below." in mock_message_user.call_args.args[1] @@ -250,7 +250,7 @@ def test_post_valid_without_permission_redirects_to_change(self, transformer_adm ) as mock_get_change_url, patch.object(transformer_admin, "message_user") as mock_message_user, ): - response = transformer_admin.run_on_existing_records(request, "1") + response = transformer_admin.run_on_existing_records(transformer_admin, request, "1") assert response.status_code == 302 assert response.url == "/change/" @@ -285,7 +285,7 @@ def test_post_valid_schedules_job_for_both_in_master_detail(self, transformer_ad patch("country_workspace.workspaces.admin.transformer.reverse", return_value="/batch/"), patch.object(transformer_admin, "message_user") as mock_message_user, ): - response = transformer_admin.run_on_existing_records(request, "99") + response = transformer_admin.run_on_existing_records(transformer_admin, request, "99") assert response.status_code == 302 assert response.url == "/batch/" From 8c04507e91a83babb8c437abe7d2dc46c2595cb0 Mon Sep 17 00:00:00 2001 From: Arsen P Date: Fri, 5 Jun 2026 11:13:14 +0300 Subject: [PATCH 6/7] Update test cases in test_transformer.py to handle None state values and improve mock form usage - Modified the test cases to set `state.tenant` and `state.program` to None, ensuring proper handling of these values during testing. - Enhanced the mock setup for `RunTransformerForm` to explicitly define `ApplyToOptions`, improving clarity and consistency in the tests. - These changes contribute to more robust and maintainable test cases for the `run_on_existing_records` functionality. --- tests/workspace/admin/test_transformer.py | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/tests/workspace/admin/test_transformer.py b/tests/workspace/admin/test_transformer.py index dd5ef6bf2..40b052c4c 100644 --- a/tests/workspace/admin/test_transformer.py +++ b/tests/workspace/admin/test_transformer.py @@ -183,14 +183,16 @@ def test_returns_404_when_transformer_not_found(self, transformer_admin): def test_get_renders_form(self, transformer_admin): request = self._build_request("GET") transformer = MagicMock(pk=1, name="T1") - state.tenant = MagicMock() - state.program = MagicMock() + state.tenant = None + state.program = None + mock_form = MagicMock() with ( patch( "country_workspace.workspaces.admin.transformer.CountryTransformerAdmin.get_object", return_value=transformer, ), + patch("country_workspace.workspaces.admin.transformer.RunTransformerForm", return_value=mock_form), patch( "country_workspace.workspaces.admin.transformer.render", return_value=HttpResponse("ok") ) as mock_render, @@ -243,13 +245,16 @@ def test_post_valid_without_permission_redirects_to_change(self, transformer_adm "country_workspace.workspaces.admin.transformer.CountryTransformerAdmin.get_object", return_value=transformer, ), - patch("country_workspace.workspaces.admin.transformer.RunTransformerForm", return_value=mock_form), + patch( + "country_workspace.workspaces.admin.transformer.RunTransformerForm", return_value=mock_form + ) as mock_form_class, patch( "country_workspace.workspaces.admin.transformer.CountryTransformerAdmin.get_change_url", return_value="/change/", ) as mock_get_change_url, patch.object(transformer_admin, "message_user") as mock_message_user, ): + mock_form_class.ApplyToOptions = RunTransformerForm.ApplyToOptions response = transformer_admin.run_on_existing_records(transformer_admin, request, "1") assert response.status_code == 302 @@ -278,13 +283,16 @@ def test_post_valid_schedules_job_for_both_in_master_detail(self, transformer_ad "country_workspace.workspaces.admin.transformer.CountryTransformerAdmin.get_object", return_value=transformer, ), - patch("country_workspace.workspaces.admin.transformer.RunTransformerForm", return_value=mock_form), + patch( + "country_workspace.workspaces.admin.transformer.RunTransformerForm", return_value=mock_form + ) as mock_form_class, patch( "country_workspace.workspaces.admin.transformer.AsyncJob.objects.create", return_value=job ) as mock_create, patch("country_workspace.workspaces.admin.transformer.reverse", return_value="/batch/"), patch.object(transformer_admin, "message_user") as mock_message_user, ): + mock_form_class.ApplyToOptions = RunTransformerForm.ApplyToOptions response = transformer_admin.run_on_existing_records(transformer_admin, request, "99") assert response.status_code == 302 From fd386dc28801a96a8956ad3779c4f27621637bf6 Mon Sep 17 00:00:00 2001 From: Arsen P Date: Fri, 5 Jun 2026 11:46:22 +0300 Subject: [PATCH 7/7] Add test for batch queryset filtering by office in test_transformer.py - Introduced a new test case to verify that the batch queryset is filtered correctly by the office parameter in the RunTransformerForm. - Utilized mocking to simulate the queryset behavior, ensuring that the filter method is called with the expected office argument. - This addition enhances the test coverage for the RunTransformerForm functionality, contributing to more robust testing practices. --- tests/workspace/admin/test_transformer.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/tests/workspace/admin/test_transformer.py b/tests/workspace/admin/test_transformer.py index 40b052c4c..65d5ef40f 100644 --- a/tests/workspace/admin/test_transformer.py +++ b/tests/workspace/admin/test_transformer.py @@ -135,6 +135,16 @@ def test_choices_without_program(self, mock_batch_objects): RunTransformerForm.ApplyToOptions.BOTH, ] + @patch("country_workspace.workspaces.admin.transformer.Batch.objects") + def test_batch_queryset_filters_by_office(self, mock_batch_objects): + qs = self._build_mock_queryset() + mock_batch_objects.order_by.return_value = qs + office = MagicMock() + + RunTransformerForm(office=office) + + qs.filter.assert_any_call(country_office=office) + @patch("country_workspace.workspaces.admin.transformer.Batch.objects") def test_choices_master_detail_program(self, mock_batch_objects): qs = self._build_mock_queryset()