Add minimum dependency testing to CI

.github/workflows/conda.yml

@@ -13,7 +13,7 @@ on:
 # When this workflow is queued, automatically cancel any previous running
 # or pending jobs from the same branch
 concurrency:
-  group: conda-${{ github.head_ref }}
+  group: conda-${{ github.ref }}
   cancel-in-progress: true
 
 # Required shell entrypoint to have properly activated conda environments

.github/workflows/tests.yaml

@@ -9,7 +9,7 @@ on:
 # When this workflow is queued, automatically cancel any previous running
 # or pending jobs from the same branch
 concurrency:
-  group: ${{ github.ref }}
+  group: tests-${{ github.ref }}
   cancel-in-progress: true
 
 jobs:
@@ -23,20 +23,24 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-latest, windows-latest, macos-latest]
-        python-version: ["3.8", "3.9", "3.10"]
+        environment: ["3.8", "3.9", "3.10", mindeps]
         queuing: [queue]
         # Cherry-pick test modules to split the overall runtime roughly in half
         partition: [ci1, not ci1]
         exclude:
           - os: macos-latest
-            python-version: 3.9
+            environment: "3.9"
+          - os: macos-latest
+            environment: mindeps
+          - os: windows-latest
+            environment: mindeps
         include:
           - os: ubuntu-latest
-            python-version: 3.9
+            environment: "3.9"
             queuing: no_queue
             partition: "ci1"
           - os: ubuntu-latest
-            python-version: 3.9
+            environment: "3.9"
             queuing: no_queue
             partition: "not ci1"
 
@@ -49,15 +53,15 @@ jobs:
         # run: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20]
 
     env:
-      CONDA_FILE: continuous_integration/environment-${{ matrix.python-version }}.yaml
+      CONDA_FILE: continuous_integration/environment-${{ matrix.environment }}.yaml
 
     steps:
       - name: Set $TEST_ID
         run: |
           export PARTITION_LABEL=$( echo "${{ matrix.partition }}" | sed "s/ //" )
-          export TEST_ID="${{ matrix.os }}-${{ matrix.python-version }}-${{ matrix.queuing }}-$PARTITION_LABEL"
+          export TEST_ID="${{ matrix.os }}-${{ matrix.environment }}-${{ matrix.queuing }}-$PARTITION_LABEL"
           # Switch to this version for stress-test:
-          # export TEST_ID="${{ matrix.os }}-${{ matrix.python-version }}-${{ matrix.queuing }}-$PARTITION_LABEL-${{ matrix.run }}"
+          # export TEST_ID="${{ matrix.os }}-${{ matrix.environment }}-${{ matrix.queuing }}-$PARTITION_LABEL-${{ matrix.run }}"
           echo "TEST_ID: $TEST_ID"
           echo "TEST_ID=$TEST_ID" >> $GITHUB_ENV
         shell: bash
@@ -99,7 +103,7 @@ jobs:
           path: ${{ env.CONDA }}/envs
           key: conda-${{ matrix.os }}-${{ steps.get-date.outputs.today }}-${{ hashFiles(env.CONDA_FILE) }}-${{ env.CACHE_NUMBER }}
         env:
-          # Increase this value to reset cache if continuous_integration/environment-${{ matrix.python-version }}.yaml has not changed
+          # Increase this value to reset cache if continuous_integration/environment-${{ matrix.environment }}.yaml has not changed
           CACHE_NUMBER: 0
         id: cache
 
@@ -148,6 +152,8 @@ jobs:
         run: echo "DASK_DISTRIBUTED__SCHEDULER__WORKER_SATURATION=inf" >> $GITHUB_ENV
 
       - name: Print host info
+        # host_info.py imports numpy, which isn't a direct dependency of distributed
+        if: matrix.environment != 'mindeps'
         shell: bash -l {0}
         run: |
           python continuous_integration/scripts/host_info.py

continuous_integration/environment-3.10.yaml

@@ -17,7 +17,7 @@ dependencies:
   - h5py
   - ipykernel
   - ipywidgets
-  - jinja2
+  - jinja2 >=2.10.3
   - locket >=1.0
   - msgpack-python
   - netcdf4

continuous_integration/environment-3.8.yaml

@@ -18,7 +18,7 @@ dependencies:
   - h5py
   - ipykernel
   - ipywidgets
-  - jinja2
+  - jinja2 >=2.10.3
   - locket >=1.0
   - msgpack-python
   - netcdf4

continuous_integration/environment-3.9.yaml

@@ -18,7 +18,7 @@ dependencies:
   - h5py
   - ipykernel
   - ipywidgets
-  - jinja2
+  - jinja2 >=2.10.3
   - locket >=1.0
   - lz4 >=0.23.1  # Only tested here
   - msgpack-python

continuous_integration/environment-mindeps.yaml

@@ -0,0 +1,32 @@
+name: dask-distributed
+channels:
+  - conda-forge
+  - defaults
+dependencies:
+  - python=3.8
+  - click=7.0
+  - cloudpickle=1.5.0
+  - cytoolz=0.10.1
+  - jinja2=2.10.3
+  - locket=1.0.0
+  - msgpack-python=0.6.2
+  - packaging=20.0
+  - psutil=5.6.3
+  - pyyaml=5.3.1
+  - sortedcontainers=2.0.5
+  - tblib=1.6.0
+  - toolz=0.10.0
+  - tornado=6.0.3
+  - urllib3=1.24.3
+  - zict=0.1.4
+  # Distributed depends on the latest version of Dask
+  - pip
+  - pip:
+    - git+https://github.com/dask/dask
+  # test dependencies
+  - pytest
+  - pytest-cov
+  - pytest-faulthandler
+  - pytest-repeat
+  - pytest-rerunfailures
+  - pytest-timeout

continuous_integration/recipes/dask/meta.yaml

@@ -30,7 +30,7 @@ requirements:
     - numpy >=1.18
     - pandas >=1.0
     - bokeh >=2.4.2,<3
-    - jinja2
+    - jinja2 >=2.10.3
 
   run_constrained:
     - openssl !=1.1.1e

continuous_integration/recipes/distributed/meta.yaml

@@ -31,20 +31,20 @@ requirements:
     - python >=3.8
     - click >=7.0
     - cloudpickle >=1.5.0
-    - cytoolz >=0.8.2
+    - cytoolz >=0.10.1
     - {{ pin_compatible('dask-core', max_pin='x.x.x.x') }}
-    - jinja2
+    - jinja2 >=2.10.3
     - locket >=1.0.0
-    - msgpack-python >=0.6.0
+    - msgpack-python >=0.6.2
     - packaging >=20.0
-    - psutil >=5.0
-    - pyyaml
-    - sortedcontainers !=2.0.0,!=2.0.1
+    - psutil >=5.6.3
+    - pyyaml >=5.3.1
+    - sortedcontainers >=2.0.5
     - tblib >=1.6.0
-    - toolz >=0.8.2
+    - toolz >=0.10.0
     - tornado >=6.0.3,<6.2
-    - urllib3
-    - zict >=0.1.3
+    - urllib3 >=1.24.3
+    - zict >=0.1.4
   run_constrained:
     - openssl !=1.1.1e
 

distributed/http/scheduler/tests/test_scheduler_http.py

@@ -4,11 +4,7 @@
 import json
 import re
 
-import aiohttp
 import pytest
-
-pytest.importorskip("bokeh")
-
 from tornado.escape import url_escape
 from tornado.httpclient import AsyncHTTPClient, HTTPClientError
 
@@ -81,6 +77,8 @@ async def test_worker_404(c, s):
 
 @gen_cluster(client=True, scheduler_kwargs={"http_prefix": "/foo", "dashboard": True})
 async def test_prefix(c, s, a, b):
+    pytest.importorskip("bokeh")
+
     http_client = AsyncHTTPClient()
     for suffix in ["foo/info/main/workers.html", "foo/json/index.html", "foo/system"]:
         response = await http_client.fetch(
@@ -274,6 +272,8 @@ async def test_task_page(c, s, a, b):
     },
 )
 async def test_allow_websocket_origin(c, s, a, b):
+    pytest.importorskip("bokeh")
+
     from tornado.httpclient import HTTPRequest
     from tornado.websocket import websocket_connect
 
@@ -314,6 +314,8 @@ def test_api_disabled_by_default():
     },
 )
 async def test_api(c, s, a, b):
+    aiohttp = pytest.importorskip("aiohttp")
+
     async with aiohttp.ClientSession() as session:
         async with session.get(
             "http://localhost:%d/api/v1" % s.http_server.port
@@ -332,6 +334,8 @@ async def test_api(c, s, a, b):
     },
 )
 async def test_retire_workers(c, s, a, b):
+    aiohttp = pytest.importorskip("aiohttp")
+
     async with aiohttp.ClientSession() as session:
         params = {"workers": [a.address, b.address]}
         async with session.post(
@@ -353,6 +357,8 @@ async def test_retire_workers(c, s, a, b):
     },
 )
 async def test_get_workers(c, s, a, b):
+    aiohttp = pytest.importorskip("aiohttp")
+
     async with aiohttp.ClientSession() as session:
         async with session.get(
             "http://localhost:%d/api/v1/get_workers" % s.http_server.port
@@ -373,6 +379,8 @@ async def test_get_workers(c, s, a, b):
     },
 )
 async def test_adaptive_target(c, s, a, b):
+    aiohttp = pytest.importorskip("aiohttp")
+
     async with aiohttp.ClientSession() as session:
         async with session.get(
             "http://localhost:%d/api/v1/adaptive_target" % s.http_server.port

distributed/protocol/tests/test_collection_cuda.py

@@ -3,9 +3,6 @@
 import pytest
 
 pytestmark = pytest.mark.gpu
-
-from dask.dataframe.utils import assert_eq
-
 from distributed.protocol import deserialize, serialize
 
 
@@ -46,6 +43,7 @@ def test_serialize_cupy(collection, y, y_serializer):
 def test_serialize_pandas_pandas(collection, df2, df2_serializer):
     cudf = pytest.importorskip("cudf")
     pd = pytest.importorskip("pandas")
+    dd = pytest.importorskip("dask.dataframe")
     df1 = cudf.DataFrame({"A": [1, 2, None], "B": [1.0, 2.0, None]})
     if df2 is not None:
         df2 = cudf.from_pandas(pd.DataFrame(df2))
@@ -63,8 +61,8 @@ def test_serialize_pandas_pandas(collection, df2, df2_serializer):
     assert sub_headers[1]["serializer"] == df2_serializer
     assert isinstance(t, collection)
 
-    assert_eq(t["df1"] if isinstance(t, dict) else t[0], df1)
+    dd.assert_eq(t["df1"] if isinstance(t, dict) else t[0], df1)
     if df2 is None:
         assert (t["df2"] if isinstance(t, dict) else t[1]) is None
     else:
-        assert_eq(t["df2"] if isinstance(t, dict) else t[1], df2)
+        dd.assert_eq(t["df2"] if isinstance(t, dict) else t[1], df2)

distributed/protocol/tests/test_highlevelgraph.py

@@ -4,18 +4,18 @@
 
 import pytest
 
+np = pytest.importorskip("numpy")
+pd = pytest.importorskip("pandas")
+
+from numpy.testing import assert_array_equal
+
 import dask
 import dask.array as da
 import dask.dataframe as dd
 
 from distributed.diagnostics import SchedulerPlugin
 from distributed.utils_test import gen_cluster
 
-np = pytest.importorskip("numpy")
-pd = pytest.importorskip("pandas")
-
-from numpy.testing import assert_array_equal
-
 
 @gen_cluster(client=True)
 async def test_combo_of_layer_types(c, s, a, b):

distributed/protocol/tests/test_protocol_utils.py

@@ -66,6 +66,7 @@ def test_readonly_buffer(self):
         assert result == base
 
     def test_catch_non_memoryview(self):
+        pytest.importorskip("numpy")
         with pytest.raises(TypeError, match="Expected memoryview"):
             merge_memoryviews([b"1234", memoryview(b"4567")])
 

distributed/shuffle/tests/test_shuffle.py

@@ -8,9 +8,10 @@
 from collections import defaultdict
 from typing import Any
 
-import pandas as pd
 import pytest
 
+pd = pytest.importorskip("pandas")
+
 import dask
 import dask.dataframe as dd
 from dask.distributed import Worker
@@ -30,8 +31,6 @@
 )
 from distributed.utils_test import gen_cluster, gen_test
 
-pa = pytest.importorskip("pyarrow")
-
 
 def clean_worker(worker):
     """Assert that the worker has no shuffle state"""
@@ -191,6 +190,8 @@ def test_processing_chain():
     In practice this takes place on many different workers.
     Here we verify its accuracy in a single threaded situation.
     """
+    pa = pytest.importorskip("pyarrow")
+
     workers = ["a", "b", "c"]
     npartitions = 5
     df = pd.DataFrame({"x": range(100), "y": range(100)})
@@ -552,6 +553,8 @@ async def test_basic_lowlevel_shuffle(
     npartitions,
     barrier_first_worker,
 ):
+    pa = pytest.importorskip("pyarrow")
+
     dfs = []
     rows_per_df = 10
     for ix in range(n_input_partitions):
@@ -626,6 +629,8 @@ def _done():
 
 @gen_test()
 async def test_error_offload(tmpdir, loop_in_thread):
+    pa = pytest.importorskip("pyarrow")
+
     dfs = []
     rows_per_df = 10
     n_input_partitions = 2

distributed/tests/test_client.py

@@ -29,6 +29,7 @@
 from typing import Any
 from unittest import mock
 
+import msgpack
 import psutil
 import pytest
 import yaml
@@ -7508,11 +7509,23 @@ def _verify_cluster_dump(
     url = str(url) + (".msgpack.gz" if format == "msgpack" else ".yaml")
     state = load_cluster_dump(url)
 
+    # msgpack < 1.0.0 returns state dict with bytes key/values
+    MSGPACK_LT_1_0_0 = format == "msgpack" and msgpack.version < (1, 0, 0)
+
+    scheduler_key = b"scheduler" if MSGPACK_LT_1_0_0 else "scheduler"
+    workers_key = b"workers" if MSGPACK_LT_1_0_0 else "workers"
+    versions_key = b"versions" if MSGPACK_LT_1_0_0 else "versions"
+    state_addresses = (
+        {k.decode("utf-8") for k in state[workers_key].keys()}
+        if MSGPACK_LT_1_0_0
+        else state[workers_key].keys()
+    )
+
     assert isinstance(state, dict)
-    assert "scheduler" in state
-    assert "workers" in state
-    assert "versions" in state
-    assert state["workers"].keys() == addresses
+    assert scheduler_key in state
+    assert workers_key in state
+    assert versions_key in state
+    assert state_addresses == addresses
     return state