TestDataLoader: load_text and more powerful preprocess

Author: soxofaan

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `openeo.testing.io.TestDataLoader`: unit test utility to compactly load (and optionally preprocess) tests data (text/JSON/...)
+
 ### Changed
 
 - `DataCube.apply_dimension()`: not explicitly specifying the `dimension` argument is deprecated and will trigger warnings ([#774](https://github.com/Open-EO/openeo-python-client/issues/774))

openeo/_version.py

@@ -1 +1 @@
-__version__ = "0.42.0a1"
+__version__ = "0.42.0a2"

openeo/testing/__init__.py

@@ -2,36 +2,6 @@
 Utilities for testing of openEO client workflows.
 """
 
-import json
-from pathlib import Path
-from typing import Callable, Optional, Union
 
-
-class TestDataLoader:
-    """
-    Helper to resolve paths to test data files, load them as JSON, optionally preprocess them, etc.
-
-    It's intended to be used as a pytest fixture, e.g. from ``conftest.py``:
-
-    .. code-block:: python
-
-        @pytest.fixture
-        def test_data() -> TestDataLoader:
-            return TestDataLoader(root=Path(__file__).parent / "data")
-
-    .. versionadded:: 0.30.0
-    """
-
-    def __init__(self, root: Union[str, Path]):
-        self.data_root = Path(root)
-
-    def get_path(self, filename: Union[str, Path]) -> Path:
-        """Get absolute path to a test data file"""
-        return self.data_root / filename
-
-    def load_json(self, filename: Union[str, Path], preprocess: Optional[Callable[[str], str]] = None) -> dict:
-        """Parse data from a test JSON file"""
-        data = self.get_path(filename).read_text(encoding="utf8")
-        if preprocess:
-            data = preprocess(data)
-        return json.loads(data)
+# Legacy import for backwards compatibility
+from openeo.testing.io import TestDataLoader

openeo/testing/io.py

@@ -0,0 +1,103 @@
+import json
+import re
+from pathlib import Path
+from typing import Callable, Optional, Union
+
+
+class TestDataLoader:
+    """
+    Test data helper: easily resolve paths to test data files,
+    load them as bytes, text, JSON,
+    optionally preprocess them, etc.
+
+    It's intended to be used as a pytest fixture, e.g. from ``conftest.py``:
+
+    .. code-block:: python
+
+        @pytest.fixture
+        def test_data() -> TestDataLoader:
+            return TestDataLoader(root=Path(__file__).parent / "data")
+
+    .. versionadded:: 0.30.0
+
+    .. versionchanged:: 0.42.0
+        Moved to ``openeo.testing.io``.
+        Added ``load_bytes()`` and ``load_text()``.
+        Improved ``preprocess``: can now be a replacement dict (with regex support).
+    """
+
+    def __init__(self, root: Union[str, Path]):
+        self.data_root = Path(root)
+
+    def get_path(self, filename: Union[str, Path]) -> Path:
+        """Get absolute path to a test data file"""
+        return self.data_root / filename
+
+    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]:
+        """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)
+                    else:
+                        raise ValueError(key)
+                return text
+
+            return replace
+        else:
+            return preprocess
+
+    def load_text(
+        self,
+        filename: Union[str, Path],
+        *,
+        preprocess: Union[None, dict, Callable[[str], str]] = None,
+        encoding: str = "utf8",
+    ) -> str:
+        """
+        Load text file, optionally with some text based preprocessing
+
+        :param filename: Path to the file relative to the test data root
+        :param preprocess: Optional preprocessing to do on the text, given as
+
+            - Callable that takes a string and returns a string
+            - Dictionary mapping needles to replacements.
+              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 encoding: Encoding to use when reading the file
+        """
+        text = self.get_path(filename).read_text(encoding=encoding)
+        text = self._get_preprocess(preprocess)(text)
+        return text
+
+    def load_json(
+        self,
+        filename: Union[str, Path],
+        *,
+        preprocess: Union[None, dict, Callable[[str], str]] = None,
+    ) -> dict:
+        """
+        Load data from a JSON file, optionally with some text based preprocessing
+
+        :param filename: Path to the file relative to the test data root
+        :param preprocess: Optional preprocessing to do on the text, given as
+
+            - Callable that takes a string and returns a string
+            - Dictionary mapping needles to replacements.
+              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)
+        """
+        raw = self.get_path(filename).read_text(encoding="utf8")
+        raw = self._get_preprocess(preprocess)(raw)
+        return json.loads(raw)

tests/conftest.py

@@ -9,7 +9,7 @@
 )
 
 
-from openeo.testing import TestDataLoader
+from openeo.testing.io import TestDataLoader
 from openeo.util import ensure_dir
 
 pytest_plugins = "pytester"

tests/testing/test_io.py

@@ -0,0 +1,75 @@
+import json
+import re
+from pathlib import Path
+
+import pytest
+
+from openeo.testing.io import TestDataLoader
+
+
+class TestTestDataLoader:
+    def test_get_path(self, tmp_path):
+        path = tmp_path / "hello" / "world.txt"
+        loader = TestDataLoader(root=tmp_path)
+        assert loader.get_path("hello/world.txt") == path
+        assert loader.get_path(Path("hello/world.txt")) == path
+
+    def test_load_bytes(self, tmp_path):
+        path = tmp_path / "hello" / "world.txt"
+        path.parent.mkdir(parents=True)
+        path.write_bytes(b"Hello W\x00rld")
+
+        loader = TestDataLoader(root=tmp_path)
+        assert loader.load_bytes("hello/world.txt") == b"Hello W\x00rld"
+
+    @pytest.mark.parametrize(
+        ["preprocess", "expected"],
+        [
+            (None, "Hello, World!"),
+            (lambda s: s.lower(), "hello, world!"),
+            (lambda s: s.replace("World", "Earth"), "Hello, Earth!"),
+            ({"World": "Earth"}, "Hello, Earth!"),
+            ({"Hello": "Greetings", "World": "Terra"}, "Greetings, Terra!"),
+            ({re.compile("l+"): "|_"}, "He|_o, Wor|_d!"),
+            ({re.compile("([A-Z])"): r"\1\1\1"}, "HHHello, WWWorld!"),
+        ],
+    )
+    def test_load_text(self, tmp_path, preprocess, expected):
+        (tmp_path / "hello.txt").write_text("Hello, World!", encoding="utf8")
+
+        loader = TestDataLoader(root=tmp_path)
+        assert loader.load_text("hello.txt", preprocess=preprocess) == expected
+
+    @pytest.mark.parametrize(
+        ["preprocess", "expected"],
+        [
+            (None, {"salutation": "Hello", "target": "World"}),
+            (lambda s: s.upper(), {"SALUTATION": "HELLO", "TARGET": "WORLD"}),
+            (
+                lambda s: s.replace("World", "Terra"),
+                {"salutation": "Hello", "target": "Terra"},
+            ),
+            (
+                lambda s: s.replace('"World"', '["Terra","Earth"]'),
+                {"salutation": "Hello", "target": ["Terra", "Earth"]},
+            ),
+            (
+                {"World": "Earth", "salutation": "say", "target": "to"},
+                {"say": "Hello", "to": "Earth"},
+            ),
+            (
+                {"Hello": "Greetings", '"World"': '["Terra","Earth"]'},
+                {"salutation": "Greetings", "target": ["Terra", "Earth"]},
+            ),
+            (
+                {re.compile("([aeoiu])"): r"\1\1\1"},
+                {"saaaluuutaaatiiiooon": "Heeellooo", "taaargeeet": "Wooorld"},
+            ),
+        ],
+    )
+    def test_load_json(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)
+        assert loader.load_json("data.json", preprocess=preprocess) == expected