Merge branch 'main' of github.com:ExaWorks/psij-python into main

Author: hategan

CONTRIBUTING.md

@@ -11,13 +11,6 @@ details how to contribute in a standardized and efficient manner.
 - Ensure that you've opened an Issue on Github and consensus around the
   solution has be reached.
   - Minor changes (e.g., grammatical fixes) do not require an Issue first.
-- [Create your own
-  fork](https://docs.github.com/en/github/getting-started-with-github/fork-a-repo).
-- When you are starting a new set of changes, [fetch any changes from the
-  upstream
-  repo](https://matplotlib.org/stable/devel/gitwash/development_workflow.html#update-the-mirror-of-trunk),
-  and [start a new feature branch on your fork from
-  that](https://matplotlib.org/stable/devel/gitwash/development_workflow.html#make-a-new-feature-branch).
 - Make a new branch for each separable set of changes — ["one task, one
   branch"](https://mail.python.org/pipermail/ipython-dev/2010-October/005632.html).
 - [Each commit should make one change](https://dev.to/ruanbrandao/how-to-make-good-git-commits-256k).

requirements-dev.txt

@@ -7,7 +7,7 @@
 # Testing / QA
 -r requirements-tests.txt
 mypy >=0.931
-flake8
+flake8==5.0.4
 flake8-docstrings
 autopep8
 types-requests

scripts/WORKFLOW-EXAMPLES.md

@@ -100,7 +100,7 @@ RUN pip install wordcloud
 
 ## Executing a singularity container
 
-We will run the PDF2Wordcloud example in a container. We will use the docker conatiner described above and execute it with singularity. In this case the spec defines the execution of a container with singularity. The actual command line tool is passed as an optional argument after all requiered singularity options.
+We will run the PDF2Wordcloud example in a container. We will use the docker container described above and execute it with singularity. In this case the spec defines the execution of a container with singularity. The actual command line tool is passed as an optional argument after all requiered singularity options.
 
 ```
 ...

src/psij/executors/flux.py

@@ -177,7 +177,15 @@ def list(self) -> List[str]:
 
         :return: The list of known tasks.
         """
-        return [x["id"] for x in flux.job.job_list(self._fh)]
+        return [
+            x["id"]
+            for x in flux.job.job_list(
+                self._fh,
+                max_entries=100000,
+                attrs=[],
+                states=flux.constants.FLUX_JOB_STATE_ACTIVE,
+            ).get()["jobs"]
+        ]
 
     def attach(self, job: Job, native_id: str) -> None:
         """

tests/_test_tools.py

@@ -38,8 +38,9 @@ def assert_completed(job: Job, status: Optional[JobStatus]) -> None:
                              % (status.message, stdout, stderr))
 
 
-def _get_executor_instance(ep: ExecutorTestParams, job: Job) -> JobExecutor:
-    assert job.spec is not None
-    job.spec.launcher = ep.launcher
-    job.spec.attributes = JobAttributes(custom_attributes=ep.custom_attributes)
+def _get_executor_instance(ep: ExecutorTestParams, job: Optional[Job] = None) -> JobExecutor:
+    if job is not None:
+        assert job.spec is not None
+        job.spec.launcher = ep.launcher
+        job.spec.attributes = JobAttributes(custom_attributes=ep.custom_attributes)
     return JobExecutor.get_instance(ep.executor, url=ep.url)

tests/test_executor.py

@@ -32,6 +32,22 @@ def test_simple_job_redirect(execparams: ExecutorTestParams) -> None:
 
 
 def test_attach(execparams: ExecutorTestParams) -> None:
+    job1 = Job(JobSpec(executable='/bin/sleep', arguments=['1']))
+    ex = _get_executor_instance(execparams, job1)
+    ex.submit(job1)
+    job1.wait(target_states=[JobState.ACTIVE, JobState.COMPLETED])
+    native_id = job1.native_id
+
+    assert native_id is not None
+    job2 = Job()
+    ex.attach(job2, native_id)
+    status2 = job2.wait(timeout=_get_timeout(execparams))
+    assert_completed(job2, status2)
+    status1 = job1.wait(timeout=_get_timeout(execparams))
+    assert_completed(job1, status1)
+
+
+def test_attach2(execparams: ExecutorTestParams) -> None:
     job = Job(JobSpec(executable='/bin/sleep', arguments=['1']))
     ex = _get_executor_instance(execparams, job)
     ex.submit(job)
@@ -40,7 +56,8 @@ def test_attach(execparams: ExecutorTestParams) -> None:
 
     assert native_id is not None
     job2 = Job()
-    ex.attach(job2, native_id)
+    ex2 = _get_executor_instance(execparams)
+    ex2.attach(job2, native_id)
     status = job2.wait(timeout=_get_timeout(execparams))
     assert_completed(job2, status)