tweak fixtures and tests
MODIFY
tests/conftest.py
tests/conftest.py
@@ -51,8 +51,6 @@ def daily_xarray_dataset():
def _split_up_files_by_day(ds, day_param):
- items_per_file = {"D": 1, "2D": 2}
- ds.attrs["items_per_file"] = items_per_file[day_param]
gb = ds.resample(time=day_param)
_, datasets = zip(*gb)
fnames = [f"{n:03d}.nc" for n in range(len(datasets))]
@@ -78,7 +76,8 @@ def netcdf_local_paths(daily_xarray_dataset, tmpdir_factory, request):
datasets, fnames = _split_up_files_by_day(daily_xarray_dataset.copy(), request.param)
full_paths = [tmp_path.join(fname) for fname in fnames]
xr.save_mfdataset(datasets, [str(path) for path in full_paths])
- return full_paths
+ items_per_file = {"D": 1, "2D": 2}[request.param]
+ return full_paths, items_per_file
# TODO: this is quite repetetive of the fixture above. Replace with parametrization.
@@ -98,11 +97,13 @@ def netcdf_local_paths_by_variable(daily_xarray_dataset, tmpdir_factory, request
# duplicating the whole test.
@pytest.fixture()
def netcdf_http_server(netcdf_local_paths, request):
+ paths, items_per_file = netcdf_local_paths
+
def make_netcdf_http_server(username="", password=""):
- first_path = netcdf_local_paths[0]
+ first_path = paths[0]
# assume that all files are in the same directory
basedir = first_path.dirpath()
- fnames = [path.basename for path in netcdf_local_paths]
+ fnames = [path.basename for path in paths]
this_dir = os.path.dirname(os.path.abspath(__file__))
port = get_open_port()
@@ -124,7 +125,7 @@ def netcdf_http_server(netcdf_local_paths, request):
p.kill()
request.addfinalizer(teardown)
- return url, fnames
+ return url, fnames, items_per_file
return make_netcdf_http_server
@@ -153,11 +154,12 @@ def uninitialized_target():
@pytest.fixture
def netCDFtoZarr_sequential_recipe(daily_xarray_dataset, netcdf_local_paths, tmp_target, tmp_cache):
+ paths, items_per_file = netcdf_local_paths
r = recipe.NetCDFtoZarrSequentialRecipe(
- input_urls=netcdf_local_paths,
+ input_urls=paths,
sequence_dim="time",
inputs_per_chunk=1,
- nitems_per_input=daily_xarray_dataset.attrs["items_per_file"],
+ nitems_per_input=items_per_file,
target=tmp_target,
input_cache=tmp_cache,
)

MODIFY
tests/test_fixtures.py
tests/test_fixtures.py
@@ -7,9 +7,9 @@ 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]
+ paths, items_per_file = netcdf_local_paths
+ paths = [str(path) for path in paths]
ds = xr.open_mfdataset(paths, combine="nested", concat_dim="time")
- del ds.attrs["items_per_file"] # avoid metadata conflicts
assert ds.identical(daily_xarray_dataset)
@@ -17,18 +17,15 @@ def test_fixture_local_files(daily_xarray_dataset, netcdf_local_paths):
def test_fixture_local_files_by_variable(daily_xarray_dataset, netcdf_local_paths_by_variable):
paths = [str(path) for path in netcdf_local_paths_by_variable]
ds = xr.open_mfdataset(paths, combine="by_coords", concat_dim="time")
- del ds.attrs["items_per_file"] # avoid metadata conflicts
assert ds.identical(daily_xarray_dataset)
def test_fixture_http_files(daily_xarray_dataset, netcdf_http_server):
- url, paths = netcdf_http_server()
+ url, paths, items_per_file = 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)

MODIFY
tests/test_recipe.py
tests/test_recipe.py
@@ -43,6 +43,16 @@ def test_sequence_recipe(file_urls, files_per_chunk, expected_keys, expected_fil
assert fnames == expected
+def _manually_execute_recipe(r):
+ # this is the cannonical way to manually execute a recipe
+ 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()
+
+
@pytest.mark.parametrize(
"username, password", [("foo", "bar"), ("foo", "wrong"),], # noqa: E231
)
@@ -50,13 +60,13 @@ def test_NetCDFtoZarrSequentialRecipeHttpAuth(
daily_xarray_dataset, netcdf_http_server, tmp_target, tmp_cache, username, password
):
- url, fnames = netcdf_http_server("foo", "bar")
+ url, fnames, items_per_file = netcdf_http_server("foo", "bar")
urls = [f"{url}/{fname}" for fname in fnames]
r = recipe.NetCDFtoZarrSequentialRecipe(
input_urls=urls,
sequence_dim="time",
inputs_per_chunk=1,
- nitems_per_input=daily_xarray_dataset.attrs["items_per_file"],
+ nitems_per_input=items_per_file,
target=tmp_target,
input_cache=tmp_cache,
fsspec_open_kwargs={"client_kwargs": {"auth": aiohttp.BasicAuth(username, password)}},
@@ -66,13 +76,7 @@ def test_NetCDFtoZarrSequentialRecipeHttpAuth(
with pytest.raises(aiohttp.client_exceptions.ClientResponseError):
r.cache_input(next(r.iter_inputs()))
else:
- # this is the cannonical way to manually execute a recipe
- 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()
+ _manually_execute_recipe(r)
ds_target = xr.open_zarr(tmp_target.get_mapper(), consolidated=True).load()
ds_expected = daily_xarray_dataset.compute()
@@ -89,11 +93,12 @@ def test_NetCDFtoZarrSequentialRecipe(
# the same recipe is created as a fixture in conftest.py
# I left it here explicitly because it makes the test easier to read.
+ paths, items_per_file = netcdf_local_paths
r = recipe.NetCDFtoZarrSequentialRecipe(
- input_urls=netcdf_local_paths,
+ input_urls=paths,
sequence_dim="time",
inputs_per_chunk=1,
- nitems_per_input=daily_xarray_dataset.attrs["items_per_file"],
+ nitems_per_input=items_per_file,
target=tmp_target,
input_cache=tmp_cache,
process_input=process_input,
@@ -101,12 +106,7 @@ def test_NetCDFtoZarrSequentialRecipe(
)
# this is the cannonical way to manually execute a recipe
- 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()
+ _manually_execute_recipe(r)
ds_target = xr.open_zarr(tmp_target.get_mapper(), consolidated=True).load()
ds_expected = daily_xarray_dataset.compute()
@@ -129,11 +129,9 @@ def test_NetCDFtoZarrSequentialRecipeNoTarget(
daily_xarray_dataset, netcdf_local_paths, tmp_target, tmp_cache
):
+ paths, items_per_file = netcdf_local_paths
r = recipe.NetCDFtoZarrSequentialRecipe(
- input_urls=netcdf_local_paths,
- sequence_dim="time",
- inputs_per_chunk=1,
- nitems_per_input=daily_xarray_dataset.attrs["items_per_file"],
+ input_urls=paths, sequence_dim="time", inputs_per_chunk=1, nitems_per_input=items_per_file,
)
# this is the cannonical way to manually execute a recipe

tweak fixtures and tests