Skip to content

Commit

Permalink
fix: dump of groups
Browse files Browse the repository at this point in the history
  • Loading branch information
doctrino committed Jun 30, 2024
1 parent 9340541 commit 70b84c0
Showing 1 changed file with 43 additions and 0 deletions.
43 changes: 43 additions & 0 deletions cognite_toolkit/_cdf_tk/loaders/_resource_loaders.py
Original file line number Diff line number Diff line change
Expand Up @@ -363,6 +363,49 @@ def load_resource(
return group_write_list[0]
return group_write_list

def _are_equal(
self, local: GroupWrite, cdf_resource: Group, return_dumped: bool = False
) -> bool | tuple[bool, dict[str, Any], dict[str, Any]]:
local_dumped = local.dump()
cdf_dumped = cdf_resource.as_write().dump()

scope_names = ["datasetScope", "idScope", "extractionPipelineScope"]

ids_by_acl_by_actions_by_scope: dict[str, dict[frozenset[str], dict[str, list[str]]]] = {}
for capability in cdf_dumped.get("capabilities", []):
for acl, values in capability.items():
ids_by_actions_by_scope = ids_by_acl_by_actions_by_scope.setdefault(acl, {})
actions = values.get("actions", [])
ids_by_scope = ids_by_actions_by_scope.setdefault(frozenset(actions), {})
scope = values.get("scope", {})
for scope_name in scope_names:
if ids := scope.get(scope_name, {}).get("ids", []):
if scope_name in ids_by_scope:
# Duplicated
ids_by_scope[scope_name].extend(ids)
else:
ids_by_scope[scope_name] = ids

for capability in local_dumped.get("capabilities", []):
for acl, values in capability.items():
if acl not in ids_by_acl_by_actions_by_scope:
continue
ids_by_actions_by_scope = ids_by_acl_by_actions_by_scope[acl]
actions = frozenset(values.get("actions", []))
if actions not in ids_by_actions_by_scope:
continue
ids_by_scope = ids_by_actions_by_scope[actions]
scope = values.get("scope", {})
for scope_name in scope_names:
if ids := scope.get(scope_name, {}).get("ids", []):
is_dry_run = all(id_ == -1 for id_ in ids)
cdf_ids = ids_by_scope.get(scope_name, [])
are_equal_length = len(ids) == len(cdf_ids)
if is_dry_run and are_equal_length:
values["scope"][scope_name]["ids"] = list(cdf_ids)

return self._return_are_equal(local_dumped, cdf_dumped, return_dumped)

def _upsert(self, items: Sequence[GroupWrite]) -> GroupList:
if len(items) == 0:
return GroupList([])
Expand Down

0 comments on commit 70b84c0

Please sign in to comment.