fix: validate operations in ha_bulk_control and report errors (#385) (#473)

* fix: validate operations in ha_bulk_control and report errors (#385)

Root cause: Silent `continue` statements in bulk_device_control skipped
operations missing entity_id or action without logging or reporting errors.
This caused empty operation lists when all operations had malformed data.

Changes:
- Add validate_operation() helper to check each operation
- Track skipped operations with index, original data, and error message
- Log warnings for each skipped operation
- Include skipped_operations count and skipped_details in response
- Add suggestions when operations are skipped
- Handle non-dict operations gracefully

The response now includes:
- skipped_operations: count of skipped operations
- skipped_details: list with index, error, and original operation
- suggestions: helpful messages for fixing malformed operations

* refactor: centralize validation to eliminate duplication per review

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: kingpanther13

src/ha_mcp/tools/device_control.py

@@ -486,44 +486,93 @@ async def bulk_device_control(
         if not operations:
             return {"success": False, "error": "No operations provided", "results": []}
 
-        results = []
-        operation_ids = []
+        results: list[dict[str, Any]] = []
+        operation_ids: list[str] = []
+        skipped_operations: list[dict[str, Any]] = []
+
+        def validate_operation(
+            op: Any, index: int
+        ) -> tuple[str | None, str | None, str | None]:
+            """Validate operation and return (entity_id, action, error) tuple."""
+            if not isinstance(op, dict):
+                error = f"Operation at index {index} is not a dict: {type(op).__name__}"
+                logger.warning(f"Bulk control: {error}")
+                return None, None, error
+
+            entity_id = op.get("entity_id")
+            action = op.get("action")
+
+            missing_fields = []
+            if not entity_id:
+                missing_fields.append("entity_id")
+            if not action:
+                missing_fields.append("action")
+
+            if missing_fields:
+                error = (
+                    f"Operation at index {index} missing required fields: "
+                    f"{', '.join(missing_fields)}"
+                )
+                logger.warning(f"Bulk control: {error}")
+                return None, None, error
+
+            return str(entity_id), str(action), None
 
         try:
+            # Validate all operations first (centralized validation)
+            valid_operations: list[tuple[int, dict[str, Any], str, str]] = []
+            for i, op in enumerate(operations):
+                entity_id, action, error = validate_operation(op, i)
+                if error:
+                    skipped_operations.append(
+                        {
+                            "index": i,
+                            "operation": op,
+                            "error": error,
+                            "success": False,
+                        }
+                    )
+                else:
+                    # Store (index, original_op, entity_id, action) for execution
+                    valid_operations.append((i, op, entity_id, action))  # type: ignore[arg-type]
+
+            # Execute only valid operations
             if parallel:
-                # Execute all operations in parallel
+                # Build tasks for parallel execution
                 tasks = []
-                for op in operations:
-                    entity_id = op.get("entity_id")
-                    action = op.get("action")
-                    if not entity_id or not action:
-                        continue
+                for _i, op, entity_id, action in valid_operations:
                     task = self.control_device_smart(
-                        entity_id=str(entity_id),
-                        action=str(action),
+                        entity_id=entity_id,
+                        action=action,
                         parameters=op.get("parameters"),
                         timeout_seconds=op.get("timeout_seconds", 10),
                         validate_first=op.get("validate_first", True),
                     )
                     tasks.append(task)
 
-                results = await asyncio.gather(*tasks, return_exceptions=True)
-
-                # Extract operation IDs
-                for result in results:
-                    if isinstance(result, dict) and "operation_id" in result:
-                        operation_ids.append(result["operation_id"])
+                if tasks:
+                    task_results = await asyncio.gather(*tasks, return_exceptions=True)
+
+                    # Process results and extract operation IDs
+                    for result in task_results:
+                        if isinstance(result, Exception):
+                            results.append(
+                                {
+                                    "success": False,
+                                    "error": f"Exception during execution: {result!s}",
+                                }
+                            )
+                        elif isinstance(result, dict):
+                            results.append(result)
+                            if "operation_id" in result:
+                                operation_ids.append(result["operation_id"])
 
             else:
-                # Execute operations sequentially
-                for op in operations:
-                    entity_id = op.get("entity_id")
-                    action = op.get("action")
-                    if not entity_id or not action:
-                        continue
+                # Execute valid operations sequentially
+                for _i, op, entity_id, action in valid_operations:
                     result = await self.control_device_smart(
-                        entity_id=str(entity_id),
-                        action=str(action),
+                        entity_id=entity_id,
+                        action=action,
                         parameters=op.get("parameters"),
                         timeout_seconds=op.get("timeout_seconds", 10),
                         validate_first=op.get("validate_first", True),
@@ -533,34 +582,52 @@ async def bulk_device_control(
                     if "operation_id" in result:
                         operation_ids.append(result["operation_id"])
 
-            # Count successes and failures
+            # Count successes and failures from executed operations
             successful = len(
                 [r for r in results if isinstance(r, dict) and r.get("command_sent")]
             )
-            failed = len(results) - successful
+            executed_failed = len(results) - successful
+            # Total failed includes both execution failures and skipped operations
+            total_failed = executed_failed + len(skipped_operations)
 
-            return {
+            response: dict[str, Any] = {
                 "total_operations": len(operations),
                 "successful_commands": successful,
-                "failed_commands": failed,
+                "failed_commands": total_failed,
+                "skipped_operations": len(skipped_operations),
                 "execution_mode": "parallel" if parallel else "sequential",
                 "operation_ids": operation_ids,
                 "results": results,
                 "follow_up": (
                     {
-                        "message": f"Use get_bulk_operation_status() to check all {len(operation_ids)} operations",
+                        "message": (
+                            f"Use get_bulk_operation_status() to check all "
+                            f"{len(operation_ids)} operations"
+                        ),
                         "operation_ids": operation_ids,
                     }
                     if operation_ids
                     else None
                 ),
             }
 
+            # Include skipped operation details if any were skipped
+            if skipped_operations:
+                response["skipped_details"] = skipped_operations
+                response["suggestions"] = [
+                    "Some operations were skipped due to validation errors",
+                    "Each operation requires 'entity_id' and 'action' fields",
+                    "Check skipped_details for specific errors",
+                    "Example format: {'entity_id': 'light.living_room', 'action': 'on'}",
+                ]
+
+            return response
+
         except Exception as e:
             logger.error(f"Error in bulk_device_control: {e}")
             return {
                 "success": False,
-                "error": f"Bulk operation failed: {str(e)}",
+                "error": f"Bulk operation failed: {e!s}",
                 "results": results,
             }
 

tests/src/unit/test_bulk_device_control.py

@@ -0,0 +1,137 @@
+"""Unit tests for bulk_device_control validation in device_control module."""
+
+import pytest
+
+from ha_mcp.tools.device_control import DeviceControlTools
+
+
+class TestBulkDeviceControlValidation:
+    """Test bulk_device_control validation logic."""
+
+    @pytest.fixture
+    def device_control_tools(self):
+        """Create DeviceControlTools with mocked client."""
+        # Pass None client - we won't actually make calls for validation tests
+        return DeviceControlTools(client=None)
+
+    @pytest.mark.asyncio
+    async def test_empty_operations_returns_error(self, device_control_tools):
+        """Empty operations list returns error."""
+        result = await device_control_tools.bulk_device_control([])
+        assert result["success"] is False
+        assert "No operations provided" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_missing_entity_id_reports_error(self, device_control_tools):
+        """Operations missing entity_id are reported in skipped_operations."""
+        operations = [
+            {"action": "on"},  # Missing entity_id
+        ]
+        result = await device_control_tools.bulk_device_control(operations)
+
+        assert result["total_operations"] == 1
+        assert result["skipped_operations"] == 1
+        assert len(result["skipped_details"]) == 1
+        assert "entity_id" in result["skipped_details"][0]["error"]
+        assert result["skipped_details"][0]["index"] == 0
+
+    @pytest.mark.asyncio
+    async def test_missing_action_reports_error(self, device_control_tools):
+        """Operations missing action are reported in skipped_operations."""
+        operations = [
+            {"entity_id": "light.test"},  # Missing action
+        ]
+        result = await device_control_tools.bulk_device_control(operations)
+
+        assert result["total_operations"] == 1
+        assert result["skipped_operations"] == 1
+        assert len(result["skipped_details"]) == 1
+        assert "action" in result["skipped_details"][0]["error"]
+
+    @pytest.mark.asyncio
+    async def test_missing_both_fields_reports_both(self, device_control_tools):
+        """Operations missing both fields report both missing fields."""
+        operations = [
+            {},  # Missing both entity_id and action
+        ]
+        result = await device_control_tools.bulk_device_control(operations)
+
+        assert result["skipped_operations"] == 1
+        error_msg = result["skipped_details"][0]["error"]
+        assert "entity_id" in error_msg
+        assert "action" in error_msg
+
+    @pytest.mark.asyncio
+    async def test_non_dict_operation_reports_error(self, device_control_tools):
+        """Non-dict operations are reported as errors."""
+        operations = [
+            "not a dict",
+            123,
+            None,
+        ]
+        result = await device_control_tools.bulk_device_control(operations)
+
+        assert result["total_operations"] == 3
+        assert result["skipped_operations"] == 3
+        assert len(result["skipped_details"]) == 3
+        for detail in result["skipped_details"]:
+            assert "not a dict" in detail["error"]
+
+    @pytest.mark.asyncio
+    async def test_mixed_valid_and_invalid_operations(self, device_control_tools):
+        """Mix of valid and invalid operations reports skipped ones.
+
+        Note: This test only validates that invalid operations are tracked.
+        Valid operations would require a real HA connection to execute.
+        """
+        operations = [
+            {"entity_id": "light.test", "action": "on"},  # Valid (but will fail without HA)
+            {"action": "off"},  # Invalid - missing entity_id
+            {"entity_id": "switch.test"},  # Invalid - missing action
+        ]
+        result = await device_control_tools.bulk_device_control(operations)
+
+        assert result["total_operations"] == 3
+        assert result["skipped_operations"] == 2
+        # The valid operation would be attempted but fail (no client)
+        # so we check that skipped operations are properly tracked
+        assert len(result["skipped_details"]) == 2
+
+        # Verify indices are tracked correctly
+        skipped_indices = [d["index"] for d in result["skipped_details"]]
+        assert 1 in skipped_indices  # Missing entity_id
+        assert 2 in skipped_indices  # Missing action
+
+    @pytest.mark.asyncio
+    async def test_all_invalid_operations_has_suggestions(self, device_control_tools):
+        """When operations are skipped, response includes suggestions."""
+        operations = [
+            {"action": "on"},  # Invalid
+        ]
+        result = await device_control_tools.bulk_device_control(operations)
+
+        assert "suggestions" in result
+        assert any("entity_id" in s for s in result["suggestions"])
+        assert any("action" in s for s in result["suggestions"])
+
+    @pytest.mark.asyncio
+    async def test_skipped_details_includes_original_operation(self, device_control_tools):
+        """Skipped details include the original operation for debugging."""
+        original_op = {"action": "on", "parameters": {"brightness": 100}}
+        operations = [original_op]
+        result = await device_control_tools.bulk_device_control(operations)
+
+        assert result["skipped_details"][0]["operation"] == original_op
+
+    @pytest.mark.asyncio
+    async def test_sequential_execution_validates_operations(self, device_control_tools):
+        """Sequential execution mode also validates operations."""
+        operations = [
+            {"action": "on"},  # Missing entity_id
+        ]
+        result = await device_control_tools.bulk_device_control(
+            operations, parallel=False
+        )
+
+        assert result["skipped_operations"] == 1
+        assert result["execution_mode"] == "sequential"