pytest-y API test fixtures

lib/galaxy_test/api/conftest.py

@@ -0,0 +1,161 @@
+"""Fixtures for a version of API testing that relies more heavily on pytest injection."""
+
+import os
+from dataclasses import dataclass
+from typing import (
+    Any,
+    Iterator,
+    List,
+    Optional,
+)
+
+import pytest
+
+from galaxy.tool_util.verify.test_data import TestDataResolver
+from galaxy_test.base.api import (
+    AnonymousGalaxyInteractor,
+    ApiTestInteractor,
+)
+from galaxy_test.base.api_util import (
+    get_admin_api_key,
+    get_user_api_key,
+)
+from galaxy_test.base.env import setup_keep_outdir
+from galaxy_test.base.populators import (
+    _raise_skip_if,
+    DatasetCollectionPopulator,
+    DatasetPopulator,
+    get_tool_ids,
+    RequiredTool,
+    TargetHistory,
+)
+from galaxy_test.base.testcase import host_port_and_url
+
+
+@dataclass
+class ApiConfigObject:
+    host: str
+    port: Optional[str]
+    url: str
+    user_api_key: Optional[str]
+    admin_api_key: Optional[str]
+    test_data_resolver: Any
+    keepOutdir: Any
+
+
+@pytest.fixture(scope="session")
+def api_test_config_object(real_driver) -> ApiConfigObject:
+    host, port, url = host_port_and_url(real_driver)
+    user_api_key = get_user_api_key()
+    admin_api_key = get_admin_api_key()
+    test_data_resolver = TestDataResolver()
+    keepOutdir = setup_keep_outdir()
+    return ApiConfigObject(
+        host,
+        port,
+        url,
+        user_api_key,
+        admin_api_key,
+        test_data_resolver,
+        keepOutdir,
+    )
+
+
+@pytest.fixture(scope="session")
+def galaxy_interactor(api_test_config_object: ApiConfigObject) -> ApiTestInteractor:
+    return ApiTestInteractor(api_test_config_object)
+
+
+@pytest.fixture(scope="session")
+def dataset_populator(galaxy_interactor: ApiTestInteractor) -> DatasetPopulator:
+    return DatasetPopulator(galaxy_interactor)
+
+
+@pytest.fixture(scope="session")
+def dataset_collection_populator(galaxy_interactor: ApiTestInteractor) -> DatasetCollectionPopulator:
+    return DatasetCollectionPopulator(galaxy_interactor)
+
+
+@pytest.fixture(scope="session")
+def anonymous_galaxy_interactor(api_test_config_object: ApiConfigObject) -> Iterator[AnonymousGalaxyInteractor]:
+    return AnonymousGalaxyInteractor(api_test_config_object)
+
+
+_celery_app = None
+_celery_worker = None
+
+
+@pytest.fixture(autouse=True, scope="session")
+def request_celery_app(celery_session_app, celery_config):
+    try:
+        global _celery_app
+        _celery_app = celery_session_app
+        yield
+    finally:
+        if os.environ.get("GALAXY_TEST_EXTERNAL") is None:
+            from galaxy.celery import celery_app
+
+            celery_app.fork_pool.stop()
+            celery_app.fork_pool.join(timeout=5)
+
+
+@pytest.fixture(autouse=True, scope="session")
+def request_celery_worker(celery_session_worker, celery_config, celery_worker_parameters):
+    global _celery_worker
+    _celery_worker = celery_session_worker
+
+
+@pytest.fixture(scope="session", autouse=True)
+def celery_worker_parameters():
+    return {
+        "queues": ("galaxy.internal", "galaxy.external"),
+    }
+
+
+@pytest.fixture(scope="session")
+def celery_parameters():
+    return {
+        "task_create_missing_queues": True,
+        "task_default_queue": "galaxy.internal",
+    }
+
+
+@pytest.fixture
+def history_id(dataset_populator: DatasetPopulator, request) -> Iterator[str]:
+    history_name = f"API Test History for {request.node.nodeid}"
+    with dataset_populator.test_history(name=history_name) as history_id:
+        yield history_id
+
+
+@pytest.fixture
+def target_history(
+    dataset_populator: DatasetPopulator, dataset_collection_populator: DatasetCollectionPopulator, history_id: str
+) -> TargetHistory:
+    return TargetHistory(dataset_populator, dataset_collection_populator, history_id)
+
+
+@pytest.fixture
+def required_tool(dataset_populator: DatasetPopulator, history_id: str, required_tool_ids: List[str]) -> RequiredTool:
+    if len(required_tool_ids) != 1:
+        raise AssertionError("required_tool fixture must only be used on methods that require a single tool")
+    tool_id = required_tool_ids[0]
+    tool = RequiredTool(dataset_populator, tool_id, history_id)
+    return tool
+
+
+@pytest.fixture(autouse=True)
+def check_required_tools(anonymous_galaxy_interactor, request):
+    for marker in request.node.iter_markers():
+        if marker.name == "requires_tool_id":
+            tool_id = marker.args[0]
+            _raise_skip_if(tool_id not in get_tool_ids(anonymous_galaxy_interactor))
+
+
+@pytest.fixture
+def required_tool_ids(request) -> List[str]:
+    tool_ids = []
+    for marker in request.node.iter_markers():
+        if marker.name == "requires_tool_id":
+            tool_id = marker.args[0]
+            tool_ids.append(tool_id)
+    return tool_ids

lib/galaxy_test/base/api_asserts.py

@@ -24,8 +24,8 @@ def assert_status_code_is(response: Response, expected_status_code: int, failure
 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
-    preferable generally.
+    This is an alternative to ``response.raise_for_status()`` with a more detailed
+    error message.
 
     .. seealso:: :py:meth:`requests.Response.raise_for_status()`
     """
@@ -35,6 +35,17 @@ def assert_status_code_is_ok(response: Response, failure_message: Optional[str]
         _report_status_code_error(response, "2XX", failure_message)
 
 
+def assert_status_code_is_not_ok(response: Response, failure_message: Optional[str] = None):
+    """Assert that the supplied response is not okay.
+
+    .. seealso:: :py:meth:`assert_status_code_is_ok`
+    """
+    response_status_code = response.status_code
+    is_two_hundred_status_code = response_status_code >= 200 and response_status_code <= 300
+    if is_two_hundred_status_code:
+        _report_status_code_error(response, "2XX", failure_message)
+
+
 def _report_status_code_error(
     response: Response, expected_status_code: Union[str, int], failure_message: Optional[str]
 ):

lib/galaxy_test/base/decorators.py

@@ -67,6 +67,14 @@ def wrapped_method(*args, **kwargs):
     return wrapped_method
 
 
+def requires_tool_id(tool_id: str):
+
+    def method_wrapper(method):
+        return getattr(pytest.mark, "requires_tool_id")(tool_id)(method)
+
+    return method_wrapper
+
+
 def requires_new_history(method):
     return _wrap_method_with_galaxy_requirement(method, "new_history")
 

lib/galaxy_test/base/interactor.py

@@ -4,9 +4,12 @@
 class TestCaseGalaxyInteractor(GalaxyInteractorApi):
     def __init__(self, functional_test_case, test_user=None, api_key=None):
         self.functional_test_case = functional_test_case
+        admin_api_key = getattr(functional_test_case, "master_api_key", None) or getattr(
+            functional_test_case, "admin_api_key", None
+        )
         super().__init__(
             galaxy_url=functional_test_case.url,
-            master_api_key=getattr(functional_test_case, "master_api_key", None),
+            master_api_key=admin_api_key,
             api_key=api_key or getattr(functional_test_case, "user_api_key", None),
             test_user=test_user,
             keep_outputs_dir=getattr(functional_test_case, "keepOutdir", None),