Merge pull request #16220 from jmchilton/test_decorators

Small test decorator improvements.

lib/galaxy/selenium/navigates_galaxy.py

@@ -534,22 +534,48 @@ def published_grid_search_for(self, search_term=None):
             search_term,
         )
 
-    def get_logged_in_user(self):
-        return self.api_get("users/current")
+    def get_logged_in_user(self) -> Optional[Dict[str, Any]]:
+        user_dict = self.api_get("users/current")
+        # for user's not logged in - this just returns a {} so lets
+        # key this on an id being available?
+        if "id" in user_dict:
+            return user_dict
+        else:
+            return None
 
-    def get_api_key(self, force=False):
-        if not force:
-            response = self.api_get(f"users/{self.get_user_id()}/api_key/detailed")
+    def get_api_key(self, force=False) -> Optional[str]:
+        user_id = self.get_user_id()
+        if user_id is None:
+            if force:
+                raise Exception("Attempting to get_api_key but no user logged in.")
+            else:
+                return None
+        elif not force:
+            response = self.api_get(f"users/{user_id}/api_key/detailed")
             return response["key"] if response else None
         else:
-            return self.api_post(f"users/{self.get_user_id()}/api_key")
+            return self.api_post(f"users/{user_id}/api_key")
 
-    def get_user_id(self):
+    def get_user_id(self) -> Optional[str]:
         user = self.get_logged_in_user()
-        return user["id"]
+        if user is not None:
+            return user["id"]
+        else:
+            return None
 
-    def is_logged_in(self):
-        return "email" in self.get_logged_in_user()
+    def get_user_email(self) -> str:
+        user = self.get_logged_in_user()
+        if user is None:
+            raise Exception("No user is logged in, cannot fetch user e-mail.")
+        else:
+            return user["email"]
+
+    def is_logged_in(self) -> bool:
+        user_object = self.get_logged_in_user()
+        if not user_object:
+            return False
+        else:
+            return "email" in user_object
 
     @retry_during_transitions
     def _inline_search_for(self, selector, search_term=None, escape_to_clear=False):
@@ -653,7 +679,9 @@ def register(self, email=None, password=None, username=None, confirm=None, asser
 
             text = username_element.text
             assert username in text
-            assert self.get_logged_in_user()["email"] == email
+            user_object = self.get_logged_in_user()
+            assert user_object and "email" in user_object
+            assert user_object["email"] == email
 
             # clicking away no longer closes menu post Masthead -> VueJS
             self.click_masthead_user()

lib/galaxy/tool_util/verify/interactor.py

@@ -179,7 +179,7 @@ def __init__(self, **kwds):
         self.api_url = f"{kwds['galaxy_url'].rstrip('/')}/api"
         self.cookies = None
         self.master_api_key = kwds["master_api_key"]
-        self.api_key = self.__get_user_key(
+        self.api_key = self._get_user_key(
             kwds.get("api_key"), kwds.get("master_api_key"), test_user=kwds.get("test_user")
         )
         if kwds.get("user_api_key_is_admin_key", False):
@@ -204,7 +204,9 @@ def target_galaxy_version(self):
     def supports_test_data_download(self):
         return self.target_galaxy_version >= Version("19.01")
 
-    def __get_user_key(self, user_key: Optional[str], admin_key: Optional[str], test_user: Optional[str] = None) -> str:
+    def _get_user_key(
+        self, user_key: Optional[str], admin_key: Optional[str], test_user: Optional[str] = None
+    ) -> Optional[str]:
         if not test_user:
             test_user = "[email protected]"
         if user_key:

lib/galaxy/webapps/galaxy/api/plugins.py

@@ -9,7 +9,10 @@
     histories,
 )
 from galaxy.util import asbool
-from galaxy.web import expose_api
+from galaxy.web import (
+    expose_api,
+    expose_api_anonymous_and_sessionless,
+)
 from . import (
     BaseGalaxyAPIController,
     depends,
@@ -26,12 +29,12 @@ class PluginsController(BaseGalaxyAPIController):
     hda_manager: hdas.HDAManager = depends(hdas.HDAManager)
     history_manager: histories.HistoryManager = depends(histories.HistoryManager)
 
-    @expose_api
+    @expose_api_anonymous_and_sessionless
     def index(self, trans, **kwargs):
         """
         GET /api/plugins:
         """
-        registry = self._get_registry(trans)
+        registry = self._get_registry()
         dataset_id = kwargs.get("dataset_id")
         if dataset_id is not None:
             hda = self.hda_manager.get_accessible(self.decode_id(dataset_id), trans.user)
@@ -45,7 +48,7 @@ def show(self, trans, id, **kwargs):
         """
         GET /api/plugins/{id}:
         """
-        registry = self._get_registry(trans)
+        registry = self._get_registry()
         history_id = kwargs.get("history_id")
         if history_id is not None:
             history = self.history_manager.get_owned(
@@ -59,7 +62,7 @@ def show(self, trans, id, **kwargs):
             result = registry.get_plugin(id).to_dict()
         return result
 
-    def _get_registry(self, trans):
-        if not trans.app.visualizations_registry:
+    def _get_registry(self):
+        if not self.app.visualizations_registry:
             raise exceptions.MessageException("The visualization registry has not been configured.")
-        return trans.app.visualizations_registry
+        return self.app.visualizations_registry

lib/galaxy_test/base/api.py

@@ -12,6 +12,7 @@
 
 import pytest
 import requests
+from typing_extensions import Protocol
 
 from galaxy.util.properties import get_from_env
 from .api_asserts import (
@@ -84,6 +85,12 @@ def celery_parameters(self):
         }
 
 
+class HasAnonymousGalaxyInteractor(Protocol):
+    @property
+    def anonymous_galaxy_interactor(self) -> "ApiTestInteractor":
+        """Return an optionally anonymous galaxy interactor."""
+
+
 class UsesApiTestCaseMixin:
     url: str
     _galaxy_interactor: Optional["ApiTestInteractor"] = None
@@ -112,13 +119,24 @@ def _setup_interactor(self):
         self.master_api_key = get_admin_api_key()
         self._galaxy_interactor = self._get_interactor()
 
+    @property
+    def anonymous_galaxy_interactor(self) -> "ApiTestInteractor":
+        """Return an optionally anonymous galaxy interactor.
+
+        Lighter requirements for use with API requests that may not required an API key.
+        """
+        return self.galaxy_interactor
+
     @property
     def galaxy_interactor(self) -> "ApiTestInteractor":
         assert self._galaxy_interactor is not None
         return self._galaxy_interactor
 
-    def _get_interactor(self, api_key=None) -> "ApiTestInteractor":
-        return ApiTestInteractor(self, api_key=api_key)
+    def _get_interactor(self, api_key=None, allow_anonymous=False) -> "ApiTestInteractor":
+        if allow_anonymous and api_key is None:
+            return AnonymousGalaxyInteractor(self)
+        else:
+            return ApiTestInteractor(self, api_key=api_key)
 
     def _setup_user(self, email, password=None):
         return self.galaxy_interactor.ensure_user_with_email(email, password=password)
@@ -226,3 +244,13 @@ def patch(self, *args, **kwds):
 
     def put(self, *args, **kwds):
         return self._put(*args, **kwds)
+
+
+class AnonymousGalaxyInteractor(ApiTestInteractor):
+    def __init__(self, test_case):
+        super().__init__(test_case)
+
+    def _get_user_key(
+        self, user_key: Optional[str], admin_key: Optional[str], test_user: Optional[str] = None
+    ) -> Optional[str]:
+        return None

lib/galaxy_test/base/api_asserts.py

@@ -4,6 +4,7 @@
     Any,
     cast,
     Dict,
+    Optional,
     Union,
 )
 
@@ -15,33 +16,37 @@
 ASSERT_FAIL_STATUS_CODE = "Request status code (%d) was not expected value %s. Body was %s"
 
 
-def assert_status_code_is(response: Response, expected_status_code: int):
+def assert_status_code_is(response: Response, expected_status_code: int, failure_message: Optional[str] = None):
     """Assert that the supplied response has the expect status code."""
     response_status_code = response.status_code
     if expected_status_code != response_status_code:
-        _report_status_code_error(response, expected_status_code)
+        _report_status_code_error(response, expected_status_code, failure_message)
 
 
-def assert_status_code_is_ok(response: Response):
+def assert_status_code_is_ok(response: Response, failure_message: Optional[str] = None):
     """Assert that the supplied response is okay.
 
     The easier alternative ``response.raise_for_status()`` might be
-    perferable generally.
+    preferable generally.
 
     .. seealso:: :py:meth:`requests.Response.raise_for_status()`
     """
     response_status_code = response.status_code
     is_two_hundred_status_code = response_status_code >= 200 and response_status_code <= 300
     if not is_two_hundred_status_code:
-        _report_status_code_error(response, "2XX")
+        _report_status_code_error(response, "2XX", failure_message)
 
 
-def _report_status_code_error(response: Response, expected_status_code: Union[str, int]):
+def _report_status_code_error(
+    response: Response, expected_status_code: Union[str, int], failure_message: Optional[str]
+):
     try:
         body = response.json()
     except Exception:
         body = f"INVALID JSON RESPONSE <{response.text}>"
     assertion_message = ASSERT_FAIL_STATUS_CODE % (response.status_code, expected_status_code, body)
+    if failure_message:
+        assertion_message = f"{failure_message}. {assertion_message}"
     raise AssertionError(assertion_message)
 
 

lib/galaxy_test/base/populators.py

@@ -108,7 +108,10 @@
 )
 from galaxy_test.base.json_schema_utils import JsonSchemaValidator
 from . import api_asserts
-from .api import ApiTestInteractor
+from .api import (
+    ApiTestInteractor,
+    HasAnonymousGalaxyInteractor,
+)
 from .api_util import random_name
 
 FILE_URL = "https://raw.githubusercontent.com/galaxyproject/galaxy/dev/test-data/4.bed"
@@ -143,15 +146,17 @@ def wrapped_method(*args, **kwargs):
     return wrapped_method
 
 
-def skip_without_tool(tool_id):
+def skip_without_tool(tool_id: str):
     """Decorate an API test method as requiring a specific tool.
 
     Have test framework skip the test case if the tool is unavailable.
     """
 
     def method_wrapper(method):
-        def get_tool_ids(api_test_case):
-            index = api_test_case.galaxy_interactor.get("tools", data=dict(in_panel=False))
+        def get_tool_ids(api_test_case: HasAnonymousGalaxyInteractor):
+            interactor = api_test_case.anonymous_galaxy_interactor
+            index = interactor.get("tools", data=dict(in_panel=False))
+            api_asserts.assert_status_code_is_ok(index, "Failed to fetch toolbox for target Galaxy.")
             tools = index.json()
             # In panels by default, so flatten out sections...
             tool_ids = [itemgetter("id")(_) for _ in tools]
@@ -169,8 +174,11 @@ def wrapped_method(api_test_case, *args, **kwargs):
 
 def skip_without_asgi(method):
     @wraps(method)
-    def wrapped_method(api_test_case, *args, **kwd):
-        config = api_test_case.galaxy_interactor.get("configuration").json()
+    def wrapped_method(api_test_case: HasAnonymousGalaxyInteractor, *args, **kwd):
+        interactor = api_test_case.anonymous_galaxy_interactor
+        config_response = interactor.get("configuration")
+        api_asserts.assert_status_code_is_ok(config_response, "Failed to fetch configuration for target Galaxy.")
+        config = config_response.json()
         asgi_enabled = config.get("asgi_enabled", False)
         if not asgi_enabled:
             raise unittest.SkipTest("ASGI not enabled, skipping test")
@@ -179,15 +187,16 @@ def wrapped_method(api_test_case, *args, **kwd):
     return wrapped_method
 
 
-def skip_without_datatype(extension):
+def skip_without_datatype(extension: str):
     """Decorate an API test method as requiring a specific datatype.
 
     Have test framework skip the test case if the datatype is unavailable.
     """
 
-    def has_datatype(api_test_case):
-        index_response = api_test_case.galaxy_interactor.get("datatypes")
-        assert index_response.status_code == 200, "Failed to fetch datatypes for target Galaxy."
+    def has_datatype(api_test_case: HasAnonymousGalaxyInteractor):
+        interactor = api_test_case.anonymous_galaxy_interactor
+        index_response = interactor.get("datatypes", anon=True)
+        api_asserts.assert_status_code_is_ok(index_response, "Failed to fetch datatypes for target Galaxy.")
         datatypes = index_response.json()
         assert isinstance(datatypes, list)
         return extension in datatypes
@@ -203,6 +212,26 @@ def wrapped_method(api_test_case, *args, **kwargs):
     return method_wrapper
 
 
+def skip_without_visualization_plugin(plugin_name: str):
+    def has_plugin(api_test_case: HasAnonymousGalaxyInteractor):
+        interactor = api_test_case.anonymous_galaxy_interactor
+        index_response = interactor.get("plugins", anon=True)
+        api_asserts.assert_status_code_is_ok(index_response, "Failed to fetch visualizations for target Galaxy.")
+        plugins = index_response.json()
+        assert isinstance(plugins, list)
+        return plugin_name in [p["name"] for p in plugins]
+
+    def method_wrapper(method):
+        @wraps(method)
+        def wrapped_method(api_test_case, *args, **kwargs):
+            _raise_skip_if(not has_plugin(api_test_case))
+            method(api_test_case, *args, **kwargs)
+
+        return wrapped_method
+
+    return method_wrapper
+
+
 skip_if_toolshed_down = skip_if_site_down("https://toolshed.g2.bx.psu.edu")
 
 

lib/galaxy_test/selenium/framework.py

@@ -343,6 +343,12 @@ def get_download_path(self):
         """Returns default download path"""
         return DEFAULT_DOWNLOAD_PATH
 
+    @property
+    def anonymous_galaxy_interactor(self):
+        api_key = self.get_api_key(force=False)
+        interactor = self._get_interactor(api_key=api_key, allow_anonymous=True)
+        return interactor
+
     def api_interactor_for_logged_in_user(self):
         api_key = self.get_api_key(force=True)
         interactor = self._get_interactor(api_key=api_key)

lib/galaxy_test/selenium/test_history_sharing.py

@@ -135,7 +135,7 @@ class TestHistoryRequiresLoginSelenium(SeleniumTestCase):
 
     @selenium_test
     def test_share_history_login_redirect(self):
-        user_email = self.get_logged_in_user()["email"]
+        user_email = self.get_user_email()
         history_id = self.current_history_id()
         self.logout()
         self.go_to_history_sharing(history_id)