change utcnow() to Galaxy owned now() function

Author: bgruening

test/integration/test_notifications.py

@@ -8,6 +8,7 @@
 )
 from uuid import uuid4
 
+from galaxy.model.orm.now import now
 from galaxy_test.base.populators import (
     DatasetPopulator,
     WorkflowPopulator,
@@ -61,7 +62,7 @@ def test_notification_status(self):
         user1 = self._create_test_user()
         user2 = self._create_test_user()
 
-        before_creating_notifications = datetime.utcnow()
+        before_creating_notifications = now()
 
         # Only user1 will receive this notification
         subject1 = f"notification_{uuid4()}"
@@ -81,7 +82,7 @@ def test_notification_status(self):
         created_response_3 = self._send_broadcast_notification("test_notification_status 3")
         assert created_response_3["total_notifications_sent"] == 1
 
-        after_creating_notifications = datetime.utcnow()
+        after_creating_notifications = now()
 
         # The default user should have received only the broadcasted notifications
         status = self._get_notifications_status_since(before_creating_notifications)
@@ -158,7 +159,7 @@ def test_delete_notification_by_user(self):
         user1 = self._create_test_user()
         user2 = self._create_test_user()
 
-        before_creating_notifications = datetime.utcnow()
+        before_creating_notifications = now()
 
         subject = f"notification_{uuid4()}"
         created_response = self._send_test_notification_to(
@@ -265,8 +266,8 @@ def test_update_notifications(self):
         assert updated_response["source"] == "updated_source"
 
     def test_admins_get_all_broadcasted_even_inactive(self):
-        tomorrow = datetime.utcnow() + timedelta(days=1)
-        yesterday = datetime.utcnow() - timedelta(days=1)
+        tomorrow = now() + timedelta(days=1)
+        yesterday = now() - timedelta(days=1)
         self._send_broadcast_notification(subject="Active")
         self._send_broadcast_notification(subject="Scheduled", publication_time=tomorrow)
         self._send_broadcast_notification(subject="Expired", expiration_time=yesterday)

test/unit/app/managers/test_NotificationManager.py

@@ -24,6 +24,7 @@
     Role,
     User,
 )
+from galaxy.model.orm.now import now
 from galaxy.schema.notifications import (
     BroadcastNotificationContent,
     BroadcastNotificationCreateRequest,
@@ -88,7 +89,7 @@ def _send_message_notification_to_users(self, users: list[User], notification: O
         return created_notification, notifications_sent
 
     def _has_expired(self, expiration_time: Optional[datetime]) -> bool:
-        return expiration_time < datetime.utcnow() if expiration_time else False
+        return expiration_time < now() if expiration_time else False
 
     def _assert_notification_expected(self, actual_notification: Any, expected_notification: dict[str, Any]):
         assert actual_notification
@@ -146,13 +147,13 @@ def test_get_broadcasted_notification(self):
         assert actual_notification.id == created_notification.id
 
     def test_get_all_broadcasted_notifications(self):
-        now = datetime.utcnow()
-        next_week = now + timedelta(days=7)
-        next_month = now + timedelta(days=30)
+        current_time = now()
+        next_week = current_time + timedelta(days=7)
+        next_month = current_time + timedelta(days=30)
 
         notification_data = self._default_broadcast_notification_data()
         notification_data["content"]["subject"] = "Recent Notification"
-        notification_data["publication_time"] = now
+        notification_data["publication_time"] = current_time
         self._send_broadcast_notification(notification_data)
 
         notification_data = self._default_broadcast_notification_data()
@@ -179,18 +180,18 @@ def test_get_all_broadcasted_notifications(self):
         assert notifications[0].content["subject"] == "Scheduled Next Month Notification"
 
     def test_update_broadcasted_notification(self):
-        next_month = datetime.utcnow() + timedelta(days=30)
+        next_month = now() + timedelta(days=30)
         notification_data = self._default_broadcast_notification_data()
         notification_data["content"]["subject"] = "Old Scheduled Notification"
         notification_data["publication_time"] = next_month
         actual_notification = self._send_broadcast_notification(notification_data)
 
-        now = datetime.utcnow()
+        current_time = now()
         expected_content = BroadcastNotificationContent(subject="Updated Notification", message="Updated Message")
         update_request = NotificationBroadcastUpdateRequest(
             source="updated_source",
             variant=NotificationVariant.warning,
-            publication_time=now,
+            publication_time=current_time,
             content=expected_content,
         )
         updated_count = self.notification_manager.update_broadcasted_notification(
@@ -206,7 +207,7 @@ def test_update_broadcasted_notification(self):
         assert content["message"] == expected_content.message
 
     def test_cleanup_expired_broadcast_notifications(self):
-        one_hour_ago = datetime.utcnow() - timedelta(hours=1)
+        one_hour_ago = now() - timedelta(hours=1)
         notification_data = self._default_broadcast_notification_data()
         notification_data["expiration_time"] = one_hour_ago
         actual_notification = self._send_broadcast_notification(notification_data)
@@ -290,7 +291,7 @@ def test_update_user_notifications(self):
 
     def test_scheduled_notifications(self):
         user = self._create_test_user()
-        tomorrow = datetime.utcnow() + timedelta(hours=24)
+        tomorrow = now() + timedelta(hours=24)
         expected_notification = self._default_test_notification_data()
         expected_notification["source"] = "test_scheduled"
         expected_notification["publication_time"] = tomorrow
@@ -347,8 +348,10 @@ def test_update_user_notification_preferences(self):
 
     def test_cleanup_expired_notifications(self):
         user = self._create_test_user()
-        now = datetime.utcnow()
-        notification, _ = self._send_message_notification_to_users([user], notification={"expiration_time": now})
+        current_time = now()
+        notification, _ = self._send_message_notification_to_users(
+            [user], notification={"expiration_time": current_time}
+        )
         user_notification = self.notification_manager.get_user_notification(user, notification.id, active_only=False)
         assert user_notification
         assert self._has_expired(user_notification.expiration_time) is True

test/unit/app/managers/test_UserManager.py

@@ -4,7 +4,6 @@
 Executable directly using: python -m test.unit.managers.test_UserManager
 """
 
-from datetime import datetime
 from unittest.mock import patch
 
 from sqlalchemy import (
@@ -21,6 +20,7 @@
     histories,
     users,
 )
+from galaxy.model.orm.now import now
 from galaxy.security.passwords import check_password
 from .base import BaseTestCase
 
@@ -178,7 +178,7 @@ def test_change_password(self):
         )
         assert check_password(default_password, user2.password)
         assert not check_password(changed_password, user2.password)
-        prt.expiration_time = datetime.utcnow()
+        prt.expiration_time = now()
         user, message = self.user_manager.change_password(
             self.trans, token=prt.token, password=default_password, confirm=default_password
         )