Typing improvements for galaxy.tool_util.verify and related code.

lib/galaxy/tool_util/verify/__init__.py

@@ -10,6 +10,12 @@ import os.path
 import re
 import shutil
 import tempfile
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Optional,
+)
 
 try:
     import pysam
@@ -31,14 +37,14 @@ DEFAULT_TEST_DATA_RESOLVER = TestDataResolver()
 
 
 def verify(
-    item_label,
-    output_content,
-    attributes,
-    filename=None,
-    get_filecontent=None,
-    get_filename=None,
-    keep_outputs_dir=None,
-    verify_extra_files=None,
+    item_label: str,
+    output_content: bytes,
+    attributes: Dict[str, Any],
+    filename: Optional[str] = None,
+    get_filecontent: Optional[Callable[[str], bytes]] = None,
+    get_filename: Optional[Callable[[str], str]] = None,
+    keep_outputs_dir: Optional[str] = None,
+    verify_extra_files: Optional[Callable] = None,
     mode="file",
 ):
     """Verify the content of a test output using test definitions described by attributes.
@@ -46,16 +52,21 @@ def verify(
     Throw an informative assertion error if any of these tests fail.
     """
     if get_filename is None:
+        get_filecontent_: Callable[[str], bytes]
         if get_filecontent is None:
-            get_filecontent = DEFAULT_TEST_DATA_RESOLVER.get_filecontent
+            get_filecontent_ = DEFAULT_TEST_DATA_RESOLVER.get_filecontent
+        else:
+            get_filecontent_ = get_filecontent
 
-        def get_filename(filename):
-            file_content = get_filecontent(filename)
+        def get_filename(filename: str) -> str:
+            file_content = get_filecontent_(filename)
             local_name = make_temp_fname(fname=filename)
             with open(local_name, "wb") as f:
                 f.write(file_content)
             return local_name
 
+    assert get_filename
+
     # Check assertions...
     assertions = attributes.get("assert_list", None)
     if attributes is not None and assertions is not None:
@@ -136,7 +147,9 @@ def verify(
             # filename already point to a file that exists on disk
             local_name = filename
         else:
-            local_name = get_filename(filename)
+            filename_ = get_filename(filename)
+            assert filename_, f"Failed to find output target for test {filename_}"
+            local_name = filename_
 
         compare = attributes.get("compare", "diff")
         try:

lib/galaxy/tool_util/verify/asserts/__init__.py

@@ -30,14 +30,14 @@ for assertion_module_name in assertion_module_names:
             assertion_functions[member] = value
 
 
-def verify_assertions(data, assertion_description_list):
+def verify_assertions(data: bytes, assertion_description_list):
     """This function takes a list of assertions and a string to check
     these assertions against."""
     for assertion_description in assertion_description_list:
         verify_assertion(data, assertion_description)
 
 
-def verify_assertion(data, assertion_description):
+def verify_assertion(data: bytes, assertion_description):
     tag = assertion_description["tag"]
     assert_function_name = "assert_" + tag
     assert_function = assertion_functions.get(assert_function_name)

lib/galaxy/tool_util/verify/interactor.py

@@ -11,7 +11,13 @@ import urllib.parse
 import zipfile
 from json import dumps
 from logging import getLogger
-from typing import Optional
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    List,
+    Optional,
+)
 
 import requests
 from packaging.version import (
@@ -19,6 +25,7 @@ from packaging.version import (
     Version,
 )
 from requests.cookies import RequestsCookieJar
+from typing_extensions import Protocol
 
 try:
     from nose.tools import nottest
@@ -71,9 +78,15 @@ class OutputsDict(dict):
             return super().__getitem__(item)
 
 
+JobDataT = Dict[str, Any]
+JobDataCallbackT = Callable[[JobDataT], None]
+ToolTestDictT = Dict[str, Any]
+ToolTestDictsT = List[ToolTestDictT]
+
+
 def stage_data_in_history(
-    galaxy_interactor,
-    tool_id,
+    galaxy_interactor: "GalaxyInteractorApi",
+    tool_id: str,
     all_test_data,
     history=None,
     force_path_paste=False,
@@ -113,6 +126,8 @@ def stage_data_in_history(
 
 
 class GalaxyInteractorApi:
+    api_key: Optional[str]
+
     def __init__(self, **kwds):
         self.api_url = f"{kwds['galaxy_url'].rstrip('/')}/api"
         self.cookies = None
@@ -157,7 +172,7 @@ class GalaxyInteractorApi:
         assert response.status_code == 200, f"Non 200 response from tool tests available API. [{response.content}]"
         return response.json()
 
-    def get_tool_tests(self, tool_id, tool_version=None):
+    def get_tool_tests(self, tool_id: str, tool_version: Optional[str] = None) -> ToolTestDictsT:
         url = f"tools/{tool_id}/test_data"
         params = {"tool_version": tool_version} if tool_version else None
         response = self._get(url, data=params)
@@ -430,13 +445,13 @@ class GalaxyInteractorApi:
 
     def stage_data_async(
         self,
-        test_data,
-        history_id,
-        tool_id,
-        force_path_paste=False,
-        maxseconds=DEFAULT_TOOL_TEST_WAIT,
-        tool_version=None,
-    ):
+        test_data: Dict[str, Any],
+        history_id: str,
+        tool_id: str,
+        force_path_paste: bool = False,
+        maxseconds: int = DEFAULT_TOOL_TEST_WAIT,
+        tool_version: Optional[str] = None,
+    ) -> Callable[[], None]:
         fname = test_data["fname"]
         tool_input = {
             "file_type": test_data["ftype"],
@@ -761,6 +776,7 @@ class GalaxyInteractorApi:
                 else:
                     break
 
+            assert response
             response.raise_for_status()
             return response.content
 
@@ -1097,12 +1113,17 @@ def _verify_extra_files_content(extra_files, hda_id, dataset_fetcher, test_data_
             shutil.rmtree(path)
 
 
-class NullClientTestConfig:
+class TestConfig(Protocol):
+    def get_test_config(self, job_data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+        ...
+
+
+class NullClientTestConfig(TestConfig):
     def get_test_config(self, job_data):
         return None
 
 
-class DictClientTestConfig:
+class DictClientTestConfig(TestConfig):
     def __init__(self, tools):
         self._tools = tools or {}
 
@@ -1140,22 +1161,22 @@ class DictClientTestConfig:
 
 
 def verify_tool(
-    tool_id,
-    galaxy_interactor,
-    resource_parameters=None,
-    register_job_data=None,
-    test_index=0,
-    tool_version=None,
-    quiet=False,
-    test_history=None,
-    no_history_cleanup=False,
-    publish_history=False,
-    force_path_paste=False,
-    maxseconds=DEFAULT_TOOL_TEST_WAIT,
-    tool_test_dicts=None,
-    client_test_config=None,
-    skip_with_reference_data=False,
-    skip_on_dynamic_param_errors=False,
+    tool_id: str,
+    galaxy_interactor: "GalaxyInteractorApi",
+    resource_parameters: Optional[Dict[str, Any]] = None,
+    register_job_data: Optional[JobDataCallbackT] = None,
+    test_index: int = 0,
+    tool_version: Optional[str] = None,
+    quiet: bool = False,
+    test_history: Optional[str] = None,
+    no_history_cleanup: bool = False,
+    publish_history: bool = False,
+    force_path_paste: bool = False,
+    maxseconds: int = DEFAULT_TOOL_TEST_WAIT,
+    tool_test_dicts: Optional[ToolTestDictsT] = None,
+    client_test_config: Optional[TestConfig] = None,
+    skip_with_reference_data: bool = False,
+    skip_on_dynamic_param_errors: bool = False,
 ):
     if resource_parameters is None:
         resource_parameters = {}
@@ -1170,7 +1191,7 @@ def verify_tool(
     if tool_version is None and "tool_version" in tool_test_dict:
         tool_version = tool_test_dict.get("tool_version")
 
-    job_data = {
+    job_data: JobDataT = {
         "tool_id": tool_id,
         "tool_version": tool_version,
         "test_index": test_index,
@@ -1190,7 +1211,7 @@ def verify_tool(
         if required_loc_files:
             skip_message = f"Skipping test because of required loc files ({required_loc_files})"
 
-    if skip_message:
+    if skip_message and register_job_data:
         job_data["status"] = "skip"
         register_job_data(job_data)
         return
@@ -1212,7 +1233,7 @@ def verify_tool(
     tool_inputs = None
     job_stdio = None
     job_output_exceptions = None
-    tool_execution_exception = None
+    tool_execution_exception: Optional[Exception] = None
     input_staging_exception = None
     expected_failure_occurred = False
     begin_time = time.time()
@@ -1302,9 +1323,9 @@ def _verify_outputs(testdef, history, jobs, data_list, data_collection_list, gal
     assert len(jobs) == 1, "Test framework logic error, somehow tool test resulted in more than one job."
     job = jobs[0]
 
-    found_exceptions = []
+    found_exceptions: List[Exception] = []
 
-    def register_exception(e):
+    def register_exception(e: Exception):
         if not found_exceptions and not quiet:
             # Only print this stuff out once.
             for stream in ["stdout", "stderr"]:

lib/galaxy/tool_util/verify/script.py

@@ -12,6 +12,7 @@ from concurrent.futures import (
     thread,
     ThreadPoolExecutor,
 )
+from typing import List
 
 import yaml
 
@@ -34,11 +35,13 @@ TestException = namedtuple("TestException", ["tool_id", "exception", "was_record
 
 
 class Results:
+    test_exceptions: List[Exception]
+
     def __init__(self, default_suitename, test_json, append=False, galaxy_url=None):
         self.test_json = test_json or "-"
         self.galaxy_url = galaxy_url
         test_results = []
-        test_exceptions = []
+        test_exceptions: List[Exception] = []
         suitename = default_suitename
         if append:
             assert test_json != "-"

lib/galaxy/tool_util/verify/test_data.py

@@ -35,23 +35,28 @@ class TestDataResolver:
         else:
             self.resolvers = []
 
-    def get_filename(self, name):
+    def get_filename(self, name: str) -> str:
         for resolver in self.resolvers or []:
             if not resolver.exists(name):
                 continue
             filename = resolver.path(name)
             if filename:
                 return os.path.abspath(filename)
+        raise TestDataNotFoundError(f"Failed to find test file {name} against any test data resolvers")
 
-    def get_filecontent(self, name):
+    def get_filecontent(self, name: str) -> bytes:
         filename = self.get_filename(name=name)
         with open(filename, mode="rb") as f:
             return f.read()
 
-    def get_directory(self, name):
+    def get_directory(self, name: str) -> str:
         return self.get_filename(name=name)
 
 
+class TestDataNotFoundError(ValueError):
+    pass
+
+
 def build_resolver(uri, environ):
     if uri.startswith("http") and uri.endswith(".git"):
         return GitDataResolver(uri, environ)

lib/galaxy/tools/__init__.py

@@ -1228,7 +1228,10 @@ class Tool(Dictifiable):
             # Fallback to Galaxy test data directory for builtin tools, tools
             # under development, and some older ToolShed published tools that
             # used stock test data.
-            test_data = self.app.test_data_resolver.get_filename(filename)
+            try:
+                test_data = self.app.test_data_resolver.get_filename(filename)
+            except ValueError:
+                test_data = None
         return test_data
 
     def __walk_test_data(self, dir, filename):

lib/galaxy_test/base/api.py

@@ -1,6 +1,11 @@
 import os
 from contextlib import contextmanager
-from typing import Optional
+from typing import (
+    Any,
+    Dict,
+    Optional,
+    Tuple,
+)
 from urllib.parse import (
     urlencode,
     urljoin,
@@ -122,7 +127,7 @@ class UsesApiTestCaseMixin:
         user = [user for user in users if user["email"] == email][0]
         return user
 
-    def _setup_user_get_key(self, email, password=None, is_admin=True):
+    def _setup_user_get_key(self, email, password=None, is_admin=True) -> Tuple[Dict[str, Any], str]:
         user = self._setup_user(email, password, is_admin)
         return user, self._post(f"users/{user['id']}/api_key", admin=True).json()
 

mypy.ini

@@ -294,8 +294,6 @@ check_untyped_defs = False
 check_untyped_defs = False
 [mypy-galaxy.tools.repositories]
 check_untyped_defs = False
-[mypy-galaxy.tool_util.verify.interactor]
-check_untyped_defs = False
 [mypy-galaxy.objectstore.s3]
 check_untyped_defs = False
 [mypy-galaxy.objectstore.pithos]