From f9e70aca798882ae94738a484254fbf31fb25788 Mon Sep 17 00:00:00 2001 From: Matthias Bernt Date: Mon, 28 Oct 2024 17:09:01 +0100 Subject: [PATCH] sending notifications is task based only sinde 24.1 --- bioblend/_tests/TestGalaxyNotifications.py | 22 +++++++++++++++++----- 1 file changed, 17 insertions(+), 5 deletions(-) diff --git a/bioblend/_tests/TestGalaxyNotifications.py b/bioblend/_tests/TestGalaxyNotifications.py index a67b8b591..d28a8dc57 100755 --- a/bioblend/_tests/TestGalaxyNotifications.py +++ b/bioblend/_tests/TestGalaxyNotifications.py @@ -2,6 +2,7 @@ datetime, timedelta, ) +from packaging.version import Version from typing import ( Any, Dict, @@ -34,7 +35,9 @@ def test_notification_status(self): if not self.gi.users.get_current_user()["is_admin"]: self.skipTest("This tests requires the current user to be an admin, which is not the case.") - task_based = self.gi.config.get_config()["enable_celery_tasks"] + # sending notifications are handled by tasks since 24.1 if celery is enabled + version = self.gi.config.get_config()["version_major"] + task_based = self.gi.config.get_config()["enable_celery_tasks"] and (version == "dev" or Version(version) >= Version("24.1")) # user creation for the test user1 = self._create_local_test_user(password="password") @@ -150,7 +153,9 @@ def test_get_user_notifications(self): if not self.gi.users.get_current_user()["is_admin"]: self.skipTest("This tests requires the current user to be an admin, which is not the case.") - task_based = self.gi.config.get_config()["enable_celery_tasks"] + # sending notifications are handled by tasks since 24.1 if celery is enabled + version = self.gi.config.get_config()["version_major"] + task_based = self.gi.config.get_config()["enable_celery_tasks"] and (version == "dev" or Version(version) >= Version("24.1")) # send the notifications user_id = [self.gi.users.get_current_user()["id"]] @@ -248,7 +253,10 @@ def test_show_notification(self): self.skipTest("This Galaxy instance is not configured to use notifications.") if not self.gi.users.get_current_user()["is_admin"]: self.skipTest("This tests requires the current user to be an admin, which is not the case.") - task_based = self.gi.config.get_config()["enable_celery_tasks"] + + # sending notifications are handled by tasks since 24.1 if celery is enabled + version = self.gi.config.get_config()["version_major"] + task_based = self.gi.config.get_config()["enable_celery_tasks"] and (version == "dev" or Version(version) >= Version("24.1")) # user creation for the test user = self._create_local_test_user(password="password") @@ -280,7 +288,9 @@ def test_update_notifications(self): self.skipTest("This Galaxy instance is not configured to use notifications.") if not self.gi.users.get_current_user()["is_admin"]: self.skipTest("This tests requires the current user to be an admin, which is not the case.") - task_based = self.gi.config.get_config()["enable_celery_tasks"] + # sending notifications are handled by tasks since 24.1 if celery is enabled + version = self.gi.config.get_config()["version_major"] + task_based = self.gi.config.get_config()["enable_celery_tasks"] and (version == "dev" or Version(version) >= Version("24.1")) # user creation for the test user = self._create_local_test_user(password="password") @@ -363,7 +373,9 @@ def test_delete_notifications(self): self.skipTest("This Galaxy instance is not configured to use notifications.") if not self.gi.users.get_current_user()["is_admin"]: self.skipTest("This tests requires the current user to be an admin, which is not the case.") - task_based = self.gi.config.get_config()["enable_celery_tasks"] + # sending notifications are handled by tasks since 24.1 if celery is enabled + version = self.gi.config.get_config()["version_major"] + task_based = self.gi.config.get_config()["enable_celery_tasks"] and (version == "dev" or Version(version) >= Version("24.1")) # user creation for the test user = self._create_local_test_user(password="password")