TestDataLoader: add verify_dict_keys feature (on by default)

Author: soxofaan

openeo/testing/io.py

@@ -3,6 +3,11 @@
 from pathlib import Path
 from typing import Callable, Optional, Union
 
+from openeo.util import repr_truncate
+
+
+class PreprocessError(ValueError):
+    pass
 
 class TestDataLoader:
     """
@@ -36,20 +41,28 @@ def get_path(self, filename: Union[str, Path]) -> Path:
     def load_bytes(self, filename: Union[str, Path]) -> bytes:
         return self.get_path(filename).read_bytes()
 
-    def _get_preprocess(self, preprocess: Union[None, dict, Callable[[str], str]]) -> Callable[[str], str]:
+    def _get_preprocess(
+        self,
+        preprocess: Union[None, dict, Callable[[str], str]],
+        verify_dict_keys: bool = True,
+    ) -> Callable[[str], str]:
         """Normalize preprocess argument to a callable"""
         if preprocess is None:
             return lambda x: x
         elif isinstance(preprocess, dict):
 
             def replace(text: str) -> str:
-                for key, value in preprocess.items():
-                    if isinstance(key, re.Pattern):
-                        text = key.sub(value, text)
-                    elif isinstance(key, str):
-                        text = text.replace(key, value)
+                for needle, replacement in preprocess.items():
+                    if isinstance(needle, re.Pattern):
+                        if verify_dict_keys and not needle.search(text):
+                            raise PreprocessError(f"{needle!r} not found in {repr_truncate(text, width=265)}")
+                        text = needle.sub(repl=replacement, string=text)
+                    elif isinstance(needle, str):
+                        if verify_dict_keys and needle not in text:
+                            raise PreprocessError(f"{needle!r} not found in {repr_truncate(text, width=256)}")
+                        text = text.replace(needle, replacement)
                     else:
-                        raise ValueError(key)
+                        raise ValueError(needle)
                 return text
 
             return replace
@@ -61,6 +74,7 @@ def load_text(
         filename: Union[str, Path],
         *,
         preprocess: Union[None, dict, Callable[[str], str]] = None,
+        verify_dict_keys: bool = True,
         encoding: str = "utf8",
     ) -> str:
         """
@@ -74,17 +88,20 @@ def load_text(
               Needle can be a simple string that will be replaced with the replacement value,
               or it can be a ``re.Pattern`` that will be used in ``re.sub()`` style
               (which supports group references, e.g. "\1" for first group in match)
+        :param verify_dict_keys: when ``preprocess`` is specified as dict:
+            whether to verify that the keys actually exist in the text before replacing them.
         :param encoding: Encoding to use when reading the file
         """
         text = self.get_path(filename).read_text(encoding=encoding)
-        text = self._get_preprocess(preprocess)(text)
+        text = self._get_preprocess(preprocess, verify_dict_keys=verify_dict_keys)(text)
         return text
 
     def load_json(
         self,
         filename: Union[str, Path],
         *,
         preprocess: Union[None, dict, Callable[[str], str]] = None,
+        verify_dict_keys: bool = True,
     ) -> dict:
         """
         Load data from a JSON file, optionally with some text based preprocessing
@@ -97,7 +114,9 @@ def load_json(
               Needle can be a simple string that will be replaced with the replacement value,
               or it can be a ``re.Pattern`` that will be used in ``re.sub()`` style
               (which supports group references, e.g. "\1" for first group in match)
+        :param verify_dict_keys: when ``preprocess`` is specified as dict:
+            whether to verify that the keys actually exist in the text before replacing them.
         """
         raw = self.get_path(filename).read_text(encoding="utf8")
-        raw = self._get_preprocess(preprocess)(raw)
+        raw = self._get_preprocess(preprocess, verify_dict_keys=verify_dict_keys)(raw)
         return json.loads(raw)

tests/testing/test_io.py

@@ -4,7 +4,7 @@
 
 import pytest
 
-from openeo.testing.io import TestDataLoader
+from openeo.testing.io import PreprocessError, TestDataLoader
 
 
 class TestTestDataLoader:
@@ -40,6 +40,23 @@ def test_load_text(self, tmp_path, preprocess, expected):
         loader = TestDataLoader(root=tmp_path)
         assert loader.load_text("hello.txt", preprocess=preprocess) == expected
 
+    @pytest.mark.parametrize(
+        ["preprocess", "expected"],
+        [
+            ({"apple": "banana"}, "'apple' not found in 'Hello, World!'"),
+            ({re.compile("ap+le"): "banana"}, "re.compile('ap+le') not found in 'Hello, World!'"),
+        ],
+    )
+    def test_load_text_missing_key(self, tmp_path, preprocess, expected):
+        (tmp_path / "hello.txt").write_text("Hello, World!", encoding="utf8")
+
+        loader = TestDataLoader(root=tmp_path)
+        with pytest.raises(PreprocessError, match=re.escape(expected)):
+            loader.load_text("hello.txt", preprocess=preprocess)
+
+        # Don't fail when verify_dict_keys=False
+        assert loader.load_text("hello.txt", preprocess=preprocess, verify_dict_keys=False) == "Hello, World!"
+
     @pytest.mark.parametrize(
         ["preprocess", "expected"],
         [
@@ -73,3 +90,30 @@ def test_load_json(self, tmp_path, preprocess, expected):
 
         loader = TestDataLoader(root=tmp_path)
         assert loader.load_json("data.json", preprocess=preprocess) == expected
+
+    @pytest.mark.parametrize(
+        ["preprocess", "expected"],
+        [
+            (
+                {"apple": "banana"},
+                '\'apple\' not found in \'{"salutation": "Hello", "target": "World"}\'',
+            ),
+            (
+                {re.compile("ap+le"): "banana"},
+                're.compile(\'ap+le\') not found in \'{"salutation": "Hello", "target": "World"}\'',
+            ),
+        ],
+    )
+    def test_load_json_missing_key(self, tmp_path, preprocess, expected):
+        with (tmp_path / "data.json").open("w", encoding="utf8") as f:
+            json.dump({"salutation": "Hello", "target": "World"}, f)
+
+        loader = TestDataLoader(root=tmp_path)
+        with pytest.raises(PreprocessError, match=re.escape(expected)):
+            loader.load_json("data.json", preprocess=preprocess)
+
+        # Don't fail when verify_dict_keys=False
+        assert loader.load_json("data.json", preprocess=preprocess, verify_dict_keys=False) == {
+            "salutation": "Hello",
+            "target": "World",
+        }