Feature/h2025 01 v0.9.26 hiroba

docs/source/developer/database.rst

@@ -6161,7 +6161,7 @@ shibboleth_user
      - 
    * - shib_eppn
      - shib_eppn
-     - VARCHAR(128)
+     - VARCHAR(2310)
      - True
      - False
      - None
@@ -6189,7 +6189,7 @@ shibboleth_user
      - 
    * - shib_page_name
      - shib_page_name
-     - VARCHAR(255)
+     - VARCHAR(1024)
      - False
      - False
      - None
@@ -6222,6 +6222,13 @@ shibboleth_user
      - False
      - None
      - 
+   * - shib_organization
+     - shib_organization
+     - VARCHAR(255)
+     - False
+     - False
+     - None
+     - 
 
 Keys
 ^^^^

modules/invenio-records-rest/invenio_records_rest/views.py

@@ -469,6 +469,8 @@ def redirect_to_search(page, size):
     if size:
         url += "&size=" + str(size)
     get_args = request.args.to_dict()
+    if get_args.get('format') != 'html':
+        get_args['format'] = 'html'
     for key, param in get_args.items():
         if key == "page_no" or key == "list_view_num" \
             or key == "log_term" or key == "lang":
@@ -583,8 +585,8 @@ def get(self, **kwargs):
                                       'list_view_num', 10, type=int),
                                   type=int)
         size = RecordsListResource.adjust_list_view_num(size)
-        format = request.values.get('format')
-        if (not format or format == "html") and request.values.get('q') == None:
+        formats = request.values.getlist('format')
+        if (not formats or 'html' in formats) and request.values.get('q') == None:
             return redirect_to_search(page, size)
 
         # if page * size >= self.max_result_window:

modules/invenio-records-rest/tests/test_views_listresource.py

@@ -0,0 +1,166 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of Invenio.
+# Copyright (C) 2016-2018 CERN.
+#
+# Invenio is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+"""Tests for patching records."""
+
+import pytest
+from flask import Flask, url_for
+from flask_login import AnonymousUserMixin
+import flask_security
+from invenio_records_rest.views import RecordsListResource
+from werkzeug.test import EnvironBuilder
+from werkzeug.wrappers import Request
+
+class DummyUser(AnonymousUserMixin):
+    is_authenticated = False
+    id = None
+
+class DummySearch:
+    def with_preference_param(self):
+        return self
+    def params(self, **kwargs):
+        return self
+    def to_dict(self):
+        return {"sort": [{"control_number": {"order": "desc"}}]}
+    def __getitem__(self, key):
+        return self
+    def execute(self):
+        class Result:
+            hits = type('hits', (), {'total': 0})()
+            def to_dict(self):
+                return {}
+        return Result()
+
+def make_request(app, query_string=None):
+    builder = EnvironBuilder(method='GET', query_string=query_string)
+    env = builder.get_environ()
+    req = Request(env)
+    return req
+
+@pytest.fixture
+def resource_with_dummy_search(app):
+    def dummy_search_factory(self, search):
+        return search, {}
+
+    return RecordsListResource(
+        minter_name="recid",
+        pid_type="recid",
+        pid_fetcher="recid",
+        read_permission_factory=None,
+        create_permission_factory=None,
+        list_permission_factory=None,
+        search_class=DummySearch,
+        record_serializers={},
+        record_loaders=None,
+        search_serializers={'application/json': lambda *a, **k: {}},
+        default_media_type='application/json',
+        max_result_window=10000,
+        search_factory=dummy_search_factory,
+        item_links_factory=None,
+        record_class=None,
+        indexer_class=None
+    )
+
+def test_format_html_redirect(monkeypatch, app, resource_with_dummy_search):
+    # Should redirect when format=html is specified
+    with app.test_request_context('/?format=html'):
+        monkeypatch.setattr(flask_security, "current_user", DummyUser())
+        monkeypatch.setattr('flask.url_for', lambda endpoint, **kwargs: '/dummy_url')
+        monkeypatch.setattr('invenio_records_rest.views.url_for', lambda endpoint, **kwargs: '/dummy_url')
+        monkeypatch.setattr('invenio_accounts.models.User', type('User', (), {'query': type('query', (), {'get': staticmethod(lambda x: type('U', (), {'email': 'dummy@example.com'})())})()}) )
+        resp = resource_with_dummy_search.get()
+        assert resp.status_code == 302
+        assert 'search' in resp.location
+
+def test_format_rss(monkeypatch, app, resource_with_dummy_search):
+    # Should NOT redirect when format=rss is specified
+    with app.test_request_context('/?format=rss&q=test'):
+        monkeypatch.setattr(flask_security, "current_user", DummyUser())
+        monkeypatch.setattr('flask.url_for', lambda endpoint, **kwargs: '/dummy_url')
+        monkeypatch.setattr('invenio_records_rest.views.url_for', lambda endpoint, **kwargs: '/dummy_url')
+        monkeypatch.setattr('invenio_accounts.models.User', type('User', (), {'query': type('query', (), {'get': staticmethod(lambda x: type('U', (), {'email': 'dummy@example.com'})())})()}) )
+        resp = resource_with_dummy_search.get()
+        # Should return a dict as search result, not a redirect
+        assert isinstance(resp, dict)
+
+def test_format_multiple(monkeypatch, app, resource_with_dummy_search):
+    # Should prioritize html when both format=html&format=rss are specified
+    with app.test_request_context('/?format=html&format=rss'):
+        monkeypatch.setattr('flask_security.current_user', DummyUser())
+        monkeypatch.setattr('flask.url_for', lambda endpoint, **kwargs: '/dummy_url')
+        monkeypatch.setattr('invenio_records_rest.views.url_for', lambda endpoint, **kwargs: '/dummy_url')
+        monkeypatch.setattr('invenio_accounts.models.User', type('User', (), {'query': type('query', (), {'get': staticmethod(lambda x: type('U', (), {'email': 'dummy@example.com'})())})()}) )
+        resp = resource_with_dummy_search.get()
+        assert resp.status_code == 302
+        assert 'search' in resp.location
+
+def test_no_format_no_query(monkeypatch, app, resource_with_dummy_search):
+    # Should redirect when neither format nor q is specified
+    with app.test_request_context('/'):
+        monkeypatch.setattr('flask_security.current_user', DummyUser())
+        monkeypatch.setattr('flask.url_for', lambda endpoint, **kwargs: '/dummy_url')
+        monkeypatch.setattr('invenio_records_rest.views.url_for', lambda endpoint, **kwargs: '/dummy_url')
+        monkeypatch.setattr('invenio_accounts.models.User', type('User', (), {'query': type('query', (), {'get': staticmethod(lambda x: type('U', (), {'email': 'dummy@example.com'})())})()}) )
+        resp = resource_with_dummy_search.get()
+        assert resp.status_code == 302
+        assert 'search' in resp.location
+
+# def test_redirect_to_search_sets_format_html(app):
+    # Should always set format=html in redirect URL, even if other formats are specified
+    from invenio_records_rest.views import redirect_to_search
+    with app.test_request_context('/?format=rss&q=test&foo=bar'):
+        resp = redirect_to_search(page=2, size=50)
+        assert resp.status_code == 302
+        location = resp.location
+        assert 'format=html' in location
+        assert 'page=2' in location
+        assert 'size=50' in location
+        assert 'foo=bar' in location
+        assert 'q=test' in location
+        assert location.startswith('http')
+
+
+def test_redirect_to_search_no_format(app):
+    # Should add format=html if no format is specified, and preserve other parameters
+    from invenio_records_rest.views import redirect_to_search
+    with app.test_request_context('/?q=xyz'):
+        resp = redirect_to_search(page=3, size=75)
+        assert resp.status_code == 302
+        location = resp.location
+        assert 'format=html' in location
+        assert 'q=xyz' in location
+        assert 'page=3' in location
+        assert 'size=75' in location
+
+def test_redirect_to_search_page_size_falsy(app):
+    # Should not include page or size in URL if they are None, but should include format=html and other params
+    from invenio_records_rest.views import redirect_to_search
+    with app.test_request_context('/?format=xml&q=abc'):
+        resp = redirect_to_search(page=None, size=None)
+        location = resp.location
+        assert 'page=None' not in location
+        assert 'size=None' not in location
+        assert 'format=html' in location
+        assert 'q=abc' in location
+
+def test_redirect_to_search_exclude_keys(app):
+    # Should exclude specific keys (page_no, list_view_num, log_term, lang) from the redirect URL, but include others
+    from invenio_records_rest.views import redirect_to_search
+    with app.test_request_context('/?page_no=5&list_view_num=99&log_term=foo&lang=ja&q=zzz'):
+        resp = redirect_to_search(page=1, size=20)
+        location = resp.location
+        # Excluded keys should not be in the URL
+        assert 'page_no=' not in location
+        assert 'list_view_num=' not in location
+        assert 'log_term=' not in location
+        assert 'lang=' not in location
+        # Other parameters should be present
+        assert 'q=zzz' in location
+        assert 'page=1' in location
+        assert 'size=20' in location
+        assert 'format=html' in location
+

modules/weko-accounts/tests/test_api.py

@@ -6,6 +6,7 @@
 from flask_login.utils import login_user
 from invenio_accounts.models import Role, User
 from weko_user_profiles.models import UserProfile
+from sqlalchemy.exc import SQLAlchemyError
 from weko_accounts.models import ShibbolethUser
 from weko_accounts.api import ShibUser,get_user_info_by_role_name
 
@@ -100,27 +101,26 @@ def test__create_unknown_roles(self, app, users, mocker):
 #    def get_relation_info(self):
 # .tox/c1/bin/pytest --cov=weko_accounts tests/test_api.py::TestShibUser::test_get_relation_info -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-accounts/.tox/c1/tmp
     def test_get_relation_info(self,app,db,users):
-        
+
         user1 = users[0]["obj"]
         user2 = users[1]["obj"]
+
         attr = {
-            "shib_eppn":"test_eppn"
+            "shib_eppn": "test_eppn",
+            "shib_mail": None,
+            "shib_user_name": None,
+            "shib_role_authority_name": None,
+            "shib_page_name": None,
+            "shib_active_flag": None,
+            "shib_ip_range_flag": None,
+            "shib_organization": None
         }
         s_user1 = ShibbolethUser(weko_uid=user1.id,weko_user=user1,**attr)
         db.session.add(s_user1)
         db.session.commit()
 
         # exist shib_eppn,exist shib_user.weko_user,not exist self.user
         # attribute does not exist
-        attr = {
-            "shib_eppn":"test_eppn",
-            "shib_mail":None,
-            "shib_user_name":None,
-            "shib_role_authority_name":None,
-            "shib_page_name":None,
-            "shib_active_flag":None,
-            "shib_ip_range_flag":None,
-        }
         shibuser = ShibUser(attr)
         result = shibuser.get_relation_info()
         assert result.shib_mail == None
@@ -129,6 +129,7 @@ def test_get_relation_info(self,app,db,users):
         assert result.shib_page_name == None
         assert result.shib_active_flag == None
         assert result.shib_ip_range_flag == None
+        assert result.shib_organization == None
 
         # attribute exists
         attr = {
@@ -139,6 +140,7 @@ def test_get_relation_info(self,app,db,users):
             "shib_page_name":"shib page",
             "shib_active_flag":"TRUE",
             "shib_ip_range_flag":"TRUE",
+            "shib_organization":"shib org"
         }
         shibuser = ShibUser(attr)
         result = shibuser.get_relation_info()
@@ -148,11 +150,18 @@ def test_get_relation_info(self,app,db,users):
         assert result.shib_page_name == "shib page"
         assert result.shib_active_flag == "TRUE"
         assert result.shib_ip_range_flag == "TRUE"
-
+        assert result.shib_organization == "shib org"
+
         # not exist shib_eppn,not exist shib_user.weko_user
         attr = {
             "shib_eppn":"",
-            "shib_user_name":"shib name2"
+            "shib_user_name":"shib name2",
+            "shib_mail":None,
+            "shib_role_authority_name":None,
+            "shib_page_name":None,
+            "shib_active_flag":None,
+            "shib_ip_range_flag":None,
+            "shib_organization":None
         }
         s_user2 = ShibbolethUser(**attr)
         db.session.add(s_user2)
@@ -161,15 +170,15 @@ def test_get_relation_info(self,app,db,users):
         result = shibuser.get_relation_info()
         assert result == None
 
-        # not exist shib_eppn, exist shib_user.weko_user,exist self.user, raise Exception
+        # not exist shib_eppn, exist shib_user.weko_user,exist self.user, raise SQLAlchemyError
         s_user2.weko_user = user2
         s_user2.weko_uid = user2.id
         db.session.merge(s_user2)
         db.session.commit()
         shibuser.user = user2
-        with patch("weko_accounts.api.db.session.commit",side_effect=Exception):
-            result = shibuser.get_relation_info()
-            assert result == None
+        with patch("weko_accounts.api.db.session.commit",side_effect=SQLAlchemyError):
+            with pytest.raises(SQLAlchemyError):
+                shibuser.get_relation_info()
 #    def check_weko_user(self, account, pwd):
 # .tox/c1/bin/pytest --cov=weko_accounts tests/test_api.py::TestShibUser::test_check_weko_user -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-accounts/.tox/c1/tmp
     def test_check_weko_user(self,app,users):

modules/weko-accounts/weko_accounts/api.py

@@ -147,6 +147,8 @@ def get_relation_info(self):
                     shib_user.shib_active_flag = self.shib_attr['shib_active_flag']
                 if self.shib_attr['shib_ip_range_flag']:
                     shib_user.shib_ip_range_flag = self.shib_attr['shib_ip_range_flag']
+                if self.shib_attr['shib_organization']:
+                    shib_user.shib_organization = self.shib_attr['shib_organization']
             db.session.commit()
         except SQLAlchemyError as ex:
             current_app.logger.error("SQLAlchemyError: {}".format(ex))

modules/weko-accounts/weko_accounts/config.py

@@ -71,6 +71,7 @@
     'SHIB_ATTR_SITE_USER_WITHIN_IP_RANGE_FLAG': (False, 'shib_ip_range_flag'),
     'SHIB_ATTR_MAIL': (False, 'shib_mail'),
     'SHIB_ATTR_USER_NAME': (False, 'shib_user_name'),
+    'SHIB_ATTR_ORGANIZATION': (False, 'shib_organization'),
 }
 """IdP attribute map."""
 

modules/weko-accounts/weko_accounts/models.py

@@ -71,6 +71,9 @@ class ShibbolethUser(db.Model):
     shib_ip_range_flag = db.Column(db.String(255), nullable=True)
     """SHIB_ATTR_SITE_USER_WITHIN_IP_RANGE_FLAG"""
 
+    shib_organization = db.Column(db.String(255), nullable=True)
+    """SHIB_ATTR_ORGANIZATION"""
+
     shib_roles = db.relationship(
         'Role',
         secondary=shibuserrole,

modules/weko-items-ui/tests/conftest.py

@@ -345,26 +345,31 @@ def db(app):
 
 @pytest.fixture()
 def esindex(app,db_records):
+    current_search_client.indices.delete(index='test-*')
     with open("tests/data/mappings/item-v1.0.0.json","r") as f:
         mapping = json.load(f)
+    with open("tests/data/mappings/record-view-v1.json","r") as f:
+        mapping_record_view = json.load(f)
 
     search = LocalProxy(lambda: app.extensions["invenio-search"])
 
     with app.test_request_context():
-        search.client.indices.create(app.config["INDEXER_DEFAULT_INDEX"],body=mapping)
-        search.client.indices.put_alias(index=app.config["INDEXER_DEFAULT_INDEX"], name="test-weko")
+        current_search_client.indices.create(app.config["INDEXER_DEFAULT_INDEX"],body=mapping)
+        current_search_client.indices.put_alias(index=app.config["INDEXER_DEFAULT_INDEX"], name="test-weko")
         # print(current_search_client.indices.get_alias())
+        current_search_client.indices.create("test-stats-record-view-000001",body=mapping_record_view)
+        current_search_client.indices.put_alias(index="test-stats-record-view-000001",name="test-stats-record-view")
 
     for depid, recid, parent, doi, record, item in db_records:
-        search.client.index(index='test-weko-item-v1.0.0', doc_type='item-v1.0.0', id=record.id, body=record,refresh='true')
+        current_search_client.index(index='test-weko-item-v1.0.0', doc_type='item-v1.0.0', id=record.id, body=record,refresh='true')
 
-
-    yield search
+    yield current_search_client
 
     with app.test_request_context():
-        search.client.indices.delete_alias(index=app.config["INDEXER_DEFAULT_INDEX"], name="test-weko")
-        search.client.indices.delete(index=app.config["INDEXER_DEFAULT_INDEX"], ignore=[400, 404])
-
+        current_search_client.indices.delete_alias(index=app.config["INDEXER_DEFAULT_INDEX"], name="test-weko")
+        current_search_client.indices.delete(index=app.config["INDEXER_DEFAULT_INDEX"], ignore=[400, 404])
+        #current_search_client.indices.delete_alias(index="test-stats-record-view-000001",name="test-stats-record-view")
+        #current_search_client.indices.delete(index="test-stats-record-view-000001", ignore=[400,400])
 
 
 @pytest.fixture()