Pipfile: Upgrade to invenio-app-rdm v13.0.0b1.dev19

invenio.cfg

@@ -279,7 +279,7 @@ CDS_EOS_OFFLOAD_REDIRECT_BASE_PATH = ""
 
 # CDS Migration
 CDS_REDIRECTION_COLLECTIONS_MAPPING = {
-    "CERN Student Project Notes": "", # uuid
+    "CERN Students Projects": "", # uuid
 }
 
 RDM_PERMISSION_POLICY = CDSRDMRecordPermissionPolicy

site/cds_rdm/legacy/redirector.py

@@ -11,16 +11,10 @@
 from flask import Blueprint, current_app, redirect, render_template, request, url_for
 from invenio_communities.views.ui import not_found_error
 from invenio_rdm_records.resources.urls import record_url_for
-
-# from invenio_records_resources.services.errors import RecordPermissionDeniedError
 from sqlalchemy.orm.exc import NoResultFound
 
 from .errors import VersionNotFound
-from .resolver import (
-    get_community_by_uuid,
-    get_pid_by_legacy_recid,
-    get_record_by_version,
-)
+from .resolver import get_pid_by_legacy_recid, get_record_by_version
 
 
 def version_not_found_error(error):
@@ -68,46 +62,41 @@ def legacy_collection_redirect(collection_name):
     )
     if not cds_community_uuid:
         raise NoResultFound
-    community = get_community_by_uuid(cds_community_uuid)
-    query_params = request.args.copy()
-    query_params["q"] = query_params.pop("p", None)
-    if query_params["q"]:
-        url_path = url_for(
-            "invenio_app_rdm_communities.communities_detail",
-            pid_value=community.data["slug"],
-            **query_params,
-        )
-    else:
-        url_path = url_for(
-            "invenio_app_rdm_communities.communities_home",
-            pid_value=community.data["slug"],
-            **query_params,
-        )
+    url_path = url_for(
+        "invenio_app_rdm_communities.communities_home",
+        pid_value=cds_community_uuid,
+        **request.args,
+    )
     return redirect(url_path)
 
 
 def legacy_search_redirect():
     """
     Redirection for legacy search. Transforms the legacy URL syntax into RDM URL syntax.
 
-        /legacy?cc=<legacy collection name>... -> /communities/<rdm_community_id>?...
-        /legacy?c=<legacy collection name>... -> /communities/<rdm_community_id>?...
+        /legacy?cc=<legacy collection name>... -> /communities/<rdm_community_id>/records?...
+        /legacy?c=<legacy collection name>... -> /communities/<rdm_community_id>/records?...
         /legacy?c=<legacy collection name>&p=<query>... -> /communities/<rdm_community_id>/records?q=<query>...
     """
-    query_params = request.args.copy()
+    query_params = {"q": request.args.get("p")}  # `p` in legacy is the search query
     # Fetch current collection if it exists
-    collection_name = query_params.pop("cc", None)
-    # If not, then fetch from collection list, URLs with only single 'c' will be redirected for now
+    collection_name = request.args.get("cc", None)
+    # If not, then fetch from collection list (query param 'c')
     if not collection_name:
-        collection_name = query_params.pop("c", None)
+        collections_list = request.args.getlist("c")
+        collection_name = collections_list[0] if collections_list else None
+    # If collection not found, i.e. 'c' or 'cc' not found, redirect to global record search with search query
     if not collection_name:
-        raise not_found_error()
-    # Add logic for other redirections from search params when we get there
-    url = url_for(
-        "cds_rdm.legacy_collection_redirect",
-        collection_name=collection_name,
-        **query_params,
-    )
+        url = url_for(
+            "invenio_search_ui.search",
+            **query_params,
+        )
+    else:
+        url = url_for(
+            "cds_rdm.legacy_collection_redirect",
+            collection_name=collection_name,
+            **query_params,
+        )
     return redirect(url)
 
 

site/cds_rdm/legacy/resolver.py

@@ -51,12 +51,3 @@ def get_record_by_version(parent_pid_value, version):
         # If record is not found, that means the version doesn't exist
         raise VersionNotFound(version=version, latest_record=latest_record)
     return hits[0]
-
-
-def get_community_by_uuid(community_uuid):
-    """Get community by uuid."""
-    community = current_communities.service.read(
-        id_=community_uuid,
-        identity=system_identity,  # using system_identity here as the permission check on user identity will be performed by the read service method
-    )
-    return community

site/tests/conftest.py

@@ -10,11 +10,6 @@
 from collections import namedtuple
 
 import pytest
-from cds_rdm.permissions import (
-    CDSCommunitiesPermissionPolicy,
-    CDSRDMRecordPermissionPolicy,
-)
-from cds_rdm.schemes import is_cern
 from flask_webpackext.manifest import (
     JinjaManifest,
     JinjaManifestEntry,
@@ -39,6 +34,12 @@
 from invenio_vocabularies.proxies import current_service as vocabulary_service
 from invenio_vocabularies.records.api import Vocabulary
 
+from cds_rdm.permissions import (
+    CDSCommunitiesPermissionPolicy,
+    CDSRDMRecordPermissionPolicy,
+)
+from cds_rdm.schemes import is_cern
+
 
 class MockJinjaManifest(JinjaManifest):
     """Mock manifest."""

site/tests/legacy/test_redirector.py

@@ -133,6 +133,7 @@ def test_legacy_record_redirection(
 
 def test_legacy_collection_redirection(
     superuser_identity,
+    uploader,
     client,
     app,
     monkeypatch,
@@ -154,37 +155,67 @@ def test_legacy_collection_redirection(
         },
     )
 
-    response = client.get("/legacy/collection/Legacy%20Collection")
-    assert response.status_code == 302
-    assert response.location == "/communities/legacy-community/"
-
-    response = client.get("/legacy?cc=Legacy%20Collection")
-    assert response.status_code == 302
-    response = client.get(response.location)
-    assert response.status_code == 302
-    assert response.location == "/communities/legacy-community/"
+    response = client.get(
+        "/legacy/collection/Legacy%20Collection", follow_redirects=True
+    )
+    assert response.status_code == 200
+    assert len(response.history) > 0
+    assert response.request.path == "/communities/legacy-community/records"
+
+    response = client.get("/legacy?cc=Legacy%20Collection", follow_redirects=True)
+    assert response.status_code == 200
+    assert len(response.history) > 0
+    assert response.request.path == "/communities/legacy-community/records"
+
+    response = client.get("/legacy?c=Legacy%20Collection", follow_redirects=True)
+    assert response.status_code == 200
+    assert len(response.history) > 0
+    assert response.request.path == "/communities/legacy-community/records"
+
+    response = client.get(
+        "/legacy?c=Legacy%20Collection&p=something&not=passed", follow_redirects=True
+    )
+    assert response.status_code == 200
+    assert len(response.history) > 0
+    assert response.request.path == "/communities/legacy-community/records"
+    assert response.request.query_string.decode() == "q=something"
+
+    response = client.get(
+        "/legacy?c=Legacy%20Collection&c=Legacy%20Restricted%20Collection&p=something&not=passed",
+        follow_redirects=True,
+    )
+    assert response.status_code == 200
+    assert len(response.history) > 0
+    assert response.request.path == "/communities/legacy-community/records"
+    assert response.request.query_string.decode() == "q=something"
+
+    response = client.get(
+        "/legacy?p=something&not=passed",
+        follow_redirects=True,
+    )
+    assert response.status_code == 200
+    assert len(response.history) > 0
+    assert response.request.path == "/search"
+    assert response.request.query_string.decode() == "q=something"
 
-    response = client.get("/legacy?c=Legacy%20Collection")
-    assert response.status_code == 302
-    response = client.get(response.location)
-    assert response.status_code == 302
-    assert response.location == "/communities/legacy-community/"
+    response = client.get(
+        "/legacy/collection/Legacy%20Wrong%20Collection", follow_redirects=True
+    )
+    assert response.status_code == 404
 
-    response = client.get("/legacy?c=Legacy%20Collection&p=something")
-    assert response.status_code == 302
-    response = client.get(response.location)
-    assert response.status_code == 302
-    assert response.location == "/communities/legacy-community/records?q=something"
+    response = client.get(
+        "/legacy?cc=Legacy%20Wrong%20Collection", follow_redirects=True
+    )
+    assert response.status_code == 404
 
-    response = client.get("/legacy/collection/Legacy%20Restricted%20Collection")
-    assert response.status_code == 302
-    response = client.get(response.location)
+    client = uploader.login(client)
+    response = client.get(
+        "/legacy/collection/Legacy%20Restricted%20Collection", follow_redirects=True
+    )
     assert response.status_code == 403
 
-    response = client.get("/legacy/collection/Legacy%20Wrong%20Collection")
-    assert response.status_code == 404
-
-    response = client.get("/legacy?cc=Legacy%20Wrong%20Collection")
-    assert response.status_code == 302
-    response = client.get(response.location)
-    assert response.status_code == 404
+    response = client.get(
+        "/legacy?c=Legacy%20Restricted%20Collection&c=Legacy%20Wrong%20Collection",
+        follow_redirects=True,
+    )
+    assert response.status_code == 403

site/tests/test_tasks.py

@@ -9,13 +9,14 @@
 from datetime import datetime, timedelta
 
 import pytest
-from cds_rdm.tasks import merge_duplicate_names_vocabulary, sync_local_accounts_to_names
 from invenio_access.permissions import system_identity
 from invenio_accounts import testutils
 from invenio_records_resources.proxies import current_service_registry
 from invenio_search.engine import dsl
 from invenio_vocabularies.contrib.names.api import Name
 
+from cds_rdm.tasks import merge_duplicate_names_vocabulary, sync_local_accounts_to_names
+
 
 @pytest.fixture(scope="function")
 def user_1(app):