(scim_url, token,
keeper_groups,
external_groups,
dry_run=False,
**kwargs)
| 483 | |
| 484 | @staticmethod |
| 485 | def sync_groups(scim_url, token, |
| 486 | keeper_groups, |
| 487 | external_groups, |
| 488 | dry_run=False, |
| 489 | **kwargs): # type: (str, str, Dict[str, ScimGroup], Dict[str, ScimGroup], bool, Any) -> None |
| 490 | keeper_group_copy = keeper_groups.copy() |
| 491 | external_group_copy = external_groups.copy() |
| 492 | for match_round in range(3): # 0 - external ID, 1 - name, 2 - reuse groups |
| 493 | if len(keeper_group_copy) == 0 or len(external_group_copy) == 0: |
| 494 | break |
| 495 | if match_round == 0: |
| 496 | group_lookup = {x.external_id: x for x in keeper_group_copy.values() if x.external_id} |
| 497 | elif match_round == 1: |
| 498 | group_lookup = {x.name.casefold(): x for x in keeper_group_copy.values()} |
| 499 | elif match_round == 2: |
| 500 | group_lookup = {key: value for key, value in |
| 501 | zip(external_group_copy.keys(), keeper_group_copy.values()) if value.external_id} |
| 502 | else: |
| 503 | continue |
| 504 | |
| 505 | for group_id in list(external_group_copy.keys()): |
| 506 | group = external_group_copy[group_id] |
| 507 | if match_round in (0, 2): |
| 508 | key = group.id |
| 509 | elif match_round == 1: |
| 510 | key = group.name.casefold() |
| 511 | else: |
| 512 | continue |
| 513 | if key in group_lookup: |
| 514 | keeper_group = group_lookup[key] |
| 515 | op = { |
| 516 | 'op': 'replace', |
| 517 | 'value': {} |
| 518 | } |
| 519 | if keeper_group.external_id != group.id: |
| 520 | op['value']['externalId'] = group.id |
| 521 | if keeper_group.name != group.name: |
| 522 | op['value']['displayName'] = group.name |
| 523 | if len(op['value']) > 0: |
| 524 | payload = { |
| 525 | 'schemas': ['urn:ietf:params:scim:api:messages:2.0:PatchOp'], |
| 526 | 'Operations': [op] |
| 527 | } |
| 528 | try: |
| 529 | ScimPushCommand.patch_scim_resource( |
| 530 | f'{scim_url}/Groups', keeper_group.id, token, payload, dry_run) |
| 531 | keeper_group.external_id = group.id |
| 532 | keeper_group.name = group.name |
| 533 | logging.info('SCIM updated group "%s"', group.name) |
| 534 | except Exception as e: |
| 535 | logging.warning('PATCH group "%s" error: %s', group.name, e) |
| 536 | |
| 537 | del keeper_group_copy[keeper_group.id] |
| 538 | del external_group_copy[group.id] |
| 539 | |
| 540 | if len(external_group_copy) > 0: # add groups |
| 541 | for group in external_group_copy.values(): |
| 542 | payload = { |
no test coverage detected