make capability-compare-functions understand WRITE_PROPERTIES (#1725)

CHANGELOG.md

@@ -17,6 +17,11 @@ Changes are grouped as follows
 - `Fixed` for any bug fixes.
 - `Security` in case of vulnerabilities.
 
+## [7.37.3] - 2024-04-18
+### Improved
+- Minor quality of life change for comparing capabilities involving `DataModelInstancesAcl.WRITE_PROPERTIES`; any
+  ACL already covered by `WRITE` will not be reported as missing.
+
 ## [7.37.2] - 2024-04-18
 ### Fixed
 - Datapoints inserted into non-existent time series, no longer get their identifier hidden in the `failed` attribute

cognite/client/_api/iam.py

@@ -25,10 +25,13 @@
 from cognite.client.data_classes.capabilities import (
     AllScope,
     Capability,
+    CapabilityTuple,
+    DataModelInstancesAcl,
     LegacyCapability,
     ProjectCapability,
     ProjectCapabilityList,
     RawAcl,
+    SpaceIDScope,
     UnknownAcl,
 )
 from cognite.client.data_classes.iam import GroupWrite, SecurityCategoryWrite, SessionStatus, TokenInspection
@@ -50,7 +53,9 @@
 ]
 
 
-def _convert_capability_to_tuples(capabilities: ComparableCapability, project: str | None = None) -> set[tuple]:
+def _convert_capability_to_tuples(
+    capabilities: ComparableCapability, project: str | None = None
+) -> set[CapabilityTuple]:
     if isinstance(capabilities, ProjectCapability):
         return ProjectCapabilityList([capabilities]).as_tuples(project)
     if isinstance(capabilities, ProjectCapabilityList):
@@ -63,7 +68,7 @@ def _convert_capability_to_tuples(capabilities: ComparableCapability, project: s
     elif isinstance(capabilities, GroupList):
         capabilities = [cap for grp in capabilities for cap in grp.capabilities or []]
     if isinstance(capabilities, Sequence):
-        tpls: set[tuple] = set()
+        tpls: set[CapabilityTuple] = set()
         has_skipped = False
         for cap in capabilities:
             if isinstance(cap, dict):
@@ -174,27 +179,41 @@ def compare_capabilities(
         to_check_lookup = {k: set(grp) for k, grp in groupby(sorted(missing), key=itemgetter(slice(2)))}
 
         missing.clear()
-        raw_group, raw_check_grp = set(), set()
-        for key, check_grp in to_check_lookup.items():
+        raw_group, raw_check_group = set(), set()
+        for key, check_group in to_check_lookup.items():
             group = has_capabilties_lookup.get(key, set())
-            if any(AllScope._scope_name == grp[2] for grp in group):
+            if any(AllScope._scope_name == tpl.scope_name for tpl in group):
                 continue  # If allScope exists for capability, we safely skip ahead
-            elif RawAcl._capability_name == next(iter(check_grp))[0]:
+
+            cap_name, _ = key
+            if cap_name == RawAcl._capability_name:
+                # rawAcl needs specialized handling (below):
                 raw_group.update(group)
-                raw_check_grp.update(check_grp)
+                raw_check_group.update(check_group)
+            elif key == (DataModelInstancesAcl._capability_name, DataModelInstancesAcl.Action.Write_Properties.value):
+                # For dataModelInstancesAcl, 'WRITE_PROPERTIES' may covered by 'WRITE', so we must check AllScope:
+                write_grp = has_capabilties_lookup.get((cap_name, DataModelInstancesAcl.Action.Write.value), set())
+                if any(AllScope._scope_name == grp.scope_name for grp in write_grp):
+                    continue
+                # ...and if no AllScope, check individual SpaceIDScope:
+                for check_tpl in check_group:
+                    to_find = (SpaceIDScope._scope_name, check_tpl.scope_id)
+                    if any(to_find == (tpl.scope_name, tpl.scope_id) for tpl in write_grp):
+                        continue
+                    missing.add(check_tpl)
             else:
-                missing.update(check_grp)
+                missing.update(check_group)
 
         # Special handling of rawAcl which has a "hidden" database scope between "all" and "tables":
-        raw_to_check = {k: sorted(grp) for k, grp in groupby(sorted(raw_check_grp), key=itemgetter(slice(4)))}
+        raw_to_check = {k: sorted(grp) for k, grp in groupby(sorted(raw_check_group), key=itemgetter(slice(4)))}
         raw_has_capabs = {k: sorted(grp) for k, grp in groupby(sorted(raw_group), key=itemgetter(slice(4)))}
         for key, check_db_grp in raw_to_check.items():
-            if (db_group := raw_has_capabs.get(key)) and not db_group[0][-1]:
-                # [0] because empty string sorts first; [-1] is table; if no table -> db scope -> skip ahead
+            if (db_group := raw_has_capabs.get(key)) and not db_group[0].table:
+                # [0] because empty string sorts first; if no table -> db scope -> skip ahead
                 continue
             missing.update(check_db_grp)
 
-        return [Capability.from_tuple(tpl) for tpl in missing]
+        return [Capability.from_tuple(tpl) for tpl in sorted(missing)]
 
     def verify_capabilities(
         self,

cognite/client/_version.py

@@ -1,4 +1,4 @@
 from __future__ import annotations
 
-__version__ = "7.37.2"
+__version__ = "7.37.3"
 __api_subversion__ = "20230101"

cognite/client/data_classes/capabilities.py

@@ -9,7 +9,7 @@
 from dataclasses import asdict, dataclass, field
 from itertools import product
 from types import MappingProxyType
-from typing import TYPE_CHECKING, Any, ClassVar, Iterable, Literal, Sequence, cast
+from typing import TYPE_CHECKING, Any, ClassVar, Iterable, Literal, NamedTuple, NoReturn, Sequence, cast
 
 from typing_extensions import Self
 
@@ -28,6 +28,24 @@
 logger = logging.getLogger(__name__)
 
 
+class CapabilityTuple(NamedTuple):
+    acl_name: str
+    action: str
+    scope_name: str
+    scope_id: int | str | None = None
+    scope_extra: str | None = None
+
+    @property
+    def database(self) -> str:
+        assert self.acl_name == RawAcl._capability_name and isinstance(self.scope_id, str)
+        return self.scope_id
+
+    @property
+    def table(self) -> str:
+        assert self.acl_name == RawAcl._capability_name and isinstance(self.scope_extra, str)
+        return self.scope_extra
+
+
 @dataclass
 class Capability(ABC):
     _capability_name: ClassVar[str]
@@ -122,27 +140,27 @@ def dump(self, camel_case: bool = True) -> dict[str, Any]:
                 data = convert_all_keys_to_camel_case(data)
             return {self._scope_name: data}
 
-        def as_tuples(self) -> set[tuple]:
+        def as_tuples(
+            self,
+        ) -> set[tuple[str]] | set[tuple[str, str]] | set[tuple[str, str, str]] | set[tuple[str, int]]:
             # Basic implementation for all simple Scopes (e.g. all or currentuser)
             return {(self._scope_name,)}
 
     @classmethod
-    def from_tuple(cls, tpl: tuple) -> Self:
-        acl_name, action, scope_name, *scope_params = tpl
-        capability_cls = _CAPABILITY_CLASS_BY_NAME[acl_name]
-        scope_cls = _SCOPE_CLASS_BY_NAME[scope_name]
+    def from_tuple(cls, tpl: CapabilityTuple) -> Self:
+        capability_cls = _CAPABILITY_CLASS_BY_NAME[tpl.acl_name]
+        scope_cls = _SCOPE_CLASS_BY_NAME[tpl.scope_name]
 
-        if not scope_params:
+        if tpl.scope_id is None:
             scope = scope_cls()
-        elif len(scope_params) == 1:
-            scope = scope_cls(scope_params)  # type: ignore [call-arg]
-        elif len(scope_params) == 2 and scope_cls is TableScope:
-            db, tbl = scope_params
-            scope = scope_cls({db: [tbl] if tbl else []})  # type: ignore [call-arg]
+        elif tpl.scope_extra is None:
+            scope = scope_cls([tpl.scope_id])  # type: ignore [call-arg]
+        elif scope_cls is TableScope:
+            scope = scope_cls({tpl.database: [tpl.table] if tpl.table else []})  # type: ignore [call-arg]
         else:
-            raise ValueError(f"tuple not understood as capability: {tpl}")
+            raise ValueError(f"CapabilityTuple not understood: {tpl}")
 
-        return cast(Self, capability_cls(actions=[capability_cls.Action(action)], scope=scope))  # type: ignore [call-arg]
+        return cast(Self, capability_cls(actions=[capability_cls.Action(tpl.action)], scope=scope, allow_unknown=False))
 
     @classmethod
     def load(cls, resource: dict | str, allow_unknown: bool = False) -> Self:
@@ -189,9 +207,9 @@ def dump(self, camel_case: bool = True) -> dict[str, Any]:
         capability_name = self._capability_name
         return {to_camel_case(capability_name) if camel_case else to_snake_case(capability_name): data}
 
-    def as_tuples(self) -> set[tuple]:
+    def as_tuples(self) -> set[CapabilityTuple]:
         return set(
-            (acl, action, *scope_tpl)
+            CapabilityTuple(acl, action, *scope_tpl)  # type: ignore [arg-type]
             for acl, action, scope_tpl in product(
                 [self._capability_name], [action.value for action in self.actions], self.scope.as_tuples()
             )
@@ -284,10 +302,10 @@ def _infer_project(self, project: str | None = None) -> str:
             return self._cognite_client.config.project
         return project
 
-    def as_tuples(self, project: str | None = None) -> set[tuple]:
+    def as_tuples(self, project: str | None = None) -> set[CapabilityTuple]:
         project = self._infer_project(project)
 
-        output: set[tuple] = set()
+        output: set[CapabilityTuple] = set()
         for proj_cap in self:
             cap = proj_cap.capability
             if isinstance(cap, UnknownAcl):
@@ -323,7 +341,7 @@ class IDScope(Capability.Scope):
     def __post_init__(self) -> None:
         object.__setattr__(self, "ids", [int(i) for i in self.ids])
 
-    def as_tuples(self) -> set[tuple]:
+    def as_tuples(self) -> set[tuple[str, int]]:
         return {(self._scope_name, i) for i in self.ids}
 
 
@@ -337,7 +355,7 @@ class IDScopeLowerCase(Capability.Scope):
     def __post_init__(self) -> None:
         object.__setattr__(self, "ids", [int(i) for i in self.ids])
 
-    def as_tuples(self) -> set[tuple]:
+    def as_tuples(self) -> set[tuple[str, int]]:
         return {(self._scope_name, i) for i in self.ids}
 
 
@@ -349,7 +367,7 @@ class ExtractionPipelineScope(Capability.Scope):
     def __post_init__(self) -> None:
         object.__setattr__(self, "ids", [int(i) for i in self.ids])
 
-    def as_tuples(self) -> set[tuple]:
+    def as_tuples(self) -> set[tuple[str, int]]:
         return {(self._scope_name, i) for i in self.ids}
 
 
@@ -361,7 +379,7 @@ class DataSetScope(Capability.Scope):
     def __post_init__(self) -> None:
         object.__setattr__(self, "ids", [int(i) for i in self.ids])
 
-    def as_tuples(self) -> set[tuple]:
+    def as_tuples(self) -> set[tuple[str, int]]:
         return {(self._scope_name, i) for i in self.ids}
 
 
@@ -385,7 +403,7 @@ def dump(self, camel_case: bool = True) -> dict[str, Any]:
         key = "dbsToTables" if camel_case else "dbs_to_tables"
         return {self._scope_name: {key: {k: {"tables": v} for k, v in self.dbs_to_tables.items()}}}
 
-    def as_tuples(self) -> set[tuple]:
+    def as_tuples(self) -> set[tuple[str, str, str]]:
         # When the scope contains no tables, it means all tables... since database name must be at least 1
         # character, we represent this internally with the empty string:
         return {(self._scope_name, db, tbl) for db, tables in self.dbs_to_tables.items() for tbl in tables or [""]}
@@ -399,7 +417,7 @@ class AssetRootIDScope(Capability.Scope):
     def __post_init__(self) -> None:
         object.__setattr__(self, "root_ids", [int(i) for i in self.root_ids])
 
-    def as_tuples(self) -> set[tuple]:
+    def as_tuples(self) -> set[tuple[str, int]]:
         return {(self._scope_name, i) for i in self.root_ids}
 
 
@@ -408,7 +426,7 @@ class ExperimentsScope(Capability.Scope):
     _scope_name = "experimentscope"
     experiments: list[str]
 
-    def as_tuples(self) -> set[tuple]:
+    def as_tuples(self) -> set[tuple[str, str]]:
         return {(self._scope_name, s) for s in self.experiments}
 
 
@@ -417,7 +435,7 @@ class SpaceIDScope(Capability.Scope):
     _scope_name = "spaceIdScope"
     space_ids: list[str]
 
-    def as_tuples(self) -> set[tuple]:
+    def as_tuples(self) -> set[tuple[str, str]]:
         return {(self._scope_name, s) for s in self.space_ids}
 
 
@@ -429,7 +447,7 @@ class PartitionScope(Capability.Scope):
     def __post_init__(self) -> None:
         object.__setattr__(self, "partition_ids", [int(i) for i in self.partition_ids])
 
-    def as_tuples(self) -> set[tuple]:
+    def as_tuples(self) -> set[tuple[str, int]]:
         return {(self._scope_name, i) for i in self.partition_ids}
 
 
@@ -438,7 +456,7 @@ class LegacySpaceScope(Capability.Scope):
     _scope_name = "spaceScope"
     external_ids: list[str]
 
-    def as_tuples(self) -> set[tuple]:
+    def as_tuples(self) -> set[tuple[str, str]]:
         return {(self._scope_name, s) for s in self.external_ids}
 
 
@@ -447,7 +465,7 @@ class LegacyDataModelScope(Capability.Scope):
     _scope_name = "dataModelScope"
     external_ids: list[str]
 
-    def as_tuples(self) -> set[tuple]:
+    def as_tuples(self) -> set[tuple[str, str]]:
         return {(self._scope_name, s) for s in self.external_ids}
 
 
@@ -464,7 +482,7 @@ class UnknownScope(Capability.Scope):
     def __getitem__(self, item: str) -> Any:
         return self.data[item]
 
-    def as_tuples(self) -> set[tuple]:
+    def as_tuples(self) -> NoReturn:
         raise NotImplementedError("Unknown scope cannot be converted to tuples (needed for comparisons)")
 
 

pyproject.toml

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "cognite-sdk"
 
-version = "7.37.2"
+version = "7.37.3"
 description = "Cognite Python SDK"
 readme = "README.md"
 documentation = "https://cognite-sdk-python.readthedocs-hosted.com"

tests/tests_unit/test_data_classes/test_capabilities.py

@@ -12,6 +12,7 @@
     AllScope,
     Capability,
     CurrentUserScope,
+    DataModelInstancesAcl,
     DataSetsAcl,
     EventsAcl,
     ExperimentsAcl,
@@ -20,6 +21,7 @@
     ProjectCapabilityList,
     ProjectsAcl,
     RawAcl,
+    SpaceIDScope,
     TableScope,
     UnknownAcl,
     UnknownScope,
@@ -247,6 +249,39 @@ def test_load__action_does_not_exist(self, acl_cls_name: str, bad_action: str, d
 
         assert Capability.load(dumped, allow_unknown=True)
 
+    @pytest.mark.parametrize("has_write_allscope", (True, False))
+    @pytest.mark.parametrize("has_write_props_allscope", (True, False))
+    @pytest.mark.parametrize("has_write_in_same_scope", (True, False))
+    def test_load_data_model_instances__with_write_properties(
+        self, cognite_client, has_write_allscope, has_write_props_allscope, has_write_in_same_scope
+    ):
+        # WRITE_PROPERTIES grants a subset of capabilities of WRITE, so we ensure that having just WRITE
+        # won't cause WRITE_PROPERTIES to be reported as missing.
+        existing = [
+            DataModelInstancesAcl([DataModelInstancesAcl.Action.Write], SpaceIDScope(["foo", "this"])),
+            DataModelInstancesAcl([DataModelInstancesAcl.Action.Write_Properties], SpaceIDScope(["bar"])),
+        ]
+        if has_write_allscope:
+            existing.append(DataModelInstancesAcl([DataModelInstancesAcl.Action.Write], AllScope()))
+        if has_write_in_same_scope:
+            existing.append(DataModelInstancesAcl([DataModelInstancesAcl.Action.Write], SpaceIDScope(["too_much"])))
+        if has_write_props_allscope:
+            existing.append(DataModelInstancesAcl([DataModelInstancesAcl.Action.Write_Properties], AllScope()))
+
+        desired = [
+            DataModelInstancesAcl([DataModelInstancesAcl.Action.Write_Properties], SpaceIDScope(["foo", "too_much"])),
+            DataModelInstancesAcl([DataModelInstancesAcl.Action.Write_Properties], SpaceIDScope(["bar"])),
+        ]
+        if has_write_allscope or has_write_props_allscope or has_write_in_same_scope:
+            expected_missing = []
+        else:
+            expected_missing = [
+                DataModelInstancesAcl([DataModelInstancesAcl.Action.Write_Properties], SpaceIDScope(["too_much"]))
+            ]
+
+        missing = cognite_client.iam.compare_capabilities(existing_capabilities=existing, desired_capabilities=desired)
+        assert missing == expected_missing
+
     def test_create_capability_forget_initializing_scope(self):
         # Ensure these do not raise. All other scopes require arguments and so will
         # raise appropriate errors if not initialized.