Add unit tests for transformer application in batch processing

- Introduced multiple test cases for the `_apply_transformer_only` and `apply_batch_transformers` functions to ensure correct behavior when applying transformations to individual and household records.
- Added tests to verify that transformers are only applied when necessary and handle various flex field types appropriately.
- Included checks for error handling when transformer IDs are missing, enhancing the robustness of batch processing functionality.

Author: Arsen P

tests/workspace/test_batch_reprocessing_task.py

@@ -1142,3 +1142,188 @@ def test_reprocess_kobo_resolves_household_refs(program, user: User, force_migra
     assert household.flex_fields["head_of_household_id"] == head.pk
     assert household.flex_fields["primary_collector_id"] == primary.pk
     assert household.flex_fields["alternate_collector_id"] == alternate.pk
+
+
+def test_apply_transformer_only_updates_record(program, user: User, force_migrated_records) -> None:
+    from country_workspace.workspaces.admin.batch.reprocessing import _apply_transformer_only
+    from testutils.factories import CountryIndividualFactory, TransformerFactory
+
+    individual = CountryIndividualFactory(
+        household=None,
+        batch__program=program,
+        batch__country_office=program.country_office,
+        flex_fields={"gender": "F"},
+    )
+    transformer = TransformerFactory(
+        office=program.country_office,
+        value_transformations="function t(d) { d['gender'] = 'FEMALE'; return d; }",
+    )
+
+    applied = _apply_transformer_only(individual, transformer)
+
+    assert applied is True
+    individual.refresh_from_db()
+    assert individual.flex_fields["gender"] == "FEMALE"
+
+
+def test_apply_transformer_only_returns_false_when_unchanged(program, user: User, force_migrated_records) -> None:
+    from country_workspace.workspaces.admin.batch.reprocessing import _apply_transformer_only
+    from testutils.factories import CountryIndividualFactory, TransformerFactory
+
+    individual = CountryIndividualFactory(
+        household=None,
+        batch__program=program,
+        batch__country_office=program.country_office,
+        flex_fields={"gender": "F"},
+    )
+    transformer = TransformerFactory(
+        office=program.country_office,
+        value_transformations="function t(d) { return d; }",
+    )
+
+    applied = _apply_transformer_only(individual, transformer)
+
+    assert applied is False
+
+
+def test_apply_transformer_only_uses_empty_dict_for_non_dict_flex_fields(
+    program, user: User, force_migrated_records
+) -> None:
+    from country_workspace.workspaces.admin.batch.reprocessing import _apply_transformer_only
+    from testutils.factories import CountryIndividualFactory, TransformerFactory
+
+    individual = CountryIndividualFactory(
+        household=None,
+        batch__program=program,
+        batch__country_office=program.country_office,
+        flex_fields="unexpected",
+    )
+    transformer = TransformerFactory(
+        office=program.country_office,
+        value_transformations="function t(d) { d['created'] = true; return d; }",
+    )
+
+    applied = _apply_transformer_only(individual, transformer)
+
+    assert applied is True
+    individual.refresh_from_db()
+    assert individual.flex_fields["created"] is True
+
+
+def test_apply_batch_transformers_requires_transformer_ids(program, user: User) -> None:
+    from country_workspace.workspaces.admin.batch.reprocessing import apply_batch_transformers
+    from testutils.factories import AsyncJobFactory, CountryBatchFactory
+
+    batch = CountryBatchFactory(program=program, country_office=program.country_office)
+    job = AsyncJobFactory(program=program, batch=batch, owner=user, config={"batch_id": batch.pk})
+
+    with pytest.raises(ValueError, match="At least one transformer id is required"):
+        apply_batch_transformers(job)
+
+
+def test_apply_batch_transformers_applies_individual_only(program, user: User, force_migrated_records) -> None:
+    from country_workspace.workspaces.admin.batch.reprocessing import apply_batch_transformers
+    from testutils.factories import AsyncJobFactory, CountryIndividualFactory, TransformerFactory
+
+    individual = CountryIndividualFactory(
+        household=None,
+        batch__program=program,
+        batch__country_office=program.country_office,
+        flex_fields={"status": "1"},
+    )
+    batch = individual.batch
+    transformer = TransformerFactory(
+        office=program.country_office,
+        value_transformations="function t(d) { d['status'] = 'ACTIVE'; return d; }",
+    )
+    job = AsyncJobFactory(
+        program=program,
+        batch=batch,
+        owner=user,
+        config={"batch_id": batch.pk, "individual_transformer_id": transformer.pk},
+    )
+
+    with patch("country_workspace.workspaces.admin.batch.reprocessing.sync_collector_links") as mock_sync:
+        result = apply_batch_transformers(job)
+
+    individual.refresh_from_db()
+    assert individual.flex_fields["status"] == "ACTIVE"
+    assert result["batch_id"] == batch.pk
+    assert result["transformed_individuals"] == 1
+    mock_sync.assert_called_once()
+
+
+def test_apply_batch_transformers_applies_household_only_master_detail(
+    program, user: User, force_migrated_records
+) -> None:
+    from country_workspace.workspaces.admin.batch.reprocessing import apply_batch_transformers
+    from testutils.factories import AsyncJobFactory, CountryHouseholdFactory, TransformerFactory
+    from testutils.factories.program import BeneficiaryGroupFactory
+
+    bg = BeneficiaryGroupFactory(master_detail=True)
+    program.beneficiary_group = bg
+    program.save()
+
+    household = CountryHouseholdFactory(
+        batch__program=program,
+        batch__country_office=program.country_office,
+        flex_fields={"category": "A"},
+    )
+    batch = household.batch
+    transformer = TransformerFactory(
+        office=program.country_office,
+        value_transformations="function t(d) { d['category'] = 'B'; return d; }",
+    )
+    job = AsyncJobFactory(
+        program=program,
+        batch=batch,
+        owner=user,
+        config={"batch_id": batch.pk, "household_transformer_id": transformer.pk},
+    )
+
+    with patch("country_workspace.workspaces.admin.batch.reprocessing._sync_household_refs") as mock_sync:
+        result = apply_batch_transformers(job)
+
+    household.refresh_from_db()
+    assert household.flex_fields["category"] == "B"
+    assert result["transformed_households"] == 1
+    assert result["transformed_individuals"] == 0
+    mock_sync.assert_called_once_with(batch)
+
+
+def test_apply_batch_transformers_skips_households_for_non_master_detail(
+    program, user: User, force_migrated_records
+) -> None:
+    from country_workspace.workspaces.admin.batch.reprocessing import apply_batch_transformers
+    from testutils.factories import AsyncJobFactory, CountryHouseholdFactory, TransformerFactory
+    from testutils.factories.program import BeneficiaryGroupFactory
+
+    bg = BeneficiaryGroupFactory(master_detail=False)
+    program.beneficiary_group = bg
+    program.save()
+
+    household = CountryHouseholdFactory(
+        batch__program=program,
+        batch__country_office=program.country_office,
+        flex_fields={"category": "A"},
+    )
+    batch = household.batch
+    transformer = TransformerFactory(
+        office=program.country_office,
+        value_transformations="function t(d) { d['category'] = 'B'; return d; }",
+    )
+    job = AsyncJobFactory(
+        program=program,
+        batch=batch,
+        owner=user,
+        config={"batch_id": batch.pk, "household_transformer_id": transformer.pk},
+    )
+
+    with patch("country_workspace.workspaces.admin.batch.reprocessing._sync_household_refs") as mock_sync:
+        result = apply_batch_transformers(job)
+
+    household.refresh_from_db()
+    assert household.flex_fields["category"] == "A"
+    assert "transformed_households" not in result
+    assert result["transformed_individuals"] == 0
+    mock_sync.assert_not_called()