Code style fixes (#3)

* Fix unit test failure
* Fix code style + code cleanup
* Add a proper comment before the `assert_process_died()`assertions
* Revert change in remote autodetect code

Author: vkarak

reframe/core/schedulers/flux.py

@@ -25,31 +25,22 @@
     import flux.job
     from flux.job import JobspecV1
 except ImportError:
-    error = "no flux Python bindings found"
+    error = 'no flux Python bindings found'
 else:
     error = None
 
-waiting_states = ["QUEUED", "HELD", "WAITING", "PENDING"]
+WAITING_STATES = ('QUEUED', 'HELD', 'WAITING', 'PENDING')
 
 
 class _FluxJob(Job):
     def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-
-        self._cancelled = False
+        '''Create the flux job (and future) to watch.'''
 
-        # This is set by the scheduler when both the job's state is
-        # 'COMPLETED' and the job's stdout and stderr are written back
-        self._completed = False
-        self.create_job()
+        super().__init__(*args, **kwargs)
 
-    def create_job(self):
-        """
-        Create the flux job (and future) to watch.
-        """
         # Generate the flux job
         self.fluxjob = JobspecV1.from_command(
-            command=["/bin/bash", self.script_filename],
+            command=['/bin/bash', self.script_filename],
             num_tasks=self.num_tasks_per_core or 1,
             cores_per_task=self.num_cpus_per_task or 1,
         )
@@ -65,21 +56,17 @@ def create_job(self):
         self.fluxjob.cwd = os.path.abspath(self.workdir)
         self.fluxjob.environment = dict(os.environ)
 
-    @property
-    def cancelled(self):
-        return self._cancelled
-
     @property
     def completed(self):
-        return self._completed
+        return not self.state in WAITING_STATES
 
 
-@register_scheduler("flux", error=error)
+@register_scheduler('flux', error=error)
 class FluxJobScheduler(JobScheduler):
     def __init__(self):
         self._fexecutor = flux.job.FluxExecutor()
         self._submit_timeout = rt.runtime().get_option(
-            f"schedulers/@{self.registered_name}/job_submit_timeout"
+            f'schedulers/@{self.registered_name}/job_submit_timeout'
         )
 
     def emit_preamble(self, job):
@@ -90,36 +77,24 @@ def make_job(self, *args, **kwargs):
         return _FluxJob(*args, **kwargs)
 
     def submit(self, job):
-        """
-        Submit a job to the flux executor.
-        """
+        '''Submit a job to the flux executor.'''
+
         flux_future = self._fexecutor.submit(job.fluxjob)
         job._jobid = str(flux_future.jobid())
         job._submit_time = time.time()
         job._flux_future = flux_future
 
     def cancel(self, job):
-        """
-        Cancel a running Flux job.
-        """
+        '''Cancel a running Flux job.'''
+
         # Job future cannot cancel once running or completed
         if not job._flux_future.cancel():
             # This will raise JobException with event=cancel (on poll)
             flux.job.cancel(flux.Flux(), job._flux_future.jobid())
 
-        # In testing, cancel is too "soft" - or often the pid is still remaining
-        # after. We thus do a cancel in addition to a kill.
-        try:
-            flux.job.kill(flux.Flux(), job._flux_future.jobid())
-
-        # Job is inactive
-        except EnvironmentError:
-            pass
-
     def poll(self, *jobs):
-        """
-        Poll running Flux jobs for updated states.
-        """
+        '''Poll running Flux jobs for updated states.'''
+
         if jobs:
             # filter out non-jobs
             jobs = [job for job in jobs if job is not None]
@@ -135,41 +110,43 @@ def poll(self, *jobs):
                     exit_code = job._flux_future.result(0)
                 except flux.job.JobException:
                     # Currently the only state we see is cancelled here
-                    self.log(f"Job {job.jobid} was likely cancelled.")
-                    job._state = "CANCELLED"
-                    job._cancelled = True
+                    self.log(f'Job {job.jobid} was likely cancelled.')
+                    job._state = 'CANCELLED'
                 except RuntimeError:
                     # Assume some runtime issue (suspended)
-                    self.log(f"Job {job.jobid} was likely suspended.")
-                    job._state = "SUSPENDED"
+                    self.log(f'Job {job.jobid} was likely suspended.')
+                    job._state = 'SUSPENDED'
                 else:
                     # the job finished (but possibly with nonzero exit code)
                     if exit_code != 0:
-                        self.log(f"Job {job.jobid} did not finish successfully")
+                        self.log(
+                            f'Job {job.jobid} did not finish successfully'
+                        )
 
-                    job._state = "COMPLETED"
+                    job._state = 'COMPLETED'
 
                 job._completed = True
-            elif job.state in waiting_states and job.max_pending_time:
+            elif job.state in WAITING_STATES and job.max_pending_time:
                 if time.time() - job.submit_time >= job.max_pending_time:
                     self.cancel(job)
                     job._exception = JobError(
-                        "maximum pending time exceeded", job.jobid
+                        'maximum pending time exceeded', job.jobid
                     )
             else:
                 # Otherwise, we are still running
-                job._state = "RUNNING"
+                job._state = 'RUNNING'
 
     def allnodes(self):
-        raise NotImplementedError("flux backend does not support node listing")
+        raise NotImplementedError('flux backend does not support node listing')
 
     def filternodes(self, job, nodes):
-        raise NotImplementedError("flux backend does not support node filtering")
+        raise NotImplementedError(
+            'flux backend does not support node filtering'
+        )
 
     def wait(self, job):
-        """
-        Wait until a job is finished.
-        """
+        '''Wait until a job is finished.'''
+
         intervals = itertools.cycle([1, 2, 3])
         while not self.finished(job):
             self.poll(job)
@@ -178,4 +155,5 @@ def wait(self, job):
     def finished(self, job):
         if job.exception:
             raise job.exception
+
         return job.completed

unittests/test_schedulers.py

@@ -397,7 +397,7 @@ def test_submit(make_job, exec_ctx):
         assert [socket.gethostname()] == minimal_job.nodelist
         assert minimal_job.exitcode == 0
         assert minimal_job.state == 'SUCCESS'
-    elif sched_name in ('slurm', 'squeue', 'pbs', 'torque', 'flux'):
+    elif sched_name in ('slurm', 'squeue', 'pbs', 'torque'):
         num_tasks_per_node = minimal_job.num_tasks_per_node or 1
         num_nodes = minimal_job.num_tasks // num_tasks_per_node
         assert num_nodes == len(minimal_job.nodelist)
@@ -667,8 +667,12 @@ def test_cancel_with_grace(minimal_job, scheduler, local_only):
     assert minimal_job.signal == signal.SIGKILL
 
     # Verify that the spawned sleep is killed, too, but back off a bit in
-    # order to allow the sleep process to wake up and get the signal
-    time.sleep(0.1)
+    # order to allow the init process to reap it.
+    #
+    # NOTE: If this unit test is run inside a container, make sure that the
+    # PID 1 process is able to reap zombie processes; if not, make sure that
+    # the container is launched with the proper options, e.g., `docker --init`.
+    time.sleep(0.2)
     assert_process_died(sleep_pid)
 
 
@@ -710,8 +714,12 @@ def test_cancel_term_ignore(minimal_job, scheduler, local_only):
     assert minimal_job.signal == signal.SIGKILL
 
     # Verify that the spawned sleep is killed, too, but back off a bit in
-    # order to allow the sleep process to wake up and get the signal
-    time.sleep(0.1)
+    # order to allow the init process to reap it.
+    #
+    # NOTE: If this unit test is run inside a container, make sure that the
+    # PID 1 process is able to reap zombie processes; if not, make sure that
+    # the container is launched with the proper options, e.g., `docker --init`.
+    time.sleep(0.2)
     assert_process_died(sleep_pid)