サブリポジトリ対応プルリクエスト

modules/invenio-accounts/invenio_accounts/admin.py

@@ -15,16 +15,21 @@
 from flask_admin.babel import lazy_gettext
 from flask_admin.contrib.sqla import ModelView
 from flask_admin.contrib.sqla.ajax import QueryAjaxModelLoader
+from flask_admin.contrib.sqla.fields import QuerySelectMultipleField
+from flask_admin.form import Select2Widget
 from flask_admin.form.fields import DateTimeField
 from flask_admin.model.fields import AjaxSelectMultipleField
 from flask_babelex import gettext as _
 from flask_security import current_user
 from flask_security.recoverable import send_reset_password_instructions
 from flask_security.utils import hash_password
+from invenio_communities.models import Community
 from invenio_db import db
 from passlib import pwd
+from sqlalchemy import func
 from werkzeug.local import LocalProxy
-from wtforms.fields import BooleanField
+from collections import OrderedDict
+from wtforms.fields import BooleanField, SelectMultipleField
 from wtforms.validators import DataRequired
 
 from weko_workflow.models import WorkFlow, WorkflowRole
@@ -52,7 +57,7 @@ class UserView(ModelView):
         column_details_list = \
         list_all
 
-    form_columns = ('email', 'password', 'active', 'roles', 'notification')
+    form_columns = ('email', 'password', 'active', 'notification')
 
     form_args = dict(
         email=dict(label='Email', validators=[DataRequired()]),
@@ -79,18 +84,77 @@ class UserView(ModelView):
         'last_login_ip': _('Last Login IP')
     }
 
+    def scaffold_form(self):
+        form_class = super(UserView, self).scaffold_form()
+        form_class.role = QuerySelectMultipleField(
+            'Roles',
+            query_factory=lambda: Role.query.filter(~Role.name.like('%_groups_%')).all(),
+            get_label='name',
+            widget=Select2Widget(multiple=True)
+        )
+        form_class.group = QuerySelectMultipleField(
+            'Groups',
+            query_factory=lambda: Role.query.filter(Role.name.like('%_groups_%')).all(),
+            get_label='name',
+            widget=Select2Widget(multiple=True)
+        )
+
+        return form_class
+
+    def _order_fields(self, form):
+        custom_order = ['email', 'password', 'active', 'role', 'group', 'notification']
+        ordered_fields = OrderedDict()
+        for field_name in custom_order:
+            ordered_fields[field_name] = form._fields[field_name]
+        form._fields = ordered_fields
+        return form
+
+    def create_form(self, obj=None):
+        form = super(UserView, self).create_form(obj)
+        return self._order_fields(form)
+
+    def edit_form(self, obj=None):
+        form = super(UserView, self).edit_form(obj)
+        return self._order_fields(form)
+
+    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]
+
     def on_model_change(self, form, User, is_created):
         """Hash password when saving."""
         if form.password.data is not None:
             pwd_ctx = current_app.extensions['security'].pwd_context
             if pwd_ctx.identify(form.password.data) is None:
                 User.password = hash_password(form.password.data)
 
+        roles = form.role.data + form.group.data
+        User.roles = roles
+
     def after_model_change(self, form, User, is_created):
         """Send password instructions if desired."""
         if is_created and form.notification.data is True:
             send_reset_password_instructions(User)
 
+    def get_query(self):
+        """Return a query for the model type."""
+        if any(role.name in current_app.config['WEKO_PERMISSION_SUPER_ROLE_USER'] for role in current_user.roles):
+            return self.session.query(self.model)
+        else:
+            repositories = Community.get_repositories_by_user(current_user)
+            groups = [repository.group for repository in repositories]
+            return self.session.query(self.model).filter(self.model.roles.any(Role.id.in_([group.id for group in groups])))
+
+    def get_count_query(self):
+        """Return a the count query for the model type"""
+        if any(role.name in current_app.config['WEKO_PERMISSION_SUPER_ROLE_USER'] for role in current_user.roles):
+            return self.session.query(func.count('*')).select_from(self.model)
+        else:
+            repositories = Community.get_repositories_by_user(current_user)
+            groups = [repository.group for repository in repositories]
+            return self.session.query(func.count('*')).select_from(self.model).filter(self.model.roles.any(Role.id.in_([group.id for group in groups])))
+
     @action('inactivate', _('Inactivate'),
             _('Are you sure you want to inactivate selected users?'))
     @commit
@@ -138,7 +202,12 @@ def action_activate(self, ids):
 
     _system_role = os.environ.get('INVENIO_ROLE_SYSTEM',
                                   'System Administrator')
-
+    _repo_role = os.environ.get('INVENIO_ROLE_REPOSITORY'
+                                'Repository Administrator')
+    _com_role = os.environ.get('INVENIO_ROLE_COMMUNITY',
+                               'Community Administrator')
+    _admin_roles = [_system_role, _repo_role, _com_role]
+
     @property
     def can_create(self):
         """Check permission for creating."""
@@ -147,12 +216,12 @@ def can_create(self):
     @property
     def can_edit(self):
         """Check permission for Editing."""
-        return self._system_role in [role.name for role in current_user.roles]
+        return any(role.name in self._admin_roles for role in current_user.roles)
 
     @property
     def can_delete(self):
         """Check permission for Deleting."""
-        return self._system_role in [role.name for role in current_user.roles]
+        return any(role.name in self._admin_roles for role in current_user.roles)
 
 
 class RoleView(ModelView):

modules/invenio-accounts/invenio_accounts/utils.py

@@ -17,6 +17,7 @@
 from jwt import DecodeError, ExpiredSignatureError, decode, encode
 
 from .errors import JWTDecodeError, JWTExpiredToken
+from .models import User, userrole, Role
 
 
 def jwt_create_token(user_id=None, additional_data=None):
@@ -85,3 +86,15 @@ def set_session_info(app, response, **extra):
         response.headers['X-Session-ID'] = session_id
     if current_user.is_authenticated:
         response.headers['X-User-ID'] = current_user.get_id()
+
+
+def get_user_ids_by_role(role_id):
+    """Get user IDs by role ID.
+
+    Args:
+        role_id (int): The ID of the role.
+
+    Returns:
+        list: A list of user IDs associated with the given role.
+    """
+    return [str(user.id) for user in User.query.join(userrole).join(Role).filter(Role.id == role_id).all()]

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_records_ui.config import WEKO_PERMISSION_SUPER_ROLE_USER
 
 from invenio_accounts import InvenioAccounts
 from invenio_accounts.admin import role_adminview, session_adminview, \
@@ -69,7 +70,8 @@ def _app_factory(config=None):
         TESTING=True,
         WTF_CSRF_ENABLED=False,
         ACCOUNTS_JWT_ALOGORITHM = 'HS256',
-        ACCOUNTS_JWT_SECRET_KEY = 'None'
+        ACCOUNTS_JWT_SECRET_KEY = 'None',
+        WEKO_PERMISSION_SUPER_ROLE_USER = WEKO_PERMISSION_SUPER_ROLE_USER,
     )
 
     # Set key value session store to use Redis when running on TravisCI.

modules/invenio-accounts/tests/test_admin.py

@@ -7,7 +7,7 @@
 # under the terms of the MIT License; see LICENSE file for more details.
 
 import pytest
-from mock import patch
+from mock import patch, MagicMock
 from flask import current_app, session, url_for
 from flask_admin import menu
 from flask_security import url_for_security
@@ -17,9 +17,9 @@
 
 from invenio_accounts import InvenioAccounts
 from invenio_accounts.cli import users_create
-from invenio_accounts.models import SessionActivity
+from invenio_accounts.models import SessionActivity, User, Role
 from invenio_accounts.testutils import create_test_user, login_user_via_view
-from invenio_accounts.admin import SessionActivityView
+from invenio_accounts.admin import SessionActivityView, UserView
 
 _datastore = LocalProxy(
     lambda: current_app.extensions['security'].datastore
@@ -220,4 +220,66 @@ def test_admin_sessions_action_delete(app):
             view.action_delete([user1_sid])
             sessions = SessionActivity.query.all()
             assert len(sessions) == 1
-
+
+
+# .tox/c1/bin/pytest --cov=invenio_accounts tests/test_admin.py::test_userview_get_query -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/invenio-accounts/.tox/c1/tmp
+def test_userview_get_query(app, users):
+        """Test get_query for super role user."""
+        view = UserView(User, db.session)
+        mock_repo = MagicMock(group=MagicMock(id=1))
+        with app.test_request_context():
+            with patch("flask_login.utils._get_user", return_value=users[2]['obj']):
+                query = view.get_query()
+                assert query is not None
+                assert query.count() == User.query.count()
+
+            with patch("flask_login.utils._get_user", return_value=users[0]['obj']):
+                with patch("invenio_communities.models.Community.get_repositories_by_user", return_value=[mock_repo]):
+                    db.session.add(users[0]['obj'])
+                    db.session.commit()
+                    query = view.get_query()
+                    assert query is not None
+                    assert query.count() == 0
+
+
+# .tox/c1/bin/pytest --cov=invenio_accounts tests/test_admin.py::test_userview_get_count_query -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/invenio-accounts/.tox/c1/tmp
+def test_userview_get_count_query(app, users):
+        """Test get_count_query for super role user."""
+        view = UserView(User, db.session)
+        mock_repo = MagicMock(group=MagicMock(id=1))
+        with app.test_request_context():
+            with patch("flask_login.utils._get_user", return_value=users[2]['obj']):
+                query = view.get_count_query()
+                assert query is not None
+                assert query.scalar() == User.query.count()
+
+            with patch("flask_login.utils._get_user", return_value=users[0]['obj']):
+                with patch("invenio_communities.models.Community.get_repositories_by_user", return_value=[mock_repo]):
+                    db.session.add(users[0]['obj'])
+                    db.session.commit()
+                    query = view.get_count_query()
+                    assert query is not None
+                    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):
+    """Test on_form_prefill for super role user."""
+    view = UserView(User, db.session)
+    form = view.create_form()
+    user = users[2]['obj']
+    user.roles = [
+        Role(name='role1'),
+        Role(name='role2_groups_1')
+    ]
+    view.get_one = MagicMock(return_value=user)
+    view.on_form_prefill(form, user.id)
+    assert form.data['active'] is False
+    assert form.role.data == [user.roles[0]]
+    assert form.group.data == [user.roles[1]]
+
+# .tox/c1/bin/pytest --cov=invenio_accounts tests/test_admin.py::test_userview_edit_form -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/invenio-accounts/.tox/c1/tmp
+def test_userview_edit_form(app, users):
+    """Test edit_form for super role user."""
+    view = UserView(User, db.session)
+    form = view.edit_form()
+    assert form.data['active'] is False

modules/invenio-accounts/tests/test_utils.py

@@ -20,7 +20,7 @@
 
 from invenio_accounts import testutils
 from invenio_accounts.errors import JWTDecodeError, JWTExpiredToken
-from invenio_accounts.utils import jwt_create_token, jwt_decode_token
+from invenio_accounts.utils import jwt_create_token, jwt_decode_token, get_user_ids_by_role
 
 
 def test_client_authenticated(app):
@@ -172,3 +172,33 @@ def test_jwt_expired_token(app):
         # Random token
         with pytest.raises(JWTDecodeError):
             jwt_decode_token('Roadster SV')
+
+
+def test_get_user_ids_by_role_returns_user_ids(app):
+    """Test get_user_ids_by_role."""
+    with app.app_context():
+        ds = app.extensions['invenio-accounts'].datastore
+
+        user1 = ds.create_user(email='test1@test.org', active=True)
+        user2 = ds.create_user(email='test2@test.org', active=True)
+        role = ds.create_role(name='superuser', description='1234')
+        ds.add_role_to_user(user1, role)
+        ds.add_role_to_user(user2, role)
+        ds.commit()
+
+        user_ids = get_user_ids_by_role(role.id)
+        assert str(user1.id) in user_ids
+        assert str(user2.id) in user_ids
+        assert len(user_ids) == 2
+
+
+@pytest.mark.parametrize("role_id, expected", [
+    (999, []),
+    (None, []),
+    ('1', []),
+])
+def test_get_user_ids_by_role_returns_empty_list(app, role_id, expected):
+    """Test get_user_ids_by_role."""
+    with app.app_context():
+        user_ids = get_user_ids_by_role(role_id)
+        assert user_ids == expected