Improve Capability.load (#1769)

cognite/client/data_classes/capabilities.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import difflib
 import enum
 import inspect
 import itertools
@@ -75,7 +76,7 @@ def _validate(self) -> None:
         if bad_actions := [a for a in self.actions if a not in capability_cls.Action]:
             full_action_examples = ", ".join(f"{acl}.Action.{action.name}" for action in capability_cls.Action)
             raise ValueError(
-                f"{acl} got unknown action(s): {bad_actions}, expected a subset of: [{full_action_examples}]."
+                f"{acl} got unknown action(s): {bad_actions}, expected a subset of: [{full_action_examples}]"
             )
         allowed_scopes, scope_names = _VALID_SCOPES_BY_CAPABILITY[capability_cls]
         if type(self.scope) in allowed_scopes or not allowed_scopes:
@@ -87,7 +88,7 @@ def _validate(self) -> None:
         else:
             full_scope_examples = ", ".join(f"{acl}.Scope.{name}" for name in scope_names)
             raise ValueError(
-                f"{acl} got an unknown scope: {self.scope}, expected an instance of one of: [{full_scope_examples}]."
+                f"{acl} got an unknown scope: {self.scope}, expected an instance of one of: [{full_scope_examples}]"
             )
 
     @classmethod
@@ -116,12 +117,13 @@ class Scope(ABC):
         _scope_name: ClassVar[str]
 
         @classmethod
-        def load(cls, resource: dict | str) -> Self:
+        def load(cls, resource: dict | str, allow_unknown: bool = False) -> Self:
             resource = resource if isinstance(resource, dict) else load_yaml_or_json(resource)
-            for cls_name, scope_cls in _SCOPE_CLASS_BY_NAME.items():
-                if cls_name not in resource:
-                    continue
-                data = convert_all_keys_to_snake_case(resource[cls_name])
+            known_scopes = set(resource).intersection(_SCOPE_CLASS_BY_NAME)
+            if len(known_scopes) == 1:
+                (name,) = known_scopes
+                scope_cls = _SCOPE_CLASS_BY_NAME[name]
+                data = convert_all_keys_to_snake_case(resource[name])
                 try:
                     return cast(Self, scope_cls(**data))
                 except TypeError:
@@ -133,9 +135,19 @@ def load(cls, resource: dict | str) -> Self:
                     return cast(Self, scope_cls(**rename_and_exclude_keys(data, exclude=not_supported)))
 
             # We infer this as an unknown scope not yet added to the SDK:
-            name, data = next(iter(resource.items()))
-            data = convert_all_keys_to_snake_case(data)
-            return cast(Self, UnknownScope(name=name, data=data))
+            if len(resource) == 1:
+                ((name, data),) = resource.items()
+                if allow_unknown:
+                    return cast(Self, UnknownScope(name=name, data=data))
+
+                raise ValueError(
+                    f"Unable to parse Scope, {name!r} is not a known scope. Pass `allow_unknown=True` to force "
+                    f"loading it as an unknown scope. List of known: {sorted(_SCOPE_CLASS_BY_NAME)}."
+                )
+            raise ValueError(
+                f"Unable to parse Scope, none of the top-level keys in the input, {sorted(resource)}, "
+                f"matched known Scopes, - or - multiple was found. List of known: {sorted(_SCOPE_CLASS_BY_NAME)}."
+            )
 
         def dump(self, camel_case: bool = True) -> dict[str, Any]:
             if isinstance(self, UnknownScope):
@@ -176,32 +188,45 @@ def load(cls, resource: dict | str, allow_unknown: bool = False) -> Self:
             (name,) = known_acls
             capability_cls = _CAPABILITY_CLASS_BY_NAME[name]
             # TODO: We ignore extra keys that might be present like 'projectUrlNames' - are these needed?
+            try:
+                scopes = Capability.Scope.load(resource[name]["scope"], allow_unknown)
+            except Exception as err:
+                raise ValueError(f"Could not instantiate {capability_cls.__name__} due to: {err}")
+
             return cast(
                 Self,
                 capability_cls(
                     actions=[capability_cls.Action.load(act, allow_unknown) for act in resource[name]["actions"]],
-                    scope=Capability.Scope.load(resource[name]["scope"]),
+                    scope=scopes,
                     allow_unknown=allow_unknown,
                 ),
             )
-        elif not known_acls and len(resource) == 1:
-            # We infer this as an unknown acl not yet added to the SDK:
-            ((name, data),) = resource.items()
-            return cast(
-                Self,
-                UnknownAcl(
+        # We infer this as an unknown acl not yet added to the SDK. All API-responses are neccessarily
+        # loaded with 'allow_unknown=True' (to future-proof):
+        if allow_unknown:
+            if len(resource) != 1:
+                unknown_acl = UnknownAcl(actions=None, scope=None, raw_data=resource, allow_unknown=True)  # type: ignore [arg-type]
+            else:
+                ((name, data),) = resource.items()
+                unknown_acl = UnknownAcl(
                     capability_name=name,
-                    actions=[UnknownAcl.Action.load(act, allow_unknown) for act in data["actions"]],
-                    scope=Capability.Scope.load(data["scope"]),
+                    actions=[UnknownAcl.Action.load(act, allow_unknown=True) for act in data["actions"]],
+                    scope=Capability.Scope.load(data["scope"], allow_unknown=True),
                     raw_data=resource,
-                    allow_unknown=allow_unknown,
-                ),
-            )
-        # We got more than one acl (highly unlikely) or we got an unknown acl + extra keys in the response:
-        raise ValueError(
-            "Unable to parse capability from API-response, please create an issue on Github: "
-            "https://github.com/cognitedata/cognite-sdk-python"
+                    allow_unknown=True,
+                )
+            return cast(Self, unknown_acl)
+
+        top_lvl_keys = sorted(resource)
+        err_msg = (
+            f"Unable to parse Capability, none of the top-level keys in the input, {top_lvl_keys}, "
+            f"matched known ACLs, - or - multiple was found. Pass `allow_unknown=True` to force loading it "
+            f"as an unknown capability."
         )
+        if matches := [match for key in top_lvl_keys for match in difflib.get_close_matches(key, ALL_CAPABILITIES)]:
+            err_msg += f" Did you mean one of: {matches}?"
+        err_msg += " List of all ACLs: from cognite.client.data_classes.capabilities import ALL_CAPABILITIES"
+        raise ValueError(err_msg)
 
     def dump(self, camel_case: bool = True) -> dict[str, Any]:
         if isinstance(self, UnknownAcl):
@@ -1310,6 +1335,8 @@ class Scope:
         if c not in (UnknownAcl, LegacyCapability)
     }
 )
+ALL_CAPABILITIES = sorted(_CAPABILITY_CLASS_BY_NAME)
+
 # Give all Actions a better error message (instead of implementing __missing__ for all):
 for acl in _CAPABILITY_CLASS_BY_NAME.values():
     if acl.Action.__members__:

tests/tests_unit/test_data_classes/test_capabilities.py

@@ -209,9 +209,17 @@ def test_load_dump__extra_top_level_keys(self, cap_with_extra_key) -> None:
         assert capability.dump(camel_case=True) == cap_with_extra_key
 
     def test_load_dump_unknown__extra_top_level_keys(self, unknown_cap_with_extra_key) -> None:
-        with pytest.raises(ValueError, match="^Unable to parse capability from API-response"):
+        match = (
+            "^Unable to parse Capability, none of the top-level keys in the input, "
+            r"\['extra-key', 'veryUnknownAcl'\], matched known ACLs"
+        )
+        with pytest.raises(ValueError, match=match):
             Capability.load(unknown_cap_with_extra_key)
 
+        # API responses should never fail to load, even when crazy:
+        loaded = Capability.load(unknown_cap_with_extra_key, allow_unknown=True)
+        assert loaded.dump() == unknown_cap_with_extra_key
+
     @pytest.mark.parametrize(
         "raw", [{"dataproductAcl": {"actions": ["UTILIZE"], "scope": {"components": {"ids": [1, 2, 3]}}}}]
     )
@@ -223,6 +231,26 @@ def test_load_dump_unknown(self, raw: dict[str, Any]) -> None:
         assert capability.raw_data == raw
         assert capability.dump(camel_case=True) == raw
 
+    def test_load_capability_misspelled_acl(self, unknown_acls_items):
+        unknown_acl, *_ = unknown_acls_items
+        exp_err_msg = re.escape(
+            "Unable to parse Capability, none of the top-level keys in the input, ['funkyAssetsAcl'], "
+            "matched known ACLs, - or - multiple was found. Pass `allow_unknown=True` to force loading "
+            "it as an unknown capability. Did you mean one of: ['assetsAcl', 'functionsAcl']? List of all "
+            "ACLs: from cognite.client.data_classes.capabilities import ALL_CAPABILITIES"
+        )
+        # difflib should give some nice suggestions for misspelling:
+        # - funkyAssetsAcl -> [assetsAcl, functionsAcl]
+        with pytest.raises(ValueError, match=f"^{exp_err_msg}$"):
+            Capability.load(unknown_acl, allow_unknown=False)
+
+        # when difflib doesnt find any matches, it should be omitted from the err. msg:
+        with pytest.raises(ValueError, match="force loading it as an unknown capability. List of all ACLs"):
+            Capability.load(
+                {"does not match anything really": {"actions": ["READ"], "scope": {"all": {}}}},
+                allow_unknown=False,
+            )
+
     @pytest.mark.parametrize(
         "acl_cls_name, bad_action, dumped",
         [
@@ -397,13 +425,23 @@ def test_groups_list(self, cognite_client, mock_groups_resp, unknown_acls_items)
         assert expected == [g["capabilities"] for g in groups.dump()]
 
         # Ensure that the capabilities that did -not- raise from groups/list, would raise for a normal user:
-        with pytest.raises(ValueError, match=r"^'READ' is not a valid Capability.Action"):
+        acl_err_match = r"top-level keys in the input, \['funkyAssetsAcl'\], matched known ACLs"
+        action_err_match = "^'UN-KN-OWN' is not a valid AssetsAcl.Action$"
+        scope_err_match = "^Could not instantiate AssetsAcl due to: Unable to parse Scope, 'astronautSpace' is not"
+        with pytest.raises(ValueError, match=acl_err_match):
             GroupList.load(groups.dump(camel_case=True))
 
         # ...and ensure each individual (acl/action/scope) raises:
-        for unknown_acl in unknown_acls_items:
-            with pytest.raises(ValueError, match="is not a valid |^Could not instantiate "):
-                Group.load({"name": "me", "id": 123, "source_id": "huh", "capabilities": [unknown_acl]})
+        u1, u2, u3, u4 = unknown_acls_items
+        group = {"name": "me", "id": 123, "source_id": "huh"}
+        with pytest.raises(ValueError, match=acl_err_match):
+            Group.load({**group, "capabilities": [u1]})  # Unknown capability
+        with pytest.raises(ValueError, match=action_err_match):
+            Group.load({**group, "capabilities": [u2]})  # Unknown action
+        with pytest.raises(ValueError, match=scope_err_match):
+            Group.load({**group, "capabilities": [u3]})  # Unknown scope
+        with pytest.raises(ValueError, match=acl_err_match):
+            Group.load({**group, "capabilities": [u4]})  # Unknown -everything-
 
     def test_token_inspect(self, cognite_client, mock_token_inspect_resp):
         # Mostly a repeat of test_groups_list, ensuring token/inspect won't ever raise