Reformatting and type hint updates

Author: AlexTate

cwltool/executors.py

@@ -439,7 +439,9 @@ def run_jobs(
         logger: logging.Logger,
         runtime_context: RuntimeContext,
     ) -> None:
-        self.taskqueue: TaskQueue = TaskQueue(threading.Lock(), int(math.ceil(self.max_cores)), runtime_context)
+        self.taskqueue: TaskQueue = TaskQueue(
+            threading.Lock(), int(math.ceil(self.max_cores)), runtime_context
+        )
         try:
             jobiter = process.job(job_order_object, self.output_callback, runtime_context)
 

cwltool/workflow.py

@@ -1,8 +1,8 @@
-import os
 import copy
 import datetime
 import functools
 import logging
+import os
 import random
 from collections.abc import Mapping, MutableMapping, MutableSequence
 from typing import Callable, Optional, cast
@@ -454,5 +454,5 @@ def job(
     def visit(self, op: Callable[[CommentedMap], None]) -> None:
         self.embedded_tool.visit(op)
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"<{self.__class__.__name__} [{os.path.basename(self.id)}]>"

cwltool/workflow_job.py

@@ -66,7 +66,7 @@ def job(
 
         yield from self.step.job(joborder, output_callback, runtimeContext)
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"<{self.__class__.__name__} [{self.name}]>"
 
 
@@ -92,7 +92,9 @@ def completed(self) -> int:
         """The number of completed internal jobs."""
         return len(self._completed)
 
-    def receive_scatter_output(self, index: int, runtimeContext: RuntimeContext, jobout: CWLObjectType, processStatus: str) -> None:
+    def receive_scatter_output(
+        self, index: int, runtimeContext: RuntimeContext, jobout: CWLObjectType, processStatus: str
+    ) -> None:
         """Record the results of a scatter operation."""
         for key, val in jobout.items():
             self.dest[key][index] = val
@@ -147,7 +149,10 @@ def parallel_steps(
                 continue
             try:
                 for j in step:
-                    if runtimeContext.on_error != "continue" and rc.processStatus not in ("success", "skipped"):
+                    if runtimeContext.on_error != "continue" and rc.processStatus not in (
+                        "success",
+                        "skipped",
+                    ):
                         break
                     if j is not None:
                         made_progress = True
@@ -260,7 +265,9 @@ def _flat_crossproduct_scatter(
         if len(scatter_keys) == 1:
             if runtimeContext.postScatterEval is not None:
                 sjob = runtimeContext.postScatterEval(sjob)
-            curriedcallback = functools.partial(callback.receive_scatter_output, put, runtimeContext)
+            curriedcallback = functools.partial(
+                callback.receive_scatter_output, put, runtimeContext
+            )
             if sjob is not None:
                 steps.append(process.job(sjob, curriedcallback, runtimeContext))
             else:
@@ -808,10 +815,7 @@ def job(
             self.made_progress = False
 
             for step in self.steps:
-                if (
-                    runtimeContext.on_error  != "continue"
-                    and self.processStatus != "success"
-                ):
+                if runtimeContext.on_error != "continue" and self.processStatus != "success":
                     break
 
                 if not step.submitted:
@@ -855,8 +859,10 @@ def job(
             # or all outputs have been produced.
 
     def __repr__(self):
+    def __repr__(self) -> str:
         return f"<{self.__class__.__name__} [{self.name}]>"
 
+
 class WorkflowJobLoopStep:
     """Generated for each step in Workflow.steps() containing a `loop` directive."""
 

tests/test_parallel.py

@@ -2,7 +2,7 @@
 import math
 import time
 from pathlib import Path
-from typing import Union
+from typing import Union, cast
 
 from cwltool.context import RuntimeContext
 from cwltool.executors import MultithreadedJobExecutor
@@ -58,7 +58,7 @@ def selectResources(
     # arbitrary test values
     sleep_time = 3333  # a "sufficiently large" timeout
     n_sleepers = 4
-    start_time = 0
+    start_time = 0.0
 
     try:
         start_time = time.time()
@@ -68,6 +68,6 @@ def selectResources(
         )
     except WorkflowStatus as e:
         end_time = time.time()
-        output = e.out["roulette_mask"]
+        output = cast(dict[str, list[bool]], e.out)["roulette_mask"]
         assert len(output) == n_sleepers and sum(output) == 1
-        assert end_time - start_time < sleep_time
+        assert end_time - start_time < sleep_time