testing executors
MODIFY
pangeo_forge/executors/__init__.py
pangeo_forge/executors/__init__.py
+"""
+Executors know how to run recipes.
+"""
+
+from .prefect import PrefectExecutor
+from .python import PythonExecutor
+
+__all__ = [PythonExecutor, PrefectExecutor]

MODIFY
pangeo_forge/recipe.py
pangeo_forge/recipe.py
@@ -318,6 +318,3 @@ class StandardSequentialRecipe(
ZarrConsolidatorMixin,
):
pass
-
-
-# helper utilities

MODIFY
setup.cfg
setup.cfg
@@ -11,5 +11,4 @@ combine_as_imports=True
line_length=100
[tool:pytest]
-log_cli = True
-log_level = INFO
+log_cli = False

MODIFY
tests/conftest.py
tests/conftest.py
@@ -7,6 +7,7 @@ import pandas as pd
import pytest
import xarray as xr
+from pangeo_forge import recipe
from pangeo_forge.storage import InputCache, Target
# where to run the http server
@@ -88,3 +89,21 @@ def tmp_cache(tmpdir_factory):
fs = fsspec.get_filesystem_class("file")()
cache = InputCache(fs, prefix=path)
return cache
+
+
+@pytest.fixture
+def sequential_recipe(daily_xarray_dataset, netcdf_local_paths, tmp_target, tmp_cache):
+ r = recipe.StandardSequentialRecipe(
+ consolidate_zarr=True,
+ xarray_open_kwargs={},
+ xarray_concat_kwargs={},
+ require_cache=False,
+ input_cache=tmp_cache,
+ target=tmp_target,
+ chunk_preprocess_funcs=[],
+ input_urls=netcdf_local_paths,
+ sequence_dim="time",
+ inputs_per_chunk=1,
+ nitems_per_input=daily_xarray_dataset.attrs["items_per_file"],
+ )
+ return r, daily_xarray_dataset, tmp_target

ADD
tests/test_executors.py
+import pytest
+import xarray as xr
+
+from pangeo_forge.executors import PrefectExecutor, PythonExecutor
+
+
+@pytest.mark.parametrize("Executor", [PythonExecutor, PrefectExecutor])
+def test_recipe_w_executor(Executor, sequential_recipe):
+ rec, ds_expected, target = sequential_recipe
+ ex = Executor()
+ plan = ex.prepare_plan(rec)
+ ex.execute_plan(plan)
+ ds_actual = xr.open_zarr(target.get_mapper()).load()
+ assert ds_actual.identical(ds_expected)

ADD
tests/test_fixtures.py
+import fsspec
+import xarray as xr
+
+from pangeo_forge.utils import fix_scalar_attr_encoding
+
+
+def test_fixture_local_files(daily_xarray_dataset, netcdf_local_paths):
+ paths = [str(path) for path in netcdf_local_paths]
+ ds = xr.open_mfdataset(paths, combine="nested", concat_dim="time")
+ assert ds.identical(daily_xarray_dataset)
+
+
+def test_fixture_http_files(daily_xarray_dataset, netcdf_http_server):
+ url, paths = netcdf_http_server
+ urls = ["/".join([url, str(path)]) for path in paths]
+ open_files = [fsspec.open(url).open() for url in urls]
+ ds = xr.open_mfdataset(open_files, combine="nested", concat_dim="time").load()
+ ds = fix_scalar_attr_encoding(ds)
+ print(ds)
+ print(daily_xarray_dataset)
+ assert ds.identical(daily_xarray_dataset)
+
+
+def test_target(tmp_target):
+ mapper = tmp_target.get_mapper()
+ mapper["foo"] = b"bar"
+ with open(tmp_target.path + "/foo") as f:
+ res = f.read()
+ assert res == "bar"
+
+
+def test_cache(tmp_cache):
+ assert not tmp_cache.exists("foo")
+ with tmp_cache.open("foo", mode="w") as f:
+ f.write("bar")
+ assert tmp_cache.exists("foo")
+ with tmp_cache.open("foo", mode="r") as f:
+ assert f.read() == "bar"
+ tmp_cache.rm("foo")
+ assert not tmp_cache.exists("foo")

MODIFY
tests/test_recipe.py
tests/test_recipe.py
@@ -46,6 +46,8 @@ def test_sequence_recipe(file_urls, files_per_chunk, expected_keys, expected_fil
def test_full_recipe(daily_xarray_dataset, netcdf_local_paths, tmp_target, tmp_cache):
+ # the same recipe is created as a fixture in conftest.py
+ # I left it here explicitly because it makes the test easier to read.
r = recipe.StandardSequentialRecipe(
consolidate_zarr=True,
xarray_open_kwargs={},

testing executors