tests: add tests for sync groups

invenio_cern_sync/groups/__init__.py

@@ -0,0 +1,8 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2024 CERN.
+#
+# Invenio-CERN-sync is free software; you can redistribute it and/or modify it under
+# the terms of the MIT License; see LICENSE file for more details.
+
+"""Invenio-CERN-sync groups module."""

invenio_cern_sync/groups/sync.py

@@ -0,0 +1,49 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2024 CERN.
+#
+# Invenio-CERN-sync is free software; you can redistribute it and/or modify it under
+# the terms of the MIT License; see LICENSE file for more details.
+
+"""Invenio-CERN-sync groups sync API."""
+
+import time
+import uuid
+
+from invenio_oauthclient.handlers.utils import create_or_update_roles
+
+from ..authz.client import AuthZService, KeycloakService
+from ..logging import log_info
+
+
+def _serialize_groups(groups):
+    """Serialize groups."""
+    for group in groups:
+        yield {
+            "id": group["groupIdentifier"],
+            "name": group["displayName"],
+            "description": group["description"],
+        }
+
+
+def sync(**kwargs):
+    """Sync CERN groups with local db."""
+    log_uuid = str(uuid.uuid4())
+    log_info(log_uuid, "groups_sync", dict(status="fetching-cern-groups"))
+    start_time = time.time()
+
+    overridden_params = kwargs.get("keycloak_service", dict())
+    keycloak_service = KeycloakService(**overridden_params)
+
+    overridden_params = kwargs.get("authz_service", dict())
+    authz_client = AuthZService(keycloak_service, **overridden_params)
+
+    overridden_params = kwargs.get("groups", dict())
+    groups = authz_client.get_groups(**overridden_params)
+
+    roles_ids = create_or_update_roles(_serialize_groups(groups))
+
+    total_time = time.time() - start_time
+    log_info(log_uuid, "groups_sync", dict(status="completed", time=total_time))
+
+    return list(roles_ids)

invenio_cern_sync/users/sync.py

@@ -198,7 +198,7 @@ def sync(method="AuthZ", **kwargs):
         overridden_params = kwargs.get("authz_service", dict())
         authz_client = AuthZService(keycloak_service, **overridden_params)
 
-        overridden_params = kwargs.get("identities_fields", dict())
+        overridden_params = kwargs.get("identities", dict())
         users = authz_client.get_identities(**overridden_params)
         serializer_fn = serialize_cern_identities
     elif method == "LDAP":

tests/conftest.py

@@ -108,3 +108,18 @@ def ldap_users():
             }
         )
     return users
+
+
+@pytest.fixture()
+def authz_groups():
+    """Return CERN groups test data."""
+    groups = []
+    for i in range(10):
+        groups.append(
+            {
+                "groupIdentifier": f"cern-accounts{i}",
+                "displayName": f"CERN Accounts {i}",
+                "description": "This is a test group {i}",
+            }
+        )
+    return groups

tests/test_sync_groups.py

@@ -0,0 +1,69 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2024 CERN.
+#
+# Invenio-CERN-sync is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Sync groups tests."""
+
+from unittest import mock
+from unittest.mock import patch
+
+from invenio_accounts.proxies import current_datastore
+
+from invenio_cern_sync.groups.sync import sync
+
+
+@patch("invenio_cern_sync.groups.sync.KeycloakService")
+@patch("invenio_cern_sync.groups.sync.AuthZService")
+def test_sync_groups(
+    MockAuthZService,
+    MockKeycloakService,
+    app,
+    authz_groups,
+):
+    """Test sync with AuthZ."""
+    MockAuthZService.return_value.get_groups.return_value = authz_groups
+
+    results = sync()
+
+    for expected_group in list(authz_groups):
+        role = current_datastore.find_role_by_id(expected_group["groupIdentifier"])
+        assert role.name == expected_group["displayName"]
+        assert role.description == expected_group["description"]
+
+    assert len(results) == len(authz_groups)
+
+
+@patch("invenio_cern_sync.groups.sync.KeycloakService")
+@patch("invenio_cern_sync.groups.sync.AuthZService")
+def test_sync_groups_update(
+    MockAuthZService,
+    MockKeycloakService,
+    app,
+    authz_groups,
+):
+    """Test sync with AuthZ."""
+    # prepare the db with the initial data
+    MockAuthZService.return_value.get_groups.return_value = authz_groups
+    sync()
+
+    new_group = {
+        "groupIdentifier": "cern-primary-accounts",
+        "displayName": "CERN Primary Accounts",
+        "description": "Group for primary CERN accounts",
+    }
+    updated_group = {
+        "groupIdentifier": "cern-accounts3",
+        "displayName": "New display name",
+        "description": "New description",
+    }
+
+    MockAuthZService.return_value.get_groups.return_value = [new_group, updated_group]
+    sync()
+
+    for expected_group in [new_group, updated_group]:
+        role = current_datastore.find_role_by_id(expected_group["groupIdentifier"])
+        assert role.name == expected_group["displayName"]
+        assert role.description == expected_group["description"]

tests/test_sync.py → tests/test_sync_users.py

@@ -5,7 +5,7 @@
 # Invenio-CERN-sync is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
 
-"""Sync tests."""
+"""Sync users tests."""
 
 from unittest import mock
 from unittest.mock import patch