Repository: pangeo-forge/pangeo-forge-recipes (language: python)

Issue:
Tests slowed by > 2x following merge of #167
[Checks](https://github.com/pangeo-forge/pangeo-forge-recipes/actions/runs/1183164556) for #174 took < 15 mins:
<img width="773" alt="Screen Shot 2021-09-02 at 1 49 33 PM" src="https://user-images.githubusercontent.com/62192187/131914215-c2baf038-a2dc-4928-932b-830877ebaca8.png">

[Checks](https://github.com/pangeo-forge/pangeo-forge-recipes/actions/runs/1192494723) for #167 took > 30 mins:
<img width="798" alt="Screen Shot 2021-09-02 at 1 53 01 PM" src="https://user-images.githubusercontent.com/62192187/131914700-c5aa8a3e-a569-409d-8b34-11e5f9e8c7e6.png">

Output of [`pytest --durations=10 --durations-min=1.0`](https://docs.pytest.org/en/6.2.x/usage.html#profiling-test-execution-duration):
```
========================================= slowest 10 durations ==========================================
5.37s teardown tests/test_utils.py::test_calc_subsets
2.74s call     tests/test_recipes.py::test_chunks[prefect-netCDFtoZarr_recipe-target_chunks6-False-error_expectation6-1-subset_inputs1-netcdf_local_file_pattern_sequential_multi_variable-D]
2.67s setup    tests/test_recipes.py::test_recipe_caching_copying[manual-netCDFtoZarr_http_recipe-True-True-netcdf_http_paths0-netcdf_local_paths_sequential-D]
2.33s call     tests/test_recipes.py::test_chunks[prefect-netCDFtoZarr_recipe-target_chunks6-False-error_expectation6-1-subset_inputs0-netcdf_local_file_pattern_sequential_multi_variable-D]
2.28s setup    tests/test_fixtures.py::test_fixture_http_files[netcdf_http_paths1-netcdf_local_paths_sequential_multi_variable-D]
2.28s setup    tests/test_fixtures.py::test_fixture_http_files[netcdf_http_paths2-netcdf_local_paths_sequential_multi_variable-D]
2.25s call     tests/test_recipes.py::test_chunks[prefect-netCDFtoZarr_recipe-target_chunks5-False-error_expectation5-1-subset_inputs0-netcdf_local_file_pattern_sequential_multi_variable-D]
2.25s setup    tests/test_fixtures.py::test_fixture_http_files[netcdf_http_paths0-netcdf_local_paths_sequential_multi_variable-D]
2.23s call     tests/test_recipes.py::test_chunks[prefect-netCDFtoZarr_recipe-target_chunks5-False-error_expectation5-1-subset_inputs1-netcdf_local_file_pattern_sequential_multi_variable-D]
2.17s setup    tests/test_fixtures.py::test_fixture_http_files[netcdf_http_paths2-netcdf_local_paths_sequential-D]
```
Setup for `tests/test_fixtures.py::test_fixture_http_files` is [obviously time-consuming](https://github.com/pangeo-forge/pangeo-forge-recipes/blob/e748febdb9a5111d43a9ba7cf837812614894f27/tests/conftest.py#L227), but there are only 8 total tests run from `test_fixtures.py` and they cumulatively take < 30 seconds to run:
```
$ pytest test_fixtures.py
Results (28.12s):
       8 passed
       8 skipped
```
The slow `tests/test_recipes.py::test_chunks` calls are of greater interest, because with current fixturing, that test gets called *a lot*, as shown by the output of [`python parametrization_counter.py details plot`](https://github.com/pangeo-forge/pangeo-forge-recipes/pull/198/files#diff-0e7d92f16205081086bc83ea6d0b8d0611e07e7b201061d5f76b7daf42962d7eR10):
![test_counts](https://user-images.githubusercontent.com/62192187/131917144-b32ffec9-d5d9-4c88-ad13-51d2061cb9c7.png)



Discussion / hints:

Fix patch:
diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml
index 13cefe12..d564fae4 100644
--- a/.github/workflows/main.yaml
+++ b/.github/workflows/main.yaml
@@ -59,7 +59,8 @@ jobs:
           py.test tests -v --redirect-dask-worker-logs-to-stdout=DEBUG \
             --cov=pangeo_forge_recipes --cov-config .coveragerc \
             --cov-report term-missing \
-            --cov-report xml
+            --cov-report xml \
+            --durations=10 --durations-min=1.0
       - name: Codecov
         uses: codecov/codecov-action@v2.0.2
         with:


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index ed50440d..d3af0b8e 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,3 +1,18 @@
+"""
+Objects in this module belong to the following groups, delimited by inline comments:
+    - Helper functions:
+        Functions used in the creation of fixtures
+    - Dataset + path fixtures:
+        Create a dataset or a path to dataset(s)
+    - FilePattern fixtures:
+        Create `pangeo_forge_recipes.patterns.FilePattern` instances
+    - Storage fixtures:
+        Create temporary storage locations for caching, writing, etc.
+    - Execution fixtures:
+        Create infrastructure or define steps for executing recipes
+Note:
+   Recipe fixtures are defined in their respective test modules, e.g. `test_recipes.py`
+"""
 import logging
 import os
 import socket
@@ -29,6 +44,8 @@
 )
 from pangeo_forge_recipes.storage import CacheFSSpecTarget, FSSpecTarget, MetadataTarget
 
+# Helper functions --------------------------------------------------------------------------------
+
 
 # to use this feature, e.g.
 # $ pytest --redirect-dask-worker-logs-to-stdout=DEBUG
@@ -38,45 +55,6 @@ def pytest_addoption(parser):
     )
 
 
-def get_open_port():
-    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-    s.bind(("", 0))
-    s.listen(1)
-    port = str(s.getsockname()[1])
-    s.close()
-    return port
-
-
-@pytest.fixture(scope="session")
-def daily_xarray_dataset():
-    """Return a synthetic random xarray dataset."""
-    np.random.seed(1)
-    # TODO: change nt to 11 in order to catch the edge case where
-    # items_per_input does not evenly divide the length of the sequence dimension
-    nt, ny, nx = 10, 18, 36
-    time = pd.date_range(start="2010-01-01", periods=nt, freq="D")
-    lon = (np.arange(nx) + 0.5) * 360 / nx
-    lon_attrs = {"units": "degrees_east", "long_name": "longitude"}
-    lat = (np.arange(ny) + 0.5) * 180 / ny
-    lat_attrs = {"units": "degrees_north", "long_name": "latitude"}
-    foo = np.random.rand(nt, ny, nx)
-    foo_attrs = {"long_name": "Fantastic Foo"}
-    # make sure things work with heterogenous data types
-    bar = np.random.randint(0, 10, size=(nt, ny, nx))
-    bar_attrs = {"long_name": "Beautiful Bar"}
-    dims = ("time", "lat", "lon")
-    ds = xr.Dataset(
-        {"bar": (dims, bar, bar_attrs), "foo": (dims, foo, foo_attrs)},
-        coords={
-            "time": ("time", time),
-            "lat": ("lat", lat, lat_attrs),
-            "lon": ("lon", lon, lon_attrs),
-        },
-        attrs={"conventions": "CF 1.6"},
-    )
-    return ds
-
-
 def split_up_files_by_day(ds, day_param):
     gb = ds.resample(time=day_param)
     _, datasets = zip(*gb)
@@ -103,7 +81,7 @@ def make_file_pattern(path_fixture):
     Parameters
     ----------
     path_fixture : callable
-        One of `netcdf_local_paths` or `netcdf_http_paths`
+        `netcdf_local_paths`, `netcdf_http_paths`, or similar
     """
     paths, items_per_file, fnames_by_variable, path_format, kwargs = path_fixture
 
@@ -127,11 +105,6 @@ def format_function(variable, time):
     return file_pattern
 
 
-@pytest.fixture(scope="session", params=["D", "2D"])
-def items_per_file(request):
-    return request.param
-
-
 def make_netcdf_local_paths(daily_xarray_dataset, tmpdir_factory, items_per_file, file_splitter):
     tmp_path = tmpdir_factory.mktemp("netcdf_data")
     file_splitter_tuple = file_splitter(daily_xarray_dataset.copy(), items_per_file)
@@ -149,114 +122,213 @@ def make_netcdf_local_paths(daily_xarray_dataset, tmpdir_factory, items_per_file
     return full_paths, items_per_file, fnames_by_variable, path_format, kwargs
 
 
+def get_open_port():
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.bind(("", 0))
+    s.listen(1)
+    port = str(s.getsockname()[1])
+    s.close()
+    return port
+
+
+def start_http_server(paths, request, username=None, password=None, required_query_string=None):
+
+    first_path = paths[0]
+    # assume that all files are in the same directory
+    basedir = first_path.dirpath()
+
+    this_dir = os.path.dirname(os.path.abspath(__file__))
+    port = get_open_port()
+    command_list = [
+        "python",
+        os.path.join(this_dir, "http_auth_server.py"),
+        f"--port={port}",
+        "--address=127.0.0.1",
+    ]
+    if username:
+        command_list += [f"--username={username}", f"--password={password}"]
+    if required_query_string:
+        command_list += [f"--required-query-string={required_query_string}"]
+    p = subprocess.Popen(command_list, cwd=basedir)
+    url = f"http://127.0.0.1:{port}"
+    time.sleep(2)  # let the server start up
+
+    def teardown():
+        p.kill()
+
+    request.addfinalizer(teardown)
+
+    return url
+
+
+def make_netcdf_http_paths(netcdf_local_paths, request):
+    paths, items_per_file, fnames_by_variable, _, kwargs = netcdf_local_paths
+
+    url = start_http_server(paths, request, **request.param)
+    path_format = url + "/{variable}_{time:03d}.nc" if fnames_by_variable else None
+
+    fnames = [path.basename for path in paths]
+    all_urls = ["/".join([url, str(fname)]) for fname in fnames]
+
+    if "username" in request.param.keys():
+        kwargs.update(dict(fsspec_open_kwargs={"auth": aiohttp.BasicAuth("foo", "bar")}))
+    if "required_query_string" in request.param.keys():
+        kwargs.update(dict(query_string_secrets={"foo": "foo", "bar": "bar"}))
+
+    return all_urls, items_per_file, fnames_by_variable, path_format, kwargs
+
+
+# Dataset + path fixtures -------------------------------------------------------------------------
+
+
 @pytest.fixture(scope="session")
-def netcdf_local_paths_sequential(daily_xarray_dataset, tmpdir_factory, items_per_file):
-    return make_netcdf_local_paths(
-        daily_xarray_dataset, tmpdir_factory, items_per_file, split_up_files_by_day
+def daily_xarray_dataset():
+    """Return a synthetic random xarray dataset."""
+    np.random.seed(1)
+    # TODO: change nt to 11 in order to catch the edge case where
+    # items_per_input does not evenly divide the length of the sequence dimension
+    nt, ny, nx = 10, 18, 36
+    time = pd.date_range(start="2010-01-01", periods=nt, freq="D")
+    lon = (np.arange(nx) + 0.5) * 360 / nx
+    lon_attrs = {"units": "degrees_east", "long_name": "longitude"}
+    lat = (np.arange(ny) + 0.5) * 180 / ny
+    lat_attrs = {"units": "degrees_north", "long_name": "latitude"}
+    foo = np.random.rand(nt, ny, nx)
+    foo_attrs = {"long_name": "Fantastic Foo"}
+    # make sure things work with heterogenous data types
+    bar = np.random.randint(0, 10, size=(nt, ny, nx))
+    bar_attrs = {"long_name": "Beautiful Bar"}
+    dims = ("time", "lat", "lon")
+    ds = xr.Dataset(
+        {"bar": (dims, bar, bar_attrs), "foo": (dims, foo, foo_attrs)},
+        coords={
+            "time": ("time", time),
+            "lat": ("lat", lat, lat_attrs),
+            "lon": ("lon", lon, lon_attrs),
+        },
+        attrs={"conventions": "CF 1.6"},
     )
+    return ds
 
 
 @pytest.fixture(scope="session")
-def netcdf_local_paths_sequential_multi_variable(
-    daily_xarray_dataset, tmpdir_factory, items_per_file
-):
+def netcdf_local_paths_sequential_1d(daily_xarray_dataset, tmpdir_factory):
+    return make_netcdf_local_paths(daily_xarray_dataset, tmpdir_factory, "D", split_up_files_by_day)
+
+
+@pytest.fixture(scope="session")
+def netcdf_local_paths_sequential_2d(daily_xarray_dataset, tmpdir_factory):
     return make_netcdf_local_paths(
-        daily_xarray_dataset, tmpdir_factory, items_per_file, split_up_files_by_variable_and_day
+        daily_xarray_dataset, tmpdir_factory, "2D", split_up_files_by_day,
     )
 
 
 @pytest.fixture(
     scope="session",
     params=[
-        lazy_fixture("netcdf_local_paths_sequential"),
-        lazy_fixture("netcdf_local_paths_sequential_multi_variable"),
+        lazy_fixture("netcdf_local_paths_sequential_1d"),
+        lazy_fixture("netcdf_local_paths_sequential_2d"),
     ],
 )
-def netcdf_local_paths(request):
+def netcdf_local_paths_sequential(request):
     return request.param
 
 
 @pytest.fixture(scope="session")
-def netcdf_local_file_pattern_sequential(netcdf_local_paths_sequential):
-    return make_file_pattern(netcdf_local_paths_sequential)
+def netcdf_local_paths_sequential_multivariable_1d(daily_xarray_dataset, tmpdir_factory):
+    return make_netcdf_local_paths(
+        daily_xarray_dataset, tmpdir_factory, "D", split_up_files_by_variable_and_day,
+    )
 
 
 @pytest.fixture(scope="session")
-def netcdf_local_file_pattern_sequential_multi_variable(
-    netcdf_local_paths_sequential_multi_variable,
-):
-    return make_file_pattern(netcdf_local_paths_sequential_multi_variable)
+def netcdf_local_paths_sequential_multivariable_2d(daily_xarray_dataset, tmpdir_factory):
+    return make_netcdf_local_paths(
+        daily_xarray_dataset, tmpdir_factory, "2D", split_up_files_by_variable_and_day,
+    )
 
 
 @pytest.fixture(
     scope="session",
     params=[
-        lazy_fixture("netcdf_local_file_pattern_sequential"),
-        lazy_fixture("netcdf_local_file_pattern_sequential_multi_variable"),
+        lazy_fixture("netcdf_local_paths_sequential_multivariable_1d"),
+        lazy_fixture("netcdf_local_paths_sequential_multivariable_2d"),
     ],
 )
-def netcdf_local_file_pattern(request):
+def netcdf_local_paths_sequential_multivariable(request):
     return request.param
 
 
-@pytest.fixture(scope="session")
-def netcdf_http_file_pattern(netcdf_http_paths):
-    return make_file_pattern(netcdf_http_paths)
+@pytest.fixture(
+    scope="session",
+    params=[
+        lazy_fixture("netcdf_local_paths_sequential"),
+        lazy_fixture("netcdf_local_paths_sequential_multivariable"),
+    ],
+)
+def netcdf_local_paths(request):
+    return request.param
 
 
-def start_http_server(paths, request, username=None, password=None, required_query_string=None):
+http_auth_params = [
+    dict(username="foo", password="bar"),
+    dict(required_query_string="foo=foo&bar=bar"),
+]
 
-    first_path = paths[0]
-    # assume that all files are in the same directory
-    basedir = first_path.dirpath()
 
-    this_dir = os.path.dirname(os.path.abspath(__file__))
-    port = get_open_port()
-    command_list = [
-        "python",
-        os.path.join(this_dir, "http_auth_server.py"),
-        f"--port={port}",
-        "--address=127.0.0.1",
-    ]
-    if username:
-        command_list += [f"--username={username}", f"--password={password}"]
-    if required_query_string:
-        command_list += [f"--required-query-string={required_query_string}"]
-    p = subprocess.Popen(command_list, cwd=basedir)
-    url = f"http://127.0.0.1:{port}"
-    time.sleep(2)  # let the server start up
+@pytest.fixture(scope="session", params=[dict()])
+def netcdf_public_http_paths_sequential_1d(netcdf_local_paths_sequential_1d, request):
+    return make_netcdf_http_paths(netcdf_local_paths_sequential_1d, request)
 
-    def teardown():
-        p.kill()
 
-    request.addfinalizer(teardown)
+@pytest.fixture(scope="session", params=[*http_auth_params])
+def netcdf_private_http_paths_sequential_1d(netcdf_local_paths_sequential_1d, request):
+    return make_netcdf_http_paths(netcdf_local_paths_sequential_1d, request)
+
+
+@pytest.fixture(
+    scope="session",
+    params=[
+        lazy_fixture("netcdf_public_http_paths_sequential_1d"),
+        lazy_fixture("netcdf_private_http_paths_sequential_1d"),
+    ],
+)
+def netcdf_http_paths_sequential_1d(request):
+    return request.param
+
+
+# FilePattern fixtures ----------------------------------------------------------------------------
+
+
+@pytest.fixture(scope="session")
+def netcdf_local_file_pattern_sequential(netcdf_local_paths_sequential):
+    return make_file_pattern(netcdf_local_paths_sequential)
 
-    return url
+
+@pytest.fixture(scope="session")
+def netcdf_local_file_pattern_sequential_multivariable(
+    netcdf_local_paths_sequential_multivariable,
+):
+    return make_file_pattern(netcdf_local_paths_sequential_multivariable)
 
 
 @pytest.fixture(
     scope="session",
     params=[
-        {},
-        dict(username="foo", password="bar"),
-        dict(required_query_string="foo=foo&bar=bar"),
+        lazy_fixture("netcdf_local_file_pattern_sequential"),
+        lazy_fixture("netcdf_local_file_pattern_sequential_multivariable"),
     ],
 )
-def netcdf_http_paths(netcdf_local_paths, request):
-    paths, items_per_file, fnames_by_variable, _, kwargs = netcdf_local_paths
+def netcdf_local_file_pattern(request):
+    return request.param
 
-    url = start_http_server(paths, request, **request.param)
-    path_format = url + "/{variable}_{time:03d}.nc" if fnames_by_variable else None
 
-    fnames = [path.basename for path in paths]
-    all_urls = ["/".join([url, str(fname)]) for fname in fnames]
+@pytest.fixture(scope="session")
+def netcdf_http_file_pattern_sequential_1d(netcdf_http_paths_sequential_1d):
+    return make_file_pattern(netcdf_http_paths_sequential_1d)
 
-    if "username" in request.param.keys():
-        kwargs.update(dict(fsspec_open_kwargs={"auth": aiohttp.BasicAuth("foo", "bar")}))
-    if "required_query_string" in request.param.keys():
-        kwargs.update(dict(query_string_secrets={"foo": "foo", "bar": "bar"}))
 
-    return all_urls, items_per_file, fnames_by_variable, path_format, kwargs
+# Storage fixtures --------------------------------------------------------------------------------
 
 
 @pytest.fixture()
@@ -282,6 +354,9 @@ def tmp_metadata_target(tmpdir_factory):
     return cache
 
 
+# Execution fixtures ------------------------------------------------------------------------------
+
+
 @pytest.fixture(scope="session")
 def dask_cluster(request):
     cluster = LocalCluster(n_workers=2, threads_per_worker=1, silence_logs=False)
@@ -328,46 +403,100 @@ def redirect_logs():
 }
 
 
-@pytest.fixture(params=["manual", "python", "dask", "prefect", "prefect-dask"])
-def execute_recipe(request, dask_cluster):
-    if request.param == "manual":
+@pytest.fixture()
+def execute_recipe_manual():
+    def execute(r):
+        if r.cache_inputs:
+            for input_key in r.iter_inputs():
+                r.cache_input(input_key)
+        r.prepare_target()
+        for chunk_key in r.iter_chunks():
+            r.store_chunk(chunk_key)
+        r.finalize_target()
+
+    return execute
+
+
+@pytest.fixture()
+def execute_recipe_python():
+    def execute(recipe):
+        return recipe.to_function()()
 
-        def execute(r):
-            if r.cache_inputs:
-                for input_key in r.iter_inputs():
-                    r.cache_input(input_key)
-            r.prepare_target()
-            for chunk_key in r.iter_chunks():
-                r.store_chunk(chunk_key)
-            r.finalize_target()
+    return execute
 
-    elif request.param == "python":
 
-        def execute(recipe):
-            return recipe.to_function()()
+@pytest.fixture()
+def execute_recipe_dask(dask_cluster):
+    def execute(recipe):
+        with Client(dask_cluster):
+            return recipe.to_dask().compute()
 
-    elif request.param == "dask":
+    return execute
 
-        def execute(recipe):
-            with Client(dask_cluster):
-                return recipe.to_dask().compute()
 
-    elif request.param == "prefect":
+@pytest.fixture()
+def execute_recipe_prefect():
+    def execute(recipe):
+        state = recipe.to_prefect().run()
+        if state.is_failed():
+            raise ValueError(f"Prefect flow run failed with message {state.message}")
 
-        def execute(recipe):
-            state = recipe.to_prefect().run()
-            if state.is_failed():
-                raise ValueError(f"Prefect flow run failed with message {state.message}")
+    return execute
 
-    else:
-        assert request.param == "prefect-dask"
 
-        def execute(recipe):
-            flow = recipe.to_prefect()
-            executor = DaskExecutor(address=dask_cluster.scheduler_address)
-            state = flow.run(executor=executor)
-            if state.is_failed():
-                raise ValueError(f"Prefect flow run failed with message {state.message}")
+@pytest.fixture()
+def execute_recipe_prefect_dask(dask_cluster):
+    def execute(recipe):
+        flow = recipe.to_prefect()
+        executor = DaskExecutor(address=dask_cluster.scheduler_address)
+        state = flow.run(executor=executor)
+        if state.is_failed():
+            raise ValueError(f"Prefect flow run failed with message {state.message}")
 
-    execute.param = request.param
     return execute
+
+
+@pytest.fixture(
+    params=[
+        lazy_fixture("execute_recipe_manual"),
+        lazy_fixture("execute_recipe_python"),
+        lazy_fixture("execute_recipe_dask"),
+    ],
+)
+def execute_recipe_no_prefect(request):
+    return request.param
+
+
+@pytest.fixture(
+    params=[lazy_fixture("execute_recipe_prefect"), lazy_fixture("execute_recipe_prefect_dask")],
+)
+def execute_recipe_with_prefect(request):
+    return request.param
+
+
+@pytest.fixture(
+    params=[
+        lazy_fixture("execute_recipe_manual"),
+        lazy_fixture("execute_recipe_python"),
+        lazy_fixture("execute_recipe_prefect"),
+    ],
+)
+def execute_recipe_no_dask(request):
+    return request.param
+
+
+@pytest.fixture(
+    params=[lazy_fixture("execute_recipe_dask"), lazy_fixture("execute_recipe_prefect_dask")],
+)
+def execute_recipe_with_dask(request):
+    return request.param
+
+
+@pytest.fixture(
+    params=[
+        lazy_fixture("execute_recipe_no_prefect"),
+        lazy_fixture("execute_recipe_with_prefect"),
+    ],
+)
+def execute_recipe(request):
+    return request.param
diff --git a/tests/test_fixtures.py b/tests/test_fixtures.py
index ad43e720..351ee17e 100644
--- a/tests/test_fixtures.py
+++ b/tests/test_fixtures.py
@@ -1,7 +1,7 @@
 import fsspec
-import pytest
 import xarray as xr
 
+from pangeo_forge_recipes.storage import _add_query_string_secrets
 from pangeo_forge_recipes.utils import fix_scalar_attr_encoding
 
 
@@ -12,14 +12,13 @@ def test_fixture_local_files(daily_xarray_dataset, netcdf_local_paths):
     assert ds.identical(daily_xarray_dataset)
 
 
-def test_fixture_http_files(daily_xarray_dataset, netcdf_http_paths):
-    urls = netcdf_http_paths[0]
-    if (
-        "auth" in netcdf_http_paths[-1]["fsspec_open_kwargs"].keys()
-        or netcdf_http_paths[-1]["query_string_secrets"].keys()
-    ):
-        pytest.skip("Authentication and required query strings are tested in test_storage.py")
-    open_files = [fsspec.open(url).open() for url in urls]
+def test_fixture_http_files(daily_xarray_dataset, netcdf_http_paths_sequential_1d):
+    urls = netcdf_http_paths_sequential_1d[0]
+    open_kwargs = netcdf_http_paths_sequential_1d[-1]["fsspec_open_kwargs"]
+    secrets = netcdf_http_paths_sequential_1d[-1]["query_string_secrets"]
+    if secrets:
+        urls = [_add_query_string_secrets(url, secrets) for url in urls]
+    open_files = [fsspec.open(url, **open_kwargs).open() for url in urls]
     ds = xr.open_mfdataset(
         open_files, combine="by_coords", concat_dim="time", engine="h5netcdf"
     ).load()
diff --git a/tests/test_recipes.py b/tests/test_recipes.py
index f1c215fd..38297456 100644
--- a/tests/test_recipes.py
+++ b/tests/test_recipes.py
@@ -59,6 +59,23 @@ def netCDFtoZarr_http_recipe(
     )
 
 
+@pytest.fixture
+def netCDFtoZarr_http_recipe_sequential_1d(
+    netcdf_http_file_pattern_sequential_1d,
+    daily_xarray_dataset,
+    tmp_target,
+    tmp_cache,
+    tmp_metadata_target,
+):
+    return make_netCDFtoZarr_recipe(
+        netcdf_http_file_pattern_sequential_1d,
+        daily_xarray_dataset,
+        tmp_target,
+        tmp_cache,
+        tmp_metadata_target,
+    )
+
+
 @pytest.fixture
 def netCDFtoZarr_subset_recipe(
     netcdf_local_file_pattern, daily_xarray_dataset, tmp_target, tmp_cache, tmp_metadata_target
@@ -132,7 +149,11 @@ def test_prune_recipe(recipe_fixture, execute_recipe, nkeep):
 @pytest.mark.parametrize("cache_inputs", [True, False])
 @pytest.mark.parametrize("copy_input_to_local_file", [True, False])
 @pytest.mark.parametrize(
-    "recipe", [lazy_fixture("netCDFtoZarr_recipe"), lazy_fixture("netCDFtoZarr_http_recipe")],
+    "recipe",
+    [
+        lazy_fixture("netCDFtoZarr_recipe_sequential_only"),
+        lazy_fixture("netCDFtoZarr_http_recipe_sequential_1d"),
+    ],
 )
 def test_recipe_caching_copying(recipe, execute_recipe, cache_inputs, copy_input_to_local_file):
     """Test that caching and copying to local file work."""
@@ -185,26 +206,9 @@ def test_process(recipe_fixture, execute_recipe, process_input, process_chunk):
     xr.testing.assert_identical(ds_actual, ds_expected)
 
 
-@pytest.mark.parametrize("inputs_per_chunk,subset_inputs", [(1, {}), (1, {"time": 2}), (2, {})])
-@pytest.mark.parametrize(
-    "target_chunks,specify_nitems_per_input,error_expectation",
-    [
-        ({}, True, does_not_raise()),
-        ({"lon": 12}, True, does_not_raise()),
-        ({"lon": 12, "time": 1}, True, does_not_raise()),
-        ({"lon": 12, "time": 3}, True, does_not_raise()),
-        ({"time": 10}, True, does_not_raise()),  # only one big chunk
-        ({"lon": 12, "time": 1}, False, does_not_raise()),
-        ({"lon": 12, "time": 3}, False, does_not_raise()),
-        # can't determine target chunks for the next two because 'time' missing from target_chunks
-        ({}, False, pytest.raises(ValueError)),
-        ({"lon": 12}, False, pytest.raises(ValueError)),
-    ],
-)
-@pytest.mark.parametrize("recipe_fixture", recipes_no_subset)
-def test_chunks(
+def do_actual_chunks_test(
     recipe_fixture,
-    execute_recipe,
+    executor,
     inputs_per_chunk,
     target_chunks,
     error_expectation,
@@ -250,7 +254,7 @@ def test_chunks(
     with subset_error_expectation as excinfo:
         # error is raised at execution stage because we don't generally know a priori how
         # many items in each file
-        execute_recipe(rec)
+        executor(rec)
     if excinfo:
         # don't continue if we got an exception
         return
@@ -273,20 +277,82 @@ def test_chunks(
     xr.testing.assert_identical(ds_actual, ds_expected)
 
 
-def test_lock_timeout(netCDFtoZarr_recipe_sequential_only, execute_recipe):
+@pytest.mark.parametrize("inputs_per_chunk,subset_inputs", [(1, {}), (1, {"time": 2}), (2, {})])
+@pytest.mark.parametrize(
+    "target_chunks,specify_nitems_per_input,error_expectation",
+    [
+        ({}, True, does_not_raise()),
+        ({"lon": 12}, True, does_not_raise()),
+        ({"lon": 12, "time": 1}, True, does_not_raise()),
+        ({"lon": 12, "time": 3}, True, does_not_raise()),
+        ({"time": 10}, True, does_not_raise()),  # only one big chunk
+        ({"lon": 12, "time": 1}, False, does_not_raise()),
+        ({"lon": 12, "time": 3}, False, does_not_raise()),
+        # can't determine target chunks for the next two because 'time' missing from target_chunks
+        ({}, False, pytest.raises(ValueError)),
+        ({"lon": 12}, False, pytest.raises(ValueError)),
+    ],
+)
+@pytest.mark.parametrize("recipe_fixture", recipes_no_subset)
+def test_chunks(
+    recipe_fixture,
+    execute_recipe_python,
+    inputs_per_chunk,
+    target_chunks,
+    error_expectation,
+    subset_inputs,
+    specify_nitems_per_input,
+):
+    do_actual_chunks_test(
+        recipe_fixture,
+        execute_recipe_python,
+        inputs_per_chunk,
+        target_chunks,
+        error_expectation,
+        subset_inputs,
+        specify_nitems_per_input,
+    )
+
+
+@pytest.mark.parametrize("inputs_per_chunk,subset_inputs", [(1, {}), (1, {"time": 2}), (2, {})])
+@pytest.mark.parametrize(
+    "target_chunks,specify_nitems_per_input,error_expectation",
+    [({"lon": 12, "time": 3}, False, does_not_raise())],
+)
+@pytest.mark.parametrize("recipe_fixture", recipes_no_subset)
+def test_chunks_distributed_locking(
+    recipe_fixture,
+    execute_recipe_with_dask,
+    inputs_per_chunk,
+    target_chunks,
+    error_expectation,
+    subset_inputs,
+    specify_nitems_per_input,
+):
+    do_actual_chunks_test(
+        recipe_fixture,
+        execute_recipe_with_dask,
+        inputs_per_chunk,
+        target_chunks,
+        error_expectation,
+        subset_inputs,
+        specify_nitems_per_input,
+    )
+
+
+def test_lock_timeout(netCDFtoZarr_recipe_sequential_only, execute_recipe_no_dask):
     RecipeClass, file_pattern, kwargs, ds_expected, target = netCDFtoZarr_recipe_sequential_only
 
     recipe = RecipeClass(file_pattern=file_pattern, lock_timeout=1, **kwargs)
 
     with patch("pangeo_forge_recipes.recipes.xarray_zarr.lock_for_conflicts") as p:
-        execute_recipe(recipe)
-
-    # We can only check that the mock object is called with the right parameters if
-    # the function is called in the same processes as our mock object. We can't
-    # observe anything that happens when the function is executed in subprocess, i.e.
-    # if we're using a Dask executor.
-    if execute_recipe.param in {"manual", "python", "prefect"}:
-        assert p.call_args[1]["timeout"] == 1
+        # We can only check that the mock object is called with the right parameters if
+        # the function is called in the same processes as our mock object. We can't
+        # observe anything that happens when the function is executed in subprocess, i.e.
+        # if we're using a Dask executor.
+        execute_recipe_no_dask(recipe)
+
+    assert p.call_args[1]["timeout"] == 1
 
 
 class MyRecipe(BaseRecipe):
diff --git a/tests/test_storage.py b/tests/test_storage.py
index 90f015a1..c620615a 100644
--- a/tests/test_storage.py
+++ b/tests/test_storage.py
@@ -40,7 +40,11 @@ def test_metadata_target(tmp_metadata_target):
 
 
 @pytest.mark.parametrize(
-    "file_paths", [lazy_fixture("netcdf_local_paths"), lazy_fixture("netcdf_http_paths")],
+    "file_paths",
+    [
+        lazy_fixture("netcdf_local_paths_sequential"),
+        lazy_fixture("netcdf_http_paths_sequential_1d"),
+    ],
 )
 @pytest.mark.parametrize("copy_to_local", [False, True])
 @pytest.mark.parametrize("use_cache, cache_first", [(False, False), (True, False), (True, True)])