Add test with http basic auth
MODIFY
setup.cfg
setup.cfg
@@ -3,7 +3,7 @@ max-line-length = 100
[isort]
known_first_party=pangeo_forge
-known_third_party=fsspec,numpy,pandas,pkg_resources,prefect,pytest,rechunker,setuptools,sphinx_book_theme,xarray,zarr
+known_third_party=aiohttp,fsspec,numpy,pandas,pkg_resources,prefect,pytest,rechunker,setuptools,sphinx_book_theme,xarray,zarr
multi_line_output=3
include_trailing_comma=True
force_grid_wrap=0

MODIFY
tests/conftest.py
tests/conftest.py
+import os
import subprocess
import time
@@ -58,20 +59,36 @@ def netcdf_local_paths(daily_xarray_dataset, tmpdir_factory, request):
@pytest.fixture()
-def netcdf_http_server(netcdf_local_paths):
+def netcdf_http_server(netcdf_local_paths, request):
+ def make_netcdf_http_server(username="", password=""):
first_path = netcdf_local_paths[0]
# assume that all files are in the same directory
basedir = first_path.dirpath()
fnames = [path.basename for path in netcdf_local_paths]
- # this feels very hacky
- command_list = ["python", "-m", "http.server", _PORT, "--bind", _ADDRESS]
+ this_dir = os.path.dirname(os.path.abspath(__file__))
+ command_list = [
+ "python",
+ os.path.join(this_dir, "http_auth_server.py"),
+ _PORT,
+ _ADDRESS,
+ username,
+ password,
+ ]
+ if username:
+ command_list += [username, password]
p = subprocess.Popen(command_list, cwd=basedir)
url = f"http://{_ADDRESS}:{_PORT}"
time.sleep(1) # let the server start up
- yield url, fnames
+
+ def teardown():
p.kill()
+ request.addfinalizer(teardown)
+ return url, fnames
+
+ return make_netcdf_http_server
+
@pytest.fixture()
def tmp_target(tmpdir_factory):

ADD
tests/http_auth_server.py
+import base64
+import http.server
+import socketserver
+import sys
+
+port_str, ADDRESS = sys.argv[1:3]
+PORT = int(port_str)
+if len(sys.argv) > 3:
+ username, password = sys.argv[3:5]
+else:
+ username, password = "", ""
+
+
+class Handler(http.server.SimpleHTTPRequestHandler):
+ def do_GET(self):
+ if username:
+ auth = self.headers.get("Authorization")
+ if (
+ auth is None
+ or not auth.startswith("Basic")
+ or auth[6:]
+ != str(base64.b64encode((username + ":" + password).encode("utf-8")), "utf-8")
+ ):
+ self.send_response(401)
+ self.send_header("WWW-Authenticate", "Basic")
+ self.end_headers()
+ return
+ return http.server.SimpleHTTPRequestHandler.do_GET(self)
+
+
+socketserver.TCPServer.allow_reuse_address = True
+with socketserver.TCPServer((ADDRESS, PORT), Handler) as httpd:
+ httpd.serve_forever()

MODIFY
tests/test_fixtures.py
tests/test_fixtures.py
@@ -13,7 +13,7 @@ def test_fixture_local_files(daily_xarray_dataset, netcdf_local_paths):
def test_fixture_http_files(daily_xarray_dataset, netcdf_http_server):
- url, paths = 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()

MODIFY
tests/test_recipe.py
tests/test_recipe.py
+import aiohttp
import pytest
import xarray as xr
@@ -35,6 +36,42 @@ def test_sequence_recipe(file_urls, files_per_chunk, expected_keys, expected_fil
assert fnames == expected
+@pytest.mark.parametrize(
+ "username, password", [("foo", "bar"), ("foo", "wrong"),], # noqa: E231
+)
+def test_NetCDFtoZarrSequentialRecipeHttpAuth(
+ daily_xarray_dataset, netcdf_http_server, tmp_target, tmp_cache, username, password
+):
+
+ url, fnames = 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"],
+ target=tmp_target,
+ input_cache=tmp_cache,
+ fsspec_open_kwargs={"client_kwargs": {"auth": aiohttp.BasicAuth(username, password)}},
+ )
+
+ if password == "wrong":
+ 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()
+
+ ds_target = xr.open_zarr(tmp_target.get_mapper(), consolidated=True).load()
+ ds_expected = daily_xarray_dataset.compute()
+ assert ds_target.identical(ds_expected)
+
+
def test_NetCDFtoZarrSequentialRecipe(
daily_xarray_dataset, netcdf_local_paths, tmp_target, tmp_cache
):

Add test with http basic auth