refactor download data function

tests/helper_functions.py

@@ -0,0 +1,21 @@
+import os
+
+import requests
+
+
+def download_test_data(nexus_url, filename):
+    local_directory = "./tests/data"
+
+    if not os.path.exists(local_directory):
+        os.makedirs(local_directory)
+
+    # Construct the full path for the local file
+    local_file_path = os.path.join(local_directory, filename)
+
+    if not os.path.exists(local_file_path):
+        session = requests.Session()
+        response = session.get(nexus_url)
+        if response.status_code == 200:
+            # Save the downloaded data to the local file
+            with open(local_file_path, "wb") as f:
+                f.write(response.content)

tests/test_datacube_axes_init.py

@@ -1,7 +1,4 @@
-import os
-
 import pytest
-import requests
 from earthkit import data
 
 from polytope.datacube.backends.xarray import XArrayDatacube
@@ -10,26 +7,13 @@
 from polytope.polytope import Polytope, Request
 from polytope.shapes import Box, Select
 
+from .helper_functions import download_test_data
+
 
 class TestInitDatacubeAxes:
     def setup_method(self, method):
         nexus_url = "https://get.ecmwf.int/test-data/polytope/test-data/foo.grib"
-
-        local_directory = "./tests/data"
-
-        if not os.path.exists(local_directory):
-            os.makedirs(local_directory)
-
-        # Construct the full path for the local file
-        local_file_path = os.path.join(local_directory, "foo.grib")
-
-        if not os.path.exists(local_file_path):
-            session = requests.Session()
-            response = session.get(nexus_url)
-            if response.status_code == 200:
-                # Save the downloaded data to the local file
-                with open(local_file_path, "wb") as f:
-                    f.write(response.content)
+        download_test_data(nexus_url, "foo.grib")
 
         ds = data.from_source("file", "./tests/data/foo.grib")
         latlon_array = ds.to_xarray().isel(step=0).isel(number=0).isel(surface=0).isel(time=0)

tests/test_healpix_mapper.py

@@ -1,7 +1,4 @@
-import os
-
 import pytest
-import requests
 from earthkit import data
 from eccodes import codes_grib_find_nearest, codes_grib_new_from_file
 
@@ -10,26 +7,13 @@
 from polytope.polytope import Polytope, Request
 from polytope.shapes import Box, Select
 
+from .helper_functions import download_test_data
+
 
 class TestOctahedralGrid:
     def setup_method(self, method):
         nexus_url = "https://get.ecmwf.int/test-data/polytope/test-data/healpix.grib"
-
-        local_directory = "./tests/data"
-
-        if not os.path.exists(local_directory):
-            os.makedirs(local_directory)
-
-        # Construct the full path for the local file
-        local_file_path = os.path.join(local_directory, "healpix.grib")
-
-        if not os.path.exists(local_file_path):
-            session = requests.Session()
-            response = session.get(nexus_url)
-            if response.status_code == 200:
-                # Save the downloaded data to the local file
-                with open(local_file_path, "wb") as f:
-                    f.write(response.content)
+        download_test_data(nexus_url, "healpix.grib")
 
         ds = data.from_source("file", "./tests/data/healpix.grib")
         self.latlon_array = ds.to_xarray().isel(step=0).isel(time=0).isel(isobaricInhPa=0).z

tests/test_merge_octahedral_one_axis.py

@@ -1,34 +1,18 @@
-import os
-
 import pytest
-import requests
 from earthkit import data
 
 from polytope.datacube.backends.xarray import XArrayDatacube
 from polytope.engine.hullslicer import HullSlicer
 from polytope.polytope import Polytope, Request
 from polytope.shapes import Box, Select
 
+from .helper_functions import download_test_data
+
 
 class TestSlicingMultipleTransformationsOneAxis:
     def setup_method(self, method):
         nexus_url = "https://get.ecmwf.int/test-data/polytope/test-data/foo.grib"
-
-        local_directory = "./tests/data"
-
-        if not os.path.exists(local_directory):
-            os.makedirs(local_directory)
-
-        # Construct the full path for the local file
-        local_file_path = os.path.join(local_directory, "foo.grib")
-
-        if not os.path.exists(local_file_path):
-            session = requests.Session()
-            response = session.get(nexus_url)
-            if response.status_code == 200:
-                # Save the downloaded data to the local file
-                with open(local_file_path, "wb") as f:
-                    f.write(response.content)
+        download_test_data(nexus_url, "foo.grib")
 
         ds = data.from_source("file", "./tests/data/foo.grib")
         self.latlon_array = ds.to_xarray().isel(step=0).isel(number=0).isel(surface=0).isel(time=0)

tests/test_octahedral_grid.py

@@ -1,7 +1,4 @@
-import os
-
 import pytest
-import requests
 from earthkit import data
 from eccodes import codes_grib_find_nearest, codes_grib_new_from_file
 
@@ -10,26 +7,13 @@
 from polytope.polytope import Polytope, Request
 from polytope.shapes import Box, Select
 
+from .helper_functions import download_test_data
+
 
 class TestOctahedralGrid:
     def setup_method(self, method):
         nexus_url = "https://get.ecmwf.int/test-data/polytope/test-data/foo.grib"
-
-        local_directory = "./tests/data"
-
-        if not os.path.exists(local_directory):
-            os.makedirs(local_directory)
-
-        # Construct the full path for the local file
-        local_file_path = os.path.join(local_directory, "foo.grib")
-
-        if not os.path.exists(local_file_path):
-            session = requests.Session()
-            response = session.get(nexus_url)
-            if response.status_code == 200:
-                # Save the downloaded data to the local file
-                with open(local_file_path, "wb") as f:
-                    f.write(response.content)
+        download_test_data(nexus_url, "foo.grib")
 
         ds = data.from_source("file", "./tests/data/foo.grib")
         self.latlon_array = ds.to_xarray().isel(step=0).isel(number=0).isel(surface=0).isel(time=0)

tests/test_slicer_era5.py

@@ -1,35 +1,19 @@
-import os
-
 import numpy as np
 import pytest
-import requests
 from earthkit import data
 
 from polytope.datacube.backends.xarray import XArrayDatacube
 from polytope.engine.hullslicer import HullSlicer
 from polytope.polytope import Polytope, Request
 from polytope.shapes import Box, Select
 
+from .helper_functions import download_test_data
+
 
 class TestSlicingEra5Data:
     def setup_method(self, method):
         nexus_url = "https://get.ecmwf.int/test-data/polytope/test-data/era5-levels-members.grib"
-
-        local_directory = "./tests/data"
-
-        if not os.path.exists(local_directory):
-            os.makedirs(local_directory)
-
-        # Construct the full path for the local file
-        local_file_path = os.path.join(local_directory, "era5-levels-members.grib")
-
-        if not os.path.exists(local_file_path):
-            session = requests.Session()
-            response = session.get(nexus_url)
-            if response.status_code == 200:
-                # Save the downloaded data to the local file
-                with open(local_file_path, "wb") as f:
-                    f.write(response.content)
+        download_test_data(nexus_url, "era5-levels-members.grib")
 
         ds = data.from_source("file", "./tests/data/era5-levels-members.grib")
         array = ds.to_xarray().isel(step=0).t

tests/test_snapping_real_data.py

@@ -1,37 +1,22 @@
 # import geopandas as gpd
 # import matplotlib.pyplot as plt
-import os
 
 import numpy as np
 import pytest
-import requests
 from earthkit import data
 
 from polytope.datacube.backends.xarray import XArrayDatacube
 from polytope.engine.hullslicer import HullSlicer
 from polytope.polytope import Polytope, Request
 from polytope.shapes import Box, Select
 
+from .helper_functions import download_test_data
+
 
 class TestSlicingEra5Data:
     def setup_method(self, method):
         nexus_url = "https://get.ecmwf.int/test-data/polytope/test-data/era5-levels-members.grib"
-
-        local_directory = "./tests/data"
-
-        if not os.path.exists(local_directory):
-            os.makedirs(local_directory)
-
-        # Construct the full path for the local file
-        local_file_path = os.path.join(local_directory, "era5-levels-members.grib")
-
-        if not os.path.exists(local_file_path):
-            session = requests.Session()
-            response = session.get(nexus_url)
-            if response.status_code == 200:
-                # Save the downloaded data to the local file
-                with open(local_file_path, "wb") as f:
-                    f.write(response.content)
+        download_test_data(nexus_url, "era5-levels-members.grib")
 
         ds = data.from_source("file", "./tests/data/era5-levels-members.grib")
         array = ds.to_xarray().isel(step=0).t