homeassistant-ai
diff --git a/‎tests/AGENTS.md‎
Lines changed: 10 additions & 2 deletions b/‎tests/AGENTS.md‎
Lines changed: 10 additions & 2 deletions
diff --git a/‎tests/src/e2e/utilities/assertions.py‎
Lines changed: 36 additions & 16 deletions b/‎tests/src/e2e/utilities/assertions.py‎
Lines changed: 36 additions & 16 deletions
diff --git a/‎tests/src/unit/test_e2e_assertions.py‎
Lines changed: 61 additions & 1 deletion b/‎tests/src/unit/test_e2e_assertions.py‎
Lines changed: 61 additions & 1 deletion
@@ -49,8 +49,16 @@ rather than assume the e2e's always-present case.
 ## Test Patterns
 
 - Tests expecting tool **success**: use `mcp.call_tool_success()` inside `MCPAssertions` context
-- Tests expecting tool **failure**: use `safe_call_tool()` directly (catches `ToolError`, returns parsed dict)
-- Service availability checks should use `safe_call_tool` to probe, not `call_tool_success`
+- Tests expecting tool **failure**: use `mcp.call_tool_failure()` inside `MCPAssertions`
+  context. It rejects any result `assert_mcp_success()` would accept — including the
+  tools that succeed with no `success` key (`pending_restart`, bulk-operation payloads)
+  — so it genuinely proves the call failed. Prefer also passing `expected_error` to pin
+  *which* failure; about half the current call sites omit it and assert on the returned
+  dict themselves instead, which is equally fine.
+- `safe_call_tool()` is for calls whose outcome the test does **not** assert: `finally`
+  cleanup (so a cleanup failure cannot mask the real assertion) and service-availability
+  probes. It swallows `ToolError` and returns a parsed dict, so using it for an expected
+  failure means nothing verifies the call failed at all.
 
 ## E2E Test Patterns
 
 
@@ -113,8 +113,11 @@ async def safe_call_tool(
 ) -> dict[str, Any]:
     """Call an MCP tool and return parsed result, handling ToolError exceptions.
 
-    This is useful for tests that expect tools to fail and want to inspect
-    the error response without catching exceptions manually.
+    For a call whose outcome the test does NOT assert: ``finally``-block cleanup
+    (so a cleanup failure cannot mask the real assertion) and service-availability
+    probes. It swallows ``ToolError`` and returns a parsed dict either way, so a
+    test that asserts a failure should use ``MCPAssertions.call_tool_failure()``
+    with ``expected_error`` instead -- see tests/AGENTS.md "Test Patterns".
 
     Args:
         mcp_client: The MCP client instance
@@ -131,17 +134,16 @@ async def safe_call_tool(
         return tool_error_to_result(exc)
 
 
-def assert_mcp_success(result, operation_name: str = "operation"):
-    """
-    Assert that MCP tool result indicates success.
+def looks_like_success(data: dict[str, Any]) -> bool:
+    """Whether a parsed tool result is a success response.
 
-    Args:
-        result: FastMCP client result
-        operation_name: Name of operation for error message
+    Shared by ``assert_mcp_success`` and ``assert_mcp_failure`` so the two
+    cannot disagree about what "success" means. Several tools succeed
+    WITHOUT a ``success`` key (``pending_restart``, bulk-operation
+    payloads), so a bare ``data.get("success")`` check treats those as
+    failures -- which is fine for the success assertion (it lists them
+    explicitly) but silently accepted them as failures on the other side.
     """
-    data = parse_mcp_result(result)
-
-    # Handle different success indicators
     success_indicators = [
         data.get("success") is True,
         # ha_manage_app's options/network write returns
@@ -172,7 +174,20 @@ def assert_mcp_success(result, operation_name: str = "operation"):
         ),
     ]
 
-    if not any(success_indicators):
+    return any(success_indicators)
+
+
+def assert_mcp_success(result, operation_name: str = "operation"):
+    """
+    Assert that MCP tool result indicates success.
+
+    Args:
+        result: FastMCP client result
+        operation_name: Name of operation for error message
+    """
+    data = parse_mcp_result(result)
+
+    if not looks_like_success(data):
         error_msg = data.get("error", "Unknown error")
         suggestions = data.get("suggestions", [])
 
@@ -199,8 +214,12 @@ def assert_mcp_failure(
     """
     data = parse_mcp_result(result)
 
-    # Check that operation actually failed
-    if data.get("success"):
+    # Check that operation actually failed. Uses the shared success
+    # predicate, not a bare data.get("success"): a tool that succeeds
+    # without a success key (pending_restart, bulk-operation payloads)
+    # would otherwise be accepted here as a failure, so a regression that
+    # made an expected-failure call SUCCEED could pass unnoticed.
+    if looks_like_success(data):
         raise AssertionError(f"{operation_name} should have failed but succeeded")
 
     # If expected error specified, check for it
@@ -380,8 +399,9 @@ async def call_tool_failure(
         except ToolError as exc:
             # Convert ToolError to result dict and validate
             data = tool_error_to_result(exc)
-            # Verify this is actually a failure
-            if data.get("success"):
+            # Verify this is actually a failure (shared predicate, see
+            # assert_mcp_failure)
+            if looks_like_success(data):
                 raise AssertionError(
                     f"{operation_name} should have failed but succeeded"
                 ) from exc
 
@@ -29,7 +29,11 @@
 import pytest
 
 from tests.src.e2e.error_handling.test_network_errors import _hard_failures
-from tests.src.e2e.utilities.assertions import assert_mcp_success, parse_mcp_result
+from tests.src.e2e.utilities.assertions import (
+    assert_mcp_failure,
+    assert_mcp_success,
+    parse_mcp_result,
+)
 
 
 def test_pending_restart_is_a_successful_mcp_result() -> None:
@@ -221,3 +225,59 @@ def test_mixed_batch_returns_only_the_untolerated_failures(self):
         healthy_bulk = {"successful_commands": 2, "failed_commands": 0}
         results = [exc, tolerated, tool_error, healthy_bulk, {"success": True}]
         assert _hard_failures(results) == [exc, tool_error]
+
+
+class TestFailureAssertionRejectsEverySuccessShape:
+    """``assert_mcp_failure`` must reject anything ``assert_mcp_success``
+    accepts.
+
+    It used to guard on a bare ``if data.get("success")``, so the tools that
+    succeed WITHOUT that key sailed through as failures: ``ha_manage_app``'s
+    options/network write returns ``{"status": "pending_restart"}``, and
+    bulk-operation payloads carry ``total_operations``/``results`` with no
+    ``success`` field. An expected-failure test whose tool regressed to one
+    of those shapes passed silently -- and 30 of the 61 ``call_tool_failure``
+    call sites omit ``expected_error``, so nothing downstream caught it
+    either. This is the exact class of harness bug that degrades into a
+    green run, which is why it is pinned here rather than in the e2e suite.
+    """
+
+    @pytest.mark.parametrize(
+        ("shape", "why"),
+        [
+            ({"success": True}, "explicit success"),
+            ({"status": "pending_restart"}, "ha_manage_app write, no success key"),
+            (
+                {"total_operations": 2, "successful_commands": 2},
+                "bulk payload, no success key",
+            ),
+            ({"results": [], "operation_ids": ["a"]}, "bulk payload via results"),
+            ({"data": {"x": 1}}, "data present, no error, no success key"),
+        ],
+    )
+    def test_success_shapes_are_not_accepted_as_failures(self, shape, why) -> None:
+        with pytest.raises(AssertionError, match="should have failed but succeeded"):
+            assert_mcp_failure(shape, why)
+
+    @pytest.mark.parametrize(
+        "shape",
+        [
+            {"success": False, "error": {"code": "VALIDATION_FAILED"}},
+            {"success": False, "error": "boom"},
+            {"error": "boom"},
+        ],
+    )
+    def test_genuine_failures_still_pass(self, shape) -> None:
+        assert assert_mcp_failure(shape, "op") == shape
+
+    def test_expected_error_still_matched_on_a_genuine_failure(self) -> None:
+        shape = {"success": False, "error": {"message": "config_hash mismatch"}}
+        assert_mcp_failure(shape, "op", "config_hash")
+        with pytest.raises(AssertionError, match="doesn't contain"):
+            assert_mcp_failure(shape, "op", "not_present")
+
+    def test_pending_restart_with_an_explicit_error_is_a_real_failure(self) -> None:
+        """The success predicate guards ``pending_restart`` on ``error is
+        None``, so a write that reports both must stay a failure here."""
+        shape = {"status": "pending_restart", "error": "supervisor rejected it"}
+        assert assert_mcp_failure(shape, "op") == shape