Make assign_asset report only an assignment that holds

CHANGELOG.md

@@ -25,6 +25,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   its text, and an unknown id is the error `entity_not_found` instead of a
   traceback. Script refusals also keep the gem's own code (for example
   `secure_mode` when editor Python is disabled) instead of a generic `editor_error`.
+- **`assign_asset` reports only an assignment that holds.** It printed "Assigned"
+  without checking anything, then fell back to a misaddressed call that also
+  "succeeded". It also treated an unknown asset path as found (the catalog answers
+  an invalid id, not `None`) and passed the id as text. It now resolves the
+  asset with the reflected three-argument call and checks it is valid, sets the
+  property with the `AssetId` and checks the outcome, and reads the property back.
+  Failures are `asset_not_found`, `component_type_not_found`,
+  `component_not_on_entity`, `set_property_failed` or `assign_asset_failed`.
 - **Editor failures use the error envelope too.** The failure envelope from the error
   contract below was returned by the server and by some editor tools, but other editor
   tools still printed plain-text failures (add/remove component, assign_asset, set_parent,

src/o3de_mcp/tools/editor.py

@@ -1559,7 +1559,14 @@ async def set_component_property(
     async def assign_asset(
         entity_id: EntityIdArg, component_type: str, property_path: str, asset_path: str
     ) -> str:
-        """Assign an asset to a component property by resolving the asset path."""
+        """Assign an asset to a component property by resolving the asset path.
+
+        ``asset_path`` is the product path in the asset catalog (for example
+        ``objects/thing.fbx.azmodel``). The property is read back after setting,
+        and only an assignment that holds is reported as done; an unknown asset,
+        component or property, or a value that does not stick, is returned as an
+        error.
+        """
         entity_id = _validate_entity_id(entity_id)
         component_type = _validate_component_type(component_type)
         asset_path = asset_path.strip()
@@ -1579,6 +1586,8 @@ async def assign_asset(
             import azlmbr.editor as editor
             import azlmbr.bus as bus
             import azlmbr.entity as entity
+            import azlmbr.asset as asset
+            import azlmbr.math as math
             import json
 
             _params = json.loads({params!r})
@@ -1587,57 +1596,58 @@ async def assign_asset(
             prop_path = _params['property_path']
             asset_path = _params['asset_path']
 
-            # Resolve the asset ID from the project-relative path
-            try:
-                import azlmbr.asset as asset
-                asset_id = asset.AssetCatalogRequestBus(
-                    bus.Broadcast, 'GetAssetIdByPath',
-                    asset_path, azlmbr.math.Uuid(), False
-                )
-            except Exception:
-                try:
-                    asset_id = asset.AssetCatalogRequestBus(
-                        bus.Broadcast, 'GetAssetIdByPath',
-                        asset_path
-                    )
-                except Exception as e:
-                    _why = f'Failed to resolve asset path {{asset_path}}: {{e}}'
-                    asset_id = None
-
-            if asset_id is None:
+            # The catalog answers an invalid AssetId, not None, for an unknown path.
+            asset_id = asset.AssetCatalogRequestBus(
+                bus.Broadcast, 'GetAssetIdByPath', asset_path, math.Uuid(), False
+            )
+            type_ids = editor.EditorComponentAPIBus(
+                bus.Broadcast, 'FindComponentTypeIdsByEntityType',
+                [comp_type], entity.EntityType().Game
+            )
+            _null = '00000000-0000-0000-0000-000000000000'
+            if asset_id is None or not asset_id.is_valid():
                 _o3de_fail(
                     'asset_not_found',
-                    locals().get('_why') or f'Asset not found: {{asset_path}}')
+                    f'No asset at {{asset_path}} in the asset catalog; pass the product '
+                    f'path (for example objects/thing.fbx.azmodel)',
+                )
+            elif not type_ids or _null in str(type_ids[0]):
+                _o3de_fail('component_type_not_found', f'Component type "{{comp_type}}" not found')
             else:
-                asset_ref = str(asset_id)
-                type_ids = editor.EditorComponentAPIBus(
-                    bus.Broadcast, 'FindComponentTypeIdsByEntityType',
-                    [comp_type], entity.EntityType().Game
+                found = editor.EditorComponentAPIBus(
+                    bus.Broadcast, 'GetComponentOfType', eid, type_ids[0]
                 )
-                success = False
-                try:
-                    if type_ids:
-                        outcome = editor.EditorComponentAPIBus(
-                            bus.Broadcast, 'GetComponentOfType', eid, type_ids[0]
-                        )
-                        if hasattr(outcome, 'IsSuccess') and outcome.IsSuccess():
-                            pair = outcome.GetValue()
-                            result = editor.EditorComponentAPIBus(
-                                bus.Broadcast, 'SetComponentProperty', pair,
-                                prop_path, asset_ref
-                            )
-                            print(f'Assigned asset {{asset_path}} to '
-                                  f'{{prop_path}} (result={{result}})')
-                            success = True
-                except Exception:
-                    pass
-
-                if not success:
+                if not hasattr(found, 'IsSuccess') or not found.IsSuccess():
+                    _o3de_fail(
+                        'component_not_on_entity',
+                        f'Component "{{comp_type}}" is not on entity {{eid}}',
+                    )
+                else:
+                    pair = found.GetValue()
                     result = editor.EditorComponentAPIBus(
-                        bus.Event, 'SetComponentProperty', eid,
-                        prop_path, asset_ref
+                        bus.Broadcast, 'SetComponentProperty', pair, prop_path, asset_id
                     )
-                    print(f'Assigned asset {{asset_path}} to {{prop_path}} (result={{result}})')
+                    if not hasattr(result, 'IsSuccess') or not result.IsSuccess():
+                        _why = result.GetError() if hasattr(result, 'GetError') else result
+                        _o3de_fail(
+                            'set_property_failed',
+                            f'Could not set {{prop_path}} on {{comp_type}}: {{_why}}',
+                        )
+                    else:
+                        # Read it back: report only an assignment that actually holds.
+                        back = editor.EditorComponentAPIBus(
+                            bus.Broadcast, 'GetComponentProperty', pair, prop_path
+                        )
+                        got = None
+                        if hasattr(back, 'IsSuccess') and back.IsSuccess():
+                            got = back.GetValue()
+                        if got is None or str(got) != str(asset_id):
+                            _o3de_fail(
+                                'assign_asset_failed',
+                                f'{{prop_path}} reads back {{got}}, not {{asset_id}}',
+                            )
+                        else:
+                            print(f'Assigned asset {{asset_path}} to {{prop_path}}')
         """)
         return await _async_run_editor_script(script)
 

tests/test_editor_scripts.py

@@ -704,3 +704,88 @@ def test_an_unknown_entity_id_is_entity_not_found(surface: dict, tmp_path: Path)
     parsed = json.loads(out)
     assert parsed["status"] == "error" and parsed["code"] == "entity_not_found"
     assert "123" in parsed["message"]
+
+
+class _Asset(Anything):
+    """An AssetId stub: valid or not, printing as the given text."""
+
+    def __init__(self, valid: bool, text: str = "{ASSET}:0") -> None:
+        super().__init__()
+        object.__setattr__(self, "_valid", valid)
+        object.__setattr__(self, "_text", text)
+
+    def is_valid(self) -> bool:
+        return object.__getattribute__(self, "_valid")
+
+    def __str__(self) -> str:
+        return object.__getattribute__(self, "_text")
+
+
+class _Ok(Anything):
+    def __init__(self, value: object = None) -> None:
+        super().__init__()
+        object.__setattr__(self, "_value", value)
+
+    def IsSuccess(self) -> bool:  # noqa: N802
+        return True
+
+    def GetValue(self) -> object:  # noqa: N802
+        return object.__getattribute__(self, "_value")
+
+
+class TestAssignAsset:
+    """assign_asset reports only an assignment that reads back."""
+
+    def _run(self, surface: dict, tmp_path: Path, overrides: dict) -> dict | str:
+        out = _run_tool("assign_asset", surface, tmp_path, overrides)
+        try:
+            return json.loads(out)
+        except json.JSONDecodeError:
+            return out
+
+    def test_an_unknown_asset_is_asset_not_found(self, surface: dict, tmp_path: Path) -> None:
+        parsed = self._run(
+            surface,
+            tmp_path,
+            {("AssetCatalogRequestBus", "GetAssetIdByPath"): lambda *a: _Asset(False)},
+        )
+        assert parsed["code"] == "asset_not_found"
+
+    def test_a_refused_set_is_reported(self, surface: dict, tmp_path: Path) -> None:
+        parsed = self._run(
+            surface,
+            tmp_path,
+            {
+                ("AssetCatalogRequestBus", "GetAssetIdByPath"): lambda *a: _Asset(True),
+                ("EditorComponentAPIBus", "SetComponentProperty"): Failure(),
+            },
+        )
+        assert parsed["code"] == "set_property_failed"
+
+    def test_a_value_that_does_not_stick_is_reported(self, surface: dict, tmp_path: Path) -> None:
+        parsed = self._run(
+            surface,
+            tmp_path,
+            {
+                ("AssetCatalogRequestBus", "GetAssetIdByPath"): lambda *a: _Asset(True, "{A}:0"),
+                ("EditorComponentAPIBus", "SetComponentProperty"): lambda *a: _Ok(),
+                ("EditorComponentAPIBus", "GetComponentProperty"): lambda *a: _Ok(
+                    _Asset(True, "{B}:0")
+                ),
+            },
+        )
+        assert parsed["code"] == "assign_asset_failed"
+
+    def test_an_assignment_that_reads_back_succeeds(self, surface: dict, tmp_path: Path) -> None:
+        out = self._run(
+            surface,
+            tmp_path,
+            {
+                ("AssetCatalogRequestBus", "GetAssetIdByPath"): lambda *a: _Asset(True, "{A}:0"),
+                ("EditorComponentAPIBus", "SetComponentProperty"): lambda *a: _Ok(),
+                ("EditorComponentAPIBus", "GetComponentProperty"): lambda *a: _Ok(
+                    _Asset(True, "{A}:0")
+                ),
+            },
+        )
+        assert isinstance(out, str) and out.startswith("Assigned asset"), out

tests/test_live_editor.py

@@ -341,6 +341,42 @@ def test_python_path_keeps_the_scale_when_none_is_given(self, mcp_server: MCPSer
             _run(_call(mcp_server, "delete_entity", entity_id=eid))
 
 
+class TestLiveAssignAsset:
+    def test_assigns_a_real_mesh_and_refuses_an_unknown_one(self, mcp_server: MCPServer) -> None:
+        created = json.loads(_run(_call(mcp_server, "create_entity", name="AssignTest")))
+        eid = str(created["entity_id"])
+        try:
+            _run(_call(mcp_server, "add_component", entity_id=eid, component_type="Mesh"))
+            prop = "Controller|Configuration|Model Asset"
+            ok = _run(
+                _call(
+                    mcp_server,
+                    "assign_asset",
+                    entity_id=eid,
+                    component_type="Mesh",
+                    property_path=prop,
+                    asset_path="objects/shaderball/ground_plane_4x4m.fbx.azmodel",
+                )
+            )
+            # Only an assignment that reads back is reported as done.
+            assert ok.startswith("Assigned asset"), ok
+            missing = json.loads(
+                _run(
+                    _call(
+                        mcp_server,
+                        "assign_asset",
+                        entity_id=eid,
+                        component_type="Mesh",
+                        property_path=prop,
+                        asset_path="objects/does_not_exist.fbx.azmodel",
+                    )
+                )
+            )
+            assert missing["status"] == "error" and missing["code"] == "asset_not_found", missing
+        finally:
+            _run(_call(mcp_server, "delete_entity", entity_id=eid))
+
+
 class TestLiveConsole:
     def test_run_console_command(self, mcp_server: MCPServer) -> None:
         result = _run(_call(mcp_server, "run_console_command", command="r_DisplayInfo 0"))