Issue #668 some final tweaks and tests

soxofaan · soxofaan · commit 632ad1ce87eb · 2025-01-31T10:10:03.000+01:00
diff --git a/openeo/rest/models/federation_extension.py b/openeo/rest/models/federation_extension.py
@@ -3,6 +3,7 @@
 
 _log = logging.getLogger(__name__)
 
+
 def get_backend_details(data: dict) -> Union[Dict[str, dict], None]:
     """
     Get federated backend details from capabilities document (``GET /``)
@@ -15,6 +16,7 @@ def get_backend_details(data: dict) -> Union[Dict[str, dict], None]:
 def get_federation_missing(data: dict, *, resource_name: str, auto_warn: bool = False) -> Union[List[str], None]:
     """
     Get "federation:missing" field from response data, if present.
+
     :param data: response data
     :param resource_name: name of the requested resource (listing)
     :param auto_warn: whether to automatically log a warning if missing federation components are detected.
diff --git a/openeo/rest/models/general.py b/openeo/rest/models/general.py
@@ -81,6 +81,7 @@ def ext_federation_missing(self, *, auto_warn: bool = False) -> Union[None, List
             data=self._data, resource_name="collection listing", auto_warn=auto_warn
         )
 
+
 class ProcessListingResponse(list):
     """
     Container for process metadata listing received
@@ -99,7 +100,6 @@ class ProcessListingResponse(list):
     .. versionadded:: 0.38.0
     """
 
-
     __slots__ = ["_data"]
 
     def __init__(self, response_data: dict):
@@ -134,6 +134,7 @@ def ext_federation_missing(self, *, auto_warn: bool = False) -> Union[None, List
             data=self._data, resource_name="process listing", auto_warn=auto_warn
         )
 
+
 class JobListingResponse(list):
     """
     Container for job metadata listing received
@@ -153,7 +154,6 @@ class JobListingResponse(list):
     .. versionadded:: 0.38.0
     """
 
-
     __slots__ = ["_data"]
 
     def __init__(self, response_data: dict):
@@ -186,6 +186,7 @@ def ext_federation_missing(self, *, auto_warn: bool = False) -> Union[None, List
             data=self._data, resource_name="job listing", auto_warn=auto_warn
         )
 
+
 class LogsResponse(list):
     """
     Container for job/service logs as received
@@ -208,7 +209,6 @@ class LogsResponse(list):
     .. versionadded:: 0.38.0
     """
 
-
     __slots__ = ["_data"]
 
     def __init__(self, response_data: dict, *, log_level: Optional[str] = None):
@@ -236,7 +236,6 @@ def accept_level(level: str) -> bool:
 
         self.ext_federation_missing(auto_warn=True)
 
-
     def _repr_html_(self):
         return render_component(component="logs", data=self)
 
diff --git a/tests/rest/models/test_federation_extension.py b/tests/rest/models/test_federation_extension.py
@@ -0,0 +1,57 @@
+import pytest
+
+from openeo.rest.models.federation_extension import (
+    get_backend_details,
+    get_federation_missing,
+)
+
+
+def test_get_backend_details():
+    assert get_backend_details({}) is None
+    assert get_backend_details(
+        {
+            "api_version": "1.2.0",
+            "backend_version": "1.1.2",
+            "stac_version": "1.0.0",
+            "type": "Catalog",
+            "id": "cool-eo-cloud",
+            "endpoints": [
+                {"path": "/collections", "methods": ["GET"]},
+            ],
+            "federation": {
+                "eoa": {"title": "EO Answers", "url": "https://eoe.test/go"},
+                "eob": {"title": "Beyond EO", "url": "https://eoeb.example.com"},
+            },
+        }
+    ) == {
+        "eoa": {"title": "EO Answers", "url": "https://eoe.test/go"},
+        "eob": {"title": "Beyond EO", "url": "https://eoeb.example.com"},
+    }
+
+
+def test_get_federation_missing():
+    assert get_federation_missing({}, resource_name="things") is None
+    assert get_federation_missing(
+        {
+            "things": ["apple", "banana"],
+            "federation:missing": ["veggies"],
+        },
+        resource_name="things",
+    ) == ["veggies"]
+
+
+@pytest.mark.parametrize(["auto_warn"], [[True], [False]])
+def test_get_federation_missing_auto_warn(auto_warn, caplog):
+    assert get_federation_missing(
+        {
+            "things": ["apple", "banana"],
+            "federation:missing": ["veggies"],
+        },
+        resource_name="things",
+        auto_warn=auto_warn,
+    ) == ["veggies"]
+
+    if auto_warn:
+        assert "Partial things: missing federation components: ['veggies']." in caplog.text
+    else:
+        assert caplog.text == ""
diff --git a/tests/rest/models/test_general.py b/tests/rest/models/test_general.py
@@ -1,6 +1,13 @@
 import pytest
 
-from openeo.rest.models.general import CollectionListingResponse, Link
+from openeo.rest.models.general import (
+    CollectionListingResponse,
+    JobListingResponse,
+    Link,
+    LogsResponse,
+    ProcessListingResponse,
+)
+from openeo.rest.models.logs import LogEntry
 
 
 class TestLink:
@@ -60,3 +67,123 @@ def test_links(self):
     def test_federation_missing(self, data, expected):
         collections = CollectionListingResponse(data)
         assert collections.ext_federation_missing() == expected
+
+
+class TestProcessListingResponse:
+    def test_basic(self):
+        data = {"processes": [{"id": "ndvi"}, {"id": "s2mask"}]}
+        processes = ProcessListingResponse(data)
+        assert processes == [{"id": "ndvi"}, {"id": "s2mask"}]
+        assert repr(processes) == "[{'id': 'ndvi'}, {'id': 's2mask'}]"
+
+    def test_links(self):
+        data = {
+            "processes": [{"id": "ndvi"}, {"id": "s2mask"}],
+            "links": [
+                {"rel": "self", "href": "https://openeo.test/processes"},
+                {"rel": "next", "href": "https://openeo.test/processes?page=2"},
+            ],
+        }
+        processes = ProcessListingResponse(data)
+        assert processes.links == [
+            Link(rel="self", href="https://openeo.test/processes"),
+            Link(rel="next", href="https://openeo.test/processes?page=2"),
+        ]
+
+    @pytest.mark.parametrize(
+        ["data", "expected"],
+        [
+            (
+                {"processes": [{"id": "ndvi"}], "federation:missing": ["wow"]},
+                ["wow"],
+            ),
+            (
+                {"processes": [{"id": "ndvi"}]},
+                None,
+            ),
+        ],
+    )
+    def test_federation_missing(self, data, expected):
+        processes = ProcessListingResponse(data)
+        assert processes.ext_federation_missing() == expected
+
+
+class TestJobListingResponse:
+    def test_basic(self):
+        data = {"jobs": [{"id": "job-01"}, {"id": "job-02"}]}
+        jobs = JobListingResponse(data)
+        assert jobs == [{"id": "job-01"}, {"id": "job-02"}]
+        assert repr(jobs) == "[{'id': 'job-01'}, {'id': 'job-02'}]"
+
+    def test_links(self):
+        data = {
+            "jobs": [{"id": "job-01"}, {"id": "job-02"}],
+            "links": [
+                {"rel": "self", "href": "https://openeo.test/jobs"},
+                {"rel": "next", "href": "https://openeo.test/jobs?page=2"},
+            ],
+        }
+        jobs = JobListingResponse(data)
+        assert jobs.links == [
+            Link(rel="self", href="https://openeo.test/jobs"),
+            Link(rel="next", href="https://openeo.test/jobs?page=2"),
+        ]
+
+    @pytest.mark.parametrize(
+        ["data", "expected"],
+        [
+            (
+                {"jobs": [{"id": "job-01"}], "federation:missing": ["wow"]},
+                ["wow"],
+            ),
+            (
+                {"jobs": [{"id": "job-01"}]},
+                None,
+            ),
+        ],
+    )
+    def test_federation_missing(self, data, expected):
+        jobs = JobListingResponse(data)
+        assert jobs.ext_federation_missing() == expected
+
+
+class TestLogsResponse:
+    def test_basic(self):
+        data = {"logs": [{"id": "log-01", "level": "info", "message": "hello"}]}
+        logs = LogsResponse(data)
+        assert logs == [{"id": "log-01", "level": "info", "message": "hello"}]
+        assert logs == [LogEntry(id="log-01", level="info", message="hello")]
+        assert logs.logs == [{"id": "log-01", "level": "info", "message": "hello"}]
+        assert logs.logs == [LogEntry(id="log-01", level="info", message="hello")]
+        assert repr(logs) == "[{'id': 'log-01', 'level': 'info', 'message': 'hello'}]"
+
+    def test_links(self):
+        data = {
+            "logs": [{"id": "log-01", "level": "info", "message": "hello"}],
+            "links": [
+                {"rel": "self", "href": "https://openeo.test/logs"},
+                {"rel": "next", "href": "https://openeo.test/logs?page=2"},
+            ],
+        }
+        logs = LogsResponse(data)
+        assert logs.links == [
+            Link(rel="self", href="https://openeo.test/logs"),
+            Link(rel="next", href="https://openeo.test/logs?page=2"),
+        ]
+
+    @pytest.mark.parametrize(
+        ["data", "expected"],
+        [
+            (
+                {"logs": [{"id": "log-01", "level": "info", "message": "hello"}], "federation:missing": ["wow"]},
+                ["wow"],
+            ),
+            (
+                {"logs": [{"id": "log-01", "level": "info", "message": "hello"}]},
+                None,
+            ),
+        ],
+    )
+    def test_federation_missing(self, data, expected):
+        logs = LogsResponse(data)
+        assert logs.ext_federation_missing() == expected
