cleaning with pylint

Author: BerengerBerthoul

.github/workflows/test.yml

@@ -30,7 +30,7 @@ jobs:
         pip install -r *.egg-info/requires.txt
     - name: Analysing the code with pylint
       run: |
-        pylint --unsafe-load-any-extension=y --disable=fixme $(git ls-files '*.py') || true
+        pylint --unsafe-load-any-extension=y --disable=fixme $(git ls-files "pytest_parallel/*.py" "test/*.py") || true
 
   build:
     needs: [pylint]

.slurm_draft/worker.py

@@ -11,13 +11,13 @@
 test_idx = int(sys.argv[3])
 
 comm = MPI.COMM_WORLD
-print(f'start at {scheduler_ip}@{server_port} test {test_idx} at rank {comm.Get_rank()}/{comm.Get_size()} exec on {socket.gethostname()} - ',datetime.datetime.now())
+print(f'start at {scheduler_ip}@{server_port} test {test_idx} at rank {comm.rank}/{comm.size} exec on {socket.gethostname()} - ',datetime.datetime.now())
 
-if comm.Get_rank() == 0:
+if comm.rank == 0:
     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
         s.connect((scheduler_ip, server_port))
         #time.sleep(10+5*test_idx)
-        #msg = f'Hello from test {test_idx} at rank {comm.Get_rank()}/{comm.Get_size()} exec on {socket.gethostname()}'
+        #msg = f'Hello from test {test_idx} at rank {comm.rank}/{comm.size} exec on {socket.gethostname()}'
         #socket_utils.send(s, msg)
         info = {
             'test_idx': test_idx,

pytest_parallel/gather_report.py

@@ -45,8 +45,8 @@ def gather_report_on_local_rank_0(report):
     del report.sub_comm  # No need to keep it in the report
     # Furthermore we need to serialize the report
     # and mpi4py does not know how to serialize report.sub_comm
-    i_sub_rank = sub_comm.Get_rank()
-    n_sub_rank = sub_comm.Get_size()
+    i_sub_rank = sub_comm.rank
+    n_sub_rank = sub_comm.size
 
     if (
         report.outcome != "skipped"

pytest_parallel/mpi_reporter.py

@@ -1,5 +1,6 @@
-import pytest
 import sys
+
+import pytest
 from mpi4py import MPI
 
 from .algo import partition, lower_bound
@@ -11,7 +12,7 @@
 
 def mark_skip(item):
     comm = MPI.COMM_WORLD
-    n_rank = comm.Get_size()
+    n_rank = comm.size
     n_proc_test = get_n_proc_for_test(item)
     skip_msg = f"Not enough procs to execute: {n_proc_test} required but only {n_rank} available"
     item.add_marker(pytest.mark.skip(reason=skip_msg), append=False)
@@ -38,8 +39,7 @@ def create_sub_comm_of_size(global_comm, n_proc, mpi_comm_creation_function):
         assert 0, 'Unknown MPI communicator creation function. Available: `MPI_Comm_create`, `MPI_Comm_split`'
 
 def create_sub_comms_for_each_size(global_comm, mpi_comm_creation_function):
-    i_rank = global_comm.Get_rank()
-    n_rank = global_comm.Get_size()
+    n_rank = global_comm.size
     sub_comms = [None] * n_rank
     for i in range(0,n_rank):
         n_proc = i+1
@@ -48,8 +48,7 @@ def create_sub_comms_for_each_size(global_comm, mpi_comm_creation_function):
 
 
 def add_sub_comm(items, global_comm, test_comm_creation, mpi_comm_creation_function):
-    i_rank = global_comm.Get_rank()
-    n_rank = global_comm.Get_size()
+    n_rank = global_comm.size
 
     # Strategy 'by_rank': create one sub-communicator by size, from sequential (size=1) to n_rank
     if test_comm_creation == 'by_rank':
@@ -109,7 +108,7 @@ def pytest_runtestloop(self, session) -> bool:
         _ = yield
         # prevent return value being non-zero (ExitCode.NO_TESTS_COLLECTED)
         # when no test run on non-master
-        if self.global_comm.Get_rank() != 0 and session.testscollected == 0:
+        if self.global_comm.rank != 0 and session.testscollected == 0:
             session.testscollected = 1
         return True
 
@@ -132,7 +131,7 @@ def pytest_runtest_logreport(self, report):
 
 
 def prepare_items_to_run(items, comm):
-    i_rank = comm.Get_rank()
+    i_rank = comm.rank
 
     items_to_run = []
 
@@ -164,7 +163,7 @@ def prepare_items_to_run(items, comm):
 
 
 def items_to_run_on_this_proc(items_by_steps, items_to_skip, comm):
-    i_rank = comm.Get_rank()
+    i_rank = comm.rank
 
     items = []
 
@@ -207,7 +206,7 @@ def pytest_runtestloop(self, session) -> bool:
         if session.config.option.collectonly:
             return True
 
-        n_workers = self.global_comm.Get_size()
+        n_workers = self.global_comm.size
 
         add_n_procs(session.items)
 
@@ -217,12 +216,12 @@ def pytest_runtestloop(self, session) -> bool:
             items_by_steps, items_to_skip, self.global_comm
         )
 
-        for i, item in enumerate(items):
+        for item in items:
             nextitem = None
             run_item_test(item, nextitem, session)
 
         # prevent return value being non-zero (ExitCode.NO_TESTS_COLLECTED) when no test run on non-master
-        if self.global_comm.Get_rank() != 0 and session.testscollected == 0:
+        if self.global_comm.rank != 0 and session.testscollected == 0:
             session.testscollected = 1
         return True
 
@@ -244,8 +243,8 @@ def pytest_runtest_logreport(self, report):
         gather_report_on_local_rank_0(report)
 
         # master ranks of each sub_comm must send their report to rank 0
-        if sub_comm.Get_rank() == 0:  # only master are concerned
-            if self.global_comm.Get_rank() != 0:  # if master is not global master, send
+        if sub_comm.rank == 0:  # only master are concerned
+            if self.global_comm.rank != 0:  # if master is not global master, send
                 self.global_comm.send(report, dest=0)
             elif report.master_running_proc != 0:  # else, recv if test run remotely
                 # In the line below, MPI.ANY_TAG will NOT clash with communications outside the framework because self.global_comm is private
@@ -342,7 +341,7 @@ def wait_test_to_complete(items_to_run, session, available_procs, inter_comm):
     for sub_rank in sub_ranks:
         if sub_rank != first_rank_done:
             rank_original_idx = inter_comm.recv(source=sub_rank, tag=WORK_DONE_TAG)
-            assert (rank_original_idx == original_idx) # sub_rank is supposed to have worked on the same test
+            assert rank_original_idx == original_idx # sub_rank is supposed to have worked on the same test
 
     # the procs are now available
     for sub_rank in sub_ranks:
@@ -499,7 +498,7 @@ def pytest_runtest_logreport(self, report):
                 sub_comm = report.sub_comm
                 gather_report_on_local_rank_0(report)
 
-                if sub_comm.Get_rank() == 0:  # if local master proc, send
+                if sub_comm.rank == 0:  # if local master proc, send
                     # The idea of the scheduler is the following:
                     #   The server schedules test over clients
                     #   A client executes the test then report to the server it is done

pytest_parallel/plugin.py

@@ -7,6 +7,7 @@
 import tempfile
 from pathlib import Path
 import argparse
+
 import pytest
 from _pytest.terminal import TerminalReporter
 
@@ -104,9 +105,9 @@ def pytest_configure(config):
     slurm_file = config.getoption('slurm_file')
     slurm_export_env = config.getoption('slurm_export_env')
     detach = config.getoption('detach')
-    if scheduler != 'slurm' and scheduler != 'shell':
+    if not scheduler in ['slurm', 'shell']:
         assert not is_worker, f'Internal pytest_parallel error `--_worker` not available with`--scheduler={scheduler}`'
-    if (scheduler == 'slurm' or scheduler == 'shell') and not is_worker:
+    if scheduler in ['slurm', 'shell'] and not is_worker:
         if n_workers is None:
             raise PytestParallelError(f'You need to specify `--n-workers` when `--scheduler={scheduler}`')
     if scheduler != 'slurm':
@@ -119,7 +120,7 @@ def pytest_configure(config):
         if slurm_file is not None:
             raise PytestParallelError('Option `--slurm-file` only available when `--scheduler=slurm`')
 
-    if (scheduler == 'shell' or scheduler == 'slurm') and not is_worker:
+    if scheduler in ['shell', 'slurm'] and not is_worker:
         from mpi4py import MPI
         if MPI.COMM_WORLD.size != 1:
             err_msg = 'Do not launch `pytest_parallel` on more that one process when `--scheduler=shell` or `--scheduler=slurm`.\n' \
@@ -142,7 +143,7 @@ def pytest_configure(config):
                 raise PytestParallelError('You cannot specify `--slurm-init-cmds` together with `--slurm-file`')
 
         if '-n=' in slurm_options or '--ntasks=' in slurm_options:
-                raise PytestParallelError('Do not specify `-n/--ntasks` in `--slurm-options` (it is deduced from the `--n-worker` value).')
+            raise PytestParallelError('Do not specify `-n/--ntasks` in `--slurm-options` (it is deduced from the `--n-worker` value).')
 
         from .slurm_scheduler import SlurmScheduler
 
@@ -154,7 +155,7 @@ def pytest_configure(config):
         ## pull apart `--slurm-options` for special treatement
         main_invoke_params = main_invoke_params.replace(f'--slurm-options={slurm_options}', '')
         for file_or_dir in config.option.file_or_dir:
-          main_invoke_params = main_invoke_params.replace(file_or_dir, '')
+            main_invoke_params = main_invoke_params.replace(file_or_dir, '')
         slurm_option_list = slurm_options.split() if slurm_options is not None else []
         slurm_conf = {
             'options'     : slurm_option_list,
@@ -172,7 +173,7 @@ def pytest_configure(config):
         # reconstruct complete invoke string
         main_invoke_params = _invoke_params(config.invocation_params.args)
         for file_or_dir in config.option.file_or_dir:
-          main_invoke_params = main_invoke_params.replace(file_or_dir, '')
+            main_invoke_params = main_invoke_params.replace(file_or_dir, '')
         plugin = ShellStaticScheduler(main_invoke_params, n_workers, detach)
     else:
         from mpi4py import MPI
@@ -190,7 +191,7 @@ def pytest_configure(config):
         elif scheduler == 'dynamic':
             inter_comm = spawn_master_process(global_comm)
             plugin = DynamicScheduler(global_comm, inter_comm)
-        elif (scheduler == 'slurm' or scheduler == 'shell') and is_worker:
+        elif scheduler in ['shell', 'slurm'] and is_worker:
             scheduler_ip_address = config.getoption('_scheduler_ip_address')
             scheduler_port = config.getoption('_scheduler_port')
             session_folder = config.getoption('_session_folder')
@@ -209,7 +210,7 @@ def pytest_configure(config):
 
         # Pytest relies on having a terminal reporter to decide on how to create error messages, see #12
         # Hence, register a terminal reporter that outputs to /dev/null
-        null_file = open(os.devnull,'w')
+        null_file = open(os.devnull,'w', encoding="utf-8")
         terminal_reporter = TerminalReporter(config, null_file)
         config.pluginmanager.register(terminal_reporter, "terminalreporter")
 
@@ -238,7 +239,7 @@ def __init__(self, comm):
     def __enter__(self):
         from mpi4py import MPI
         if self.comm != MPI.COMM_NULL: # TODO DEL once non-participating rank do not participate in fixtures either
-            rank = self.comm.Get_rank()
+            rank = self.comm.rank
             self.tmp_dir = tempfile.TemporaryDirectory() if rank == 0 else None
             self.tmp_path = Path(self.tmp_dir.name) if rank == 0 else None
             return self.comm.bcast(self.tmp_path, root=0)
@@ -247,7 +248,7 @@ def __exit__(self, type, value, traceback):
         from mpi4py import MPI
         if self.comm != MPI.COMM_NULL: # TODO DEL once non-participating rank do not participate in fixtures either
             self.comm.barrier()
-            if self.comm.Get_rank() == 0:
+            if self.comm.rank == 0:
                 self.tmp_dir.cleanup()
 
 

pytest_parallel/process_worker.py

@@ -1,9 +1,9 @@
-import pytest
+from pathlib import Path
+import pickle
 
+import pytest
 from mpi4py import MPI
 
-from pathlib import Path
-import pickle
 from .utils.items import get_n_proc_for_test, run_item_test
 from .gather_report import gather_report_on_local_rank_0
 
@@ -32,15 +32,15 @@ def pytest_runtestloop(self, session) -> bool:
 
         # check there is no file from a previous run
         if comm.rank == 0:
-            for when in {'fatal_error', 'setup', 'call', 'teardown'}:
+            for when in ['fatal_error', 'setup', 'call', 'teardown']:
                 path = self._file_path(when)
                 assert not path.exists(), f'INTERNAL FATAL ERROR in pytest_parallel: file "{path}" should not exist at this point'
 
         # check the number of procs matches the one specified by the test
         if comm.size != test_comm_size: # fatal error, SLURM and MPI do not interoperate correctly
             if comm.rank == 0:
                 error_info = f'FATAL ERROR in pytest_parallel with slurm scheduling: test `{item.nodeid}`' \
-                             f' uses a `comm` of size {test_comm_size} but was launched with size {comm.Get_size()}.\n' \
+                             f' uses a `comm` of size {test_comm_size} but was launched with size {comm.size}.\n' \
                              f' This generally indicates that `srun` does not interoperate correctly with MPI.'
                 file_path = self._file_path('fatal_error')
                 with open(file_path, "w") as f:
@@ -55,7 +55,7 @@ def pytest_runtestloop(self, session) -> bool:
             assert 0, f'{item.test_info["fatal_error"]}'
 
         return True
-      
+
     @pytest.hookimpl(hookwrapper=True)
     def pytest_runtest_makereport(self, item):
         """

pytest_parallel/send_report.py

@@ -2,12 +2,12 @@
 import socket
 import pickle
 from pathlib import Path
-from .utils.socket import send as socket_send
 from _pytest._code.code import (
     ExceptionChainRepr,
     ReprTraceback,
     ReprEntryNative,
 )
+from .utils.socket import send as socket_send
 
 
 parser = argparse.ArgumentParser(description='Send return the codes of the tests to the master pytest_parallel process')
@@ -21,52 +21,51 @@
 args = parser.parse_args()
 
 def _file_path(when):
-  return Path(f'.pytest_parallel/{args._session_folder}/_partial/{args._test_idx}_{when}')
+    return Path(f'.pytest_parallel/{args._session_folder}/_partial/{args._test_idx}_{when}')
 
 test_info = {'test_idx': args._test_idx, 'fatal_error': None} # TODO no fatal_error=None (absense means no error)
 
 # 'fatal_error' file
 file_path = _file_path('fatal_error')
 if file_path.exists():
-  with open(file_path, 'r') as file:
-    fatal_error = file.read()
-    test_info['fatal_error'] = fatal_error
+    with open(file_path, 'r') as file:
+        fatal_error = file.read()
+        test_info['fatal_error'] = fatal_error
 
 
 # 'setup/call/teardown' files
 already_failed = False
 for when in ('setup', 'call', 'teardown'):
-  file_path = _file_path(when)
-  if file_path.exists():
-    try:
-      with open(file_path, 'rb') as file:
-        report_info = file.read()
-        report_info = pickle.loads(report_info)
-        test_info[when] = report_info
-    except pickle.PickleError:
-      test_info['fatal_error'] = f'FATAL ERROR in pytest_parallel : unable to decode {file_path}'
-  else: # Supposedly not found because the test crashed before writing the file
-    collect_longrepr = []
-    msg = f'Error: the test crashed. '
-    red = 31
-    bold = 1
-    msg = f'\x1b[{red}m' + f'\x1b[{bold}m' + msg+ '\x1b[0m'
-    msg += f'Log file: {args._test_name}\n'
-    trace_back = ReprTraceback([ReprEntryNative(msg)], None, None)
-    collect_longrepr.append(
-        (trace_back, None, None)
-    )
-    longrepr = ExceptionChainRepr(collect_longrepr)
-    
-    outcome = 'passed' if already_failed else 'failed' # No need to report the error twice
-    test_info[when] = {'outcome' : outcome,
-                       'longrepr': longrepr,
-                       'duration': 0, } # unable to report accurately
+    file_path = _file_path(when)
+    if file_path.exists():
+        try:
+            with open(file_path, 'rb') as file:
+                report_info = file.read()
+                report_info = pickle.loads(report_info)
+                test_info[when] = report_info
+        except pickle.PickleError:
+            test_info['fatal_error'] = f'FATAL ERROR in pytest_parallel : unable to decode {file_path}'
+    else: # Supposedly not found because the test crashed before writing the file
+        collect_longrepr = []
+        msg = f'Error: the test crashed. '
+        red = 31
+        bold = 1
+        msg = f'\x1b[{red}m' + f'\x1b[{bold}m' + msg+ '\x1b[0m'
+        msg += f'Log file: {args._test_name}\n'
+        trace_back = ReprTraceback([ReprEntryNative(msg)], None, None)
+        collect_longrepr.append(
+            (trace_back, None, None)
+        )
+        longrepr = ExceptionChainRepr(collect_longrepr)
 
-    already_failed = True
+        outcome = 'passed' if already_failed else 'failed' # No need to report the error twice
+        test_info[when] = {'outcome' : outcome,
+                           'longrepr': longrepr,
+                           'duration': 0, } # unable to report accurately
 
+        already_failed = True
 
-with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-  s.connect((args._scheduler_ip_address, args._scheduler_port))
-  socket_send(s, pickle.dumps(test_info))
 
+with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+    s.connect((args._scheduler_ip_address, args._scheduler_port))
+    socket_send(s, pickle.dumps(test_info))