fix(spp_api_v2_simulation): add missing schema field and sudo, expand tests

- Add targeting_expression_explanation to ScenarioUpdateRequest (was causing 500 on PUT)
- Add sudo() to simulation service call in run_simulation endpoint (was causing AccessError)
- Add 27 new tests covering scenario update, run helpers, and aggregation schemas

Author: emjay0921

spp_api_v2_simulation/routers/scenario.py

@@ -438,7 +438,8 @@ async def run_simulation(
             )
 
         # Execute simulation via service
-        service = env["spp.simulation.service"]
+        # nosemgrep: odoo-sudo-without-context
+        service = env["spp.simulation.service"].sudo()
         run = service.execute_simulation(scenario)
 
         return RunSimulationResponse(

spp_api_v2_simulation/schemas/scenario.py

@@ -60,6 +60,9 @@ class ScenarioUpdateRequest(BaseModel):
     category: str | None = Field(default=None, description="Scenario category")
     target_type: str | None = Field(default=None, description="Target type: 'group' or 'individual'")
     targeting_expression: str | None = Field(default=None, description="CEL expression for targeting beneficiaries")
+    targeting_expression_explanation: str | None = Field(
+        default=None, description="Plain language explanation of targeting expression"
+    )
     ideal_population_expression: str | None = Field(
         default=None, description="CEL expression for ideal population calculation"
     )

spp_api_v2_simulation/tests/__init__.py

@@ -2,5 +2,8 @@
 
 from . import test_comparison_api, test_convert_to_program_api, test_run_api, test_scenario_api
 from . import test_scope_registration
+from . import test_aggregation_api
 from . import test_aggregation_service
+from . import test_run_helpers
+from . import test_scenario_update
 from . import test_simulation_service

spp_api_v2_simulation/tests/test_aggregation_api.py

@@ -0,0 +1,165 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+"""Tests for aggregation API schemas and endpoint logic."""
+
+from odoo.tests import tagged
+
+from .common import SimulationApiTestCase
+
+
+@tagged("-at_install", "post_install")
+class TestAggregationApi(SimulationApiTestCase):
+    """Test aggregation endpoint schemas and logic."""
+
+    def test_aggregation_scope_request_defaults(self):
+        """Test AggregationScopeRequest schema defaults."""
+        from ..schemas.aggregation import AggregationScopeRequest
+
+        scope = AggregationScopeRequest()
+        self.assertEqual(scope.target_type, "group")
+        self.assertIsNone(scope.cel_expression)
+        self.assertIsNone(scope.area_id)
+
+    def test_aggregation_scope_request_with_values(self):
+        """Test AggregationScopeRequest with explicit values."""
+        from ..schemas.aggregation import AggregationScopeRequest
+
+        scope = AggregationScopeRequest(
+            target_type="individual",
+            cel_expression="r.age >= 18",
+            area_id=42,
+        )
+        self.assertEqual(scope.target_type, "individual")
+        self.assertEqual(scope.cel_expression, "r.age >= 18")
+        self.assertEqual(scope.area_id, 42)
+
+    def test_compute_aggregation_request_minimal(self):
+        """Test ComputeAggregationRequest with minimal fields."""
+        from ..schemas.aggregation import (
+            AggregationScopeRequest,
+            ComputeAggregationRequest,
+        )
+
+        request = ComputeAggregationRequest(
+            scope=AggregationScopeRequest(),
+        )
+        self.assertEqual(request.scope.target_type, "group")
+        self.assertIsNone(request.statistics)
+        self.assertIsNone(request.group_by)
+
+    def test_compute_aggregation_request_full(self):
+        """Test ComputeAggregationRequest with all fields."""
+        from ..schemas.aggregation import (
+            AggregationScopeRequest,
+            ComputeAggregationRequest,
+        )
+
+        request = ComputeAggregationRequest(
+            scope=AggregationScopeRequest(
+                target_type="individual",
+                cel_expression="r.age >= 60",
+            ),
+            statistics=["count", "average_age"],
+            group_by=["gender", "area"],
+        )
+        self.assertEqual(request.scope.target_type, "individual")
+        self.assertEqual(len(request.statistics), 2)
+        self.assertEqual(len(request.group_by), 2)
+
+    def test_aggregation_response_schema(self):
+        """Test AggregationResponse schema."""
+        from ..schemas.aggregation import AggregationResponse
+
+        response = AggregationResponse(
+            total_count=150,
+            statistics={"average_age": 45.2, "count": 150},
+            breakdown={"gender": {"male": 70, "female": 80}},
+            from_cache=False,
+            computed_at="2024-01-01T12:00:00Z",
+            access_level="aggregate",
+        )
+        self.assertEqual(response.total_count, 150)
+        self.assertEqual(response.statistics["average_age"], 45.2)
+        self.assertIn("gender", response.breakdown)
+        self.assertFalse(response.from_cache)
+
+    def test_aggregation_response_no_breakdown(self):
+        """Test AggregationResponse without breakdown."""
+        from ..schemas.aggregation import AggregationResponse
+
+        response = AggregationResponse(
+            total_count=100,
+            statistics={},
+            from_cache=True,
+            computed_at="2024-01-01T12:00:00Z",
+            access_level="aggregate",
+        )
+        self.assertIsNone(response.breakdown)
+        self.assertTrue(response.from_cache)
+
+    def test_dimension_info_schema(self):
+        """Test DimensionInfo schema."""
+        from ..schemas.aggregation import DimensionInfo
+
+        dim = DimensionInfo(
+            name="gender",
+            label="Gender",
+            description="Biological sex",
+            dimension_type="field",
+            applies_to="all",
+            value_labels={"male": "Male", "female": "Female"},
+        )
+        self.assertEqual(dim.name, "gender")
+        self.assertEqual(dim.label, "Gender")
+        self.assertEqual(dim.dimension_type, "field")
+        self.assertEqual(dim.applies_to, "all")
+        self.assertIn("male", dim.value_labels)
+
+    def test_dimension_info_minimal(self):
+        """Test DimensionInfo with no optional fields."""
+        from ..schemas.aggregation import DimensionInfo
+
+        dim = DimensionInfo(
+            name="custom_dim",
+            label="Custom",
+            dimension_type="expression",
+            applies_to="individuals",
+        )
+        self.assertIsNone(dim.description)
+        self.assertIsNone(dim.value_labels)
+
+    def test_dimensions_list_response_schema(self):
+        """Test DimensionsListResponse schema."""
+        from ..schemas.aggregation import DimensionInfo, DimensionsListResponse
+
+        dims = [
+            DimensionInfo(
+                name="gender",
+                label="Gender",
+                dimension_type="field",
+                applies_to="all",
+            ),
+            DimensionInfo(
+                name="age_group",
+                label="Age Group",
+                dimension_type="expression",
+                applies_to="individuals",
+            ),
+        ]
+        response = DimensionsListResponse(dimensions=dims)
+        self.assertEqual(len(response.dimensions), 2)
+        self.assertEqual(response.dimensions[0].name, "gender")
+        self.assertEqual(response.dimensions[1].name, "age_group")
+
+    def test_scope_model_dump(self):
+        """Test that scope model_dump produces the expected dict."""
+        from ..schemas.aggregation import AggregationScopeRequest
+
+        scope = AggregationScopeRequest(
+            target_type="group",
+            cel_expression="r.area_id != false",
+            area_id=5,
+        )
+        dumped = scope.model_dump()
+        self.assertEqual(dumped["target_type"], "group")
+        self.assertEqual(dumped["cel_expression"], "r.area_id != false")
+        self.assertEqual(dumped["area_id"], 5)

spp_api_v2_simulation/tests/test_run_helpers.py

@@ -0,0 +1,130 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+"""Tests for run router helper functions and edge cases."""
+
+from datetime import datetime
+
+from odoo.tests import tagged
+
+from .common import SimulationApiTestCase
+
+
+@tagged("-at_install", "post_install")
+class TestRunHelpers(SimulationApiTestCase):
+    """Test run helper functions and edge cases."""
+
+    def test_datetime_to_iso_with_datetime(self):
+        """Test _datetime_to_iso with a datetime object."""
+        from ..routers.run import _datetime_to_iso
+
+        dt = datetime(2024, 6, 15, 10, 30, 0)
+        result = _datetime_to_iso(dt)
+        self.assertEqual(result, "2024-06-15T10:30:00")
+
+    def test_datetime_to_iso_with_none(self):
+        """Test _datetime_to_iso with None."""
+        from ..routers.run import _datetime_to_iso
+
+        self.assertIsNone(_datetime_to_iso(None))
+
+    def test_datetime_to_iso_with_false(self):
+        """Test _datetime_to_iso with Odoo False."""
+        from ..routers.run import _datetime_to_iso
+
+        self.assertIsNone(_datetime_to_iso(False))
+
+    def test_datetime_to_iso_with_string(self):
+        """Test _datetime_to_iso with a string (passthrough)."""
+        from ..routers.run import _datetime_to_iso
+
+        result = _datetime_to_iso("2024-01-01T00:00:00Z")
+        self.assertEqual(result, "2024-01-01T00:00:00Z")
+
+    def test_run_to_summary_with_no_duration(self):
+        """Test run to summary when execution_duration_seconds is 0."""
+        from ..routers.run import _run_to_summary
+
+        run = self.env["spp.simulation.run"].create(
+            {
+                "scenario_id": self.scenario_ready.id,
+                "state": "completed",
+                "beneficiary_count": 3,
+                "total_cost": 1500.0,
+                "coverage_rate": 30.0,
+                "equity_score": 75.0,
+                "gini_coefficient": 0.2,
+                "executed_at": datetime(2024, 3, 1, 12, 0, 0),
+                "execution_duration_seconds": 0.0,
+            }
+        )
+
+        summary = _run_to_summary(run)
+        self.assertIsNone(summary.execution_duration_seconds)
+
+    def test_run_to_response_failed_with_no_error_message(self):
+        """Test response for failed run with no error message."""
+        from ..routers.run import _run_to_response
+
+        run = self.env["spp.simulation.run"].create(
+            {
+                "scenario_id": self.scenario_ready.id,
+                "state": "failed",
+                "beneficiary_count": 0,
+                "total_cost": 0.0,
+            }
+        )
+
+        response = _run_to_response(run, include_details=False)
+        self.assertEqual(response.state, "failed")
+        self.assertIsNone(response.error_message)
+
+    def test_run_to_response_details_not_loaded_for_failed(self):
+        """Test that include_details=True doesn't load details for failed runs."""
+        from ..routers.run import _run_to_response
+
+        self.run_failed.distribution_json = {
+            "count": 0,
+            "total": 0.0,
+        }
+
+        response = _run_to_response(self.run_failed, include_details=True)
+
+        # Details should NOT be loaded for failed runs
+        self.assertIsNone(response.distribution_data)
+        self.assertIsNone(response.fairness_data)
+
+    def test_run_to_response_with_empty_json_fields(self):
+        """Test response when JSON fields are empty dicts/lists."""
+        from ..routers.run import _run_to_response
+
+        # Empty JSON fields should not create data objects
+        self.run_completed.distribution_json = {}
+        self.run_completed.fairness_json = {}
+        self.run_completed.geographic_json = []
+        self.run_completed.metric_results_json = {}
+
+        response = _run_to_response(self.run_completed, include_details=True)
+
+        # Empty dicts/lists are falsy, so data should be None
+        self.assertIsNone(response.distribution_data)
+        self.assertIsNone(response.fairness_data)
+        self.assertIsNone(response.geographic_data)
+        self.assertIsNone(response.metric_results)
+
+    def test_run_to_response_with_scenario_snapshot_no_ideal(self):
+        """Test scenario snapshot without ideal_population_expression."""
+        from ..routers.run import _run_to_response
+
+        self.run_completed.scenario_snapshot_json = {
+            "name": "Test",
+            "target_type": "group",
+            "targeting_expression": "true",
+            "budget_amount": 5000.0,
+            "budget_strategy": "none",
+            "entitlement_rules": [],
+        }
+
+        response = _run_to_response(self.run_completed, include_details=True)
+
+        self.assertIsNotNone(response.scenario_snapshot)
+        self.assertEqual(response.scenario_snapshot.name, "Test")
+        self.assertIsNone(response.scenario_snapshot.ideal_population_expression)