Fix various typing issues

Can't use classvars for a lot of the stuff, those are essentially
globals. Fixes the tests.

Author: mvdbeek

planemo/galaxy/invocations/polling.py

@@ -56,6 +56,7 @@ def summarize(invocation_id: str):
     last_subworkflow_invocation = None
     last_subworkflow_invocation_jobs = None
     last_exception = None
+    error_message: Optional[str] = None
 
     done_polling = False
     while not done_polling:
@@ -120,6 +121,7 @@ def summarize(invocation_id: str):
 
     ctx.vlog(f"The final state of all jobs and subworkflow invocations for invocation [{invocation_id}] is 'ok'")
     job_state = summary_job_state(last_invocation_jobs)
+    assert last_invocation
     return last_invocation["state"], job_state, error_message
 
 
@@ -152,7 +154,7 @@ def workflow_in_error_message(
 # we're still mocking out the old history state by just picking out a random
 # job state of interest. Seems like we should drop this.
 def summary_job_state(job_states_summary: Optional[InvocationJobsSummary]):
-    states = (job_states_summary or {"states": {}}).get("states").copy()
+    states = (job_states_summary or {"states": {}}).get("states", {}).copy()
     states.pop("ok", None)
     states.pop("skipped", None)
     if states:
@@ -165,6 +167,7 @@ def subworkflow_invocation_ids(invocation_api: InvocationApi, invocation_id: str
     invocation = invocation_api.get_invocation(invocation_id)
     subworkflow_invocation_ids = []
     for step in invocation["steps"]:
-        if step.get("subworkflow_invocation_id") is not None:
-            subworkflow_invocation_ids.append(step["subworkflow_invocation_id"])
+        subworkflow_invocation_id = step.get("subworkflow_invocation_id")
+        if subworkflow_invocation_id:
+            subworkflow_invocation_ids.append(subworkflow_invocation_id)
     return subworkflow_invocation_ids

planemo/galaxy/invocations/progress.py

@@ -43,37 +43,35 @@ class InvocationJobsSummary(TypedDict, total=False):
 
 
 class WorkflowProgress(Progress):
-    invocation_state: str = "new"
-    step_count: Optional[int] = None
-    job_count: int = 0
-    steps_color: str
-    jobs_color: str
-    subworkflows_color: str
-    step_states: Dict = {}
-    num_ok: int = 0
-    num_new: int = 0
-    num_queued: int = 0
-    num_running: int = 0
-    num_errors: int = 0
-    num_paused: int = 0
-
-    num_subworkflows: int = 0
-    num_subworkflows_complete: int = 0
 
     _jobs_task: TaskID
     _steps_task: TaskID
     _subworkflows_task: Optional[TaskID] = None
 
     def __init__(self, display: DisplayConfiguration):
+        self.invocation_state: str = "new"
+        self.step_count: Optional[int] = None
+        self.job_count: Optional[int] = 0
+        self.jobs_completed: Optional[int] = None
+        self.step_states: Dict[str, int] = {}
+        self.num_ok: int = 0
+        self.num_new: int = 0
+        self.num_queued: int = 0
+        self.num_running: int = 0
+        self.num_errors: int = 0
+        self.num_paused: int = 0
+
+        self.num_subworkflows: int = 0
+        self.num_subworkflows_complete: int = 0
         self.display = display
         bar_column = BarColumn(
             style=self.display.style_bar_back,
             finished_style=self.display.style_bar_finished,
             complete_style=self.display.style_bar_complete,
         )
-        self.jobs_color = self.display.style_initializing
-        self.steps_color = self.display.style_initializing
-        self.subworkflows_color = self.display.style_initializing
+        self.jobs_color: str = self.display.style_initializing
+        self.steps_color: str = self.display.style_initializing
+        self.subworkflows_color: str = self.display.style_initializing
         super().__init__(
             TextColumn("[progress.description]{task.description}"),
             TextColumn(display.divider),
@@ -153,7 +151,7 @@ def handle_invocation(self, invocation: Invocation, job_state_summary: Invocatio
         self.jobs_completed = self.num_ok + self.num_errors
         self.num_paused = count_states(job_state_summary, ["paused"])
         self.jobs_terminal_count = self.jobs_completed + self.num_paused
-        jobs_total = self.job_count
+        jobs_total: Optional[int] = self.job_count
         if self.num_errors > 0:
             self.jobs_color = self.display.style_error
         elif self.job_count > 0:
@@ -251,18 +249,15 @@ def running_count(job_summary: InvocationJobsSummary) -> int:
 
 
 class WorkflowProgressDisplay(Live):
-    workflow_progress: WorkflowProgress
-    subworkflow_progress: Optional[WorkflowProgress] = None
-    subworkflow_invocation_ids_seen: Set[str] = set()
-    subworkflow_invocation_ids_completed: Set[str] = set()
-    invocation_id: str
-    subworkflow_invocation_id: Optional[str] = None
 
     def __init__(
         self,
         invocation_id: str,
         display_configuration: Optional[DisplayConfiguration] = None,
     ):
+        self.subworkflow_invocation_ids_seen: Set[str] = set()
+        self.subworkflow_invocation_ids_completed: Set[str] = set()
+        self.subworkflow_invocation_id: Optional[str] = None
         self.invocation_id = invocation_id
         display = display_configuration or DisplayConfiguration()
         self.display = display
@@ -271,10 +266,12 @@ def __init__(
         super().__init__(self._panel())
 
     def _register_subworkflow_invocation_ids_from(self, invocation: Invocation):
-        subworkflow_invocation_ids = []
-        for step in invocation["steps"]:
-            if step.get("subworkflow_invocation_id") is not None:
-                subworkflow_invocation_ids.append(step["subworkflow_invocation_id"])
+        subworkflow_invocation_ids: List[str] = []
+        steps = invocation.get("steps") or []
+        for step in steps:
+            subworkflow_invocation_id = step.get("subworkflow_invocation_id")
+            if subworkflow_invocation_id:
+                subworkflow_invocation_ids.append(subworkflow_invocation_id)
         self._register_subworkflow_invocation_ids(subworkflow_invocation_ids)
 
     def _register_subworkflow_invocation_ids(self, ids: List[str]):

planemo/galaxy/invocations/simulations.py

@@ -2,6 +2,7 @@
 
 from collections import deque
 from typing import (
+    Dict,
     List,
     Optional,
 )
@@ -11,6 +12,7 @@
 
 from .api import Invocation as InvocationResponse
 from .api import InvocationJobsSummary
+from .api import InvocationStep as InvocationStepResponse
 
 
 class Ticks:
@@ -46,7 +48,7 @@ class HasState(Ticks):
     def __init__(self, after: int, states: List[StateWithDuration]):
         self.after = after or 0
         self.states = deque(states)
-        self.final_state = None
+        self.final_state: Optional[str] = None
 
     def tick_when_active(self) -> None:
         if self.final_state is not None:
@@ -128,13 +130,13 @@ def get_subworkflow_invocation(self, subworkflow_invocation_id: str) -> "Invocat
                 return step.invocation
         raise Exception(f"Unknown subworkflow invocation id ({subworkflow_invocation_id})")
 
-    def get_subworkflow_invocation_by_step_index(self, index: int) -> "Invocation":
+    def get_subworkflow_invocation_by_step_index(self, index: int) -> Optional["Invocation"]:
         return self.steps[index].invocation
 
     def get_api_invocation(self) -> InvocationResponse:
-        steps = []
+        steps: List[InvocationStepResponse] = []
         for step in self.active_steps:
-            api_step = {
+            api_step: InvocationStepResponse = {
                 "state": step.state,
             }
             if step.invocation:
@@ -148,16 +150,16 @@ def get_api_invocation(self) -> InvocationResponse:
         }
 
     def get_api_jobs_summary(self) -> InvocationJobsSummary:
-        jobs = []
+        job_states = []
         for step in self.active_steps:
             for job in step.active_jobs:
                 api_job = {
                     "state": job.state,
                 }
-                jobs.append(api_job)
-        by_state = {}
-        for job in jobs:
-            state = job["state"]
+                job_states.append(api_job)
+        by_state: Dict[str, int] = {}
+        for job_state in job_states:
+            state = job_state["state"]
             if state not in by_state:
                 by_state[state] = 0
             by_state[state] += 1
@@ -188,7 +190,7 @@ def parse_workflow_simulation_invocation_step(workflow_simulation_invocation_ste
     jobs = None
     if "jobs" in workflow_simulation_invocation_step:
         jobs = []
-        for job in workflow_simulation_invocation_step.get("jobs"):
+        for job in workflow_simulation_invocation_step.get("jobs") or []:
             jobs.append(parse_workflow_simulation_job(job))
     return InvocationStep(jobs, invocation, after, states)
 
@@ -197,7 +199,7 @@ def parse_workflow_simulation_invocation(workflow_simulation_invocation: dict) -
     states = parse_states_from(workflow_simulation_invocation)
     after = parse_after_from(workflow_simulation_invocation)
     steps = []
-    for step in workflow_simulation_invocation.get("steps"):
+    for step in workflow_simulation_invocation.get("steps") or []:
         steps.append(parse_workflow_simulation_invocation_step(step))
 
     return Invocation(steps, after, states)

tests/test_workflow_progress.py

@@ -4,6 +4,7 @@
     WorkflowProgress,
     WorkflowProgressDisplay,
 )
+from planemo.galaxy.invocations.progress_display import DisplayConfiguration
 
 STEP_NEW = {"state": "new"}
 STEP_SCHEDULED = {"state": "scheduled"}
@@ -80,7 +81,7 @@ def test_workflow_progress_typical():
 
 
 def test_workflow_progress_scheduling_state_handling():
-    with WorkflowProgress() as workflow_progress:
+    with WorkflowProgress(DisplayConfiguration()) as workflow_progress:
         workflow_progress.handle_invocation({"state": "new", "steps": [STEP_NEW]}, {"states": {"new": 1}})
         assert not workflow_progress.invocation_scheduling_terminal
 
@@ -124,7 +125,7 @@ def test_workflow_progress_scheduling_state_handling():
 def test_workflow_progress_job_state_handling():
     scheduled_invocation = {"state": "scheduled", "steps": [STEP_SCHEDULED]}
 
-    with WorkflowProgress() as workflow_progress:
+    with WorkflowProgress(DisplayConfiguration()) as workflow_progress:
         workflow_progress.handle_invocation(scheduled_invocation, {"states": {"new": 1}})
         assert not workflow_progress.jobs_terminal