global: users - renaming, refactor ui roles

Signed-off-by: pamfilos <pamfilosf@gmail.com>

Author: pamfilos

cap/modules/schemas/imp.py

@@ -25,7 +25,7 @@
 from itertools import groupby
 
 from invenio_access.models import ActionRoles, ActionUsers
-from invenio_access.permissions import Permission, superuser_access
+from invenio_access.permissions import Permission
 from invenio_cache import current_cache
 from sqlalchemy.event import listen
 
@@ -76,6 +76,11 @@ def _filter_by_deposit_create_access(schemas_list):
     ]
 
 
+def _filter_by_admin_access(schemas_list):
+    """Return only schemas that user has admin access to."""
+    return [x for x in schemas_list if AdminSchemaPermission(x).can()]
+
+
 def _filter_by_record_read_access(schemas_list):
     """Return only schemas that user has read access to."""
     return [
@@ -102,6 +107,14 @@ def get_cached_indexed_schemas_for_user_create(latest=True, user_id=None):
     return schemas
 
 
+@current_cache.memoize()
+def get_cached_indexed_schemas_for_user_admin(latest=True, user_id=None):
+    """Return all indexed schemas current user has read access to."""
+    schemas = get_indexed_schemas(latest=latest)
+    schemas = _filter_by_admin_access(schemas)
+    return schemas
+
+
 @current_cache.memoize()
 def get_cached_indexed_schemas_for_user_read(latest=True, user_id=None):
     """Return all indexed schemas current user has read access to."""
@@ -154,41 +167,6 @@ def _filter_by_read_access(schemas_list):
     return [x for x in schemas_list if ReadSchemaPermission(x).can()]
 
 
-def _filter_by_admin_access(schemas_list):
-    """Return only schemas that user has admin access to."""
-    return [x for x in schemas_list if AdminSchemaPermission(x).can()]
-
-
-def is_super_user():
-    return Permission(superuser_access).can()
-
-
-def get_admin_roles_for_user(latest=True):
-    """Return list of roles in schemas, current user has admin/superuser access to."""
-    roles = []
-    schemas = get_indexed_schemas(latest=latest)
-    schemas = _filter_by_admin_access(schemas)
-    if latest:
-        schemas = _filter_only_latest(schemas)
-
-    for schema in schemas:
-        roles.append(f"{schema.name}")
-
-    return roles
-
-
-def generate_roles(mapping):
-    roles = []
-    for method_name, method in mapping.items():
-        result = method()
-        if isinstance(result, bool) and result:
-            roles.append(method_name)
-        elif isinstance(result, list):
-            for role in result:
-                roles.append(f"{method_name}:{role}")
-    return roles
-
-
 def get_schemas_for_user(latest=True):
     """Return all schemas current user has read access to."""
     schemas = Schema.query.order_by(
@@ -233,6 +211,7 @@ def clear_schema_access_cache(mapper, connection, target):
     ):
         get_cached_indexed_schemas_for_user_create.delete_memoized()
         get_cached_indexed_schemas_for_user_read.delete_memoized()
+        get_cached_indexed_schemas_for_user_admin.delete_memoized()
         get_cached_indexed_record_schemas_for_user_create.delete_memoized()
         get_cached_indexed_record_schemas_for_user_read.delete_memoized()
 

cap/modules/user/utils.py

@@ -26,28 +26,68 @@
 import ldap
 from cachetools.func import lru_cache
 from flask import current_app
-from sqlalchemy.orm.exc import NoResultFound
-from werkzeug.local import LocalProxy
-
+from flask_login import current_user
+from invenio_access.permissions import Permission, superuser_access
 from invenio_accounts.models import Role, User
 from invenio_oauthclient.models import RemoteAccount
+from sqlalchemy.orm.exc import NoResultFound
+from werkzeug.local import LocalProxy
 
 from .errors import DoesNotExistInLDAP
 
 _datastore = LocalProxy(lambda: current_app.extensions['security'].datastore)
 
 
+def is_super_user():
+    return Permission(superuser_access).can()
+
+
+def get_admin_roles_for_user(latest=True):
+    """Return list of schemas user can admin."""
+    from cap.modules.schemas.imp import (
+        get_cached_indexed_schemas_for_user_admin,
+    )
+
+    roles = []
+    schemas = get_cached_indexed_schemas_for_user_admin(
+        latest=latest, user_id=current_user.id
+    )
+
+    for schema in schemas:
+        roles.append(f"{schema.name}")
+
+    return roles
+
+
+USER_UI_ROLES = {
+    'superuser': is_super_user,
+    'schema-admin': get_admin_roles_for_user,
+}
+
+
+def get_user_ui_roles():
+    roles = []
+    for method_name, method in USER_UI_ROLES.items():
+        result = method()
+        if isinstance(result, bool) and result:
+            roles.append(method_name)
+        elif isinstance(result, list):
+            for role in result:
+                roles.append(f"{method_name}:{role}")
+    return roles
+
+
 def _query_ldap(base, query, fields):
     lc = ldap.initialize('ldap://xldap.cern.ch')
-    lc.search_ext(base,
-                  ldap.SCOPE_ONELEVEL,
-                  query,
-                  fields,
-                  serverctrls=[
-                      ldap.controls.SimplePagedResultsControl(True,
-                                                              size=7,
-                                                              cookie='')
-                  ])
+    lc.search_ext(
+        base,
+        ldap.SCOPE_ONELEVEL,
+        query,
+        fields,
+        serverctrls=[
+            ldap.controls.SimplePagedResultsControl(True, size=7, cookie='')
+        ],
+    )
 
     return lc.result()[1]
 
@@ -60,16 +100,20 @@ def get_user_mail_from_ldap(display_name):
     res = _query_ldap(
         base='OU=Users,OU=Organic Units,DC=cern,DC=ch',
         query='(&(cernAccountType=Primary)(displayName={}))'.format(
-            display_name),
-        fields=['mail'])
+            display_name
+        ),
+        fields=['mail'],
+    )
 
     if not res:
         parts = display_name.split(' ')
         res = _query_ldap(
             base='OU=Users,OU=Organic Units,DC=cern,DC=ch',
             query='(&(cernAccountType=Primary)(givenName={}*)(sn=*{}))'.format(
-                parts[0], parts[-1]),
-            fields=['mail'])
+                parts[0], parts[-1]
+            ),
+            fields=['mail'],
+        )
 
     if len(res) != 1:
         raise DoesNotExistInLDAP
@@ -85,16 +129,19 @@ def does_user_exist_in_ldap(mail):
     res = _query_ldap(
         base='OU=Users,OU=Organic Units,DC=cern,DC=ch',
         query='(&(cernAccountType=Primary)(mail={}))'.format(mail),
-        fields=['mail'])
+        fields=['mail'],
+    )
 
     return True if res else False
 
 
 def does_egroup_exist_in_ldap(mail):
     """Query ldap to check if user exists."""
-    res = _query_ldap(base='OU=e-groups,OU=Workgroups,DC=cern,DC=ch',
-                      query='mail={}'.format(mail),
-                      fields=['mail'])
+    res = _query_ldap(
+        base='OU=e-groups,OU=Workgroups,DC=cern,DC=ch',
+        query='mail={}'.format(mail),
+        fields=['mail'],
+    )
 
     return True if res else False
 

cap/modules/user/views.py

@@ -36,25 +36,14 @@
 
 from cap.config import DEBUG
 from cap.modules.access.utils import login_required
-from cap.modules.schemas.imp import (
-    get_admin_roles_for_user,
-    get_cached_indexed_schemas_for_user_create,
-    generate_roles,
-    is_super_user,
-)
-from cap.modules.user.utils import get_remote_account_by_id
+from cap.modules.schemas.imp import get_cached_indexed_schemas_for_user_create
+from cap.modules.user.utils import get_remote_account_by_id, get_user_ui_roles
 
 _datastore = LocalProxy(lambda: current_app.extensions['security'].datastore)
 
 user_blueprint = Blueprint('cap_user', __name__, template_folder='templates')
 
 
-USER_UI_ROLES = {
-    'superuser': is_super_user,
-    'schema-admin': get_admin_roles_for_user,
-}
-
-
 @user_blueprint.route('/me')
 @login_required
 def get_user():
@@ -73,7 +62,7 @@ def get_user():
         "email": current_user.email,
         "deposit_groups": deposit_groups,
         "profile": extra_data,
-        "roles": generate_roles(USER_UI_ROLES),
+        "roles": get_user_ui_roles(),
     }
 
     response = jsonify(_user)