Repository: dask/dask-jobqueue
Pull request #560 (2022-07-19 18:54:20)

Use `--nworkers` in stead of deprecated `--nprocs` in the generated job scripts
Fix https://github.com/dask/dask-jobqueue/issues/559.

ikabadzhov: I did not leave any `--nprocs` that is right.

I think the deprecation was introduced here https://github.com/dask/distributed/blame/main/distributed/cli/dask_ssh.py#L52 (6 months ago). I am not sure how to map it to the right version.
guillaumeeb: Ok, so the new option name was introduced in 2022.02.0, see https://distributed.dask.org/en/stable/changelog.html#v2022-02-0.

So you should also update required distributed version with this one.

As we just released a dask-jobqueue version, I belive it is safe to just do that.
guillaumeeb: Also, There is on failure related to your PR in the CI: the check of formatting using black. Could you run black on your branch to format things correctly?
ikabadzhov: I just applied `black`. I am not sure for this repository to have code that is not formatted, so I kept all changes in a separate commit. Let me know if I should format only my changes.
ikabadzhov: > Ok, so the new option name was introduced in 2022.02.0, see [distributed.dask.org/en/stable/changelog.html#v2022-02-0](https://distributed.dask.org/en/stable/changelog.html#v2022-02-0).
> 
> So you should also update required distributed version with this one.
> 
> As we just released a dask-jobqueue version, I belive it is safe to just do that.

I just updated the distributed requirement, now it is:
```py
dask>=2.23
distributed>=2022.02.0
```
Is this what you mean? Shouldn't they be the same?
guillaumeeb: > Is this what you mean? Shouldn't they be the same?

This is what I mean, but yes I think they should be the same.

We may also need to wait for #562 to be in to merge this one, but not sure about it.
jacobtomlinson: _We squash and merge here so please don't squash and force push prematurely, it makes review harder as it's unclear what has changed since the last review._
jacobtomlinson: LGTM

Patch:
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 85c2815..0d680f2 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -12,7 +12,8 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        jobqueue: ["htcondor", "pbs", "sge", "slurm", "none"]
+        #HTCondor disabled for now
+        jobqueue: ["pbs", "sge", "slurm", "none"]
 
     steps:
       - name: Cancel previous runs
@@ -51,6 +52,7 @@ jobs:
           jobqueue_script
 
       - name: Cleanup
+        if: always()
         run: |
           source ci/${{ matrix.jobqueue }}.sh
           jobqueue_after_script
diff --git a/.gitignore b/.gitignore
index b4c86c2..f6e47a7 100644
--- a/.gitignore
+++ b/.gitignore
@@ -14,3 +14,4 @@ log
 .cache/
 .pytest_cache
 docs/source/generated
+dask-worker-space/
\ No newline at end of file
diff --git a/ci/environment.yml b/ci/environment.yml
index 6cb8253..abe3759 100644
--- a/ci/environment.yml
+++ b/ci/environment.yml
@@ -3,7 +3,7 @@ channels:
   - conda-forge
   - defaults
 dependencies:
-  - python=3.7
+  - python=3.8
   - dask
   - distributed
   - flake8
diff --git a/ci/htcondor.sh b/ci/htcondor.sh
index ba422a5..01f3abe 100755
--- a/ci/htcondor.sh
+++ b/ci/htcondor.sh
@@ -9,6 +9,8 @@ function jobqueue_before_install {
     docker-compose pull
     docker-compose build
     ./start-htcondor.sh
+    docker-compose exec -T submit /bin/bash -c "condor_status"
+    docker-compose exec -T submit /bin/bash -c "condor_q"
     cd -
 
     docker ps -a
@@ -23,12 +25,15 @@ function jobqueue_install {
 
 function jobqueue_script {
     cd ./ci/htcondor
-    docker-compose exec -T --user submituser submit /bin/bash -c "cd; pytest /dask-jobqueue/dask_jobqueue --verbose -E htcondor -s"
+    docker-compose exec -T --user submituser submit /bin/bash -c "cd; pytest /dask-jobqueue/dask_jobqueue --log-cli-level DEBUG --capture=tee-sys --verbose -E htcondor "
     cd -
 }
 
 function jobqueue_after_script {
     cd ./ci/htcondor
+    docker-compose exec -T submit /bin/bash -c "condor_q"
+    docker-compose exec -T submit /bin/bash -c "condor_status"
+    docker-compose exec -T submit /bin/bash -c "condor_history"
     docker-compose exec -T cm /bin/bash -c " grep -R \"\" /var/log/condor/	"
     cd -
 }
diff --git a/ci/htcondor/Dockerfile b/ci/htcondor/Dockerfile
index 937ed36..5e4af17 100644
--- a/ci/htcondor/Dockerfile
+++ b/ci/htcondor/Dockerfile
@@ -1,10 +1,13 @@
 FROM  htcondor/submit:el7 as submit
 
-RUN yum install -y gcc git 
-RUN yum install -y python3-devel python3-pip 
-RUN pip3 install dask distributed pytest
+RUN curl -o miniconda.sh https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh && \
+    bash miniconda.sh -f -b -p /opt/anaconda && \
+    /opt/anaconda/bin/conda clean -tipy && \
+    rm -f miniconda.sh
+ENV PATH /opt/anaconda/bin:$PATH
+RUN conda install --yes -c conda-forge python=3.8 dask distributed flake8 pytest pytest-asyncio
 
-FROM  htcondor/execute:el7 as execute 
-RUN yum install -y python3
-COPY --from=submit /usr/local/lib/python3.6 /usr/local/lib/python3.6
-COPY --from=submit /usr/local/lib64/python3.6 /usr/local/lib64/python3.6
+FROM  htcondor/execute:el7 as execute
+
+COPY --from=submit /opt/anaconda /opt/anaconda
+ENV PATH /opt/anaconda/bin:$PATH
diff --git a/ci/htcondor/condor_config.local b/ci/htcondor/condor_config.local
new file mode 100644
index 0000000..688eeae
--- /dev/null
+++ b/ci/htcondor/condor_config.local
@@ -0,0 +1,1 @@
+NEGOTIATOR_INTERVAL=10
\ No newline at end of file
diff --git a/ci/htcondor/docker-compose.yml b/ci/htcondor/docker-compose.yml
index b883cda..4fbad96 100644
--- a/ci/htcondor/docker-compose.yml
+++ b/ci/htcondor/docker-compose.yml
@@ -8,6 +8,7 @@ services:
       - USE_POOL_PASSWORD=yes
     volumes:
       - secrets:/root/secrets
+      - ./condor_config.local:/etc/condor/condor_config.local
     command: bash -c 'condor_store_cred -p password -f /root/secrets/pool_password ; exec bash -x /start.sh'
 
   submit:
@@ -24,6 +25,7 @@ services:
     volumes:
       - secrets:/root/secrets
       - ../..:/dask-jobqueue
+      - ./condor_config.local:/etc/condor/condor_config.local
 
   execute1:
     image: daskdev/dask-jobqueue:htcondor-execute
@@ -38,6 +40,7 @@ services:
       - cm
     volumes:
       - secrets:/root/secrets
+      - ./condor_config.local:/etc/condor/condor_config.local
 
   execute2:
     image: daskdev/dask-jobqueue:htcondor-execute
@@ -52,6 +55,7 @@ services:
       - cm
     volumes:
       - secrets:/root/secrets
+      - ./condor_config.local:/etc/condor/condor_config.local
 
 volumes:
   secrets:
diff --git a/ci/pbs/Dockerfile b/ci/pbs/Dockerfile
index 8627ee7..9b7957e 100644
--- a/ci/pbs/Dockerfile
+++ b/ci/pbs/Dockerfile
@@ -30,7 +30,7 @@ RUN curl -o miniconda.sh https://repo.anaconda.com/miniconda/Miniconda3-latest-L
     bash miniconda.sh -f -b -p /opt/anaconda && \
     /opt/anaconda/bin/conda clean -tipy && \
     rm -f miniconda.sh
-RUN conda install --yes -c conda-forge python=3.7 dask distributed flake8 pytest pytest-asyncio
+RUN conda install --yes -c conda-forge python=3.8 dask distributed flake8 pytest pytest-asyncio
 
 # Copy entrypoint and other needed scripts
 COPY ./*.sh /
diff --git a/ci/sge/docker-compose.yml b/ci/sge/docker-compose.yml
index 024ba41..8694281 100644
--- a/ci/sge/docker-compose.yml
+++ b/ci/sge/docker-compose.yml
@@ -8,7 +8,7 @@ services:
       context: .
       target: master
       args:
-        PYTHON_VERSION: 3.7
+        PYTHON_VERSION: 3.8
     container_name: sge_master
     hostname: sge_master
     #network_mode: host
@@ -22,7 +22,7 @@ services:
       context: .
       target: slave
       args:
-        PYTHON_VERSION: 3.7
+        PYTHON_VERSION: 3.8
     container_name: slave_one
     hostname: slave_one
     #network_mode: host
@@ -40,7 +40,7 @@ services:
       context: .
       target: slave
       args:
-        PYTHON_VERSION: 3.7
+        PYTHON_VERSION: 3.8
     container_name: slave_two
     hostname: slave_two
     #network_mode: host
diff --git a/ci/slurm/Dockerfile b/ci/slurm/Dockerfile
index ac1bbb2..4019625 100644
--- a/ci/slurm/Dockerfile
+++ b/ci/slurm/Dockerfile
@@ -7,7 +7,7 @@ RUN curl -o miniconda.sh https://repo.anaconda.com/miniconda/Miniconda3-latest-L
     /opt/anaconda/bin/conda clean -tipy && \
     rm -f miniconda.sh
 ENV PATH /opt/anaconda/bin:$PATH
-RUN conda install --yes -c conda-forge python=3.7 dask distributed flake8 pytest pytest-asyncio
+RUN conda install --yes -c conda-forge python=3.8 dask distributed flake8 pytest pytest-asyncio
 
 ENV LC_ALL en_US.UTF-8
 
diff --git a/dask_jobqueue/config.py b/dask_jobqueue/config.py
index 94f5cbd..32c2369 100644
--- a/dask_jobqueue/config.py
+++ b/dask_jobqueue/config.py
@@ -3,10 +3,15 @@ import os
 import dask
 import yaml
 
-fn = os.path.join(os.path.dirname(__file__), "jobqueue.yaml")
-dask.config.ensure_file(source=fn)
 
-with open(fn) as f:
-    defaults = yaml.safe_load(f)
+def reconfigure():
+    fn = os.path.join(os.path.dirname(__file__), "jobqueue.yaml")
+    dask.config.ensure_file(source=fn)
 
-dask.config.update(dask.config.config, defaults, priority="old")
+    with open(fn) as f:
+        defaults = yaml.safe_load(f)
+
+    dask.config.update_defaults(defaults)
+
+
+reconfigure()
diff --git a/dask_jobqueue/core.py b/dask_jobqueue/core.py
index baf5414..fb1ac99 100644
--- a/dask_jobqueue/core.py
+++ b/dask_jobqueue/core.py
@@ -255,7 +255,7 @@ class Job(ProcessInterface, abc.ABC):
         command_args = [dask_worker_command, self.scheduler]
         command_args += ["--nthreads", self.worker_process_threads]
         if processes is not None and processes > 1:
-            command_args += ["--nprocs", processes]
+            command_args += ["--nworkers", processes]
 
         command_args += ["--memory-limit", self.worker_process_memory]
         command_args += ["--name", str(name)]
diff --git a/dask_jobqueue/htcondor.py b/dask_jobqueue/htcondor.py
index fb7b0be..18fc74e 100644
--- a/dask_jobqueue/htcondor.py
+++ b/dask_jobqueue/htcondor.py
@@ -16,7 +16,6 @@ class HTCondorJob(Job):
 
 %(job_header)s
 
-Environment = "%(quoted_environment)s"
 Arguments = "%(quoted_arguments)s"
 Executable = %(executable)s
 
@@ -67,7 +66,14 @@ Queue
             env_extra = dask.config.get(
                 "jobqueue.%s.env-extra" % self.config_name, default=[]
             )
-        self.env_dict = self.env_lines_to_dict(env_extra)
+
+        if env_extra is not None:
+            # Overwrite command template: prepend commands from env_extra separated by semicolon.
+            # This is special for HTCondor, because lines to execute on the worker node cannot be
+            # simply added to the submit script like for other batch systems.
+            self._command_template = (
+                "; ".join(env_extra) + "; " + self._command_template
+            )
 
         self.job_header_dict = {
             "MY.DaskWorkerName": '"htcondor--$F(MY.JobId)--"',
@@ -118,31 +124,15 @@ Queue
             + " ".join(shlex.quote(arg) for arg in cancel_command_extra)
         )
 
-    def env_lines_to_dict(self, env_lines):
-        """Convert an array of export statements (what we get from env-extra
-        in the config) into a dict"""
-        env_dict = {}
-        for env_line in env_lines:
-            split_env_line = shlex.split(env_line)
-            if split_env_line[0] == "export":
-                split_env_line = split_env_line[1:]
-            for item in split_env_line:
-                if "=" in item:
-                    k, v = item.split("=", 1)
-                    env_dict[k] = v
-        return env_dict
-
     def job_script(self):
         """Construct a job submission script"""
         quoted_arguments = quote_arguments(["-c", self._command_template])
-        quoted_environment = quote_environment(self.env_dict)
         job_header_lines = "\n".join(
             "%s = %s" % (k, v) for k, v in self.job_header_dict.items()
         )
         return self._script_template % {
             "shebang": self.shebang,
             "job_header": job_header_lines,
-            "quoted_environment": quoted_environment,
             "quoted_arguments": quoted_arguments,
             "executable": self.executable,
         }
@@ -260,6 +250,17 @@ class HTCondorCluster(JobQueueCluster):
     This also works with adaptive clusters.  This automatically launches and kill workers based on load.
 
     >>> cluster.adapt(maximum_jobs=20)
+
+    If setup commands need to be run before starting the worker on the worker node, ``env_extra`` can be used,
+    e.g., to activate a virtual environment:
+
+    >>> from dask_jobqueue.htcondor import HTCondorCluster
+    >>> cluster = HTCondorCluster(cores=1, memory="2GB", disk="4GB",
+                                  env_extra=['cd /some/path/', 'source venv/bin/activate'])
+
+    Note that environment variables are no longer passed via the ``Environment`` parameter in the submit
+    description file. If you explictly want to set that, you need to use ``job_extra``.
+
     """.format(
         job=job_parameters, cluster=cluster_parameters
     )
diff --git a/docs/environment.yml b/docs/environment.yml
index c279b99..1743575 100644
--- a/docs/environment.yml
+++ b/docs/environment.yml
@@ -2,7 +2,7 @@ name: dask-jobqueue-docs
 channels:
   - conda-forge
 dependencies:
-  - python=3.7
+  - python=3.8
   - distributed
   - numpydoc
   - ipython
diff --git a/docs/source/advanced-tips-and-tricks.rst b/docs/source/advanced-tips-and-tricks.rst
index bd79810..237adc6 100644
--- a/docs/source/advanced-tips-and-tricks.rst
+++ b/docs/source/advanced-tips-and-tricks.rst
@@ -68,6 +68,36 @@ accepted option on some SLURM clusters. The error was something like this:
   sbatch: error: Memory specification can not be satisfied
   sbatch: error: Batch job submission failed: Requested node configuration is not available
 
+Run setup commands before starting the worker with ``env_extra``
+----------------------------------------------------------------
+
+Sometimes you need to run some setup commands before the actual worker can be started. This includes
+setting environment variables, loading environment modules, sourcing/activating a virtual environment,
+or activating conda/mamba environments.
+
+This can be achieved using the ``env_extra`` parameter. Example for setting up a virtual environment:
+
+.. code-block:: python
+
+   from dask_jobqueue.htcondor import HTCondorCluster
+   env_extra = ['cd /some/path', 'source venv/bin/activate']
+   cluster = HTCondorCluster(cores=1, memory="2GB", disk="4GB", log_directory = 'logs', python='python3',
+                             env_extra=env_extra)
+   print(cluster.job_script())
+
+For ``HTCondorCluster``, the commands will be prepended to the actual python call in the ``Arguments``
+parameter in the submit description file. The relevant lines will look like this:
+
+.. code-block:: text
+
+   ...
+   Arguments = "-c 'cd /some/path; source venv/bin/activate; python3 -m distributed.cli.dask_worker tcp://<IP>:<PORT> --nthreads 1 --memory-limit 2.00GB --name dummy-name --nanny --death-timeout 60'"
+   Executable = /bin/sh
+   ...
+
+For other batch systems (``*Cluster`` classes) the additional commands will be inserted as separate lines
+in the submission script.
+
 
 How to handle job queueing system walltime killing workers
 ----------------------------------------------------------
diff --git a/docs/source/debug.rst b/docs/source/debug.rst
index 943a6b2..7e22bb7 100644
--- a/docs/source/debug.rst
+++ b/docs/source/debug.rst
@@ -70,7 +70,7 @@ the right module just before:
    distributed.nanny - INFO - Worker closed
 
 This happens when you created the cluster using a different python than the one
-you want to use for your workers (here ``module load python/3.7.5``), giving
+you want to use for your workers (here ``module load python/3.8.5``), giving
 the following job script (pay attention to the last line which will show which
 ``python`` is used):
 
diff --git a/docs/source/examples.rst b/docs/source/examples.rst
index ad02971..4f9a382 100644
--- a/docs/source/examples.rst
+++ b/docs/source/examples.rst
@@ -32,7 +32,7 @@ PBS Deployments
                         interface='ib0')
 
 Moab Deployments
-~~~~~~~~~~~~~~~~
+----------------
 
 On systems which use the Moab Workload Manager, a subclass of ``PBSCluster``
 can be used, called ``MoabCluster``:
diff --git a/docs/source/howitworks.rst b/docs/source/howitworks.rst
index 3e0a2ca..282643d 100644
--- a/docs/source/howitworks.rst
+++ b/docs/source/howitworks.rst
@@ -57,7 +57,7 @@ generate as follows:
    #PBS -l walltime=02:00:00
 
    /home/username/path/to/bin/dask-worker tcp://127.0.1.1:43745
-   --nthreads 4 --nprocs 6 --memory-limit 18.66GB --name dask-worker-3
+   --nthreads 4 --nworkers 6 --memory-limit 18.66GB --name dask-worker-3
    --death-timeout 60
 
 Each of these jobs are sent to the job queue independently and, once that job
diff --git a/requirements.txt b/requirements.txt
index 3c32eca..2652294 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,2 +1,2 @@
-dask>=2.23
-distributed>=2.23
+dask>=2022.02.0
+distributed>=2022.02.0
diff --git a/setup.py b/setup.py
index a8d4a6c..988490d 100755
--- a/setup.py
+++ b/setup.py
@@ -28,7 +28,7 @@ setup(
     cmdclass=versioneer.get_cmdclass(),
     description="Deploy Dask on job queuing systems like PBS, Slurm, SGE or LSF",
     url="https://jobqueue.dask.org",
-    python_requires=">=3.7",
+    python_requires=">=3.8",
     license="BSD 3-Clause",
     packages=["dask_jobqueue"],
     include_package_data=True,
diff --git a/versioneer.py b/versioneer.py
index 0562ec5..8f1f2f1 100644
--- a/versioneer.py
+++ b/versioneer.py
@@ -1,4 +1,3 @@
-
 # Version: 0.18
 
 """The Versioneer - like a rocketeer, but for versions.
@@ -310,11 +309,13 @@ def get_root():
         setup_py = os.path.join(root, "setup.py")
         versioneer_py = os.path.join(root, "versioneer.py")
     if not (os.path.exists(setup_py) or os.path.exists(versioneer_py)):
-        err = ("Versioneer was unable to run the project root directory. "
-               "Versioneer requires setup.py to be executed from "
-               "its immediate directory (like 'python setup.py COMMAND'), "
-               "or in a way that lets it use sys.argv[0] to find the root "
-               "(like 'python path/to/setup.py COMMAND').")
+        err = (
+            "Versioneer was unable to run the project root directory. "
+            "Versioneer requires setup.py to be executed from "
+            "its immediate directory (like 'python setup.py COMMAND'), "
+            "or in a way that lets it use sys.argv[0] to find the root "
+            "(like 'python path/to/setup.py COMMAND')."
+        )
         raise VersioneerBadRootError(err)
     try:
         # Certain runtime workflows (setup.py install/develop in a setuptools
@@ -327,8 +328,10 @@ def get_root():
         me_dir = os.path.normcase(os.path.splitext(me)[0])
         vsr_dir = os.path.normcase(os.path.splitext(versioneer_py)[0])
         if me_dir != vsr_dir:
-            print("Warning: build in %s is using versioneer.py from %s"
-                  % (os.path.dirname(me), versioneer_py))
+            print(
+                "Warning: build in %s is using versioneer.py from %s"
+                % (os.path.dirname(me), versioneer_py)
+            )
     except NameError:
         pass
     return root
@@ -350,6 +353,7 @@ def get_config_from_root(root):
         if parser.has_option("versioneer", name):
             return parser.get("versioneer", name)
         return None
+
     cfg = VersioneerConfig()
     cfg.VCS = VCS
     cfg.style = get(parser, "style") or ""
@@ -374,17 +378,18 @@ HANDLERS = {}
 
 def register_vcs_handler(vcs, method):  # decorator
     """Decorator to mark a method as the handler for a particular VCS."""
+
     def decorate(f):
         """Store f in HANDLERS[vcs][method]."""
         if vcs not in HANDLERS:
             HANDLERS[vcs] = {}
         HANDLERS[vcs][method] = f
         return f
+
     return decorate
 
 
-def run_command(commands, args, cwd=None, verbose=False, hide_stderr=False,
-                env=None):
+def run_command(commands, args, cwd=None, verbose=False, hide_stderr=False, env=None):
     """Call the given command(s)."""
     assert isinstance(commands, list)
     p = None
@@ -392,10 +397,13 @@ def run_command(commands, args, cwd=None, verbose=False, hide_stderr=False,
         try:
             dispcmd = str([c] + args)
             # remember shell=False, so use git.cmd on windows, not just git
-            p = subprocess.Popen([c] + args, cwd=cwd, env=env,
-                                 stdout=subprocess.PIPE,
-                                 stderr=(subprocess.PIPE if hide_stderr
-                                         else None))
+            p = subprocess.Popen(
+                [c] + args,
+                cwd=cwd,
+                env=env,
+                stdout=subprocess.PIPE,
+                stderr=(subprocess.PIPE if hide_stderr else None),
+            )
             break
         except EnvironmentError:
             e = sys.exc_info()[1]
@@ -420,7 +428,9 @@ def run_command(commands, args, cwd=None, verbose=False, hide_stderr=False,
     return stdout, p.returncode
 
 
-LONG_VERSION_PY['git'] = '''
+LONG_VERSION_PY[
+    "git"
+] = '''
 # This file helps to compute a version number in source trees obtained from
 # git-archive tarball (such as those provided by githubs download-from-tag
 # feature). Distribution tarballs (built by setup.py sdist) and build
@@ -995,7 +1005,7 @@ def git_versions_from_keywords(keywords, tag_prefix, verbose):
     # starting in git-1.8.3, tags are listed as "tag: foo-1.0" instead of
     # just "foo-1.0". If we see a "tag: " prefix, prefer those.
     TAG = "tag: "
-    tags = set([r[len(TAG):] for r in refs if r.startswith(TAG)])
+    tags = set([r[len(TAG) :] for r in refs if r.startswith(TAG)])
     if not tags:
         # Either we're using git < 1.8.3, or there really are no tags. We use
         # a heuristic: assume all version tags have a digit. The old git %d
@@ -1004,7 +1014,7 @@ def git_versions_from_keywords(keywords, tag_prefix, verbose):
         # between branches and tags. By ignoring refnames without digits, we
         # filter out many common branch names like "release" and
         # "stabilization", as well as "HEAD" and "main".
-        tags = set([r for r in refs if re.search(r'\d', r)])
+        tags = set([r for r in refs if re.search(r"\d", r)])
         if verbose:
             print("discarding '%s', no digits" % ",".join(refs - tags))
     if verbose:
@@ -1012,19 +1022,26 @@ def git_versions_from_keywords(keywords, tag_prefix, verbose):
     for ref in sorted(tags):
         # sorting will prefer e.g. "2.0" over "2.0rc1"
         if ref.startswith(tag_prefix):
-            r = ref[len(tag_prefix):]
+            r = ref[len(tag_prefix) :]
             if verbose:
                 print("picking %s" % r)
-            return {"version": r,
-                    "full-revisionid": keywords["full"].strip(),
-                    "dirty": False, "error": None,
-                    "date": date}
+            return {
+                "version": r,
+                "full-revisionid": keywords["full"].strip(),
+                "dirty": False,
+                "error": None,
+                "date": date,
+            }
     # no suitable tags, so version is "0+unknown", but full hex is still there
     if verbose:
         print("no suitable tags, using unknown + full revision id")
-    return {"version": "0+unknown",
-            "full-revisionid": keywords["full"].strip(),
-            "dirty": False, "error": "no suitable tags", "date": None}
+    return {
+        "version": "0+unknown",
+        "full-revisionid": keywords["full"].strip(),
+        "dirty": False,
+        "error": "no suitable tags",
+        "date": None,
+    }
 
 
 @register_vcs_handler("git", "pieces_from_vcs")
@@ -1039,8 +1056,7 @@ def git_pieces_from_vcs(tag_prefix, root, verbose, run_command=run_command):
     if sys.platform == "win32":
         GITS = ["git.cmd", "git.exe"]
 
-    out, rc = run_command(GITS, ["rev-parse", "--git-dir"], cwd=root,
-                          hide_stderr=True)
+    out, rc = run_command(GITS, ["rev-parse", "--git-dir"], cwd=root, hide_stderr=True)
     if rc != 0:
         if verbose:
             print("Directory %s not under git control" % root)
@@ -1048,10 +1064,19 @@ def git_pieces_from_vcs(tag_prefix, root, verbose, run_command=run_command):
 
     # if there is a tag matching tag_prefix, this yields TAG-NUM-gHEX[-dirty]
     # if there isn't one, this yields HEX[-dirty] (no NUM)
-    describe_out, rc = run_command(GITS, ["describe", "--tags", "--dirty",
-                                          "--always", "--long",
-                                          "--match", "%s*" % tag_prefix],
-                                   cwd=root)
+    describe_out, rc = run_command(
+        GITS,
+        [
+            "describe",
+            "--tags",
+            "--dirty",
+            "--always",
+            "--long",
+            "--match",
+            "%s*" % tag_prefix,
+        ],
+        cwd=root,
+    )
     # --long was added in git-1.5.5
     if describe_out is None:
         raise NotThisMethod("'git describe' failed")
@@ -1074,17 +1099,16 @@ def git_pieces_from_vcs(tag_prefix, root, verbose, run_command=run_command):
     dirty = git_describe.endswith("-dirty")
     pieces["dirty"] = dirty
     if dirty:
-        git_describe = git_describe[:git_describe.rindex("-dirty")]
+        git_describe = git_describe[: git_describe.rindex("-dirty")]
 
     # now we have TAG-NUM-gHEX or HEX
 
     if "-" in git_describe:
         # TAG-NUM-gHEX
-        mo = re.search(r'^(.+)-(\d+)-g([0-9a-f]+)$', git_describe)
+        mo = re.search(r"^(.+)-(\d+)-g([0-9a-f]+)$", git_describe)
         if not mo:
             # unparseable. Maybe git-describe is misbehaving?
-            pieces["error"] = ("unable to parse git-describe output: '%s'"
-                               % describe_out)
+            pieces["error"] = "unable to parse git-describe output: '%s'" % describe_out
             return pieces
 
         # tag
@@ -1093,10 +1117,12 @@ def git_pieces_from_vcs(tag_prefix, root, verbose, run_command=run_command):
             if verbose:
                 fmt = "tag '%s' doesn't start with prefix '%s'"
                 print(fmt % (full_tag, tag_prefix))
-            pieces["error"] = ("tag '%s' doesn't start with prefix '%s'"
-                               % (full_tag, tag_prefix))
+            pieces["error"] = "tag '%s' doesn't start with prefix '%s'" % (
+                full_tag,
+                tag_prefix,
+            )
             return pieces
-        pieces["closest-tag"] = full_tag[len(tag_prefix):]
+        pieces["closest-tag"] = full_tag[len(tag_prefix) :]
 
         # distance: number of commits since tag
         pieces["distance"] = int(mo.group(2))
@@ -1107,13 +1133,13 @@ def git_pieces_from_vcs(tag_prefix, root, verbose, run_command=run_command):
     else:
         # HEX: no tags
         pieces["closest-tag"] = None
-        count_out, rc = run_command(GITS, ["rev-list", "HEAD", "--count"],
-                                    cwd=root)
+        count_out, rc = run_command(GITS, ["rev-list", "HEAD", "--count"], cwd=root)
         pieces["distance"] = int(count_out)  # total number of commits
 
     # commit date: see ISO-8601 comment in git_versions_from_keywords()
-    date = run_command(GITS, ["show", "-s", "--format=%ci", "HEAD"],
-                       cwd=root)[0].strip()
+    date = run_command(GITS, ["show", "-s", "--format=%ci", "HEAD"], cwd=root)[
+        0
+    ].strip()
     pieces["date"] = date.strip().replace(" ", "T", 1).replace(" ", "", 1)
 
     return pieces
@@ -1169,16 +1195,22 @@ def versions_from_parentdir(parentdir_prefix, root, verbose):
     for i in range(3):
         dirname = os.path.basename(root)
         if dirname.startswith(parentdir_prefix):
-            return {"version": dirname[len(parentdir_prefix):],
-                    "full-revisionid": None,
-                    "dirty": False, "error": None, "date": None}
+            return {
+                "version": dirname[len(parentdir_prefix) :],
+                "full-revisionid": None,
+                "dirty": False,
+                "error": None,
+                "date": None,
+            }
         else:
             rootdirs.append(root)
             root = os.path.dirname(root)  # up a level
 
     if verbose:
-        print("Tried directories %s but none started with prefix %s" %
-              (str(rootdirs), parentdir_prefix))
+        print(
+            "Tried directories %s but none started with prefix %s"
+            % (str(rootdirs), parentdir_prefix)
+        )
     raise NotThisMethod("rootdir doesn't start with parentdir_prefix")
 
 
@@ -1207,11 +1239,13 @@ def versions_from_file(filename):
             contents = f.read()
     except EnvironmentError:
         raise NotThisMethod("unable to read _version.py")
-    mo = re.search(r"version_json = '''\n(.*)'''  # END VERSION_JSON",
-                   contents, re.M | re.S)
+    mo = re.search(
+        r"version_json = '''\n(.*)'''  # END VERSION_JSON", contents, re.M | re.S
+    )
     if not mo:
-        mo = re.search(r"version_json = '''\r\n(.*)'''  # END VERSION_JSON",
-                       contents, re.M | re.S)
+        mo = re.search(
+            r"version_json = '''\r\n(.*)'''  # END VERSION_JSON", contents, re.M | re.S
+        )
     if not mo:
         raise NotThisMethod("no version_json in _version.py")
     return json.loads(mo.group(1))
@@ -1220,8 +1254,7 @@ def versions_from_file(filename):
 def write_to_version_file(filename, versions):
     """Write the given version number to the given _version.py file."""
     os.unlink(filename)
-    contents = json.dumps(versions, sort_keys=True,
-                          indent=1, separators=(",", ": "))
+    contents = json.dumps(versions, sort_keys=True, indent=1, separators=(",", ": "))
     with open(filename, "w") as f:
         f.write(SHORT_VERSION_PY % contents)
 
@@ -1253,8 +1286,7 @@ def render_pep440(pieces):
                 rendered += ".dirty"
     else:
         # exception #1
-        rendered = "0+untagged.%d.g%s" % (pieces["distance"],
-                                          pieces["short"])
+        rendered = "0+untagged.%d.g%s" % (pieces["distance"], pieces["short"])
         if pieces["dirty"]:
             rendered += ".dirty"
     return rendered
@@ -1368,11 +1400,13 @@ def render_git_describe_long(pieces):
 def render(pieces, style):
     """Render the given version pieces into the requested style."""
     if pieces["error"]:
-        return {"version": "unknown",
-                "full-revisionid": pieces.get("long"),
-                "dirty": None,
-                "error": pieces["error"],
-                "date": None}
+        return {
+            "version": "unknown",
+            "full-revisionid": pieces.get("long"),
+            "dirty": None,
+            "error": pieces["error"],
+            "date": None,
+        }
 
     if not style or style == "default":
         style = "pep440"  # the default
@@ -1392,9 +1426,13 @@ def render(pieces, style):
     else:
         raise ValueError("unknown style '%s'" % style)
 
-    return {"version": rendered, "full-revisionid": pieces["long"],
-            "dirty": pieces["dirty"], "error": None,
-            "date": pieces.get("date")}
+    return {
+        "version": rendered,
+        "full-revisionid": pieces["long"],
+        "dirty": pieces["dirty"],
+        "error": None,
+        "date": pieces.get("date"),
+    }
 
 
 class VersioneerBadRootError(Exception):
@@ -1417,8 +1455,9 @@ def get_versions(verbose=False):
     handlers = HANDLERS.get(cfg.VCS)
     assert handlers, "unrecognized VCS '%s'" % cfg.VCS
     verbose = verbose or cfg.verbose
-    assert cfg.versionfile_source is not None, \
-        "please set versioneer.versionfile_source"
+    assert (
+        cfg.versionfile_source is not None
+    ), "please set versioneer.versionfile_source"
     assert cfg.tag_prefix is not None, "please set versioneer.tag_prefix"
 
     versionfile_abs = os.path.join(root, cfg.versionfile_source)
@@ -1472,9 +1511,13 @@ def get_versions(verbose=False):
     if verbose:
         print("unable to compute version")
 
-    return {"version": "0+unknown", "full-revisionid": None,
-            "dirty": None, "error": "unable to compute version",
-            "date": None}
+    return {
+        "version": "0+unknown",
+        "full-revisionid": None,
+        "dirty": None,
+        "error": "unable to compute version",
+        "date": None,
+    }
 
 
 def get_version():
@@ -1523,6 +1566,7 @@ def get_cmdclass():
             print(" date: %s" % vers.get("date"))
             if vers["error"]:
                 print(" error: %s" % vers["error"])
+
     cmds["version"] = cmd_version
 
     # we override "build_py" in both distutils and setuptools
@@ -1555,14 +1599,15 @@ def get_cmdclass():
             # now locate _version.py in the new build/ directory and replace
             # it with an updated value
             if cfg.versionfile_build:
-                target_versionfile = os.path.join(self.build_lib,
-                                                  cfg.versionfile_build)
+                target_versionfile = os.path.join(self.build_lib, cfg.versionfile_build)
                 print("UPDATING %s" % target_versionfile)
                 write_to_version_file(target_versionfile, versions)
+
     cmds["build_py"] = cmd_build_py
 
     if "cx_Freeze" in sys.modules:  # cx_freeze enabled?
         from cx_Freeze.dist import build_exe as _build_exe
+
         # nczeczulin reports that py2exe won't like the pep440-style string
         # as FILEVERSION, but it can be used for PRODUCTVERSION, e.g.
         # setup(console=[{
@@ -1583,17 +1628,21 @@ def get_cmdclass():
                 os.unlink(target_versionfile)
                 with open(cfg.versionfile_source, "w") as f:
                     LONG = LONG_VERSION_PY[cfg.VCS]
-                    f.write(LONG %
-                            {"DOLLAR": "$",
-                             "STYLE": cfg.style,
-                             "TAG_PREFIX": cfg.tag_prefix,
-                             "PARENTDIR_PREFIX": cfg.parentdir_prefix,
-                             "VERSIONFILE_SOURCE": cfg.versionfile_source,
-                             })
+                    f.write(
+                        LONG
+                        % {
+                            "DOLLAR": "$",
+                            "STYLE": cfg.style,
+                            "TAG_PREFIX": cfg.tag_prefix,
+                            "PARENTDIR_PREFIX": cfg.parentdir_prefix,
+                            "VERSIONFILE_SOURCE": cfg.versionfile_source,
+                        }
+                    )
+
         cmds["build_exe"] = cmd_build_exe
         del cmds["build_py"]
 
-    if 'py2exe' in sys.modules:  # py2exe enabled?
+    if "py2exe" in sys.modules:  # py2exe enabled?
         try:
             from py2exe.distutils_buildexe import py2exe as _py2exe  # py3
         except ImportError:
@@ -1612,13 +1661,17 @@ def get_cmdclass():
                 os.unlink(target_versionfile)
                 with open(cfg.versionfile_source, "w") as f:
                     LONG = LONG_VERSION_PY[cfg.VCS]
-                    f.write(LONG %
-                            {"DOLLAR": "$",
-                             "STYLE": cfg.style,
-                             "TAG_PREFIX": cfg.tag_prefix,
-                             "PARENTDIR_PREFIX": cfg.parentdir_prefix,
-                             "VERSIONFILE_SOURCE": cfg.versionfile_source,
-                             })
+                    f.write(
+                        LONG
+                        % {
+                            "DOLLAR": "$",
+                            "STYLE": cfg.style,
+                            "TAG_PREFIX": cfg.tag_prefix,
+                            "PARENTDIR_PREFIX": cfg.parentdir_prefix,
+                            "VERSIONFILE_SOURCE": cfg.versionfile_source,
+                        }
+                    )
+
         cmds["py2exe"] = cmd_py2exe
 
     # we override different "sdist" commands for both environments
@@ -1645,8 +1698,10 @@ def get_cmdclass():
             # updated value
             target_versionfile = os.path.join(base_dir, cfg.versionfile_source)
             print("UPDATING %s" % target_versionfile)
-            write_to_version_file(target_versionfile,
-                                  self._versioneer_generated_versions)
+            write_to_version_file(
+                target_versionfile, self._versioneer_generated_versions
+            )
+
     cmds["sdist"] = cmd_sdist
 
     return cmds
@@ -1701,11 +1756,13 @@ def do_setup():
     root = get_root()
     try:
         cfg = get_config_from_root(root)
-    except (EnvironmentError, configparser.NoSectionError,
-            configparser.NoOptionError) as e:
+    except (
+        EnvironmentError,
+        configparser.NoSectionError,
+        configparser.NoOptionError,
+    ) as e:
         if isinstance(e, (EnvironmentError, configparser.NoSectionError)):
-            print("Adding sample versioneer config to setup.cfg",
-                  file=sys.stderr)
+            print("Adding sample versioneer config to setup.cfg", file=sys.stderr)
             with open(os.path.join(root, "setup.cfg"), "a") as f:
                 f.write(SAMPLE_CONFIG)
         print(CONFIG_ERROR, file=sys.stderr)
@@ -1714,15 +1771,18 @@ def do_setup():
     print(" creating %s" % cfg.versionfile_source)
     with open(cfg.versionfile_source, "w") as f:
         LONG = LONG_VERSION_PY[cfg.VCS]
-        f.write(LONG % {"DOLLAR": "$",
-                        "STYLE": cfg.style,
-                        "TAG_PREFIX": cfg.tag_prefix,
-                        "PARENTDIR_PREFIX": cfg.parentdir_prefix,
-                        "VERSIONFILE_SOURCE": cfg.versionfile_source,
-                        })
-
-    ipy = os.path.join(os.path.dirname(cfg.versionfile_source),
-                       "__init__.py")
+        f.write(
+            LONG
+            % {
+                "DOLLAR": "$",
+                "STYLE": cfg.style,
+                "TAG_PREFIX": cfg.tag_prefix,
+                "PARENTDIR_PREFIX": cfg.parentdir_prefix,
+                "VERSIONFILE_SOURCE": cfg.versionfile_source,
+            }
+        )
+
+    ipy = os.path.join(os.path.dirname(cfg.versionfile_source), "__init__.py")
     if os.path.exists(ipy):
         try:
             with open(ipy, "r") as f:
@@ -1764,8 +1824,10 @@ def do_setup():
     else:
         print(" 'versioneer.py' already in MANIFEST.in")
     if cfg.versionfile_source not in simple_includes:
-        print(" appending versionfile_source ('%s') to MANIFEST.in" %
-              cfg.versionfile_source)
+        print(
+            " appending versionfile_source ('%s') to MANIFEST.in"
+            % cfg.versionfile_source
+        )
         with open(manifest_in, "a") as f:
             f.write("include %s\n" % cfg.versionfile_source)
     else:


Test patch:
diff --git a/conftest.py b/conftest.py
index 1b92c19..0d3aa9c 100644
--- a/conftest.py
+++ b/conftest.py
@@ -1,12 +1,15 @@
 # content of conftest.py
 
 # Make loop fixture available in all tests
-from distributed.utils_test import loop  # noqa: F401
+from distributed.utils_test import loop, loop_in_thread  # noqa: F401
 
 import pytest
 
+import dask_jobqueue.config
 import dask_jobqueue.lsf
 import dask
+import distributed.utils_test
+import copy
 
 from dask_jobqueue import (
     PBSCluster,
@@ -94,6 +97,28 @@ all_envs = {
 }
 
 
+# Overriding cleanup method from distributed that has been added to the loop
+# fixture, because it just wipe the Main Loop in our tests, and dask-jobqueue is
+# not ready for this.
+# FIXME
+@pytest.fixture
+def cleanup():
+    dask_jobqueue.config.reconfigure()
+    yield
+
+
+# Overriding distributed.utils_test.reset_config()  method because it reset the
+# config from ou tests.
+# FIXME
+def reset_config():
+    dask.config.config.clear()
+    dask.config.config.update(copy.deepcopy(distributed.utils_test.original_config))
+    dask_jobqueue.config.reconfigure()
+
+
+distributed.utils_test.reset_config = reset_config
+
+
 @pytest.fixture(
     params=[pytest.param(v, marks=[pytest.mark.env(k)]) for (k, v) in all_envs.items()]
 )
diff --git a/dask_jobqueue/tests/test_htcondor.py b/dask_jobqueue/tests/test_htcondor.py
index 99a5573..f847507 100644
--- a/dask_jobqueue/tests/test_htcondor.py
+++ b/dask_jobqueue/tests/test_htcondor.py
@@ -11,7 +11,7 @@ from dask.utils import format_bytes, parse_bytes
 from dask_jobqueue import HTCondorCluster
 from dask_jobqueue.core import Job
 
-QUEUE_WAIT = 30  # seconds
+from . import QUEUE_WAIT
 
 
 def test_header():
@@ -27,7 +27,12 @@ def test_job_script():
         processes=2,
         memory="100MB",
         disk="100MB",
-        env_extra=['export LANG="en_US.utf8"', 'export LC_ALL="en_US.utf8"'],
+        env_extra=[
+            'export LANG="en_US.utf8"',
+            'export LC_ALL="en_US.utf8"',
+            "cd /some/path/",
+            "source venv/bin/activate",
+        ],
         job_extra={"+Extra": "True"},
         submit_command_extra=["-verbose"],
         cancel_command_extra=["-forcex"],
@@ -40,9 +45,10 @@ def test_job_script():
         assert "MY.DaskWorkerDisk = 100000000" in job_script
         assert "MY.DaskWorkerMemory = 100000000" in job_script
         assert 'MY.JobId = "$(ClusterId).$(ProcId)"' in job_script
-        assert "LANG=en_US.utf8" in job_script
-        assert "LC_ALL=en_US.utf8" in job_script
-        assert "export" not in job_script
+        assert 'export LANG=""en_US.utf8""' in job_script
+        assert 'export LC_ALL=""en_US.utf8""' in job_script
+        assert "cd /some/path/" in job_script
+        assert "source venv/bin/activate" in job_script
         assert "+Extra = True" in job_script
         assert re.search(
             r"condor_submit\s.*-verbose", cluster._dummy_job.submit_command
@@ -56,25 +62,23 @@ def test_job_script():
         formatted_bytes = format_bytes(parse_bytes("50MB")).replace(" ", "")
         assert f"--memory-limit {formatted_bytes}" in job_script
         assert "--nthreads 2" in job_script
-        assert "--nprocs 2" in job_script
+        assert "--nworkers 2" in job_script
 
 
 @pytest.mark.env("htcondor")
 def test_basic(loop):
-    with HTCondorCluster(cores=1, memory="100MB", disk="100MB", loop=loop) as cluster:
+    with HTCondorCluster(cores=1, memory="100MiB", disk="100MiB", loop=loop) as cluster:
         with Client(cluster) as client:
-
             cluster.scale(2)
 
-            start = time()
-            client.wait_for_workers(2)
+            client.wait_for_workers(2, timeout=QUEUE_WAIT)
 
             future = client.submit(lambda x: x + 1, 10)
             assert future.result(QUEUE_WAIT) == 11
 
             workers = list(client.scheduler_info()["workers"].values())
             w = workers[0]
-            assert w["memory_limit"] == 1e8
+            assert w["memory_limit"] == 100 * 1024**2
             assert w["nthreads"] == 1
 
             cluster.scale(0)
@@ -98,7 +102,7 @@ def test_extra_args_broken_cancel(loop):
 
             cluster.scale(2)
 
-            client.wait_for_workers(2)
+            client.wait_for_workers(2, timeout=QUEUE_WAIT)
             workers = Job._call(["condor_q", "-af", "jobpid"]).strip()
             assert workers, "we got dask workers"
 
@@ -114,6 +118,9 @@ def test_extra_args_broken_cancel(loop):
                 if time() > start + QUEUE_WAIT // 3:
                     return
 
+            # Remove running job as it prevents other jobs execution in subsequent tests
+            Job._call(["condor_rm", "-all"]).strip()
+
 
 def test_config_name_htcondor_takes_custom_config():
     conf = {
diff --git a/dask_jobqueue/tests/test_job.py b/dask_jobqueue/tests/test_job.py
index fa94feb..5458b78 100644
--- a/dask_jobqueue/tests/test_job.py
+++ b/dask_jobqueue/tests/test_job.py
@@ -5,6 +5,7 @@ from dask_jobqueue.local import LocalCluster
 from dask_jobqueue.pbs import PBSJob
 from dask_jobqueue.core import JobQueueCluster
 from dask.distributed import Scheduler, Client
+from dask_jobqueue.tests import QUEUE_WAIT
 from distributed.core import Status
 
 import pytest
@@ -23,7 +24,7 @@ async def test_job(EnvSpecificCluster):
         job = job_cls(scheduler=s.address, name="foo", cores=1, memory="1GB")
         job = await job
         async with Client(s.address, asynchronous=True) as client:
-            await client.wait_for_workers(1)
+            await client.wait_for_workers(1, timeout=QUEUE_WAIT)
             assert list(s.workers.values())[0].name == "foo"
 
         await job.close()
@@ -47,7 +48,7 @@ async def test_cluster(EnvSpecificCluster):
             assert len(cluster.workers) == 2
             assert all(isinstance(w, job_cls) for w in cluster.workers.values())
             assert all(w.status == Status.running for w in cluster.workers.values())
-            await client.wait_for_workers(2)
+            await client.wait_for_workers(2, timeout=QUEUE_WAIT)
 
             cluster.scale(1)
             start = time()
@@ -64,7 +65,7 @@ async def test_adapt(EnvSpecificCluster):
         1, cores=1, memory="1GB", job_cls=job_cls, asynchronous=True, name="foo"
     ) as cluster:
         async with Client(cluster, asynchronous=True) as client:
-            await client.wait_for_workers(1)
+            await client.wait_for_workers(1, timeout=QUEUE_WAIT)
             cluster.adapt(minimum=0, maximum=4, interval="10ms")
 
             start = time()
@@ -75,7 +76,7 @@ async def test_adapt(EnvSpecificCluster):
             assert not cluster.workers
 
             future = client.submit(lambda: 0)
-            await client.wait_for_workers(1)
+            await client.wait_for_workers(1, timeout=QUEUE_WAIT)
 
             del future
 
@@ -113,7 +114,7 @@ def test_header_lines_skip():
 
 
 @pytest.mark.asyncio
-async def test_nprocs_scale():
+async def test_nworkers_scale():
     async with LocalCluster(
         cores=2, memory="4GB", processes=2, asynchronous=True
     ) as cluster:
@@ -121,7 +122,7 @@ async def test_nprocs_scale():
         async with Client(cluster, asynchronous=True) as client:
             cluster.scale(cores=2)
             await cluster
-            await client.wait_for_workers(2)
+            await client.wait_for_workers(2, timeout=QUEUE_WAIT)
             assert len(cluster.workers) == 1  # two workers, one job
             assert len(s.workers) == 2
             assert cluster.plan == {ws.name for ws in s.workers.values()}
diff --git a/dask_jobqueue/tests/test_jobqueue_core.py b/dask_jobqueue/tests/test_jobqueue_core.py
index cd84039..0954eab 100644
--- a/dask_jobqueue/tests/test_jobqueue_core.py
+++ b/dask_jobqueue/tests/test_jobqueue_core.py
@@ -249,11 +249,11 @@ def test_cluster_without_job_cls():
 
 def test_default_number_of_worker_processes(Cluster):
     with Cluster(cores=4, memory="4GB") as cluster:
-        assert " --nprocs 4" in cluster.job_script()
+        assert " --nworkers 4" in cluster.job_script()
         assert " --nthreads 1" in cluster.job_script()
 
     with Cluster(cores=6, memory="4GB") as cluster:
-        assert " --nprocs 3" in cluster.job_script()
+        assert " --nworkers 3" in cluster.job_script()
         assert " --nthreads 2" in cluster.job_script()
 
 
diff --git a/dask_jobqueue/tests/test_lsf.py b/dask_jobqueue/tests/test_lsf.py
index 6e0489d..20b47c8 100644
--- a/dask_jobqueue/tests/test_lsf.py
+++ b/dask_jobqueue/tests/test_lsf.py
@@ -101,7 +101,9 @@ def test_job_script():
             in job_script
         )
         formatted_bytes = format_bytes(parse_bytes("7GB")).replace(" ", "")
-        assert f"--nthreads 2 --nprocs 4 --memory-limit {formatted_bytes}" in job_script
+        assert (
+            f"--nthreads 2 --nworkers 4 --memory-limit {formatted_bytes}" in job_script
+        )
 
     with LSFCluster(
         queue="general",
@@ -128,7 +130,9 @@ def test_job_script():
             in job_script
         )
         formatted_bytes = format_bytes(parse_bytes("7GB")).replace(" ", "")
-        assert f"--nthreads 2 --nprocs 4 --memory-limit {formatted_bytes}" in job_script
+        assert (
+            f"--nthreads 2 --nworkers 4 --memory-limit {formatted_bytes}" in job_script
+        )
 
     with LSFCluster(
         walltime="1:00",
diff --git a/dask_jobqueue/tests/test_oar.py b/dask_jobqueue/tests/test_oar.py
index 8e4c409..0e156e8 100644
--- a/dask_jobqueue/tests/test_oar.py
+++ b/dask_jobqueue/tests/test_oar.py
@@ -55,7 +55,9 @@ def test_job_script():
             in job_script
         )
         formatted_bytes = format_bytes(parse_bytes("7GB")).replace(" ", "")
-        assert f"--nthreads 2 --nprocs 4 --memory-limit {formatted_bytes}" in job_script
+        assert (
+            f"--nthreads 2 --nworkers 4 --memory-limit {formatted_bytes}" in job_script
+        )
 
     with OARCluster(
         walltime="00:02:00",
@@ -86,7 +88,9 @@ def test_job_script():
             in job_script
         )
         formatted_bytes = format_bytes(parse_bytes("7GB")).replace(" ", "")
-        assert f"--nthreads 2 --nprocs 4 --memory-limit {formatted_bytes}" in job_script
+        assert (
+            f"--nthreads 2 --nworkers 4 --memory-limit {formatted_bytes}" in job_script
+        )
 
 
 def test_config_name_oar_takes_custom_config():
diff --git a/dask_jobqueue/tests/test_pbs.py b/dask_jobqueue/tests/test_pbs.py
index 887b69f..cb6973e 100644
--- a/dask_jobqueue/tests/test_pbs.py
+++ b/dask_jobqueue/tests/test_pbs.py
@@ -76,7 +76,9 @@ def test_job_script(Cluster):
             in job_script
         )
         formatted_bytes = format_bytes(parse_bytes("7GB")).replace(" ", "")
-        assert f"--nthreads 2 --nprocs 4 --memory-limit {formatted_bytes}" in job_script
+        assert (
+            f"--nthreads 2 --nworkers 4 --memory-limit {formatted_bytes}" in job_script
+        )
 
     with Cluster(
         queue="regular",
@@ -100,7 +102,9 @@ def test_job_script(Cluster):
             in job_script
         )
         formatted_bytes = format_bytes(parse_bytes("7GB")).replace(" ", "")
-        assert f"--nthreads 2 --nprocs 4 --memory-limit {formatted_bytes}" in job_script
+        assert (
+            f"--nthreads 2 --nworkers 4 --memory-limit {formatted_bytes}" in job_script
+        )
 
 
 @pytest.mark.env("pbs")
@@ -109,7 +113,7 @@ def test_basic(loop):
         walltime="00:02:00",
         processes=1,
         cores=2,
-        memory="2GB",
+        memory="2GiB",
         local_directory="/tmp",
         job_extra=["-V"],
         loop=loop,
@@ -117,7 +121,7 @@ def test_basic(loop):
         with Client(cluster) as client:
 
             cluster.scale(2)
-            client.wait_for_workers(2)
+            client.wait_for_workers(2, timeout=QUEUE_WAIT)
 
             future = client.submit(lambda x: x + 1, 10)
             assert future.result(QUEUE_WAIT) == 11
@@ -125,7 +129,7 @@ def test_basic(loop):
 
             workers = list(client.scheduler_info()["workers"].values())
             w = workers[0]
-            assert w["memory_limit"] == 2e9
+            assert w["memory_limit"] == 2 * 1024**3
             assert w["nthreads"] == 2
 
             cluster.scale(0)
@@ -144,7 +148,7 @@ def test_scale_cores_memory(loop):
         walltime="00:02:00",
         processes=1,
         cores=2,
-        memory="2GB",
+        memory="2GiB",
         local_directory="/tmp",
         job_extra=["-V"],
         loop=loop,
@@ -152,7 +156,7 @@ def test_scale_cores_memory(loop):
         with Client(cluster) as client:
 
             cluster.scale(cores=2)
-            client.wait_for_workers(1)
+            client.wait_for_workers(1, timeout=QUEUE_WAIT)
 
             future = client.submit(lambda x: x + 1, 10)
             assert future.result(QUEUE_WAIT) == 11
@@ -160,7 +164,7 @@ def test_scale_cores_memory(loop):
 
             workers = list(client.scheduler_info()["workers"].values())
             w = workers[0]
-            assert w["memory_limit"] == 2e9
+            assert w["memory_limit"] == 2 * 1024**3
             assert w["nthreads"] == 2
 
             cluster.scale(memory="0GB")
@@ -241,7 +245,7 @@ def test_adaptive_grouped(loop):
     ) as cluster:
         cluster.adapt(minimum=1)  # at least 1 worker
         with Client(cluster) as client:
-            client.wait_for_workers(1)
+            client.wait_for_workers(1, timeout=QUEUE_WAIT)
 
             future = client.submit(lambda x: x + 1, 10)
             assert future.result(QUEUE_WAIT) == 11
@@ -289,7 +293,7 @@ def test_scale_grouped(loop):
         walltime="00:02:00",
         processes=2,
         cores=2,
-        memory="2GB",
+        memory="2GiB",
         local_directory="/tmp",
         job_extra=["-V"],
         loop=loop,
@@ -310,7 +314,7 @@ def test_scale_grouped(loop):
 
             workers = list(client.scheduler_info()["workers"].values())
             w = workers[0]
-            assert w["memory_limit"] == 1e9
+            assert w["memory_limit"] == 1024**3
             assert w["nthreads"] == 1
             assert len(workers) == 4
 
diff --git a/dask_jobqueue/tests/test_sge.py b/dask_jobqueue/tests/test_sge.py
index f743f90..d24502c 100644
--- a/dask_jobqueue/tests/test_sge.py
+++ b/dask_jobqueue/tests/test_sge.py
@@ -13,7 +13,7 @@ from . import QUEUE_WAIT
 @pytest.mark.env("sge")
 def test_basic(loop):
     with SGECluster(
-        walltime="00:02:00", cores=8, processes=4, memory="2GB", loop=loop
+        walltime="00:02:00", cores=8, processes=4, memory="2GiB", loop=loop
     ) as cluster:
         with Client(cluster, loop=loop) as client:
 
@@ -30,7 +30,7 @@ def test_basic(loop):
 
             workers = list(client.scheduler_info()["workers"].values())
             w = workers[0]
-            assert w["memory_limit"] == 2e9 / 4
+            assert w["memory_limit"] == 2 * 1024**3 / 4
             assert w["nthreads"] == 2
 
             cluster.scale(0)
@@ -88,7 +88,7 @@ def test_job_script(tmpdir):
         formatted_bytes = format_bytes(parse_bytes("6GB")).replace(" ", "")
 
         for each in [
-            "--nprocs 2",
+            "--nworkers 2",
             "--nthreads 3",
             f"--memory-limit {formatted_bytes}",
             "-q my-queue",
diff --git a/dask_jobqueue/tests/test_slurm.py b/dask_jobqueue/tests/test_slurm.py
index 3218cac..74fca2b 100644
--- a/dask_jobqueue/tests/test_slurm.py
+++ b/dask_jobqueue/tests/test_slurm.py
@@ -77,7 +77,9 @@ def test_job_script():
             in job_script
         )
         formatted_bytes = format_bytes(parse_bytes("7GB")).replace(" ", "")
-        assert f"--nthreads 2 --nprocs 4 --memory-limit {formatted_bytes}" in job_script
+        assert (
+            f"--nthreads 2 --nworkers 4 --memory-limit {formatted_bytes}" in job_script
+        )
 
     with SLURMCluster(
         walltime="00:02:00",
@@ -109,7 +111,9 @@ def test_job_script():
             in job_script
         )
         formatted_bytes = format_bytes(parse_bytes("7GB")).replace(" ", "")
-        assert f"--nthreads 2 --nprocs 4 --memory-limit {formatted_bytes}" in job_script
+        assert (
+            f"--nthreads 2 --nworkers 4 --memory-limit {formatted_bytes}" in job_script
+        )
 
 
 @pytest.mark.env("slurm")
@@ -118,7 +122,7 @@ def test_basic(loop):
         walltime="00:02:00",
         cores=2,
         processes=1,
-        memory="2GB",
+        memory="2GiB",
         # job_extra=["-D /"],
         loop=loop,
     ) as cluster:
@@ -127,14 +131,14 @@ def test_basic(loop):
             cluster.scale(2)
 
             start = time()
-            client.wait_for_workers(2)
+            client.wait_for_workers(2, timeout=QUEUE_WAIT)
 
             future = client.submit(lambda x: x + 1, 10)
             assert future.result(QUEUE_WAIT) == 11
 
             workers = list(client.scheduler_info()["workers"].values())
             w = workers[0]
-            assert w["memory_limit"] == 2e9
+            assert w["memory_limit"] == 2 * 1024**3
             assert w["nthreads"] == 2
 
             cluster.scale(0)
@@ -160,7 +164,7 @@ def test_adaptive(loop):
             future = client.submit(lambda x: x + 1, 10)
 
             start = time()
-            client.wait_for_workers(1)
+            client.wait_for_workers(1, timeout=QUEUE_WAIT)
 
             assert future.result(QUEUE_WAIT) == 11
 
@@ -214,7 +218,7 @@ def test_different_interfaces_on_scheduler_and_workers(loop):
         with Client(cluster) as client:
             future = client.submit(lambda x: x + 1, 10)
 
-            client.wait_for_workers(1)
+            client.wait_for_workers(1, timeout=QUEUE_WAIT)
 
             assert future.result(QUEUE_WAIT) == 11
 
@@ -234,7 +238,7 @@ def test_worker_name_uses_cluster_name(loop):
         with Client(cluster) as client:
             cluster.scale(jobs=2)
             print(cluster.job_script())
-            client.wait_for_workers(2)
+            client.wait_for_workers(2, timeout=QUEUE_WAIT)
             worker_names = [
                 w["id"] for w in client.scheduler_info()["workers"].values()
             ]