Feature/fix map role conditions

modules/invenio-accounts/invenio_accounts/admin.py

@@ -26,12 +26,13 @@
 from invenio_communities.models import Community
 from invenio_db import db
 from passlib import pwd
-from sqlalchemy import func
+from sqlalchemy import not_
 from werkzeug.local import LocalProxy
 from collections import OrderedDict
 from wtforms.fields import BooleanField, SelectMultipleField
 from wtforms.validators import DataRequired
 
+from weko_accounts.api import is_map_group, map_role_condition, map_group_condition
 from weko_workflow.models import WorkFlow, WorkflowRole
 
 from .cli import commit
@@ -89,19 +90,15 @@ def scaffold_form(self):
         form_class.role = QuerySelectMultipleField(
             'Roles',
             query_factory=lambda: (
-                Role.query.filter(
-                    ~(
-                        Role.name.like(f"%{current_app.config['WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT']['role_keyword']}%") &
-                        Role.name.startswith(current_app.config['WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT']['prefix'])
-                    )
-                ).filter(~Role.name.like('%_groups_%')).all()
+                Role.query.filter(not_(map_role_condition())
+                ).filter(not_(map_group_condition())).all()
             ),
             get_label='name',
             widget=Select2Widget(multiple=True)
         )
         form_class.group = QuerySelectMultipleField(
             'Groups',
-            query_factory=lambda: Role.query.filter(Role.name.like('%_groups_%')).all(),
+            query_factory=lambda: Role.query.filter(map_group_condition()).all(),
             get_label='name',
             widget=Select2Widget(multiple=True)
         )
@@ -126,8 +123,8 @@ def edit_form(self, obj=None):
 
     def on_form_prefill(self, form, id):
         obj = self.get_one(id)
-        form.role.data = [role for role in obj.roles if '_groups_' not in role.name]
-        form.group.data = [role for role in obj.roles if '_groups_' in role.name]
+        form.role.data = [role for role in obj.roles if not is_map_group(role.name)]
+        form.group.data = [role for role in obj.roles if is_map_group(role.name)]
 
     def on_model_change(self, form, User, is_created):
         """Hash password when saving."""

modules/invenio-accounts/tests/conftest.py

@@ -31,6 +31,7 @@
 from simplekv.memory.redisstore import RedisStore
 from sqlalchemy_utils.functions import create_database, database_exists, \
     drop_database
+from weko_accounts.config import WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT
 from weko_records_ui.config import WEKO_PERMISSION_SUPER_ROLE_USER
 
 from invenio_accounts import InvenioAccounts
@@ -72,6 +73,7 @@ def _app_factory(config=None):
         ACCOUNTS_JWT_ALOGORITHM = 'HS256',
         ACCOUNTS_JWT_SECRET_KEY = 'None',
         WEKO_PERMISSION_SUPER_ROLE_USER = WEKO_PERMISSION_SUPER_ROLE_USER,
+        WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT = WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT
     )
 
     # Set key value session store to use Redis when running on TravisCI.

modules/invenio-accounts/tests/test_admin.py

@@ -264,15 +264,18 @@ def test_userview_get_count_query(app, users):
                     assert query.scalar() == 0
 
 # .tox/c1/bin/pytest --cov=invenio_accounts tests/test_admin.py::test_userview_on_form_prefill -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/invenio-accounts/.tox/c1/tmp
-def test_userview_on_form_prefill(app, users):
+def test_userview_on_form_prefill(app, users, mocker):
     """Test on_form_prefill for super role user."""
     with app.app_context():
+        mocker.patch.dict(current_app.config, {
+            'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib'
+        })
         view = UserView(User, db.session)
         form = view.create_form()
         user = User.query.filter_by(email=users[2]['email']).first()
         ds = app.extensions["invenio-accounts"].datastore
         ds.add_role_to_user(user, Role(name='role1'))
-        ds.add_role_to_user(user, Role(name='role2_groups_1'))
+        ds.add_role_to_user(user, Role(name='jc_test_example_com_gr_1'))
         db.session.commit()
 
         view.get_one = MagicMock(return_value=user)
@@ -289,16 +292,36 @@ def test_userview_edit_form(app, users):
     assert form.data['active'] is False
 
 # .tox/c1/bin/pytest --cov=invenio_accounts tests/test_admin.py::test_scaffold_form -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/invenio-accounts/.tox/c1/tmp
-def test_scaffold_form(app):
+def test_scaffold_form(app, mocker):
     """Test scaffold_form method of UserView."""
     with app.app_context():
-        current_app.config['WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT'] = {
-            'role_keyword': 'roles',
-            'prefix': 'jc'
-        }
-        db.session.add(Role(id=1, name='Contributor', description=None))
-        db.session.add(Role(id=2, name='jc_xxx_roles_contributor', description=None))
-        db.session.add(Role(id=3, name='jc_xxx_groups_yyy', description=None))
+        mocker.patch.dict(current_app.config, {
+            'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib'
+        })
+
+        pattern = current_app.config.get(
+            'WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT')
+        prefix = pattern.get("prefix")
+        role_key = pattern.get("role_keyword")
+        repoid = "test_example_com"
+
+        role_name1 = f"{prefix}_{repoid}_{role_key}_radm"
+        role_name2 = pattern.get("sysadm_group")
+        role_name3 = f"{prefix}_{repoid}_{role_key}_unkwown"
+        role_name4 = f"{prefix}_{repoid}_gr_radm"
+        role_name5 = f"{prefix}_test!example!com_{role_key}_radm"
+        role_name6 = f"ng_{repoid}_{role_key}_radm"
+        role_name7 = 'Contributor'
+
+        db.session.add_all([
+            Role(id=1, name=role_name1),
+            Role(id=2, name=role_name2),
+            Role(id=3, name=role_name3),
+            Role(id=4, name=role_name4),
+            Role(id=5, name=role_name5),
+            Role(id=6, name=role_name6),
+            Role(id=7, name=role_name7)
+        ])
         db.session.commit()
 
         view = UserView(User, db.session)
@@ -310,12 +333,10 @@ def test_scaffold_form(app):
         role_names = [r.name for r in roles]
         group_names = [g.name for g in groups]
         # Check included roles
-        assert 'Contributor' in role_names
+        assert len(role_names) == 3
+        assert role_name5 in role_names
+        assert role_name6 in role_names
+        assert role_name7 in role_names
         # Check included groups
-        assert 'jc_xxx_groups_yyy' in group_names
-        # Check excluded roles
-        assert 'jc_xxx_roles_contributor' not in role_names
-        assert 'jc_xxx_groups_yyy' not in role_names
-        # Check excluded groups
-        assert 'Contributor' not in group_names
-        assert 'jc_xxx_roles_contributor' not in group_names
+        assert len(group_names) == 1
+        assert role_name4 in group_names

modules/invenio-communities/invenio_communities/admin.py

@@ -38,12 +38,13 @@
 from flask_login import current_user
 from invenio_accounts.models import Role
 from invenio_db import db
-from sqlalchemy import func, or_
+from sqlalchemy import func, or_, not_
 from weko_index_tree.models import Index
 from wtforms.validators import ValidationError, Length
 from wtforms import FileField, RadioField, StringField
 from wtforms.utils import unset_value
 from invenio_i18n.ext import current_i18n
+from weko_accounts.api import map_role_condition, map_group_condition
 from weko_gridlayout.services import WidgetDesignPageServices
 from weko_handle.api import Handle
 from weko_workflow.config import WEKO_SERVER_CNRI_HOST_LINK
@@ -563,15 +564,12 @@ def validate_community_id(self, community_id):
         'owner': {
             'allow_blank': False,
             'query_factory': lambda: db.session.query(Role).filter(
-                ~(
-                    Role.name.like(f"%{current_app.config['WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT'].get('role_keyword','')}%") &
-                    Role.name.startswith(current_app.config['WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT'].get('prefix',''))
-                )
-            ).all(),
+                not_(map_role_condition())).all(),
         },
         'group': {
             'allow_blank': False,
-            'query_factory': lambda: db.session.query(Role).filter(Role.name.like("%_groups_%")).all()
+            'query_factory': lambda: db.session.query(Role).filter(
+                map_group_condition()).all()
         }
     }
     form_extra_fields = {

modules/invenio-communities/invenio_communities/models.py

@@ -40,6 +40,7 @@
 from sqlalchemy_utils.models import Timestamp
 from sqlalchemy_utils.types import UUIDType
 from weko_index_tree.models import Index
+from weko_accounts.api import create_fqdn_from_entity_id, is_map_sysadm_role
 
 from .errors import CommunitiesError, InclusionRequestExistsError, \
     InclusionRequestExpiryTimeError, InclusionRequestMissingError, \
@@ -596,16 +597,22 @@ def owner_display(self):
             str: Display name of the community owner.
         """
         if self.owner and hasattr(self.owner, 'name'):
-            roles_key = current_app.config["WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT"]["role_keyword"]
-            role_mapping = current_app.config["WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT"]["role_mapping"]
-            sysadm_key = current_app.config["WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT"]["sysadm_group"]
             owner_name = self.owner.name
-            if owner_name == sysadm_key:
-                return "System Administrator"
-            if owner_name and roles_key in owner_name:
-                suffix = owner_name.split(roles_key + '_')[-1]
-                if suffix in role_mapping.keys():
-                    owner_name = role_mapping[suffix]
+            pattern = current_app.config.get(
+                'WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT')
+            idp_entity_id = current_app.config.get('WEKO_ACCOUNTS_IDP_ENTITY_ID')
+            if not bool(pattern and idp_entity_id):
+                return owner_name
+            prefix = pattern.get("prefix")
+            role_key = pattern.get("role_keyword")
+            role_mapping = pattern.get("role_mapping")
+            fqdn = create_fqdn_from_entity_id()
+            if is_map_sysadm_role(owner_name):
+                return current_app.config['WEKO_ADMIN_PERMISSION_ROLE_SYSTEM']
+            for suffix, display_name in role_mapping.items():
+                expected_owner_name = f'{prefix}_{fqdn}_{role_key}_{suffix}'
+                if owner_name == expected_owner_name:
+                    return display_name
             return owner_name
 
 class FeaturedCommunity(db.Model, Timestamp):

modules/invenio-communities/tests/conftest.py

@@ -66,6 +66,7 @@
 import uuid
 from invenio_pidstore.models import PersistentIdentifier,PIDStatus,RecordIdentifier
 from invenio_pidrelations.models import PIDRelation
+from weko_accounts.config import WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT
 from weko_records.api import ItemsMetadata
 
 from invenio_communities import InvenioCommunities
@@ -111,6 +112,8 @@ def base_app(instance_path, request):
         INDEXER_DEFAULT_DOCTYPE='item-v1.0.0',
         INDEXER_DEFAULT_INDEX="{}-weko-item-v1.0.0".format("test"),
         SEARCH_UI_SEARCH_INDEX="{}-weko".format("test"),
+        WEKO_ADMIN_PERMISSION_ROLE_SYSTEM = "System Administrator",
+        WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT = WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT,
     )
     FlaskCeleryExt(app_)
     Menu(app_)

modules/invenio-communities/tests/test_admin.py

@@ -8,8 +8,9 @@
 from io import BytesIO
 from mock import patch
 from invenio_accounts.testutils import login_user_via_session, create_test_user
-from invenio_access.models import ActionUsers
+from invenio_accounts.models import Role
 from invenio_communities.models import Community
+from weko_accounts.api import create_fqdn_from_entity_id
 from weko_records.models import ItemTypeProperty
 from weko_index_tree.models import IndexStyle,Index
 from invenio_accounts.testutils import login_user_via_session
@@ -80,17 +81,17 @@ def setup_view_community(app,db,users):
 # .tox/c1/bin/pytest --cov=invenio_communities tests/test_admin.py::TestInclusionRequestModelView -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/invenio-communities/.tox/c1/tmp
 class TestCommunityModelView():
     # .tox/c1/bin/pytest --cov=invenio_communities tests/test_admin.py::TestCommunityModelView::test_owner_query_factory_exclude_roles -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/invenio-communities/.tox/c1/tmp
-    def test_owner_query_factory_exclude_roles(self, app, db):
-        app.config['WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT'] = {
-            'role_keyword': 'roles',
-            'prefix': 'jc'
-        }
-        from invenio_accounts.models import Role
-
-        role_key = app.config['WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT'].get('role_keyword', '')
-        prefix = app.config['WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT'].get('prefix', '')
-
-        role_both = Role(name=f"{prefix}abc{role_key}")  # Contains both → should be excluded
+    def test_owner_query_factory_exclude_roles(self, app, db, mocker):
+        mocker.patch.dict(current_app.config, {
+            'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib'
+        })
+        pattern = app.config['WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT']
+        prefix = pattern.get('prefix', '')
+        role_key = pattern.get('role_keyword', '')
+        fqdn = create_fqdn_from_entity_id()
+
+        role_all = Role(name=f"{prefix}_{fqdn}_{role_key}_radm")  # Contains all → should be excluded
+        role_both = Role(name=f"{prefix}abc{role_key}")  # Contains key, prefix → should be included
         role_only_key = Role(name=f"abc{role_key}")      # Contains only one → should be included
         role_only_prefix = Role(name=f"{prefix}abc")     # Contains only one → should be included
         role_none = Role(name="abc")                     # Contains neither → should be included
@@ -103,8 +104,9 @@ def test_owner_query_factory_exclude_roles(self, app, db):
         owner_names = [r.name for r in view.form_args['owner']['query_factory']()]
 
         # Exclude roles that contain both
-        assert role_both.name not in owner_names
+        assert role_all.name not in owner_names
         # Include roles that contain only one or neither
+        assert role_both.name in owner_names
         assert role_only_key.name in owner_names
         assert role_only_prefix.name in owner_names
         assert role_none.name in owner_names