Move get layer names func to utils

georest/__init__.py

@@ -11,7 +11,6 @@
 import datetime as dt
 import logging
 import os
-import itertools
 
 try:
     import requests
@@ -415,22 +414,6 @@ def delete_granule(cat, workspace, store, fname):
         logger.info("Granule '%s' removed", fname)
 
 
-def get_layers_for_delete_granules(config):
-    """Get list of layers for deleting granules."""
-    if config.get("layer_id", False):
-        layers = list(config["layers"].values())
-    elif config.get("layer_name_template", False) and config.get("delete_granule_layer_options", False):
-        keys = sorted(config["delete_granule_layer_options"].keys())
-        combinations = list(itertools.product(*[config["delete_granule_layer_options"][k] for k in keys]))
-        options = [dict(zip(keys, l)) for l in combinations]
-        layers = [config["layer_name_template"].format(**opt) for opt in options]
-    else:
-        raise ValueError(
-            "Either 'layer_id' or 'layer_name_template' (with 'delete_granule_layer_options') must be defined in config"
-        )
-    return layers
-
-
 def delete_old_files_from_mosaics_and_fs(config):
     """Delete a file from image mosaic.
 
@@ -441,7 +424,7 @@ def delete_old_files_from_mosaics_and_fs(config):
     workspace = config["workspace"]
     max_age = dt.datetime.utcnow() - dt.timedelta(minutes=config["max_age"])
 
-    for store in get_layers_for_delete_granules(config):
+    for store in utils.get_layers_for_delete_granules(config):
         store_obj = cat.get_store(store, workspace)
         logger.debug("Getting coverage for %s", store)
         coverage = get_layer_coverage(cat, store, store_obj)

georest/tests/test_utils.py

@@ -364,3 +364,28 @@ def test_convert_file_path_keep_subpath_inverse():
     res = convert_file_path(config, "/geoserver/internal/path/subpath/file.tif", inverse=True, keep_subpath=True)
 
     assert res == "/external/path/subpath/file.tif"
+
+
+def test_get_layers_for_delete_granules():
+    """Test the function that provides layer names for delete granules."""
+    from georest.utils import get_layers_for_delete_granules
+
+    # Case 1: layer_id and layers provided
+    config = {"layer_id": "layer_id", "layers": {"layer_1": "layer_name_1", "layer_2": "layer_name_2"}}
+    res = get_layers_for_delete_granules(config)
+
+    assert res == ["layer_name_1", "layer_name_2"]
+
+    # Case 2: layer_name_template and delete_granule_layer_options provided
+    config = {
+        "layer_name_template": "{opt1}_{opt2}",
+        "delete_granule_layer_options": {"opt1": ["option1_1", "option1_2"], "opt2": ["layer_name_2"]},
+    }
+    res = get_layers_for_delete_granules(config)
+
+    assert res == ["option1_1_layer_name_2", "option1_2_layer_name_2"]
+
+    # Case 3: no layer_id or layer_name_template provided
+    config = {}
+    with pytest.raises(ValueError):
+        get_layers_for_delete_granules(config)

georest/utils.py

@@ -15,6 +15,7 @@
 import shutil
 import tempfile
 import zipfile
+import itertools
 
 import trollsift
 import yaml
@@ -129,6 +130,22 @@ def get_exposed_layer_directories(config):
     return dirs
 
 
+def get_layers_for_delete_granules(config):
+    """Get list of layers for deleting granules."""
+    if config.get("layer_id", False):
+        layers = list(config["layers"].values())
+    elif config.get("layer_name_template", False) and config.get("delete_granule_layer_options", False):
+        keys = sorted(config["delete_granule_layer_options"].keys())
+        combinations = list(itertools.product(*[config["delete_granule_layer_options"][k] for k in keys]))
+        options = [dict(zip(keys, l)) for l in combinations]
+        layers = [config["layer_name_template"].format(**opt) for opt in options]
+    else:
+        raise ValueError(
+            "Either 'layer_id' or 'layer_name_template' (with 'delete_granule_layer_options') must be defined in config"
+        )
+    return layers
+
+
 def write_wkt(config, image_fname):
     """Write WKT text besides the image file.