Skip to content

Commit

Permalink
sending notifications is task based only sinde 24.1
Browse files Browse the repository at this point in the history
  • Loading branch information
bernt-matthias committed Oct 28, 2024
1 parent 31ed6ba commit a1e9cfd
Showing 1 changed file with 27 additions and 5 deletions.
32 changes: 27 additions & 5 deletions bioblend/_tests/TestGalaxyNotifications.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
datetime,
timedelta,
)
from packaging.version import Version
from typing import (
Any,
Dict,
Expand Down Expand Up @@ -34,7 +35,11 @@ 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")
Expand Down Expand Up @@ -150,7 +155,11 @@ 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"]]
Expand Down Expand Up @@ -248,7 +257,12 @@ 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")
Expand Down Expand Up @@ -280,7 +294,11 @@ 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")
Expand Down Expand Up @@ -363,7 +381,11 @@ 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")
Expand Down

0 comments on commit a1e9cfd

Please sign in to comment.