context plus utc

invenio_requests/alembic/74b23178bfbe_change_datetime_types.py

@@ -0,0 +1,36 @@
+#
+# This file is part of Invenio.
+# Copyright (C) 2025 CERN.
+# Copyright (C) 2026 Graz University of Technology.
+#
+# 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.
+
+"""Alter datetime columns to utc aware datetime columns."""
+
+from invenio_db.utils import (
+    update_table_columns_column_type_to_datetime,
+    update_table_columns_column_type_to_utc_datetime,
+)
+
+# revision identifiers, used by Alembic.
+revision = "74b23178bfbe"
+down_revision = "1759321170"
+branch_labels = ()
+depends_on = None
+
+
+def upgrade():
+    """Upgrade database."""
+    for table_name in ["request_metadata", "request_events"]:
+        update_table_columns_column_type_to_utc_datetime(table_name, "created")
+        update_table_columns_column_type_to_utc_datetime(table_name, "updated")
+    update_table_columns_column_type_to_utc_datetime("request_metadata", "expires_at")
+
+
+def downgrade():
+    """Downgrade database."""
+    for table_name in ["request_metadata", "request_events"]:
+        update_table_columns_column_type_to_datetime(table_name, "created")
+        update_table_columns_column_type_to_datetime(table_name, "updated")
+    update_table_columns_column_type_to_datetime("request_metadata", "expires_at")

invenio_requests/records/models.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # Copyright (C) 2021 TU Wien.
-# Copyright (C) 2024 Graz University of Technology.
+# Copyright (C) 2024-2026 Graz University of Technology.
 #
 # Invenio-Requests is free software; you can redistribute it and/or modify
 # it under the terms of the MIT License; see LICENSE file for more details.
@@ -32,7 +32,7 @@ class RequestMetadata(db.Model, RecordMetadataBase):
     number = db.Column(String(50), unique=True, index=True, nullable=True)
 
     expires_at = db.Column(
-        db.DateTime().with_variant(mysql.DATETIME(fsp=6), "mysql"),
+        db.UTCDateTime(),
         default=None,
         nullable=True,
     )

invenio_requests/records/systemfields/expired_state.py

@@ -1,16 +1,17 @@
 # -*- coding: utf-8 -*-
 #
 # Copyright (C) 2021 - 2022 TU Wien.
+# Copyright (C) 2025 Graz University of Technology.
 #
 # Invenio-Requests is free software; you can redistribute it and/or modify
 # it under the terms of the MIT License; see LICENSE file for more details.
 
 """Systemfield for calculating the ``is_expired`` property of a request."""
 
 
-from datetime import datetime
+from datetime import datetime, timezone
 
-import pytz
+import arrow
 from invenio_records_resources.records.systemfields.calculated import CalculatedField
 
 
@@ -29,10 +30,7 @@ def calculate(self, record):
         if expires_at is None:
             return False
 
-        # comparing timezone-aware and naive datetimes results in an error
-        # https://docs.python.org/3/library/datetime.html#determining-if-an-object-is-aware-or-naive # noqa
-        now = datetime.utcnow()
-        if expires_at.tzinfo and expires_at.tzinfo.utcoffset(expires_at) is not None:
-            now = now.replace(tzinfo=pytz.utc)
+        expires_at = arrow.get(expires_at, tzinfo=timezone.utc).datetime
+        now = datetime.now(timezone.utc)
 
         return expires_at < now

invenio_requests/services/schemas.py

@@ -18,6 +18,7 @@
     fields,
 )
 from marshmallow_utils import fields as utils_fields
+from marshmallow_utils.context import context_schema
 
 from invenio_requests.proxies import current_requests
 
@@ -45,30 +46,31 @@ class RequestEventSchema(BaseRecordSchema):
     def get_permissions(self, obj):
         """Return permissions to act on comments or empty dict."""
         is_comment = obj.type == CommentEventType
+        current_identity = context_schema.get()["identity"]
+        current_request = context_schema.get().get("request", None)
         if is_comment:
             service = current_requests.request_events_service
-            request = self.context.get("request", None)
             permissions = {
                 "can_update_comment": service.check_permission(
-                    self.context["identity"],
+                    current_identity,
                     "update_comment",
                     event=obj,
-                    request=request,
+                    request=current_request,
                 ),
                 "can_delete_comment": service.check_permission(
-                    self.context["identity"],
+                    current_identity,
                     "delete_comment",
                     event=obj,
-                    request=request,
+                    request=current_request,
                 ),
             }
 
-            if request is not None:
+            if current_request is not None:
                 permissions["can_reply_comment"] = service.check_permission(
-                    self.context["identity"],
+                    current_identity,
                     "reply_comment",
                     event=obj,
-                    request=request,
+                    request=current_request,
                 )
 
             return permissions

invenio_requests/tasks.py

@@ -1,14 +1,14 @@
 # -*- coding: utf-8 -*-
 #
 # This file is part of Invenio.
-# Copyright (C) 2022 Graz University of Technology.
+# Copyright (C) 2022-2025 Graz University of Technology.
 #
 # 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.
 
 """Celery tasks for requests."""
 
-from datetime import datetime
+from datetime import datetime, timezone
 
 from celery import shared_task
 from flask import current_app
@@ -25,7 +25,7 @@
 def check_expired_requests():
     """Retrieve expired requests and perform expired action."""
     service = current_requests_service
-    now = datetime.utcnow().isoformat()
+    now = datetime.now(timezone.utc).isoformat()
 
     # using scan to get all requests
     requests_list = service.scan(

tests/customizations/test_request_types.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # Copyright (C) 2021 TU Wien.
+# Copyright (C) 2025 Graz University of Technology.
 #
 # Invenio-Requests is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -17,7 +18,6 @@
     RequestState,
     RequestType,
 )
-from invenio_requests.customizations.actions import CreateAction
 from invenio_requests.errors import NoSuchActionError
 from invenio_requests.records.api import Request
 from invenio_requests.services.permissions import PermissionPolicy
@@ -30,7 +30,7 @@ class CustomCreateAction(CreateAction):
     status_to = "not_closed"
 
 
-class TestAction(RequestAction):
+class ActionTest(RequestAction):
     """Test action."""
 
     status_from = ["not_closed"]
@@ -47,7 +47,7 @@ class CustomizedReferenceRequestType(RequestType):
 
     type_id = "customized-reference-request"
 
-    available_actions = {"custom-create": CustomCreateAction, "test": TestAction}
+    available_actions = {"custom-create": CustomCreateAction, "test": ActionTest}
     available_statuses = {
         "not_closed": RequestState.OPEN,
         "closed": RequestState.CLOSED,

tests/records/systemfields/test_calculated_systemfield.py

@@ -1,13 +1,14 @@
 # -*- coding: utf-8 -*-
 #
 # Copyright (C) 2021 TU Wien.
+# Copyright (C) 2025 Graz University of Technology.
 #
 # Invenio-Requests is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
 
 """Test the calculated systemfields."""
 
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 import pytest
 
@@ -16,7 +17,7 @@
 
 def test_expired_systemfield(example_request):
     """Test if the expired system field works as intended."""
-    now = datetime.utcnow()
+    now = datetime.now(timezone.utc)
     example_request.expires_at = None
     example_request.commit()
 

tests/resources/user_moderation/test_user_moderation_resources.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # Copyright (C) 2023 CERN.
+# Copyright (C) 2025 Graz University of Technology.
 #
 # Invenio-Requests is free software; you can redistribute it and/or modify
 # it under the terms of the MIT License; see LICENSE file for more details.
@@ -22,7 +23,7 @@ def mod_request(app, user1, moderator_user):
     return request_item
 
 
-def test_moderate(app, es_clear, client_logged_as, headers, mod_request):
+def test_moderate(app, search_clear, client_logged_as, headers, mod_request):
     # Log as moderator
     client = client_logged_as("[email protected]")
 
@@ -55,7 +56,13 @@ def test_moderate(app, es_clear, client_logged_as, headers, mod_request):
     ],
 )
 def test_moderate_invalid_user(
-    app, es_clear, client_logged_as, headers, mod_request, invalid_action, expected_code
+    app,
+    search_clear,
+    client_logged_as,
+    headers,
+    mod_request,
+    invalid_action,
+    expected_code,
 ):
     """Tests that a regular user can't moderate."""
     # Log as a normal user that can't moderate
@@ -80,7 +87,13 @@ def test_moderate_invalid_user(
     ],
 )
 def test_invalid_actions_after_submit(
-    app, es_clear, client_logged_as, headers, mod_request, invalid_action, expected_code
+    app,
+    search_clear,
+    client_logged_as,
+    headers,
+    mod_request,
+    invalid_action,
+    expected_code,
 ):
     """Test invalid actions on a user moderation request.
 
@@ -105,7 +118,7 @@ def test_invalid_actions_after_submit(
     assert response.status_code == expected_code
 
 
-def test_search_as_moderator(app, es_clear, client_logged_as, headers, mod_request):
+def test_search_as_moderator(app, search_clear, client_logged_as, headers, mod_request):
     """Test search as a moderator."""
     # Log as moderator
     mod_email = "[email protected]"
@@ -123,7 +136,7 @@ def test_search_as_moderator(app, es_clear, client_logged_as, headers, mod_reque
     assert hit["status"] == "submitted"
 
 
-def test_search_as_user(app, es_clear, client_logged_as, headers, mod_request):
+def test_search_as_user(app, search_clear, client_logged_as, headers, mod_request):
     """Test search as a regular user."""
     client = client_logged_as("[email protected]")
 
@@ -136,7 +149,7 @@ def test_search_as_user(app, es_clear, client_logged_as, headers, mod_request):
     assert len(hits) == 0
 
 
-def test_links(app, es_clear, client_logged_as, headers, mod_request):
+def test_links(app, search_clear, client_logged_as, headers, mod_request):
     """Test links on search."""
     # Log as moderator
     mod_email = "[email protected]"

tests/services/requests/test_requests_tasks.py

@@ -1,14 +1,14 @@
 # -*- coding: utf-8 -*-
 #
-# Copyright (C) 2022 Graz University of Technology.
+# Copyright (C) 2022-2026 Graz University of Technology.
 #
 # Invenio-Requests is free software; you can redistribute it and/or
 # modify it under the terms of the MIT License; see LICENSE file for more
 # details.
 
 """Tasks tests."""
 
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 from invenio_access.permissions import system_identity
 from invenio_search.engine import dsl
@@ -21,7 +21,7 @@ def test_check_expired_requests(
     app, identity_simple, create_request, submit_request, requests_service
 ):
     """Test if the expired system field works as intended."""
-    now = datetime.utcnow()
+    now = datetime.now(timezone.utc).replace(tzinfo=None)
 
     # created only should not be picked up
     created_request = create_request(

tests/services/user_moderation/test_user_moderation_service.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # Copyright (C) 2023 CERN.
+# Copyright (C) 2025 Graz University of Technology.
 #
 # Invenio-Requests is free software; you can redistribute it and/or modify
 # it under the terms of the MIT License; see LICENSE file for more details.
@@ -21,7 +22,7 @@
 from invenio_requests.services.user_moderation.errors import OpenRequestAlreadyExists
 
 
-def test_request_moderation(app, es_clear, users, identity_simple, mod_identity):
+def test_request_moderation(app, search_clear, users, identity_simple, mod_identity):
     """Tests the service for request moderation."""
 
     user1 = users["user1"]
@@ -46,7 +47,7 @@ def test_request_moderation(app, es_clear, users, identity_simple, mod_identity)
     assert request_item._request.status == "submitted"
 
 
-def test_search_moderation(app, es_clear, users, submit_request, mod_identity):
+def test_search_moderation(app, search_clear, users, submit_request, mod_identity):
     """Tests the search for request moderation."""
 
     user = users["user2"]
@@ -80,7 +81,7 @@ def test_search_moderation(app, es_clear, users, submit_request, mod_identity):
     assert search.total == 2  # System process can see both requests
 
 
-def test_moderation_accept(app, es_clear, users, mod_identity):
+def test_moderation_accept(app, search_clear, users, mod_identity):
     """Tests the service for moderation."""
     user = users["user1"]
 
@@ -103,7 +104,7 @@ def test_moderation_accept(app, es_clear, users, mod_identity):
         )
 
 
-def test_moderation_decline(app, es_clear, users, mod_identity):
+def test_moderation_decline(app, search_clear, users, mod_identity):
     """Tests the service for moderation."""
     user = users["user1"]
 
@@ -126,7 +127,7 @@ def test_moderation_decline(app, es_clear, users, mod_identity):
         )
 
 
-def test_read(app, es_clear, users, mod_identity):
+def test_read(app, search_clear, users, mod_identity):
     """Tests the service for read."""
     user = users["user1"]
 
@@ -146,7 +147,7 @@ def test_read(app, es_clear, users, mod_identity):
         current_requests_service.read(user_identity, request_item.id)
 
 
-def test_invalid_request_data(app, es_clear, users, mod_identity):
+def test_invalid_request_data(app, search_clear, users, mod_identity):
     """Test request creation with invalid data."""
     user = users["user1"]
 
@@ -158,7 +159,7 @@ def test_invalid_request_data(app, es_clear, users, mod_identity):
         )
 
 
-def test_duplicate_request(app, es_clear, users, mod_identity):
+def test_duplicate_request(app, search_clear, users, mod_identity):
     """Test request creation when a request is already open."""
     user = users["user1"]
     service = current_user_moderation_service