Merge branch 'ARG-AMBIGUITY' of https://github.com/PerfectoCode/perfecto-mcp into AI-SCRIPTLESS-ALIGNMENT

Author: diego-ferrand

tests/test_ai_scriptless_manager.py

@@ -1208,6 +1208,62 @@ def test_routes_add_command(self, perfecto_token, monkeypatch):
         assert result.result["command_id"] == "wait"
         assert len(captured["script"]["flowElements"]) == 1
 
+    def test_add_command_accepts_arguments_alias(self, perfecto_token, monkeypatch):
+        captured: dict = {}
+        _mock_load_and_mutate(monkeypatch, captured=captured)
+
+        tool = _register_tool(perfecto_token)
+        result = asyncio.run(_call_tool(tool, "add_command", {
+            "test_id": TEST_ID,
+            "command_id": "wait",
+            "arguments": {"duration": "7"},
+        }))
+
+        assert result.error is None
+        values = {
+            argument["name"]: argument["data"]["value"]
+            for argument in captured["script"]["flowElements"][0]["arguments"]
+        }
+        assert values["duration"] == "7"
+
+    def test_add_command_prefers_cmd_arguments_over_arguments_alias(
+            self, perfecto_token, monkeypatch):
+        captured: dict = {}
+        _mock_load_and_mutate(monkeypatch, captured=captured)
+
+        tool = _register_tool(perfecto_token)
+        result = asyncio.run(_call_tool(tool, "add_command", {
+            "test_id": TEST_ID,
+            "command_id": "wait",
+            "cmd_arguments": {"duration": "4"},
+            "arguments": {"duration": "9"},
+        }))
+
+        assert result.error is None
+        values = {
+            argument["name"]: argument["data"]["value"]
+            for argument in captured["script"]["flowElements"][0]["arguments"]
+        }
+        assert values["duration"] == "4"
+
+    def test_modify_command_accepts_arguments_alias(self, perfecto_token, monkeypatch):
+        captured: dict = {}
+        _mock_load_and_mutate(monkeypatch, _script_with_steps("wait"), captured)
+
+        tool = _register_tool(perfecto_token)
+        result = asyncio.run(_call_tool(tool, "modify_command", {
+            "test_id": TEST_ID,
+            "step_path": "0",
+            "arguments": {"duration": "5"},
+        }))
+
+        assert result.error is None
+        values = {
+            argument["name"]: argument["data"]["value"]
+            for argument in captured["script"]["flowElements"][0]["arguments"]
+        }
+        assert values["duration"] == "5"
+
     def test_routes_list_test_variables(self, perfecto_token, monkeypatch):
         script = new_empty_script()
         script["variables"] = [{

tests/test_normalize_action_args.py

@@ -0,0 +1,98 @@
+"""
+Copyright 2025 Perforce Software, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+    10|Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+from tools.utils import normalize_action_args
+
+
+class TestNormalizeActionArgs:
+    def test_none_returns_empty_action_and_args(self):
+        action, args = normalize_action_args(None)
+        assert action == ""
+        assert args == {}
+
+    def test_empty_dict_returns_empty_action_and_args(self):
+        action, args = normalize_action_args({})
+        assert action == ""
+        assert args == {}
+
+    def test_nested_action_and_args(self):
+        action, args = normalize_action_args({
+            "action": "add_command",
+            "args": {"test_id": "PRIVATE:Folder/Test.xml", "command_id": "wait"},
+        })
+        assert action == "add_command"
+        assert args == {"test_id": "PRIVATE:Folder/Test.xml", "command_id": "wait"}
+
+    def test_flattened_top_level_params_merge_into_args(self):
+        action, args = normalize_action_args({
+            "action": "add_command",
+            "test_id": "PRIVATE:Folder/Test.xml",
+            "command_id": "wait",
+            "cmd_arguments": {"duration": "1"},
+        })
+        assert action == "add_command"
+        assert args == {
+            "test_id": "PRIVATE:Folder/Test.xml",
+            "command_id": "wait",
+            "cmd_arguments": {"duration": "1"},
+        }
+
+    def test_unwraps_double_wrapped_arguments(self):
+        action, args = normalize_action_args({
+            "arguments": {
+                "action": "add_command",
+                "args": {"test_id": "PRIVATE:Folder/Test.xml"},
+            }
+        })
+        assert action == "add_command"
+        assert args == {"test_id": "PRIVATE:Folder/Test.xml"}
+
+    def test_does_not_unwrap_arguments_when_other_keys_present(self):
+        action, args = normalize_action_args({
+            "action": "add_command",
+            "args": {"test_id": "PRIVATE:Folder/Test.xml", "command_id": "wait"},
+            "arguments": {"duration": "3"},
+        })
+        assert action == "add_command"
+        assert args["test_id"] == "PRIVATE:Folder/Test.xml"
+        assert args["command_id"] == "wait"
+        assert args["arguments"] == {"duration": "3"}
+
+    def test_strips_action_whitespace(self):
+        action, args = normalize_action_args({"action": "  list_tests  ", "args": {}})
+        assert action == "list_tests"
+        assert args == {}
+
+    def test_none_args_value_becomes_empty_dict(self):
+        action, args = normalize_action_args({"action": "list_tests", "args": None})
+        assert action == "list_tests"
+        assert args == {}
+
+    def test_args_is_never_none(self):
+        _, args = normalize_action_args(None)
+        assert args is not None
+        _, args = normalize_action_args({"action": "x"})
+        assert args is not None
+
+    def test_top_level_keys_overlay_nested_args(self):
+        action, args = normalize_action_args({
+            "action": "add_command",
+            "args": {"test_id": "nested", "command_id": "wait"},
+            "test_id": "top-level",
+        })
+        assert action == "add_command"
+        assert args["test_id"] == "top-level"
+        assert args["command_id"] == "wait"

tools/ai_scriptless_manager.py

@@ -81,6 +81,11 @@
 )
 
 
+def _command_arguments(args: Dict[str, Any]) -> Any:
+    """Prefer cmd_arguments; accept arguments for older clients that used that key."""
+    return args.get("cmd_arguments") or args.get("arguments")
+
+
 def _unknown_action_error(action: str, args: Dict[str, Any]) -> str:
     error = f"Action {action} not found in AI Scriptless manager tool"
     # A command argument flattened to the top level overwrites the dispatcher action with free text.
@@ -875,6 +880,7 @@ def register(mcp, token: Optional[PerfectoToken]):
             'action', which would collide with the action key of this tool. Always nest them in cmd_arguments,
             e.g. {"action": "add_command", "args": {"test_id": "...", "command_id": "ai_user-action",
             "cmd_arguments": {"action": "Tap on the Login button"}}}.
+            'arguments' is accepted as a backward-compatible alias for cmd_arguments.
         after_path (str, optional): Insert after this step (step_path from view_test_structure).
         parent_path (str, optional): Insert inside a container (step_path of LogicalStep, Loop, or Branch).
 - modify_command: Update command arguments and persist.
@@ -884,6 +890,7 @@ def register(mcp, token: Optional[PerfectoToken]):
         cmd_arguments (dict): Argument names to new values. Merge semantics: only the arguments you send are
             replaced, the rest keep their current value, and arguments cannot be removed (delete_command removes
             the whole step). Same key rules as add_command (declared parameter names, optional data_source form).
+            'arguments' is accepted as a backward-compatible alias for cmd_arguments.
 - delete_command: Remove a command from a test and persist.
     args(dict): Dictionary with the following required parameters:
         test_id (str): Test itemKey from list_tests.
@@ -1040,8 +1047,6 @@ async def ai_scriptless(
             ctx: Context = Field(description="Context object providing access to MCP capabilities")
     ) -> BaseResult:
         action, args = normalize_action_args(arguments)
-        if args is None:
-            args = {}
         ai_scriptless_manager = AiScriptlessManager(token, ctx)
 
         async def _dispatch():
@@ -1069,15 +1074,15 @@ async def _dispatch():
                     return await ai_scriptless_manager.add_command(
                         args.get("test_id", ""),
                         args.get("command_id", ""),
-                        args.get("cmd_arguments"),
+                        _command_arguments(args),
                         args.get("after_path"),
                         args.get("parent_path"),
                     )
                 case "modify_command":
                     return await ai_scriptless_manager.modify_command(
                         args.get("test_id", ""),
                         args.get("step_path", ""),
-                        args.get("cmd_arguments", {}),
+                        _command_arguments(args) or {},
                     )
                 case "delete_command":
                     return await ai_scriptless_manager.delete_command(

tools/device_manager.py

@@ -80,8 +80,6 @@ async def devices(
             ctx: Context = Field(description="Context object providing access to MCP capabilities")
     ) -> BaseResult:
         action, args = normalize_action_args(arguments)
-        if args is None:
-            args = {}
         device_manager = DeviceManager(token, ctx)
 
         async def _dispatch():

tools/execution_manager.py

@@ -259,8 +259,6 @@ async def execution(
             ctx: Context = Field(description="Context object providing access to MCP capabilities")
     ) -> BaseResult:
         action, args = normalize_action_args(arguments)
-        if args is None:
-            args = {}
         execution_manager = ExecutionManager(token, ctx)
 
         async def _dispatch():

tools/help_manager.py

@@ -270,8 +270,6 @@ async def help_main(
             ctx: Context = Field(description="Context object providing access to MCP capabilities")
     ) -> BaseResult:
         action, args = normalize_action_args(arguments)
-        if args is None:
-            args = {}
         help_manager = HelpManager(token, ctx)
 
         async def _dispatch():

tools/tools_manager.py

@@ -339,8 +339,6 @@ async def tools(
             ctx: Context = Field(description="Context object providing access to MCP capabilities")
     ) -> BaseResult:
         action, args = normalize_action_args(arguments)
-        if args is None:
-            args = {}
         tools_manager = ToolsManager(token, ctx)
 
         async def _dispatch():

tools/user_manager.py

@@ -50,8 +50,6 @@ async def user(
             ctx: Context = Field(description="Context object providing access to MCP capabilities")
     ) -> BaseResult:
         action, args = normalize_action_args(arguments)
-        if args is None:
-            args = {}
         user_manager = UserManager(token, ctx)
 
         async def _dispatch():

tools/utils.py

@@ -239,33 +239,3 @@ def normalize_action_args(arguments: Optional[Dict[str, Any]] = None) -> tuple[s
             args[key] = value
     return action, args
 
-
-def validate_required_args(action: str, args: Optional[Dict[str, Any]], required: list[str]) -> Optional[BaseResult]:
-    args = args or {}
-    missing = [key for key in required if key not in args or args[key] is None]
-    if not missing:
-        return None
-    missing_str = ", ".join(missing)
-    required_str = ", ".join(required)
-    return BaseResult(
-        error=(
-            f"Missing required args for action '{action}': {missing_str} not found within 'args'. "
-            f"Required args: {required_str}. Ensure parameters are passed inside the 'args' argument."
-        )
-    )
-
-
-def validate_non_empty_str_arg(
-        action: str, args: Optional[Dict[str, Any]], key: str
-) -> Optional[BaseResult]:
-    """Return BaseResult error if args[key] is missing, not a str, or only whitespace."""
-    args = args or {}
-    value = args.get(key)
-    if not isinstance(value, str) or not value.strip():
-        return BaseResult(
-            error=(
-                f"Missing required args for action '{action}': {key} must be a non-empty string "
-                f"within 'args'. Required args: {key}."
-            )
-        )
-    return None