Add root/parent/self links to collection metadata if not already

(feature moved from aggregator, originally introduced under Open-EO/openeo-aggregator#5)

openeo_driver/_version.py

@@ -1 +1 @@
-__version__ = "0.66.0a1"
+__version__ = "0.67.0a1"

openeo_driver/dummy/dummy_backend.py

@@ -4,6 +4,7 @@
 from functools import lru_cache
 from pathlib import Path
 from typing import List, Dict, Union, Tuple, Optional, Iterable, Any, Sequence
+import unittest.mock
 from unittest.mock import Mock
 
 import flask
@@ -558,7 +559,6 @@ class DummyCatalog(CollectionCatalog):
     def __init__(self):
         super().__init__(all_metadata=self._COLLECTIONS)
 
-
     def load_collection(self, collection_id: str, load_params: LoadParameters, env: EvalEnv) -> DummyDataCube:
         return self._load_collection_cached(collection_id,load_params,WhiteListEvalEnv(env,[]))
 
@@ -574,6 +574,10 @@ def _load_collection_cached(self, collection_id: str, load_params: LoadParameter
         _collections[collection_id] = image_collection
         return image_collection
 
+    def patch_collections(self, extra_collections: dict):
+        """Context manager to (temporarily) add a collection with metadata to the catalog."""
+        return unittest.mock.patch.dict(self._catalog, extra_collections)
+
 
 class DummyProcessing(ConcreteProcessing):
     def extra_validation(

openeo_driver/views.py

@@ -1704,7 +1704,19 @@ def _normalize_collection_metadata(metadata: dict, api_version: ComparableVersio
             STAC_EXTENSION.EO,
         ],
     )
-    metadata.setdefault("links", [])
+
+    links = metadata.get("links", [])
+    link_rels = {lk.get("rel") for lk in links}
+    if "root" not in link_rels and flask.current_app:
+        links.append({"rel": "root", "href": url_for("openeo.collections", _external=True)})
+    if "parent" not in link_rels and flask.current_app:
+        links.append({"rel": "parent", "href": url_for("openeo.collections", _external=True)})
+    if "self" not in link_rels and flask.current_app:
+        links.append(
+            {"rel": "self", "href": url_for("openeo.collection_by_id", collection_id=collection_id, _external=True)}
+        )
+    metadata["links"] = links
+
     metadata.setdefault("description", collection_id)
     metadata.setdefault("license", "proprietary")
     # Warn about missing fields where simple defaults are not feasible.
@@ -1748,10 +1760,13 @@ def collections():
             _normalize_collection_metadata(metadata=m, api_version=requested_api_version(), full=False)
             for m in backend_implementation.catalog.get_all_metadata()
         ]
-        return jsonify({
-            'collections': metadata,
-            'links': []
-        })
+        return jsonify(
+            {
+                "collections": metadata,
+                # TODO: how to allow customizing this "links" field?
+                "links": [],
+            }
+        )
 
     @api_endpoint
     @blueprint.route('/collections/<collection_id>', methods=['GET'])

tests/test_views.py

@@ -64,7 +64,7 @@
         "1.1.0",
     ]
 )
-def api_version(request):
+def api_version(request) -> str:
     return request.param
 
 
@@ -952,11 +952,11 @@ def test_strip_private_fields(self, api):
         metadata = api.get('/collections/S2_FOOBAR').assert_status_code(200).json
         assert '_private' not in metadata
 
-    def test_collections_detail_invalid_collection(self, api):
+    def test_collection_full_metadata_invalid_collection(self, api):
         error = api.get('/collections/FOOBOO').assert_error(404, "CollectionNotFound").json
         assert error["message"] == "Collection 'FOOBOO' does not exist."
 
-    def test_collections_detail(self, api):
+    def test_collection_full_metadata(self, api, api_version):
         collection = api.get('/collections/S2_FOOBAR').assert_status_code(200).json
         assert collection['id'] == 'S2_FOOBAR'
         assert collection['description'] == 'S2_FOOBAR'
@@ -992,11 +992,66 @@ def test_collections_detail(self, api):
                 'temporal': ['2019-01-01T00:00:00Z', None]
             }
 
-    def test_collections_detail_caching(self, api):
+        assert collection["links"] == [
+            {"rel": "root", "href": f"http://oeo.net/openeo/{api_version}/collections"},
+            {"rel": "parent", "href": f"http://oeo.net/openeo/{api_version}/collections"},
+            {"rel": "self", "href": f"http://oeo.net/openeo/{api_version}/collections/S2_FOOBAR"},
+        ]
+
+    @pytest.mark.parametrize(
+        ["orig", "expected"],
+        [
+            (
+                [],
+                lambda api_version: [
+                    {"rel": "root", "href": f"http://oeo.net/openeo/{api_version}/collections"},
+                    {"rel": "parent", "href": f"http://oeo.net/openeo/{api_version}/collections"},
+                    {"rel": "self", "href": f"http://oeo.net/openeo/{api_version}/collections/S2_WITH_LINKS"},
+                ],
+            ),
+            (
+                [
+                    {"rel": "root", "href": "https://s2.test/foobar"},
+                    {"rel": "about", "href": "https://s2.test/about"},
+                ],
+                lambda api_version: [
+                    {"rel": "root", "href": "https://s2.test/foobar"},
+                    {"rel": "about", "href": "https://s2.test/about"},
+                    {"rel": "parent", "href": f"http://oeo.net/openeo/{api_version}/collections"},
+                    {"rel": "self", "href": f"http://oeo.net/openeo/{api_version}/collections/S2_WITH_LINKS"},
+                ],
+            ),
+            (
+                [
+                    {"rel": "root", "href": "https://s2.test/foobar"},
+                    {"rel": "parent", "href": "https://s2.test/about"},
+                    {"rel": "self", "href": "https://s2.test/s2"},
+                ],
+                lambda api_version: [
+                    {"rel": "root", "href": "https://s2.test/foobar"},
+                    {"rel": "parent", "href": "https://s2.test/about"},
+                    {"rel": "self", "href": "https://s2.test/s2"},
+                ],
+            ),
+        ],
+    )
+    def test_collection_full_metadata_links(self, api, api_version, backend_implementation, orig, expected):
+        extra_collections = {
+            "S2_WITH_LINKS": {
+                "id": "S2_WITH_LINKS",
+                "links": orig,
+            }
+        }
+        with backend_implementation.catalog.patch_collections(extra_collections):
+            collection = api.get("/collections/S2_WITH_LINKS").assert_status_code(200).json
+        assert collection["id"] == "S2_WITH_LINKS"
+        assert collection["links"] == expected(api_version)
+
+    def test_collection_full_metadata_caching(self, api):
         resp = api.get('/collections/S2_FOOBAR').assert_status_code(200)
         assert resp.headers["Cache-Control"] == "max-age=900, public"
 
-    def test_collections_detail_invalid_caching(self, api):
+    def test_collection_full_metadata_invalid_caching(self, api):
         resp = api.get('/collections/FOOBOO').assert_error(404, "CollectionNotFound")
         assert "Cache-Control" not in resp.headers