Fix mypy and format

Author: tetron

cwltool/context.py

@@ -192,7 +192,9 @@ def __init__(self, kwargs: Optional[dict[str, Any]] = None) -> None:
         self.default_stderr: Optional[Union[IO[bytes], TextIO]] = None
         self.validate_only: bool = False
         self.validate_stdout: Optional["SupportsWrite[str]"] = None
-        self.workflow_job_step_name_callback: Optional[Callable[[WorkflowJobStep, CWLObjectType], str]] = None
+        self.workflow_job_step_name_callback: Optional[
+            Callable[[WorkflowJobStep, CWLObjectType], str]
+        ] = None
 
         super().__init__(kwargs)
         if self.tmp_outdir_prefix == "":

tests/test_context.py

@@ -2,9 +2,12 @@
 import sys
 import logging
 from io import StringIO
+from typing import MutableMapping, cast
 
 from cwltool.context import RuntimeContext
 from cwltool.factory import Factory
+from cwltool.utils import CWLObjectType, CWLOutputType
+from cwltool.workflow_job import WorkflowJobStep
 
 from .util import get_data
 
@@ -29,6 +32,7 @@ def test_replace_default_stdout_stderr() -> None:
     sys.stdout = original_stdout
     sys.stderr = original_stderr
 
+
 def test_workflow_job_step_name_callback() -> None:
     """Test ability to hook custom workflow step naming"""
 
@@ -40,36 +44,47 @@ def test_workflow_job_step_name_callback() -> None:
     try:
         runtime_context = RuntimeContext()
 
-        def step_name_hook(step, job):
-            print(step, job)
-            return "%s on %s" % (step.name, job.get("revtool_input", job.get("sorted_input")).get("basename"))
+        def step_name_hook(step: WorkflowJobStep, job: CWLObjectType) -> str:
+            j1 = cast(MutableMapping[str, CWLObjectType], job)
+            inp = cast(MutableMapping[str, str], j1.get("revtool_input", j1.get("sorted_input")))
+            return "%s on %s" % (
+                step.name,
+                inp.get("basename"),
+            )
 
         runtime_context.workflow_job_step_name_callback = step_name_hook
 
         factory = Factory(None, None, runtime_context)
         revsort = factory.make(get_data("tests/wf/revsort.cwl"))
 
-        result = revsort(workflow_input={"class": "File", "location": "whale.txt", "format": "https://www.iana.org/assignments/media-types/text/plain"})
+        result = revsort(
+            workflow_input={
+                "class": "File",
+                "location": "whale.txt",
+                "format": "https://www.iana.org/assignments/media-types/text/plain",
+            }
+        )
+
+        result = cast(CWLObjectType, result)
 
-        del result["sorted_output"]["location"]
+        sorted_out = cast(MutableMapping[str, str], result["sorted_output"])
+        loc = sorted_out["location"]
 
         assert result == {
-            'sorted_output': {
-                'basename': 'output.txt ',
-                'checksum': 'sha1$b9214658cc453331b62c2282b772a5c063dbd284',
-                'class': 'File',
-                'http://commonwl.org/cwltool#generation': 0,
-                'nameext': '.txt',
-                'nameroot': 'output',
-                'size': 1111,
+            "sorted_output": {
+                "basename": "output.txt ",
+                "checksum": "sha1$b9214658cc453331b62c2282b772a5c063dbd284",
+                "class": "File",
+                "http://commonwl.org/cwltool#generation": 0,
+                "nameext": ".txt",
+                "nameroot": "output",
+                "size": 1111,
+                "location": loc,
             },
         }
 
         print(stream.getvalue())
 
-        assert (
-            "foostep"
-            in stream.getvalue()
-        )
+        assert "foostep" in stream.getvalue()
     finally:
         _logger.removeHandler(streamhandler)