diff --git a/modules/invenio-accounts/invenio_accounts/admin.py b/modules/invenio-accounts/invenio_accounts/admin.py index f6b25c0067..9ab2aef2af 100644 --- a/modules/invenio-accounts/invenio_accounts/admin.py +++ b/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.""" diff --git a/modules/invenio-accounts/tests/conftest.py b/modules/invenio-accounts/tests/conftest.py index 3917c4eefd..15d5fa1df8 100644 --- a/modules/invenio-accounts/tests/conftest.py +++ b/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. diff --git a/modules/invenio-accounts/tests/test_admin.py b/modules/invenio-accounts/tests/test_admin.py index b29ff67aba..284b52e744 100644 --- a/modules/invenio-accounts/tests/test_admin.py +++ b/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 diff --git a/modules/invenio-communities/invenio_communities/admin.py b/modules/invenio-communities/invenio_communities/admin.py index 25b92e3ac2..5fedf0c77a 100644 --- a/modules/invenio-communities/invenio_communities/admin.py +++ b/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 = { diff --git a/modules/invenio-communities/invenio_communities/models.py b/modules/invenio-communities/invenio_communities/models.py index 86e42c8553..cc5008a071 100644 --- a/modules/invenio-communities/invenio_communities/models.py +++ b/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): diff --git a/modules/invenio-communities/tests/conftest.py b/modules/invenio-communities/tests/conftest.py index a102e0fbb2..799c8346f4 100644 --- a/modules/invenio-communities/tests/conftest.py +++ b/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_) diff --git a/modules/invenio-communities/tests/test_admin.py b/modules/invenio-communities/tests/test_admin.py index cb5ee318f4..72aca6007f 100644 --- a/modules/invenio-communities/tests/test_admin.py +++ b/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 diff --git a/modules/invenio-communities/tests/test_models.py b/modules/invenio-communities/tests/test_models.py index f0da640fee..aeae546227 100644 --- a/modules/invenio-communities/tests/test_models.py +++ b/modules/invenio-communities/tests/test_models.py @@ -29,8 +29,10 @@ import os import pytest from datetime import datetime, timedelta +from flask import current_app from invenio_records.api import Record from unittest.mock import patch +from invenio_accounts.models import Role from invenio_oaiserver.models import OAISet from invenio_communities.models import Community,InclusionRequest @@ -375,51 +377,103 @@ def test_oaiset_url(self, app, communities): # @classmethod # def get_featured_or_none(cls, start_date=None): -from invenio_accounts.models import Role -from flask import Flask -from invenio_communities.models import Community - -@pytest.fixture -def app(): - app = Flask(__name__) - app.config['WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT'] = { - 'role_keyword': 'roles', - 'role_mapping': { - "repoadm":"Repository Administrator", - "comadm":"Community Administrator", - }, - "sysadm_group": "jc_roles_sysadm" - } - return app # .tox/c1/bin/pytest --cov=invenio_communities tests/test_models.py::test_owner_display -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/invenio-communities/.tox/c1/tmp -def test_owner_display(app): +def test_owner_display(app, mocker): with app.app_context(): + 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" + + comm = Community() # If owner_name contains role_keyword and is mapped by role_mapping - owner1 = Role(name="roles_repoadm") - comm1 = Community() - comm1.owner = owner1 - assert comm1.owner_display == "Repository Administrator" + ower_role_name = f"{prefix}_{repoid}_{role_key}_radm" + comm.owner = Role(name=ower_role_name) + assert comm.owner_display == "Repository Administrator" # If owner_name matches sysadm_group - owner2 = Role(name="jc_roles_sysadm") - comm2 = Community() - comm2.owner = owner2 - assert comm2.owner_display == "System Administrator" + ower_role_name = pattern.get("sysadm_group") + comm.owner = Role(name=ower_role_name) + assert comm.owner_display == "System Administrator" # If owner_name contains role_keyword but is not found in role_mapping - owner3 = Role(name="roles_unknown") - comm3 = Community() - comm3.owner = owner3 - assert comm3.owner_display == "roles_unknown" + ower_role_name = f"{prefix}_{repoid}_{role_key}_unkwown" + comm.owner = Role(name=ower_role_name) + assert comm.owner_display == ower_role_name + + ower_role_name = f"{prefix}_{repoid}_gr_radm" + comm.owner = Role(name=ower_role_name) + assert comm.owner_display == ower_role_name + + ower_role_name = f"{prefix}_test!example!com_{role_key}_radm" + comm.owner = Role(name=ower_role_name) + assert comm.owner_display == ower_role_name + + ower_role_name = f"ng_{repoid}_{role_key}_radm" + comm.owner = Role(name=ower_role_name) + assert comm.owner_display == ower_role_name + + # If owner_name does not contain role_keyword + owner = Role(name="admin") + comm.owner = owner + assert comm.owner_display == "admin" + + # If owner is None + comm.owner = None + assert comm.owner_display is None + + +# .tox/c1/bin/pytest --cov=invenio_communities tests/test_models.py::test_owner_display_no_repoid -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/invenio-communities/.tox/c1/tmp +def test_owner_display_no_repoid(app, mocker): + with app.app_context(): + mocker.patch.dict(current_app.config, { + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': '' + }) + + 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" + + comm = Community() + # If owner_name contains role_keyword and is mapped by role_mapping + ower_role_name = f"{prefix}_{repoid}_{role_key}_radm" + comm.owner = Role(name=ower_role_name) + assert comm.owner_display == ower_role_name + + # If owner_name matches sysadm_group + ower_role_name = pattern.get("sysadm_group") + comm.owner = Role(name=ower_role_name) + assert comm.owner_display == ower_role_name + # If owner_name contains role_keyword but is not found in role_mapping + ower_role_name = f"{prefix}_{repoid}_{role_key}_unkwown" + comm.owner = Role(name=ower_role_name) + assert comm.owner_display == ower_role_name + + ower_role_name = f"{prefix}_{repoid}_gr_radm" + comm.owner = Role(name=ower_role_name) + assert comm.owner_display == ower_role_name + + ower_role_name = f"{prefix}_test!example!com_{role_key}_radm" + comm.owner = Role(name=ower_role_name) + assert comm.owner_display == ower_role_name + + ower_role_name = f"ng_{repoid}_{role_key}_radm" + comm.owner = Role(name=ower_role_name) + assert comm.owner_display == ower_role_name # If owner_name does not contain role_keyword - owner4 = Role(name="admin") - comm4 = Community() - comm4.owner = owner4 - assert comm4.owner_display == "admin" + owner = Role(name="admin") + comm.owner = owner + assert comm.owner_display == "admin" # If owner is None - comm_none = Community() - comm_none.owner = None - assert comm_none.owner_display is None or comm_none.owner_display == '' + comm.owner = None + assert comm.owner_display is None diff --git a/modules/weko-accounts/tests/conftest.py b/modules/weko-accounts/tests/conftest.py index 9c52fcfef9..2391106455 100644 --- a/modules/weko-accounts/tests/conftest.py +++ b/modules/weko-accounts/tests/conftest.py @@ -52,8 +52,48 @@ from weko_logging.audit import WekoLoggingUserActivity from weko_accounts import WekoAccounts, WekoAccountsREST +from weko_accounts.config import WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT from weko_accounts.views import blueprint + +@pytest.fixture() +def map_role_names(): + """Names used to test GakuNin-managed roles.""" + pattern = WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT + prefix = pattern["prefix"] + role_key = pattern["role_keyword"] + group_key = pattern["group_keyword"] + repoid = "test_example_com" + + return { + "managed_role": f"{prefix}_{repoid}_{role_key}_radm", + "sysadm_role": pattern["sysadm_group"], + "unknown_role_suffix": f"{prefix}_{repoid}_{role_key}_unkwown", + "group_format_with_role_suffix": f"{prefix}_{repoid}_{group_key}_radm", + "invalid_repository_id_role": ( + f"{prefix}_test!example!com_{role_key}_radm" + ), + "invalid_prefix_role": f"ng_{repoid}_{role_key}_radm", + } + + +@pytest.fixture() +def map_group_names(): + """Names used to test GakuNin-managed groups.""" + pattern = WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT + prefix = pattern["prefix"] + group_key = pattern["group_keyword"] + repoid = "test_example_com" + + return { + "managed_group": f"{prefix}_{repoid}_{group_key}_library", + "invalid_repository_id_group": ( + f"{prefix}_test!example!com_{group_key}_library" + ), + "invalid_prefix_group": f"ng_{repoid}_{group_key}_library", + } + + @pytest.yield_fixture() def instance_path(): """Temporary instance path.""" @@ -91,6 +131,7 @@ def base_app(instance_path): 'HTTP_WEKOID': (False, 'shib_user_name'), }, WEKO_ACCOUNTS_SHIB_IDP_LOGIN_URL='{}secure/login.py', + WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT = WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT ) Babel(app_) InvenioI18N(app_) @@ -351,7 +392,7 @@ def weko_roles(app): Args: app (Flask): Flask application. - + Returns: dict: Dictionary of roles. """ diff --git a/modules/weko-accounts/tests/test_api.py b/modules/weko-accounts/tests/test_api.py index 832bef1095..97bd5e07d9 100644 --- a/modules/weko-accounts/tests/test_api.py +++ b/modules/weko-accounts/tests/test_api.py @@ -18,7 +18,14 @@ update_contribute_role, remove_contribute_role, bind_roles_to_indices, - create_fqdn_from_entity_id + create_fqdn_from_entity_id, + map_role_condition, + map_group_condition, + is_map_managed_name, + is_map_role, + is_map_group, + is_map_sysadm_role, + _is_gakunin_map_configured, ) from invenio_db import db as db_ from invenio_db import InvenioDB @@ -1041,10 +1048,14 @@ def test_sync_shib_gakunin_map_groups_unexpected_error(app, client): # .tox/c1/bin/pytest --cov=weko_accounts tests/test_api.py::test_update_roles -vv -s --cov-branch --cov-report=html --basetemp=/code/modules/weko-accounts/.tox/c1/tmp def test_update_roles(app, db, mocker): with app.app_context(): + mocker.patch.dict(current_app.config, { + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib' + }) mock_bind = mocker.patch('weko_accounts.api.bind_roles_to_indices') # テストデータの準備 + map_group_name = 'jc_test_example_com_gr_4' map_group_list = ['group1', 'group2', 'group3',''] - existing_role_names = {'group1', 'jc_group4'} + existing_role_names = {'group1', map_group_name} # 既存のロールを追加 existing_roles = [] @@ -1063,11 +1074,11 @@ def test_update_roles(app, db, mocker): assert 'group1' in role_names assert 'group2' in role_names # 新しいロールが追加されていることを確認 assert 'group3' in role_names # 新しいロールが追加されていることを確認 - assert 'jc_group4' not in role_names # 既存のロールが削除されていることを確認 + assert map_group_name not in role_names # 既存のロールが削除されていることを確認 assert '' not in role_names # 空のロールが追加されていないことを確認 new_roles = [r for r in roles if r.name in ['group2', 'group3']] - remove_role_ids = [r.id for r in existing_roles if r.name == 'jc_group4'] + remove_role_ids = [r.id for r in existing_roles if r.name == map_group_name] mock_bind.assert_called_once_with([], new_roles, remove_role_ids) existing_roles = Role.query.all() @@ -1319,3 +1330,222 @@ def test_create_fqdn_from_entity_id(app): app.config['WEKO_ACCOUNTS_IDP_ENTITY_ID'] = None with pytest.raises(KeyError, match='WEKO_ACCOUNTS_IDP_ENTITY_ID is missing in config'): create_fqdn_from_entity_id() + + +# .tox/c1/bin/pytest --cov=weko_accounts tests/test_api.py::test_map_role_condition -vv -s --cov-branch --cov-report=html --basetemp=/code/modules/weko-accounts/.tox/c1/tmp +def test_map_role_condition(app, db, map_role_names): + config = { + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib', + } + + role_names = list(map_role_names.values()) + expected_names = { + map_role_names["managed_role"], + map_role_names["sysadm_role"], + map_role_names["unknown_role_suffix"], + } + + with app.app_context(), patch.dict(app.config, config): + db.session.add_all([Role(name=name) for name in role_names]) + db.session.commit() + matched_names = { + role.name for role in Role.query.filter( + Role.name.in_(role_names), map_role_condition()).all() + } + assert matched_names == expected_names + + non_map_role_names = { + role.name for role in Role.query.filter( + Role.name.in_(role_names), ~map_role_condition()).all() + } + assert non_map_role_names == set(role_names) - expected_names + + +# .tox/c1/bin/pytest --cov=weko_accounts tests/test_api.py::test_map_role_condition_without_gakunin_map_config -vv -s --cov-branch --cov-report=html --basetemp=/code/modules/weko-accounts/.tox/c1/tmp +def test_map_role_condition_without_gakunin_map_config( + app, db, map_role_names): + config = {'WEKO_ACCOUNTS_IDP_ENTITY_ID': ''} + + role_names = list(map_role_names.values()) + + with app.app_context(), patch.dict(app.config, config): + db.session.add_all([Role(name=name) for name in role_names]) + db.session.commit() + matched_roles = Role.query.filter(map_role_condition()).all() + assert matched_roles == [] + + non_map_role_names = { + role.name for role in Role.query.filter( + Role.name.in_(role_names), ~map_role_condition()).all() + } + assert non_map_role_names == set(role_names) + + +# .tox/c1/bin/pytest --cov=weko_accounts tests/test_api.py::test_map_group_condition -vv -s --cov-branch --cov-report=html --basetemp=/code/modules/weko-accounts/.tox/c1/tmp +def test_map_group_condition(app, db, map_group_names): + config = { + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib', + } + + role_names = list(map_group_names.values()) + expected_names = {map_group_names["managed_group"]} + + with app.app_context(), patch.dict(app.config, config): + db.session.add_all([Role(name=name) for name in role_names]) + db.session.commit() + matched_names = { + role.name for role in Role.query.filter( + Role.name.in_(role_names), map_group_condition()).all() + } + assert matched_names == expected_names + + non_map_group_names = { + role.name for role in Role.query.filter( + Role.name.in_(role_names), ~map_group_condition()).all() + } + assert non_map_group_names == set(role_names) - expected_names + + +# .tox/c1/bin/pytest --cov=weko_accounts tests/test_api.py::test_map_group_condition_without_gakunin_map_config -vv -s --cov-branch --cov-report=html --basetemp=/code/modules/weko-accounts/.tox/c1/tmp +def test_map_group_condition_without_gakunin_map_config( + app, db, map_group_names): + config = {'WEKO_ACCOUNTS_IDP_ENTITY_ID': ''} + + role_names = list(map_group_names.values()) + + with app.app_context(), patch.dict(app.config, config): + db.session.add_all([Role(name=name) for name in role_names]) + db.session.commit() + matched_roles = Role.query.filter(map_group_condition()).all() + assert matched_roles == [] + + non_map_group_names = { + role.name for role in Role.query.filter( + Role.name.in_(role_names), ~map_group_condition()).all() + } + assert non_map_group_names == set(role_names) + + +# .tox/c1/bin/pytest --cov=weko_accounts tests/test_api.py::test_is_map_managed_name -vv -s --cov-branch --cov-report=html --basetemp=/code/modules/weko-accounts/.tox/c1/tmp +def test_is_map_managed_name(app, map_role_names, map_group_names): + config = { + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib', + } + + with app.app_context(), patch.dict(app.config, config): + assert is_map_managed_name(map_role_names["managed_role"]) is True + assert is_map_managed_name(map_role_names["sysadm_role"]) is True + assert is_map_managed_name( + map_role_names["unknown_role_suffix"]) is True + assert is_map_managed_name( + map_role_names["group_format_with_role_suffix"]) is True + assert is_map_managed_name( + map_role_names["invalid_repository_id_role"]) is False + assert is_map_managed_name( + map_role_names["invalid_prefix_role"]) is False + assert is_map_managed_name(map_group_names["managed_group"]) is True + assert is_map_managed_name( + map_group_names["invalid_repository_id_group"]) is False + assert is_map_managed_name( + map_group_names["invalid_prefix_group"]) is False + + +# .tox/c1/bin/pytest --cov=weko_accounts tests/test_api.py::test_is_map_managed_name_no_repoid -vv -s --cov-branch --cov-report=html --basetemp=/code/modules/weko-accounts/.tox/c1/tmp +def test_is_map_managed_name_no_repoid( + app, map_role_names, map_group_names): + config = {'WEKO_ACCOUNTS_IDP_ENTITY_ID': ''} + with app.app_context(), patch.dict(app.config, config): + roles = (list(map_role_names.values()) + + list(map_group_names.values())) + map_roles = [role for role in roles if is_map_managed_name(role)] + assert map_roles == [] + + +# .tox/c1/bin/pytest --cov=weko_accounts tests/test_api.py::test_is_map_role -vv -s --cov-branch --cov-report=html --basetemp=/code/modules/weko-accounts/.tox/c1/tmp +def test_is_map_role(app, map_role_names): + config = { + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib', + } + with app.app_context(), patch.dict(app.config, config): + assert is_map_role(map_role_names["managed_role"]) is True + assert is_map_role(map_role_names["sysadm_role"]) is True + assert is_map_role(map_role_names["unknown_role_suffix"]) is True + assert is_map_role( + map_role_names["group_format_with_role_suffix"]) is False + assert is_map_role( + map_role_names["invalid_repository_id_role"]) is False + assert is_map_role(map_role_names["invalid_prefix_role"]) is False + + +# .tox/c1/bin/pytest --cov=weko_accounts tests/test_api.py::test_is_map_role_no_repoid -vv -s --cov-branch --cov-report=html --basetemp=/code/modules/weko-accounts/.tox/c1/tmp +def test_is_map_role_no_repoid(app, map_role_names): + config = {'WEKO_ACCOUNTS_IDP_ENTITY_ID': ''} + with app.app_context(), patch.dict(app.config, config): + roles = list(map_role_names.values()) + map_roles = [role for role in roles if is_map_role(role)] + assert map_roles == [] + + +# .tox/c1/bin/pytest --cov=weko_accounts tests/test_api.py::test_is_map_group -vv -s --cov-branch --cov-report=html --basetemp=/code/modules/weko-accounts/.tox/c1/tmp +def test_is_map_group(app, map_group_names): + config = { + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib', + } + with app.app_context(), patch.dict(app.config, config): + assert is_map_group(map_group_names["managed_group"]) is True + assert is_map_group( + map_group_names["invalid_repository_id_group"]) is False + assert is_map_group(map_group_names["invalid_prefix_group"]) is False + + +# .tox/c1/bin/pytest --cov=weko_accounts tests/test_api.py::test_is_map_group_no_repoid -vv -s --cov-branch --cov-report=html --basetemp=/code/modules/weko-accounts/.tox/c1/tmp +def test_is_map_group_no_repoid(app, map_group_names): + config = {'WEKO_ACCOUNTS_IDP_ENTITY_ID': ''} + with app.app_context(), patch.dict(app.config, config): + roles = list(map_group_names.values()) + map_groups = [role for role in roles if is_map_group(role)] + assert map_groups == [] + + +# .tox/c1/bin/pytest --cov=weko_accounts tests/test_api.py::test_is_map_sysadm_role -vv -s --cov-branch --cov-report=html --basetemp=/code/modules/weko-accounts/.tox/c1/tmp +def test_is_map_sysadm_role(app, map_role_names): + config = { + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib', + } + with app.app_context(), patch.dict(app.config, config): + assert is_map_sysadm_role(map_role_names["managed_role"]) is False + assert is_map_sysadm_role(map_role_names["sysadm_role"]) is True + assert is_map_sysadm_role(map_role_names["unknown_role_suffix"]) is False + assert is_map_sysadm_role( + map_role_names["group_format_with_role_suffix"]) is False + assert is_map_sysadm_role( + map_role_names["invalid_repository_id_role"]) is False + assert is_map_sysadm_role(map_role_names["invalid_prefix_role"]) is False + + +# .tox/c1/bin/pytest --cov=weko_accounts tests/test_api.py::test_is_map_sysadm_role_no_repoid -vv -s --cov-branch --cov-report=html --basetemp=/code/modules/weko-accounts/.tox/c1/tmp +def test_is_map_sysadm_role_no_repoid( + app, map_role_names): + config = {'WEKO_ACCOUNTS_IDP_ENTITY_ID': ''} + with app.app_context(), patch.dict(app.config, config): + roles = list(map_role_names.values()) + sysadm_roles = [role for role in roles if is_map_sysadm_role(role)] + assert sysadm_roles == [] + + +# .tox/c1/bin/pytest --cov=weko_accounts tests/test_api.py::test_is_gakunin_map_configured -vv -s --cov-branch --cov-report=html --basetemp=/code/modules/weko-accounts/.tox/c1/tmp +@pytest.mark.parametrize('config, expected', [ + ({ + 'WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT': {'prefix': 'jc'}, + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib', + }, True), + ({}, False), + ({'WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT': {'prefix': 'jc'}}, False), + ({'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib'}, False), + ({ + 'WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT': {'prefix': 'jc'}, + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': '',}, False), +]) +def test_is_gakunin_map_configured(app, config, expected): + with app.app_context(), patch.dict(app.config, config, clear=True): + assert _is_gakunin_map_configured() is expected diff --git a/modules/weko-accounts/tests/test_views.py b/modules/weko-accounts/tests/test_views.py index 658ba64681..c6f29a36f8 100644 --- a/modules/weko-accounts/tests/test_views.py +++ b/modules/weko-accounts/tests/test_views.py @@ -110,26 +110,26 @@ def test_redirect_method(app,mocker): ams_error = 'Login is blocked.' _redirect_method(True, ams_error) mock_render.assert_called_with(\ - 'http://TEST_SERVER.localdomain/ams/login?error=Login+is+blocked.') + 'http://test_server.localdomain/ams/login?error=Login+is+blocked.') # There is no user information. mock_render = mocker.patch('weko_accounts.views.redirect',return_value=make_response()) ams_error = 'There is no user information.' _redirect_method(True, ams_error) mock_render.assert_called_with( - 'http://TEST_SERVER.localdomain/ams/login?error=There+is+no+user+information.') + 'http://test_server.localdomain/ams/login?error=There+is+no+user+information.') # server error mock_render = mocker.patch('weko_accounts.views.redirect',return_value=make_response()) ams_error = 'Server error has occurred. Please contact server administrator.' _redirect_method(True, ams_error) mock_render.assert_called_with(\ - 'http://TEST_SERVER.localdomain/ams/login?'\ + 'http://test_server.localdomain/ams/login?'\ 'error=Server+error+has+occurred.+Please+contact+server+administrator.') # other error mock_render = mocker.patch('weko_accounts.views.redirect',return_value=make_response()) ams_error = 'Error Message' _redirect_method(True, ams_error) mock_render.assert_called_with(\ - 'http://TEST_SERVER.localdomain/ams/login?error=Error+Message') + 'http://test_server.localdomain/ams/login?error=Error+Message') # .tox/c1/bin/pytest --cov=weko_accounts tests/test_views.py::test_generate_ams_login_url -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-workflow/.tox/c1/tmp def test_generate_ams_login_url(app): diff --git a/modules/weko-accounts/weko_accounts/api.py b/modules/weko-accounts/weko_accounts/api.py index 324d4c1826..081ba676e1 100644 --- a/modules/weko-accounts/weko_accounts/api.py +++ b/modules/weko-accounts/weko_accounts/api.py @@ -30,6 +30,7 @@ import urllib.parse import requests +from sqlalchemy import or_, false _datastore = LocalProxy(lambda: current_app.extensions['security'].datastore) @@ -567,9 +568,10 @@ def update_roles(map_group_list, roles, indices=[]): role_names = set({role.name for role in roles}) new_role_names = [role_name for role_name in map_group_list if role_name and role_name not in role_names] - roles_to_remove = [role_name for role_name in role_names if role_name not in map_group_list and role_name.startswith('jc_')] + roles_to_remove = [role_name for role_name in role_names + if role_name not in map_group_list + and is_map_managed_name(role_name)] - new_roles = [] remove_role_ids = [] try: @@ -608,7 +610,7 @@ def bind_roles_to_indices(indices=[], new_roles=[], remove_role_ids=[]): for role_id in remove_role_ids: if role_id in browsing_roles: browsing_roles.remove(role_id) - + # bind new roles browsing_default_permission = current_app.config.get('WEKO_INDEXTREE_GAKUNIN_GROUP_DEFAULT_BROWSING_PERMISSION', False) for role in new_roles: @@ -682,7 +684,7 @@ def remove_contribute_role(self, role_id): def create_fqdn_from_entity_id(): """Create a fully qualified domain name (FQDN) from the entity ID. - + Returns: str: FQDN derived from the entity ID. """ @@ -693,3 +695,136 @@ def create_fqdn_from_entity_id(): parsed_url = urlparse(idp_entity_id) fqdn = parsed_url.netloc.split(":")[0].replace('.', '_').replace('-', '_') return fqdn + +def map_role_condition(): + """Return a SQLAlchemy condition matching GakuNin mAP roles. + + Matches the configured system administrator and roles whose names + begin with the configured repository-specific role prefix. Returns a + false SQL expression when the required settings are absent. + + Returns: + sqlalchemy.sql.elements.ColumnElement: Condition for filtering roles. + """ + if not _is_gakunin_map_configured(): + return false() + pattern = current_app.config.get('WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT') + sysadm_key = pattern.get("sysadm_group") + prefix = pattern.get("prefix") + role_key = pattern.get("role_keyword") + repo_id = create_fqdn_from_entity_id() + role_pattern = f"{prefix}_{repo_id}_{role_key}_" + return or_( + Role.name == sysadm_key, + Role.name.startswith(role_pattern, autoescape=True) + ) + +def map_group_condition(): + """Return a SQLAlchemy condition matching GakuNin mAP groups. + + Matches roles whose names begin with the configured repository-specific + group prefix. Returns a false SQL expression when required settings are + absent. + + Returns: + sqlalchemy.sql.elements.ColumnElement: Condition for filtering roles. + """ + if not _is_gakunin_map_configured(): + return false() + pattern = current_app.config.get('WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT') + prefix = pattern.get("prefix") + group_key = pattern.get("group_keyword") + repo_id = create_fqdn_from_entity_id() + group_pattern = f"{prefix}_{repo_id}_{group_key}_" + return Role.name.startswith(group_pattern, autoescape=True) + +def is_map_managed_name(role_name): + """Return whether a role name belongs to the configured GakuNin mAP. + + A matching name is the configured system administrator group or has the + repository-specific GakuNin mAP prefix. Non-string values and missing + settings return ``False``. + + Args: + role_name(str): Role name to evaluate. + + Returns: + bool: ``True`` when the role is a GakuNin mAP role or group. + """ + if not isinstance(role_name, str) or not _is_gakunin_map_configured(): + return False + pattern = current_app.config.get('WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT') + prefix = pattern.get("prefix") + repo_id = create_fqdn_from_entity_id() + repoid_pattern = f"{prefix}_{repo_id}_" + return is_map_sysadm_role(role_name) or role_name.startswith(repoid_pattern) + +def is_map_role(role_name): + """Return whether a role name is a configured GakuNin mAP role. + + The configured system administrator group is also treated as a role. + Non-string values and missing required settings return ``False``. + + Args: + role_name(str): Role name to evaluate. + + Returns: + bool: ``True`` when the role is a GakuNin mAP role. + """ + if not isinstance(role_name, str) or not _is_gakunin_map_configured(): + return False + pattern = current_app.config.get('WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT') + prefix = pattern.get("prefix") + role_key = pattern.get("role_keyword") + repo_id = create_fqdn_from_entity_id() + role_pattern = f"{prefix}_{repo_id}_{role_key}_" + return is_map_sysadm_role(role_name) or role_name.startswith(role_pattern) + +def is_map_group(role_name): + """Return whether a role name is a configured GakuNin mAP group. + + Non-string values and missing required settings return ``False``. + + Args: + role_name(str): Role name to evaluate. + + Returns: + bool: ``True`` when the role is a GakuNin mAP group. + """ + if not isinstance(role_name, str) or not _is_gakunin_map_configured(): + return False + pattern = current_app.config.get('WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT') + prefix = pattern.get("prefix") + group_key = pattern.get("group_keyword") + repo_id = create_fqdn_from_entity_id() + group_pattern = f"{prefix}_{repo_id}_{group_key}_" + return role_name.startswith(group_pattern) + +def is_map_sysadm_role(role_name): + """Return whether a role name is the configured system administrator group. + + Non-string values and missing required settings return ``False``. + + Args: + role_name(str): Role name to evaluate. + + Returns: + bool: ``True`` when the role is the system administrator group. + """ + if not isinstance(role_name, str) or not _is_gakunin_map_configured(): + return False + pattern = current_app.config.get('WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT') + sysadm = pattern.get("sysadm_group") + return role_name == sysadm + +def _is_gakunin_map_configured(): + """Return whether the required GakuNin mAP settings are configured. + + Both the group-pattern dictionary and IdP entity ID must be truthy. + + Returns: + bool: ``True`` when both required settings are configured. + """ + pattern = current_app.config.get('WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT') + idp_entity_id = current_app.config.get('WEKO_ACCOUNTS_IDP_ENTITY_ID') + return bool(pattern and idp_entity_id) diff --git a/modules/weko-accounts/weko_accounts/config.py b/modules/weko-accounts/weko_accounts/config.py index 4ced5666e3..673f2e1ee3 100644 --- a/modules/weko-accounts/weko_accounts/config.py +++ b/modules/weko-accounts/weko_accounts/config.py @@ -233,8 +233,9 @@ "role_mapping":{ "radm":"Repository Administrator", "cadm":"Community Administrator", - "cont":"Contributor", - } + "cont":"Contributor" + }, + "group_keyword":"gr" } """Patterns for Gakunin group names.""" diff --git a/modules/weko-index-tree/tests/test_api.py b/modules/weko-index-tree/tests/test_api.py index 7b0e4409eb..165a608e75 100644 --- a/modules/weko-index-tree/tests/test_api.py +++ b/modules/weko-index-tree/tests/test_api.py @@ -37,6 +37,7 @@ from invenio_i18n.ext import current_i18n from sqlalchemy.exc import IntegrityError, SQLAlchemyError from flask import current_app +from weko_accounts.api import create_fqdn_from_entity_id from weko_deposit.api import WekoDeposit from weko_index_tree.api import Indexes from weko_index_tree.models import Index @@ -621,26 +622,35 @@ def test_update_set_info(i18n_app, db, users, test_indices): def test_filter_roles(app, mocker): with app.app_context(): # モックデータの準備 + prefix = 'group' + role_key = 'key' roles = [ {"id": 1, "name": "Contributor"}, {"id": 2, "name": "Community Administrator"}, {"id": 3, "name": "Repository Administrator"}, {"id": 4, "name": "System Administrator"}, - {"id": 5, "name": "group_key_test_role"}, - {"id": 6, "name": "key_value_role"}, + {"id": 5, "name": f"{prefix}_{role_key}_test_role"}, + {"id": 6, "name": f"{role_key}_value_role"}, {"id": 7, "name": "Authenticated User"}, {"id": 8, "name": "Guest"}, {"id": 9, "name": "General"}, - {"id": 10, "name": "group_value_role"}, + {"id": 10, "name": f"{prefix}_test_example_com_gr_xxx"}, ] # 設定をモック mocker.patch.dict(current_app.config, { 'WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT': { - 'prefix': 'group', - 'role_keyword': 'key' + 'prefix': prefix, + 'role_keyword': role_key, + 'group_keyword': 'gr', + "role_mapping":{ + "radm": "Repository Administrator", + "cadm": "Community Administrator", + "cont": "Contributor" + } }, - 'WEKO_PERMISSION_ROLE_USER': ['Contributor', 'Community Administrator', 'Repository Administrator', 'System Administrator', 'General', 'Guest', 'Authenticated User'] + 'WEKO_PERMISSION_ROLE_USER': ['Contributor', 'Community Administrator', 'Repository Administrator', 'System Administrator', 'General', 'Guest', 'Authenticated User'], + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib' }) # メソッドの呼び出し @@ -648,17 +658,18 @@ def test_filter_roles(app, mocker): # 結果の検証 assert len(filtered_roles) == 1 - assert filtered_roles[0]["name"] == "group_value_role" + assert filtered_roles[0]["name"] == f"{prefix}_test_example_com_gr_xxx" assert len(excluded_roles) == 9 - assert excluded_roles[0]["name"] == "Contributor" - assert excluded_roles[1]["name"] == "Community Administrator" - assert excluded_roles[2]["name"] == "Repository Administrator" - assert excluded_roles[3]["name"] == "System Administrator" - assert excluded_roles[4]["name"] == "key_value_role" - assert excluded_roles[5]["name"] == "Authenticated User" - assert excluded_roles[6]["name"] == "Guest" - assert excluded_roles[7]["name"] == "General" - assert excluded_roles[8]["name"] == "group_key_test_role" + excluded_role_names = [role["name"] for role in excluded_roles] + assert "Contributor" in excluded_role_names + assert "Community Administrator" in excluded_role_names + assert "Repository Administrator" in excluded_role_names + assert "System Administrator" in excluded_role_names + assert f"{role_key}_value_role" in excluded_role_names + assert "Authenticated User" in excluded_role_names + assert "Guest" in excluded_role_names + assert "General" in excluded_role_names + assert f"{prefix}_{role_key}_test_role" in excluded_role_names # リスト以外の値を渡すテストケース non_list_value = "not_a_list" @@ -675,6 +686,48 @@ def test_filter_roles(app, mocker): with pytest.raises(TypeError, match="roles must be a list"): Indexes.filter_roles(dict_value) + +# .tox/c1/bin/pytest --cov=weko_index_tree tests/test_api.py::test_filter_roles_map -v -s -vv --cov-branch --cov-report=html --cov-config=tox.ini --basetemp=/code/modules/weko-index-tree/.tox/c1/tmp +def test_filter_roles_map(app, mocker): + with app.app_context(): + 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" + + roles = [ + {"id": 1, "name": role_name1}, + {"id": 2, "name": role_name2}, + {"id": 3, "name": role_name3}, + {"id": 4, "name": role_name4}, + {"id": 5, "name": role_name5}, + {"id": 6, "name": role_name6} + ] + filtered_roles, excluded_roles = Indexes.filter_roles(roles) + + assert len(filtered_roles) == 1 + assert filtered_roles[0]["name"] == role_name4 + + assert len(excluded_roles) == 5 + excluded_role_names = [role["name"] for role in excluded_roles] + assert role_name1 in excluded_role_names + assert role_name2 in excluded_role_names + assert role_name3 in excluded_role_names + assert role_name5 in excluded_role_names + assert role_name6 in excluded_role_names + + # def delete_set_info(cls, action, index_id, id_list): # def get_public_indexes_list(cls): # .tox/c1/bin/pytest --cov=weko_index_tree tests/test_api.py::test_Indexes_get_public_indexes_list -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/weko-index-tree/.tox/c1/tmp @@ -1163,13 +1216,16 @@ def test_indexes_get_index_tree(i18n_app, db, redis_connect, users, db_records, def test_get_index_with_role_group(app, db, mocker): with app.app_context(): # 必要な設定を追加 + prefix = 'group' + role_key = 'key' mocker.patch.dict(current_app.config, { 'WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT': { - 'prefix': 'group', - 'role_keyword': 'key' + 'prefix': prefix, + 'role_keyword': role_key }, 'WEKO_PERMISSION_ROLE_USER': ['Contributor', 'Community Administrator', 'Repository Administrator', 'System Administrator', 'General', 'Guest', 'Authenticated User'], - 'WEKO_PERMISSION_SUPER_ROLE_USER': ['System Administrator', 'Repository Administrator'] + 'WEKO_PERMISSION_SUPER_ROLE_USER': ['System Administrator', 'Repository Administrator'], + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib' }) # モックデータの準備 @@ -1183,16 +1239,18 @@ def test_get_index_with_role_group(app, db, mocker): } mocker.patch.object(Indexes, 'get_index', return_value=index_data) + fqdn = create_fqdn_from_entity_id() + roles = [ {"id": 3, "name": "Contributor"}, {"id": 4, "name": "Community Administrator"}, {"id": -98, "name": "Authenticated User"}, {"id": -99, "name": "Guest"}, {"id": 5, "name": "General"}, - {"id": 6, "name": "group_key_test_role"}, - {"id": 7, "name": "key_value_role"}, - {"id": 8, "name": "group_value_role"}, - {"id": 9, "name": "group_xxx_key_user1"}, + {"id": 6, "name": f"{prefix}_{role_key}_test_role"}, + {"id": 7, "name": f"{role_key}_value_role"}, + {"id": 8, "name": f"{prefix}_{fqdn}_value_role"}, + {"id": 9, "name": f"{prefix}_test_example_com_{role_key}_user1"}, {"id": 10, "name": "System Administrator"}, ] mocker.patch.object(Indexes, 'get_account_role', return_value=roles) @@ -1209,9 +1267,9 @@ def test_get_index_with_role_group(app, db, mocker): # 結果の検証 assert result['browsing_group']['allow'] == [] - assert result['browsing_group']['deny'] == [{'id': '-89', 'name': 'No Group'}, {'id': '8gr', 'name': 'group_value_role'}] + assert result['browsing_group']['deny'] == [{'id': '-89', 'name': 'No Group'}, {'id': '8gr', 'name': f'{prefix}_{fqdn}_value_role'}] assert result['contribute_group']['allow'] == [] - assert result['contribute_group']['deny'] == [{'id': '-89', 'name': 'No Group'}, {'id': '8gr', 'name': 'group_value_role'}] + assert result['contribute_group']['deny'] == [{'id': '-89', 'name': 'No Group'}, {'id': '8gr', 'name': f'{prefix}_{fqdn}_value_role'}] # 結果が空の場合のテストケース mocker.patch.object(Indexes, 'get_account_role', return_value=[]) @@ -1233,25 +1291,67 @@ def test_indexes_get_handle_index_url(app, db, users, test_indices, mocker): assert index_url == "http://TEST_SERVER/search?search_type=2&q=1" assert handle == "https://test/handle/1" + +# .tox/c1/bin/pytest --cov=weko_index_tree tests/test_api.py::test_bind_roles_including_permission -v -s -vv --cov-branch --cov-report=html --cov-config=tox.ini --basetemp=/code/modules/weko-index-tree/.tox/c1/tmp +def test_bind_roles_including_permission(app, mocker): + with app.test_request_context(): + prefix = 'jc' + sysadm_role = "jc_roles_sysadm" + mocker.patch.dict(current_app.config, { + 'WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT': { + 'prefix': prefix, + "sysadm_group": sysadm_role + }, + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib' + }) + fqdn = create_fqdn_from_entity_id() + roles = [ + {"id": 1, "name": sysadm_role}, + {"id": 2, "name": f"{prefix}_{fqdn}_ro_radm"}, + {"id": 3, "name": f"{prefix}_{fqdn}_gr_xxxx"}, + {"id": 4, "name": f"{prefix}_test!example!com_ro_radm"}, + {"id": 5, "name": f"ng_{fqdn}_ro_radm"}, + {"id": 6, "name": "System Administrator"}, + {"id": 7, "name": "Repository Administrator"} + ] + roles_true = Indexes.bind_roles_including_permission(roles, True) + assert roles_true == roles + roles_false = Indexes.bind_roles_including_permission(roles, False) + assert roles_false == [ + {"id": 4, "name": f"{prefix}_test!example!com_ro_radm"}, + {"id": 5, "name": f"ng_{fqdn}_ro_radm"}, + {"id": 6, "name": "System Administrator"}, + {"id": 7, "name": "Repository Administrator"} + ] + roles_empty_true = Indexes.bind_roles_including_permission([], True) + assert roles_empty_true == [] + roles_empty_false = Indexes.bind_roles_including_permission([], False) + assert roles_empty_false == [] + + # .tox/c1/bin/pytest --cov=weko_index_tree tests/test_api.py::test_get_allow_deny -v -s -vv --cov-branch --cov-report=html --cov-config=tox.ini --basetemp=/code/modules/weko-index-tree/.tox/c1/tmp def test_get_allow_deny(app, db, mocker): with app.app_context(): + prefix = 'jc' + role_key = 'ro' mocker.patch.dict(current_app.config, { 'WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT': { - 'prefix': 'jc', - 'role_keyword': 'roles' + 'prefix': prefix, + 'role_keyword': role_key }, - 'WEKO_PERMISSION_SUPER_ROLE_USER': ['System Administrator', 'Repository Administrator'] + 'WEKO_PERMISSION_SUPER_ROLE_USER': ['System Administrator', 'Repository Administrator'], + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib' }) - # Patterns for super role, role_keyword, prefix, allow/deny + fqdn = create_fqdn_from_entity_id() + # Patterns for super role, role_keyword, prefix, allow/deny roles = [ {"id": 1, "name": "NormalRole"}, # subject to allow/deny {"id": 2, "name": "roles_test"}, # subject to allow/deny - {"id": 3, "name": "jcAdmin"}, # starts with 'jc' and does not contain 'roles' → excluded by filter_roles + {"id": 3, "name": f"{prefix}_{fqdn}_Admin"}, # starts with 'jc' and does not contain 'roles' → excluded by filter_roles {"id": 4, "name": "System Administrator"}, # super role → skip {"id": 5, "name": "OtherRole"}, # subject to allow/deny - {"id": 6, "name": "jc_test_roles"}, # contains prefix and role_keyword → skip + {"id": 6, "name": f"{prefix}_test_example_com_{role_key}"}, # contains prefix and role_keyword → skip ] index_data = { 'id': 1, diff --git a/modules/weko-index-tree/tests/test_utils.py b/modules/weko-index-tree/tests/test_utils.py index dff408656f..49b62560fc 100644 --- a/modules/weko-index-tree/tests/test_utils.py +++ b/modules/weko-index-tree/tests/test_utils.py @@ -49,6 +49,7 @@ get_descendant_index_names, get_item_ids_in_index, get_all_records_in_index, + get_user_roles_and_groups ) from invenio_accounts.testutils import login_user_via_session, client_authenticated @@ -70,6 +71,7 @@ from flask_babelex import gettext as _ from flask_babelex import to_user_timezone, to_utc from flask_login import current_user, login_user, LoginManager +from invenio_accounts.models import Role from invenio_cache import current_cache from invenio_communities.models import Community from invenio_i18n.ext import current_i18n @@ -216,29 +218,24 @@ def test_get_user_groups(i18n_app, client_rest, users, db): assert len(get_user_groups()) == 0 -# .tox/c1/bin/pytest --cov=weko_index_tree tests/test_utils.py::test_check_index_permission_by_role_and_group -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/weko-index-tree/.tox/c1/tmp # def check_index_permission_by_role_and_group(user_role, roles, user_group, groups): -def test_check_index_permission_by_role_and_group(app, mocker): - mock_role1 = mocker.Mock() - mock_role1.id = 1 - mock_role1.name = "role1" - mock_role2 = mocker.Mock() - mock_role2.id = 2 - mock_role2.name = "jc_groups_xxx" # role_group - mock_roles = [mock_role1, mock_role2] - - mock_filter = mocker.Mock() - mock_filter.all.return_value = mock_roles - mock_query = mocker.Mock() - mock_query.filter.return_value = mock_filter - mocker.patch("weko_index_tree.utils.Role.query", mock_query) +# .tox/c1/bin/pytest --cov=weko_index_tree tests/test_utils.py::test_check_index_permission_by_role_and_group -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/weko-index-tree/.tox/c1/tmp +def test_check_index_permission_by_role_and_group(app, db, mocker): + mocker.patch.dict(current_app.config, { + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib' + }) + + role1 = Role(name="role1") + role2 = Role(name="jc_test_example_com_gr_xxx") + db.session.add_all([role1, role2]) + db.session.commit() # Admin User assert check_index_permission_by_role_and_group((True, []), '', [], '') is True mock_check_roles = mocker.patch("weko_index_tree.utils.check_roles") mock_check_groups = mocker.patch("weko_index_tree.utils.check_groups") - # mock query returns [1,2] → role id:3 is removed + # query returns [1,2] → role id:3 is removed # role id:2 is role_group → called in check_groups check_index_permission_by_role_and_group((False, [1,2,3]), '1,2,3', [5,6], '5,6') mock_check_roles.assert_called_with(['1'], ['1']) @@ -260,8 +257,66 @@ def test_check_index_permission_by_role_and_group(app, mocker): mock_check_groups = mocker.patch("weko_index_tree.utils.check_groups", return_value=False) assert check_index_permission_by_role_and_group((False, []), '', [], '') is False -# .tox/c1/bin/pytest --cov=weko_index_tree tests/test_utils.py::test_check_roles -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/weko-index-tree/.tox/c1/tmp + +# .tox/c1/bin/pytest --cov=weko_index_tree tests/test_utils.py::test_get_user_roles_and_groups -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/weko-index-tree/.tox/c1/tmp +def test_get_user_roles_and_groups(app, db, mocker): + mocker.patch.dict(current_app.config, { + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib' + }) + + with app.app_context(): + pattern = current_app.config.get( + 'WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT') + prefix = pattern.get("prefix") + role_key = pattern.get("role_keyword") + group_key = pattern.get("group_keyword") + repoid = "test_example_com" + + roles = [ + Role(id=1, name=f"{prefix}_{repoid}_{role_key}_radm"), # map role + Role(id=2, name=pattern.get("sysadm_group")), # map role + Role(id=3, name=f"{prefix}_{repoid}_{role_key}_unkwown"), # map role + Role(id=4, name=f"{prefix}_{repoid}_{group_key}_radm"), # role-group + Role(id=5, name=f"{prefix}_test!example!com_{role_key}_radm"), # normal role + Role(id=6, name=f"ng_{repoid}_{role_key}_radm"), # normal role + + Role(id=7, name=f"{prefix}_{repoid}_{group_key}_library"), # role-group + Role(id=8, name=f"{prefix}_test!example!com_{group_key}_library"), # normal role + Role(id=9, name=f"ng_{repoid}_{group_key}_library") # normal role + ] + + role_ids = [str(role.id)for role in roles] + role_ids.extend(['-98', '-99']) + + role_without_map_roles, role_groups = get_user_roles_and_groups(role_ids) + get_role_ids = [role for role in role_without_map_roles] + + assert len(get_role_ids) == 2 + assert '-98' in get_role_ids + assert '-99' in role_ids + role_group_ids = role_groups + assert len(role_group_ids) == 0 + + db.session.add_all(roles) + db.session.commit() + + role_without_map_roles, role_groups = get_user_roles_and_groups(role_ids) + get_role_ids = [role for role in role_without_map_roles] # normal role ids + assert len(get_role_ids) == 6 + assert '5' in get_role_ids + assert '6' in get_role_ids + assert '8' in get_role_ids + assert '9' in get_role_ids + assert '-98' in get_role_ids + assert '-99' in get_role_ids + role_group_ids = role_groups # role-group ids + assert len(role_group_ids) == 2 + assert '4' in role_group_ids + assert '7' in role_group_ids + + # def check_roles(user_role_list, index_role_list): +# .tox/c1/bin/pytest --cov=weko_index_tree tests/test_utils.py::test_check_roles -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/weko-index-tree/.tox/c1/tmp def test_check_roles(app, mocker): mock_user = mocker.Mock() mock_user.is_authenticated = True @@ -278,8 +333,8 @@ def test_check_roles(app, mocker): assert check_roles([], ['-99']) is True -# .tox/c1/bin/pytest --cov=weko_index_tree tests/test_utils.py::test_check_groups -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/weko-index-tree/.tox/c1/tmp # def check_groups(user_group, index_group_list, user_role_group, index_role_group): +# .tox/c1/bin/pytest --cov=weko_index_tree tests/test_utils.py::test_check_groups -v -s -vv --cov-branch --cov-report=term --cov-config=tox.ini --basetemp=/code/modules/weko-index-tree/.tox/c1/tmp def test_check_groups(app, mocker): mock_user = mocker.Mock() diff --git a/modules/weko-index-tree/weko_index_tree/api.py b/modules/weko-index-tree/weko_index_tree/api.py index d46fcfa528..1bce772d4e 100644 --- a/modules/weko-index-tree/weko_index_tree/api.py +++ b/modules/weko-index-tree/weko_index_tree/api.py @@ -42,6 +42,7 @@ from invenio_i18n.ext import current_i18n from invenio_indexer.api import RecordIndexer +from weko_accounts.api import is_map_managed_name, is_map_role from weko_groups.api import Group from weko_redis.redis import RedisConnection from weko_handle.api import Handle @@ -854,18 +855,10 @@ def filter_roles(cls, roles): gakunin_map_general_groups = [] other_roles = [] - gakunin_map_pattern = current_app.config.get( - "WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT", {} - ) - - gakunin_map_prefix = gakunin_map_pattern.get("prefix", "jc") - role_keyword = gakunin_map_pattern.get("role_keyword", "ro") - for role in roles: role_name = role.get("name", "") - role_info = {"id": role.get("id"), "name": role_name} - if role_name.startswith(gakunin_map_prefix): - if f"_{role_keyword}_" in role_name: + if is_map_managed_name(role_name): + if is_map_role(role_name): # gakunin_map group role gakunin_map_role_groups.append(role) else: @@ -886,9 +879,7 @@ def _get_allow_deny(allow, role, browse_flag=False): while role: tmp = role.pop(0) if tmp["name"] not in current_app.config['WEKO_PERMISSION_SUPER_ROLE_USER']: - role_key = current_app.config["WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT"]["role_keyword"] - prefix = current_app.config["WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT"]["prefix"] - if role_key not in tmp["name"] or not (tmp["name"].startswith(prefix)): + if not is_map_role(tmp["name"]): if str(tmp["id"]) in allow: alw.append(tmp) else: @@ -2224,7 +2215,7 @@ def bind_roles_including_permission(cls, roles, permission): """ bind_roles = [] for role in roles: - if role.get('name').startswith('jc_') and not permission: + if is_map_managed_name(role.get('name')) and not permission: continue bind_roles.append(role) return bind_roles diff --git a/modules/weko-index-tree/weko_index_tree/utils.py b/modules/weko-index-tree/weko_index_tree/utils.py index 7ec87ef752..4ebdfcd5cf 100644 --- a/modules/weko-index-tree/weko_index_tree/utils.py +++ b/modules/weko-index-tree/weko_index_tree/utils.py @@ -33,7 +33,7 @@ from flask_babelex import gettext as _ from flask_babelex import to_user_timezone, to_utc from flask_login import current_user -from sqlalchemy import and_ +from sqlalchemy import not_ from invenio_accounts.models import Role from invenio_cache import current_cache from invenio_communities.models import Community @@ -42,6 +42,7 @@ from invenio_pidstore.models import PersistentIdentifier from invenio_search import RecordsSearch from simplekv.memory.redisstore import RedisStore +from weko_accounts.api import map_role_condition, map_group_condition from weko_admin.utils import is_exists_key_in_redis from weko_groups.models import Group from weko_logging.activity_logger import UserActivityLogger @@ -334,38 +335,47 @@ def check_index_permission_by_role_and_group(user_role, roles, user_group, group user_roles = [str(r) for r in user_role[1]] if user_role[1] else [] user_group_list = [str(g) for g in user_group] - role_key = current_app.config["WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT"]["role_keyword"] - prefix = current_app.config["WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT"]["prefix"] - without_map_role = Role.query.filter( - ~and_(Role.name.like(f"%{role_key}%"), Role.name.startswith(prefix)) - ).all() - - role_groups = [str(lst.id) for lst in without_map_role - if "_groups_" in lst.name] - without_map_role = [str(r.id) for r in without_map_role] - # Guest, Authenticated User - without_map_role.extend(['-98', '-99']) - - user_role_list = [r for r in user_roles - if r in without_map_role and r not in role_groups] - user_role_group = [r for r in user_roles - if r in without_map_role and r in role_groups] - role_list = sorted(roles if isinstance(roles, list) else (roles.split(',') if roles else [])) index_group_list = sorted(groups if isinstance(groups, list) else (groups.split(',') if groups else [])) - index_role_list = [r for r in role_list - if r in without_map_role and r not in role_groups] - index_role_group = [r for r in role_list - if r in without_map_role and r in role_groups] + user_role_list, user_role_group = get_user_roles_and_groups(user_roles) + index_role_list, index_role_group = get_user_roles_and_groups(role_list) return check_roles(user_role_list, index_role_list) and \ check_groups(user_group_list, index_group_list, user_role_group, index_role_group) +def get_user_roles_and_groups(roles): + """Split role IDs into regular roles and role-groups. + + Args: + roles (list[str]): Role IDs to classify. + + Returns: + tuple[list[str], list[str]]: A pair containing regular role IDs and + role-group IDs, respectively. Guest and authenticated + user role IDs are treated as regular roles. + """ + role_without_map_role = Role.query.filter(not_(map_role_condition())).all() + role_groups = Role.query.filter(map_group_condition()).all() + + role_without_map_role = [str(role.id) for role in role_without_map_role] + # Guest, Authenticated User + role_without_map_role.extend(['-98', '-99']) + role_groups = [str(role.id) for role in role_groups] + role_list = [] + group_list = [] + for role in roles: + if role in role_without_map_role and role not in role_groups: + role_list.append(role) + if role in role_without_map_role and role in role_groups: + group_list.append(role) + return role_list, group_list + + def check_roles(user_role_list, index_role_list): """ Determine whether the user has access permission based on role IDs. diff --git a/modules/weko-workflow/tests/conftest.py b/modules/weko-workflow/tests/conftest.py index 207c3c5e99..ae1cc62e56 100644 --- a/modules/weko-workflow/tests/conftest.py +++ b/modules/weko-workflow/tests/conftest.py @@ -43,6 +43,7 @@ from invenio_oauth2server.views import settings_blueprint as oauth2server_settings_blueprint from invenio_records_ui import InvenioRecordsUI from invenio_rest import InvenioREST +from weko_accounts.config import WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT from weko_deposit.api import WekoIndexer, WekoRecord from weko_deposit.api import WekoDeposit as WekoDepositAPI from weko_search_ui.config import WEKO_SYS_USER @@ -235,10 +236,7 @@ def base_app(instance_path, search_class, cache_config): CACHE_REDIS_DB='0', CACHE_REDIS_HOST="redis", REDIS_PORT='6379', - WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT={ - 'role_keyword': 'roles', - 'prefix': 'jc' - }, + WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT = WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT, ACCOUNTS_SESSION_REDIS_DB_NO = 1, WEKO_RECORDS_UI_LICENSE_DICT=[ { @@ -1129,6 +1127,7 @@ def item_type_usage_report(db): db.session.add(item_type) + with db.session.begin_nested(): item_type_mapping = ItemTypeMapping( id=31003, item_type_id=31003, @@ -1231,6 +1230,8 @@ def db_itemtype(app, db): with db.session.begin_nested(): db.session.add(item_type_name) db.session.add(item_type) + + with db.session.begin_nested(): db.session.add(item_type_mapping) return {"item_type_name": item_type_name, "item_type": item_type, "item_type_mapping":item_type_mapping} @@ -1274,6 +1275,8 @@ def db_itemtype2(app, db): with db.session.begin_nested(): db.session.add(item_type_name) db.session.add(item_type) + + with db.session.begin_nested(): db.session.add(item_type_mapping) return {"item_type_name": item_type_name, "item_type": item_type, "item_type_mapping":item_type_mapping} @@ -1349,6 +1352,8 @@ def item_type(db): with db.session.begin_nested(): db.session.add(item_type_name_31001) db.session.add(item_type_31001) + + with db.session.begin_nested(): db.session.add(item_type_mapping_31001) item_types.append({"id": 31001, "obj": item_type_31001}) diff --git a/modules/weko-workflow/tests/test_admin.py b/modules/weko-workflow/tests/test_admin.py index a977930c7a..c9c06ad03b 100644 --- a/modules/weko-workflow/tests/test_admin.py +++ b/modules/weko-workflow/tests/test_admin.py @@ -6,6 +6,7 @@ import uuid from mock import patch,Mock from flask import json, url_for, make_response, current_app +from invenio_accounts.models import Role from invenio_accounts.testutils import login_user_via_session as login from werkzeug.exceptions import InternalServerError ,Forbidden from weko_workflow.admin import FlowSettingView,WorkFlowSettingView @@ -185,26 +186,15 @@ def test_flow_detail_update_delete(self,app,client,users,workflow ,workflow_open assert res.status_code == 404 # .tox/c1/bin/pytest --cov=weko_workflow tests/test_admin.py::TestFlowSettingView::test_flow_detail_roles_filter -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-workflow/.tox/c1/tmp - def test_flow_detail_roles_filter(self, client, db, users): - from invenio_accounts.models import Role - from invenio_accounts.testutils import login_user_via_session as login - - client.application.config.update(dict( - WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT={ - "prefix":"jc", - "role_keyword": "roles", - "role_mapping": { - "repoadm": "Repository Administrator", - "comadm": "Community Administrator", - "contributor": "Contributor", - } - } - )) + def test_flow_detail_roles_filter(self, client, db, users, mocker): + mocker.patch.dict(current_app.config, { + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib' + }) user = users[1]['obj'] db.session.add(user) role1 = Role(name="Contributor_test", description=None) - role2 = Role(name="jc_xxx_roles_contributor", description=None) - role3 = Role(name="jc_xxx_groups_yyy", description=None) + role2 = Role(name="jc_test_example_com_ro_contributor", description=None) + role3 = Role(name="jc_test_example_com_gr_yyy", description=None) db.session.add_all([role1, role2, role3]) db.session.commit() @@ -214,8 +204,8 @@ def test_flow_detail_roles_filter(self, client, db, users): args, kwargs = mock_render.call_args context = args[1] filtered_role_names = [role.name for role in context['roles']] - assert "jc_xxx_roles_contributor" not in filtered_role_names - assert "jc_xxx_groups_yyy" in filtered_role_names + assert "jc_test_example_com_ro_contributor" not in filtered_role_names + assert "jc_test_example_com_gr_yyy" in filtered_role_names assert "Contributor_test" in filtered_role_names # def get_specified_properties(): @@ -516,17 +506,15 @@ def test_index_acl(self,client,db_register2,users,users_index,status_code): # .tox/c1/bin/pytest --cov=weko_workflow tests/test_admin.py::TestWorkFlowSettingView::test_index_role_filtering -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-workflow/.tox/c1/tmp def test_index_role_filtering(self, client, db, app, mocker, users): - from invenio_accounts.models import Role role1 = Role(name="test_role", description=None) - role2 = Role(name="jc_xxx_roles_contributor", description=None) - role3 = Role(name="jc_xxx_groups_yyy", description=None) + role2 = Role(name="jc_test_example_com_ro_contributor", description=None) + role3 = Role(name="jc_test_example_com_gr_yyy", description=None) db.session.add_all([role1, role2, role3]) db.session.commit() - app.config["WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT"] = { - "role_keyword": "roles", - "prefix": "jc" - } + mocker.patch.dict(current_app.config, { + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib' + }) wf_mock = mocker.MagicMock() wf_mock.id = 1 wf_mock.index_tree_id = None @@ -543,9 +531,9 @@ def test_index_role_filtering(self, client, db, app, mocker, users): args, kwargs = mock_render.call_args context = args[1] display_names = context['workflows'][0].display.replace(',
', ',').split(',') - assert "jc_xxx_roles_contributor" not in display_names + assert "jc_test_example_com_ro_cont" not in display_names assert "test_role" in display_names - assert "jc_xxx_groups_yyy" in display_names + assert "jc_test_example_com_gr_yyy" in display_names # def workflow_detail(self, workflow_id='0'): # .tox/c1/bin/pytest --cov=weko_workflow tests/test_admin.py::TestWorkFlowSettingView::test_workflow_detail_acl_guest -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-workflow/.tox/c1/tmp @@ -789,27 +777,16 @@ def test_update_workflow(self,client,db,db_register2,users,workflow): ) # .tox/c1/bin/pytest --cov=weko_workflow tests/test_admin.py::TestWorkFlowSettingView::test_workflow_detail_roles_filter -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-workflow/.tox/c1/tmp - def test_workflow_detail_roles_filter(self, client, db, users): - from invenio_accounts.models import Role - from invenio_accounts.testutils import login_user_via_session as login - - client.application.config.update(dict( - WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT={ - "prefix": "jc", - "role_keyword": "roles", - "role_mapping": { - "repoadm": "Repository Administrator", - "comadm": "Community Administrator", - "contributor": "Contributor", - } - } - )) + def test_workflow_detail_roles_filter(self, client, db, users, mocker): + mocker.patch.dict(current_app.config, { + 'WEKO_ACCOUNTS_IDP_ENTITY_ID': 'https://test-example.com/shib' + }) user = users[1]['obj'] db.session.add(user) role1 = Role(name="Contributor_test", description=None) - role2 = Role(name="jc_xxx_roles_contributor", description=None) - role3 = Role(name="jc_xxx_groups_yyy", description=None) + role2 = Role(name="jc_test_example_com_ro_cont", description=None) + role3 = Role(name="jc_test_example_com_gr_yyy", description=None) db.session.add_all([role1, role2, role3]) db.session.commit() @@ -819,8 +796,8 @@ def test_workflow_detail_roles_filter(self, client, db, users): args, kwargs = mock_render.call_args context = args[1] filtered_role_names = [role.name for role in context['display_list']] - assert "jc_xxx_roles_contributor" not in filtered_role_names - assert "jc_xxx_groups_yyy" in filtered_role_names + assert "jc_test_example_com_ro_cont" not in filtered_role_names + assert "jc_test_example_com_gr_yyy" in filtered_role_names assert "Contributor_test" in filtered_role_names # def delete_workflow(self, workflow_id='0'): diff --git a/modules/weko-workflow/tests/test_views.py b/modules/weko-workflow/tests/test_views.py index 8ab92284b0..5aa3d60ac3 100644 --- a/modules/weko-workflow/tests/test_views.py +++ b/modules/weko-workflow/tests/test_views.py @@ -7221,6 +7221,7 @@ def test_edit_item_direct_after_login_09(client, users, db_register_full_action, mocker.patch('celery.task.control.inspect.ping', return_value='') mock_get_workflow_activity_by_item_id = mocker.patch.object(WorkActivity, 'get_workflow_activity_by_item_id', return_value = None) mocker.patch('weko_workflow.views.get_workflow_by_item_type_id', return_value=None) + mocker.patch('weko_workflow.views.check_item_is_being_edit', return_value=False) login(client=client, email=users[users_index]["email"]) url = url_for("weko_workflow.edit_item_direct_after_login", pid_value="1") res = client.get(url) @@ -7250,6 +7251,7 @@ def test_edit_item_direct_after_login_10(client, users, db_register_full_action, mocker.patch('weko_workflow.views.get_workflow_by_item_type_id', return_value=return_data_1) return_data_2 = MagicMock(activity_id=1) mocker.patch("weko_workflow.views.prepare_edit_workflow", return_value=return_data_2) + mocker.patch('weko_workflow.views.check_item_is_being_edit', return_value=False) login(client=client, email=users[users_index]["email"]) url = url_for("weko_workflow.edit_item_direct_after_login", pid_value="1") res = client.get(url) diff --git a/modules/weko-workflow/weko_workflow/admin.py b/modules/weko-workflow/weko_workflow/admin.py index 57423a90c1..fc92ef885a 100644 --- a/modules/weko-workflow/weko_workflow/admin.py +++ b/modules/weko-workflow/weko_workflow/admin.py @@ -39,6 +39,7 @@ from weko_index_tree.models import Index from weko_records.api import ItemTypes from weko_records.models import ItemTypeProperty +from weko_accounts.api import map_role_condition from weko_admin.models import AdminSettings from . import config @@ -46,7 +47,7 @@ from .config import WEKO_WORKFLOW_SHOW_HARVESTING_ITEMS from .models import WorkflowRole from .utils import recursive_get_specified_properties, check_activity_settings -from sqlalchemy import and_ +from sqlalchemy import not_ class FlowSettingView(BaseView): @expose('/', methods=['GET']) @@ -70,11 +71,7 @@ def flow_detail(self, flow_id='0'): :return: """ users = User.query.filter_by(active=True).all() - role_key = current_app.config["WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT"]["role_keyword"] - prefix = current_app.config["WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT"]["prefix"] - roles = Role.query.filter( - ~and_(Role.name.like(f"%{role_key}%"), Role.name.startswith(prefix)) - ).all() + roles = Role.query.filter(not_(map_role_condition())).all() if set(role.name for role in current_user.roles) & \ set(current_app.config['WEKO_PERMISSION_SUPER_ROLE_USER']): repositories = [{"id": "Root Index"}] + Community.query.all() @@ -113,7 +110,7 @@ def flow_detail(self, flow_id='0'): if not self._check_auth(flow_id) : abort(403) - + if not use_restricted_item: for action in flow.flow_actions: if action.action_role: @@ -327,11 +324,7 @@ def index(self): """ workflow = WorkFlow() workflows = workflow.get_workflow_list(user=current_user) - role_key = current_app.config["WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT"]["role_keyword"] - prefix = current_app.config["WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT"]["prefix"] - role = Role.query.filter( - ~and_(Role.name.like(f"%{role_key}%"), Role.name.startswith(prefix)) - ).all() + role = Role.query.filter(not_(map_role_condition())).all() for wf in workflows: index_tree = Index().get_index_by_id(wf.index_tree_id) wf.index_tree = index_tree @@ -370,11 +363,7 @@ def workflow_detail(self, workflow_id='0'): index_list = Index().get_all() location_list = Location.query.order_by(Location.id.asc()).all() hide = [] - role_key = current_app.config["WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT"]["role_keyword"] - prefix = current_app.config["WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT"]["prefix"] - role = Role.query.filter( - ~and_(Role.name.like(f"%{role_key}%"), Role.name.startswith(prefix)) - ).all() + role = Role.query.filter(not_(map_role_condition())).all() display_label = self.get_language_workflows("display") hide_label = self.get_language_workflows("hide") display_hide = self.get_language_workflows("display_hide") @@ -394,7 +383,7 @@ def workflow_detail(self, workflow_id='0'): is_display_restricted_access_checkbox = is_sysadmin \ and current_app.config.get('WEKO_ADMIN_DISPLAY_RESTRICTED_SETTINGS', False) \ and current_app.config.get('WEKO_ADMIN_RESTRICTED_ACCESS_DISPLAY_FLAG', False) - if '0' == workflow_id: + if '0' == workflow_id: """Create new workflow""" return self.render( 'weko_workflow/admin/workflow_detail.html', @@ -426,7 +415,7 @@ def workflow_detail(self, workflow_id='0'): else: display = role hide = [] - + if workflows.open_restricted \ and not is_sysadmin \ and current_app.config["WEKO_ADMIN_RESTRICTED_ACCESS_DISPLAY_FLAG"]: diff --git a/modules/weko-workflow/weko_workflow/api.py b/modules/weko-workflow/weko_workflow/api.py index 3cdd2b4679..cff264df28 100644 --- a/modules/weko-workflow/weko_workflow/api.py +++ b/modules/weko-workflow/weko_workflow/api.py @@ -44,6 +44,7 @@ from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.orm.exc import NoResultFound from sqlalchemy.types import String +from weko_accounts.api import map_role_condition from weko_deposit.api import WekoDeposit from weko_logging.activity_logger import UserActivityLogger from weko_notifications import Notification, NotificationClient @@ -730,11 +731,7 @@ def get_display_role(list_hide, role): wfs = [] current_user_roles = [role.id for role in current_user.roles] if isinstance(workflows, list): - role_key = current_app.config["WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT"]["role_keyword"] - prefix = current_app.config["WEKO_ACCOUNTS_GAKUNIN_GROUP_PATTERN_DICT"]["prefix"] - role = Role.query.filter( - ~and_(Role.name.like(f"%{role_key}%"), Role.name.startswith(prefix)) - ).all() + role = Role.query.filter(not_(map_role_condition())).all() while workflows: tmp = workflows.pop(0) list_hide = Role.query.outerjoin(WorkflowRole) \