Fix group enumeration. Fix build.

src/pas/plugins/eea/__init__.py

@@ -9,7 +9,6 @@
 from pas.plugins.eea.plugin import EEAEntraPlugin
 from pas.plugins.eea.plugin import manage_addEEAEntraPlugin
 from pas.plugins.eea.plugin import manage_addEEAEntraPluginForm
-from pas.plugins.eea.plugin import tpl_dir
 
 _ = MessageFactory("pas.plugins.eea")
 

src/pas/plugins/eea/browser/user_sync.py

@@ -1,13 +1,14 @@
 from datetime import datetime
 from logging import getLogger
 
-from plone import api
-from plone import schema
-from plone.autoform.form import AutoExtensibleForm
 from z3c.form import button
 from z3c.form import form
 from zope.interface import Interface
 
+from plone import api
+from plone import schema
+from plone.autoform.form import AutoExtensibleForm
+
 from pas.plugins.eea.sync import SyncEntra
 
 logger = getLogger(__name__)
@@ -47,4 +48,8 @@ def do_sync(self):
         syncer.sync_groups()
         syncer.sync_group_members()
 
-        return syncer.count_users, syncer.count_groups, datetime.isoformat(datetime.now())
+        return (
+            syncer.count_users,
+            syncer.count_groups,
+            datetime.isoformat(datetime.now()),
+        )

src/pas/plugins/eea/plugin.py

@@ -3,22 +3,23 @@
 
 from AccessControl import ClassSecurityInfo
 from AccessControl.class_init import InitializeClass
+
 from BTrees.OOBTree import OOBTree
+from zope.interface import implementer
+
 from Products.PageTemplates.PageTemplateFile import PageTemplateFile
 from Products.PlonePAS.interfaces.group import IGroupIntrospection
 from Products.PlonePAS.interfaces.group import IGroupManagement
 from Products.PlonePAS.plugins.autogroup import VirtualGroup
 from Products.PluggableAuthService.interfaces import plugins as pas_interfaces
 from Products.PluggableAuthService.plugins.BasePlugin import BasePlugin
-from zope.interface import implementer
 
 logger = logging.getLogger(__name__)
 tpl_dir = Path(__file__).parent.resolve() / "browser"
 
 _marker = {}
 
 
-
 def manage_addEEAEntraPlugin(context, id, title="", RESPONSE=None, **kw):
     """Create an instance of an EEA Plugin."""
     plugin = EEAEntraPlugin(id, title, **kw)
@@ -143,7 +144,7 @@ def savedGroups(self, query=None):
                     }
                 ]
 
-        if not result:
+        else:
             result = [
                 {
                     "title": group_title,
@@ -161,15 +162,23 @@ def enumerateGroups(
         self, id=None, exact_match=False, sort_by=None, max_results=None, **kw
     ):
         result = []
-        if not id:
+        if not id and kw:
+            for group in self.savedGroups():
+                for key, value in kw.items():
+                    if value.lower() in group.get(key).lower():
+                        result.append(group)
+
+        elif not id:
             result = self.savedGroups()
 
         elif id and exact_match:
             result = self.savedGroups(id)
 
-        if not result:
+        elif id:
             query = id.lower()
-            result = [g for g in self.savedGroups() if query in g["title"].lower()]
+            result = [
+                g for g in self.savedGroups() if query in g["title"].lower()
+            ]
 
         return result
 

src/pas/plugins/eea/query.py

@@ -8,6 +8,7 @@
 from typing import TypedDict
 
 import requests
+
 from plone.memoize import ram
 
 logging.basicConfig(level=logging.DEBUG)
@@ -17,34 +18,44 @@
 logger = logging.getLogger(__name__)
 
 
-ApiUser = TypedDict("ApiUser", {
-    "@odata.type": Literal["#microsoft.graph.user"],
-    "id": str,
-    "businessPhones": List[str],
-    "displayName": str,
-    "givenName": str | None,
-    "jobTitle": str | None,
-    "mail": str | None,
-    "mobilePhone": str | None,
-    "officeLocation": str | None,
-    "preferredLanguage": str | None,
-    "surname": str | None,
-    "userPrincipalName": str,
-})
-
-ApiGroup = TypedDict("ApiGroup", {
-    "@odata.type": Literal["#microsoft.graph.group"],
-    "id": str,
-    "displayName": str,
-    "mail": str | None,
-    "mailEnabled": bool,
-    "visibility": str | None,
-})
-
-ApiMember = TypedDict("ApiMember", {
-    "@odata.type": Literal["#microsoft.graph.group"] | Literal["#microsoft.graph.user"],
-    "id": str,
-})
+ApiUser = TypedDict(
+    "ApiUser",
+    {
+        "@odata.type": Literal["#microsoft.graph.user"],
+        "id": str,
+        "businessPhones": List[str],
+        "displayName": str,
+        "givenName": str | None,
+        "jobTitle": str | None,
+        "mail": str | None,
+        "mobilePhone": str | None,
+        "officeLocation": str | None,
+        "preferredLanguage": str | None,
+        "surname": str | None,
+        "userPrincipalName": str,
+    },
+)
+
+ApiGroup = TypedDict(
+    "ApiGroup",
+    {
+        "@odata.type": Literal["#microsoft.graph.group"],
+        "id": str,
+        "displayName": str,
+        "mail": str | None,
+        "mailEnabled": bool,
+        "visibility": str | None,
+    },
+)
+
+ApiMember = TypedDict(
+    "ApiMember",
+    {
+        "@odata.type": Literal["#microsoft.graph.group"]
+        | Literal["#microsoft.graph.user"],
+        "id": str,
+    },
+)
 
 
 def _cachekey_query_api_endpoint(
@@ -89,7 +100,9 @@ def get_access_token(self):
         response = requests.post(url, headers=headers, data=data)
         token_data = response.json()
 
-        QueryEntra._token_cache = {"expires": time() + token_data["expires_in"] - 60}
+        QueryEntra._token_cache = {
+            "expires": time() + token_data["expires_in"] - 60
+        }
         QueryEntra._token_cache.update(token_data)
         return QueryEntra._token_cache["access_token"]
 
@@ -120,13 +133,17 @@ def get_url(
             return response.json()
 
     def get_all(self, url, consistent=True, extra_headers=None):
-        data = self.get_url(url, consistent=consistent, extra_headers=extra_headers)
+        data = self.get_url(
+            url, consistent=consistent, extra_headers=extra_headers
+        )
         if data:
             yield from data.get("value", [data])
             next_url = data.get("@odata.nextLink")
             if next_url:
                 yield from self.get_all(
-                    next_url, consistent=consistent, extra_headers=extra_headers
+                    next_url,
+                    consistent=consistent,
+                    extra_headers=extra_headers,
                 )
 
     def get_user(self, user_id) -> ApiUser:
@@ -155,9 +172,7 @@ def search_users(self, query, properties=None) -> Iterator[ApiUser]:
         if custom_query:
             url = f'{url}?$search="{custom_query}"'
 
-        return self.get_all(
-            url, consistent=False
-        )
+        return self.get_all(url, consistent=False)
 
     def get_user_groups(self, user_id) -> Iterator[ApiGroup]:
         url = f"https://graph.microsoft.com/v1.0/users/{user_id}/memberOf/microsoft.graph.group?$top=999&$select=id"

src/pas/plugins/eea/sync.py

@@ -1,26 +1,25 @@
 import logging
 import uuid
-
 from dataclasses import dataclass
 
+from pas.plugins.authomatic.useridentities import UserIdentities
+from pas.plugins.authomatic.useridentities import UserIdentity
+from pas.plugins.authomatic.utils import authomatic_cfg
+
 from BTrees.OOBTree import TreeSet  # noqa
 
+from Products.PluggableAuthService.UserPropertySheet import UserPropertySheet
+
 from pas.plugins.eea.query import ApiUser
-from pas.plugins.eea.query import QueryEntra
 from pas.plugins.eea.query import QueryConfig
-from pas.plugins.eea.utils import get_plugin
+from pas.plugins.eea.query import QueryEntra
 from pas.plugins.eea.utils import get_authomatic_plugin
+from pas.plugins.eea.utils import get_plugin
 from pas.plugins.eea.utils import get_provider_name
-from pas.plugins.authomatic.utils import authomatic_cfg
-
-from pas.plugins.authomatic.useridentities import UserIdentities
-from pas.plugins.authomatic.useridentities import UserIdentity
-
-from Products.PluggableAuthService.UserPropertySheet import UserPropertySheet
-
 
 logger = logging.getLogger(__name__)
 
+
 @dataclass
 class MockProvider:
     name: str
@@ -60,20 +59,26 @@ def _init_query_manager(self):
         settings = authomatic_cfg()
         provider_name = get_provider_name(settings)
         cfg = settings.get(provider_name, {}) if settings else {}
-        query_config = QueryConfig(cfg["consumer_key"], cfg["consumer_secret"], cfg["domain"])
+        query_config = QueryConfig(
+            cfg["consumer_key"], cfg["consumer_secret"], cfg["domain"]
+        )
 
         self._provider_name = provider_name
         self._qm = QueryEntra(query_config)
 
     def get_plone_uuid(self, service_uuid):
         plone_key = (self._provider_name, service_uuid)
-        return self._plugin_authomatic._userid_by_identityinfo.get(plone_key, None)
+        return self._plugin_authomatic._userid_by_identityinfo.get(
+            plone_key, None
+        )
 
     def get_service_uuid(self, plone_uuid):
         match = [
             service_uuid
-            for (name, service_uuid), p_uuid
-            in self._plugin_authomatic._userid_by_identityinfo.items()
+            for (
+                name,
+                service_uuid,
+            ), p_uuid in self._plugin_authomatic._userid_by_identityinfo.items()
             if p_uuid == plone_uuid and name == self._provider_name
         ]
         return match[0] if match else None
@@ -99,10 +104,7 @@ def remember_user(self, user: ApiUser):
         self._plugin_authomatic._userid_by_identityinfo[user_key] = plone_uuid
 
         self.count_users += 1
-        logger.info(
-            "Added new user: %s (%s)",
-            plone_uuid, user["displayName"]
-        )
+        logger.info("Added new user: %s (%s)", plone_uuid, user["displayName"])
 
     def sync_users(self):
         for user in self._qm.get_all_users():
@@ -126,9 +128,10 @@ def sync_group_members(self):
             for member in self._qm.get_group_members(group_id):
                 if member["@odata.type"] == "#microsoft.graph.user":
                     plone_uuid = self.get_plone_uuid(member["id"])
-                    self._plugin_eea._ad_group_members[group_id].add(plone_uuid)
+                    self._plugin_eea._ad_group_members[group_id].add(
+                        plone_uuid
+                    )
                     self.remember_user_group(plone_uuid, group_id)
                 elif member["@odata.type"] == "#microsoft.graph.group":
                     # TODO: remember groups as group members?
                     continue
-

src/pas/plugins/eea/tests/test_setup.py

@@ -61,7 +61,9 @@ def setUp(self):
 
     def test_product_uninstalled(self):
         """Test if pas.plugins.eea is cleanly uninstalled."""
-        self.assertFalse(self.installer.is_product_installed("pas.plugins.eea"))
+        self.assertFalse(
+            self.installer.is_product_installed("pas.plugins.eea")
+        )
 
     def test_browserlayer_removed(self):
         """Test that IPasPluginsEeaLayer is removed."""

src/pas/plugins/eea/utils.py

@@ -24,6 +24,9 @@ def get_authomatic_plugin() -> "AuthomaticPlugin":
 
 def get_provider_name(cfg, default="microsoft"):
     for name, settings in cfg.items():
-        if settings.get("class_") == "authomatic.providers.oauth2.MicrosoftOnline":
+        if (
+            settings.get("class_")
+            == "authomatic.providers.oauth2.MicrosoftOnline"
+        ):
             return name
     return default