[chore] fix ty on tests/cascade

Author: tmi

tests/cascade/benchmarks/image_processing.py

@@ -8,13 +8,13 @@
 
 import io
 
-import matplotlib.pyplot as plt
+import matplotlib.pyplot as plt# ty: ignore[unresolved-import]
 import numpy as np
 from PIL import Image
 
-from cascade.cascade import Cascade
-from cascade.fluent import Fluent, Payload  # type: ignore
-from cascade.visualise import visualise
+from cascade.cascade import Cascade # ty: ignore[unresolved-import]
+from cascade.fluent import Fluent, Payload  # ty: ignore[unresolved-import]
+from cascade.visualise import visualise # ty: ignore[unresolved-import]
 
 
 def mandelbrot(c, max_iter):

tests/cascade/executor/test_executor.py

@@ -135,7 +135,7 @@ def test_func(x: np.ndarray) -> np.ndarray:
         callback(
             m1,
             TaskSequence(
-                worker=w0, tasks=["source", "sink"], publish={sink_o}, extra_env={}
+                worker=w0, tasks=["source", "sink"], publish={sink_o}, extra_env=[]
             ),
         )
         # NOTE we need to expect source_o dataset too, because of no finegraining for host-wide and worker-only
@@ -191,7 +191,7 @@ def test_func(x: np.ndarray) -> np.ndarray:
                 logger.debug(f"about to remove received message {m}")
                 expected.remove(m)
         callback(
-            m1, TaskSequence(worker=w0, tasks=["sink"], publish={sink_o}, extra_env={})
+            m1, TaskSequence(worker=w0, tasks=["sink"], publish={sink_o}, extra_env=[])
         )
         expected = [
             DatasetPublished(w0, ds=sink_o, transmit_idx=None),

tests/cascade/executor/test_runner.py

@@ -56,7 +56,7 @@ def _send_command(
                 raise ValueError(f"double allocate on {key}")
             opened_buffers.add(key)
             purging_tracker.add(key)
-            return shm_api.AllocateResponse(shmid=key, error=None)
+            return shm_api.AllocateResponse(shmid=key, error="")
         elif isinstance(comm, shm_api.CloseCallback):
             opened_buffers.remove(key2shmid(comm.key))
         else:
@@ -79,7 +79,7 @@ def _send_command(
         worker=worker,
         tasks=[],
         publish=set(),
-        extra_env={},
+        extra_env=[],
     )
     emptyRc = entrypoint.RunnerContext(
         workerId=worker,
@@ -115,7 +115,7 @@ def test_func(x):
         worker=worker,
         tasks=["t2"],
         publish={t2ds},
-        extra_env={},
+        extra_env=[],
     )
     oneTaskJob = JobInstance(tasks={"t2": t2}, edges=[])
     oneTaskRc = entrypoint.RunnerContext(
@@ -153,7 +153,7 @@ def test_func(x):
         worker=worker,
         tasks=["t3a", "t3b"],
         publish={t3o},
-        extra_env={},
+        extra_env=[],
     )
     twoTaskJob = JobInstance(
         tasks={"t3a": t3a, "t3b": t3b},
@@ -216,7 +216,7 @@ def gen_func():
         worker=worker,
         tasks=["t4g"] + [f"t4c{i}" for i in range(N)],
         publish=set(t4pOutputs),
-        extra_env={},
+        extra_env=[],
     )
     t4Job = JobInstance(
         tasks={**{"t4g": t4g}, **{f"t4c{i}": t4c for i in range(N)}},

tests/cascade/gateway/test_run.py

@@ -88,6 +88,7 @@ def test_job():
 
         submit_job_req = api.SubmitJobRequest(job=js)
         submit_job_res = client.request_response(submit_job_req, url)
+        assert isinstance(submit_job_res, api.SubmitJobResponse)
         job_id = submit_job_res.job_id
         assert submit_job_res.error is None
         assert job_id is not None
@@ -96,11 +97,10 @@ def test_job():
         job_progress_req = api.JobProgressRequest(job_ids=[job_id])
         while tries < tries_limit:
             job_progress_res = client.request_response(job_progress_req, url)
+            assert isinstance(job_progress_res, api.JobProgressResponse)
             assert job_progress_res.error is None
-            is_computed = job_progress_res.progresses[job_id].pct == "100.00"
-            is_datasets = (
-                ji.jobInstance.ext_outputs[0] in job_progress_res.datasets[job_id]
-            )
+            is_computed = job_progress_res.progresses[job_id].pct == "100.00" # ty: ignore[possibly-missing-attribute]
+            is_datasets = ji.jobInstance.ext_outputs[0] in job_progress_res.datasets[job_id]
             if is_computed and is_datasets:
                 break
             else:
@@ -114,15 +114,17 @@ def test_job():
             job_id=job_id, dataset_id=ji.jobInstance.ext_outputs[0]
         )
         result_retrieval_res = client.request_response(result_retrieval_req, url)
+        assert isinstance(result_retrieval_res, api.ResultRetrievalResponse)
         assert result_retrieval_res.error is None
         assert result_retrieval_res.result is not None
-        deser = api.decoded_result(result_retrieval_res, ji)
+        deser = api.decoded_result(result_retrieval_res, ji.jobInstance)
         assert deser == job_func(init_value)
 
         result_deletion_req = api.ResultDeletionRequest(
             datasets={job_id: [ji.jobInstance.ext_outputs[0]]}
         )
         result_deletion_res = client.request_response(result_deletion_req, url)
+        assert isinstance(result_deletion_res, api.ResultDeletionResponse)
         assert result_deletion_res.error is None
 
         # fail job
@@ -137,6 +139,7 @@ def test_job():
 
         submit_job_req = api.SubmitJobRequest(job=js)
         submit_job_res = client.request_response(submit_job_req, url)
+        assert isinstance(submit_job_res, api.SubmitJobResponse)
         job_id = submit_job_res.job_id
         assert submit_job_res.error is None
         assert job_id is not None
@@ -145,9 +148,10 @@ def test_job():
         job_progress_req = api.JobProgressRequest(job_ids=[job_id])
         while tries < tries_limit:
             job_progress_res = client.request_response(job_progress_req, url)
+            assert isinstance(job_progress_res, api.JobProgressResponse)
             assert job_progress_res.error is None
-            assert job_progress_res.progresses[job_id].pct != "100.00"
-            if job_progress_res.progresses[job_id].failure is not None:
+            assert job_progress_res.progresses[job_id].pct != "100.00" # ty: ignore[possibly-missing-attribute]
+            if job_progress_res.progresses[job_id].failure is not None: # ty: ignore[possibly-missing-attribute]
                 break
             else:
                 tries += 1
@@ -169,6 +173,8 @@ def test_job():
         req = api.SubmitJobRequest(job=js)
         res1 = client.request_response(req, url)
         res2 = client.request_response(req, url)
+        assert isinstance(res1, api.SubmitJobResponse)
+        assert isinstance(res2, api.SubmitJobResponse)
         assert res1.error is None
         assert res2.error is None
         assert res1.job_id is not None
@@ -179,9 +185,10 @@ def test_job():
         job_progress_req = api.JobProgressRequest(job_ids=job_ids)
         while tries < tries_limit:
             job_progress_res = client.request_response(job_progress_req, url)
+            assert isinstance(job_progress_res, api.JobProgressResponse)
             assert job_progress_res.error is None
             is_computed = (
-                lambda job_id: job_progress_res.progresses[job_id].pct == "100.00"
+                    lambda job_id: job_progress_res.progresses[job_id].pct == "100.00" # ty: ignore[possibly-missing-attribute]
             )
             if all(is_computed(job_id) for job_id in job_ids):
                 break
@@ -195,7 +202,7 @@ def test_job():
         # gw shutdown
         shutdown_req = api.ShutdownRequest()
         shutdown_res = client.request_response(shutdown_req, url, 3000)
-        assert shutdown_res.error is None
+        assert hasattr(shutdown_res, 'error') and shutdown_res.error is None
         gw.join(5)
         assert gw.exitcode == 0
 

tests/cascade/low/test_builders.py

@@ -19,7 +19,7 @@ def test_func(x: int, y: str) -> int:
         "invalid static input for task: x needs int, got <class 'str'>",
         "invalid static input for task: y needs str, got <class 'int'>",
     ]
-    assert sorted(job.e) == sorted(expected)
+    assert sorted(job.e) == sorted(expected) # ty: ignore[invalid-argument-type]
 
     task_good = TaskBuilder.from_callable(test_func).with_values(x=1, y="yes")
     _ = JobBuilder().with_node("task", task_good).build().get_or_raise()
@@ -35,7 +35,7 @@ def test_func(x: int, y: str) -> int:
     expected = [
         "edge connects two incompatible nodes: source=source.0 sink_task='sink' sink_input_kw='y' sink_input_ps=None",  # noqa: E501
     ]
-    assert sorted(job.e) == sorted(expected)
+    assert sorted(job.e) == sorted(expected) # ty: ignore[invalid-argument-type]
 
     sink_good = TaskBuilder.from_callable(test_func).with_values(y="yes")
     _ = (