fix some tests

Author: ungarj

mapchete_eo/cli/static_catalog.py

@@ -10,6 +10,9 @@
 from mapchete_eo.cli import options_arguments
 from mapchete_eo.platforms.sentinel2 import S2Metadata
 from mapchete_eo.platforms.sentinel2.types import Resolution
+from mapchete_eo.platforms.sentinel2.preconfigured_sources import (
+    DEPRECATED_ARCHIVES,
+)
 from mapchete_eo.search import STACSearchCatalog, STACStaticCatalog
 from mapchete_eo.search.base import CatalogSearcher
 from mapchete_eo.types import TimeRange
@@ -117,6 +120,8 @@ def get_catalog(
         else:
             raise ValueError("collection must be provided")
     elif known_archive:
-        raise NotImplementedError()
+        return STACSearchCatalog.from_collection_url(
+            DEPRECATED_ARCHIVES[known_archive]["collection"]
+        )
     else:
         raise TypeError("cannot determine catalog")

mapchete_eo/platforms/sentinel2/metadata_parser/s2metadata.py

@@ -157,12 +157,14 @@ def from_stac_item(
         item: pystac.Item,
         metadata_xml_asset_name: List[str] = ["metadata", "granule_metadata"],
         boa_offset_field: Optional[str] = None,
-        processing_baseline_field: Optional[str] = None,
+        processing_baseline_field: Union[str, List[str]] = [
+            "s2:processing_baseline",
+            "sentinel2:processing_baseline",
+            "processing:version",
+        ],
         **kwargs,
     ) -> S2Metadata:
-        metadata_xml_asset_name = metadata_xml_asset_name
-        if processing_baseline_field is None:
-            raise NotImplementedError()
+        # try to find path to metadata.xml
         for metadata_asset in metadata_xml_asset_name:
             if metadata_asset in item.assets:
                 metadata_path = MPath(item.assets[metadata_asset].href)
@@ -172,16 +174,28 @@ def from_stac_item(
                 f"could not find path to metadata XML file in assets: {', '.join(item.assets.keys())}"
             )
 
+        # maek path absolute
         if metadata_path.is_remote() or metadata_path.is_absolute():
             metadata_xml = metadata_path
         else:
             metadata_xml = MPath(item.self_href).parent / metadata_path
-        try:
-            processing_baseline = item.properties[processing_baseline_field]
-        except KeyError:
+
+        # try to find information on processing baseline version
+        for field in (
+            processing_baseline_field
+            if isinstance(processing_baseline_field, list)
+            else [processing_baseline_field]
+        ):
+            try:
+                processing_baseline = item.properties[field]
+                break
+            except KeyError:
+                pass
+        else:  # pragma: no cover
             raise KeyError(
                 f"could not find processing baseline version in item properties: {item.properties}"
             )
+
         return S2Metadata.from_metadata_xml(
             metadata_xml=metadata_xml,
             processing_baseline=processing_baseline,

mapchete_eo/platforms/sentinel2/preconfigured_sources/__init__.py

@@ -40,15 +40,15 @@
         "data_archive": "AWSCOG",
     },
     "S2AWS_JP2": {
-        "collection": "https://stac.dataspace.copernicus.eu/v1collections/sentinel-2-l2a",
+        "collection": "https://stac.dataspace.copernicus.eu/v1/collections/sentinel-2-l2a",
         "data_archive": "AWSJP2",
     },
     "S2CDSE_AWSJP2": {
-        "collection": "https://stac.dataspace.copernicus.eu/v1collections/sentinel-2-l2a",
+        "collection": "https://stac.dataspace.copernicus.eu/v1/collections/sentinel-2-l2a",
         "data_archive": "AWSJP2",
     },
     "S2CDSE_JP2": {
-        "collection": "https://stac.dataspace.copernicus.eu/v1collections/sentinel-2-l2a",
+        "collection": "https://stac.dataspace.copernicus.eu/v1/collections/sentinel-2-l2a",
     },
 }
 MetadataArchive = Literal["roda"]

mapchete_eo/search/stac_search.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import logging
 from datetime import datetime
 from functools import cached_property
@@ -202,6 +204,13 @@ def get_collections(self):
         for collection_name in self.collections:
             yield self.client.get_collection(collection_name)
 
+    @staticmethod
+    def from_collection_url(collection_url: str) -> STACSearchCatalog:
+        return STACSearchCatalog(
+            endpoint="/".join(collection_url.rstrip("/").split("/")[:-2]),
+            collections=[collection_url.rstrip("/").split("/")[-1]],
+        )
+
 
 class SpatialSearchChunks:
     bounds: Bounds

tests/test_cli.py

@@ -20,7 +20,9 @@ def test_s2_mask(s2_stac_json_half_footprint, tmp_mpath):
             str(out_path),
         ],
     )
-    assert result.exit_code == 0
+    if result.exit_code != 0:
+        raise result.exception
+
     assert out_path.exists()
     with rasterio_open(out_path) as src:
         assert src.read().any()
@@ -40,7 +42,9 @@ def test_s2_rgb(s2_stac_json_half_footprint, tmp_mpath):
             str(out_path),
         ],
     )
-    assert result.exit_code == 0
+    if result.exit_code != 0:
+        raise result.exception
+
     assert out_path.exists()
     with rasterio_open(out_path) as src:
         assert not src.read(masked=True).mask.all()
@@ -63,7 +67,9 @@ def test_s2_brdf(s2_stac_json_half_footprint, tmp_mpath):
             str(out_path),
         ],
     )
-    assert result.exit_code == 0
+    if result.exit_code != 0:
+        raise result.exception
+
     assert len(out_path.ls()) == 2
     for path in out_path.ls():
         with rasterio_open(path) as src:
@@ -100,5 +106,6 @@ def test_static_catalog(tmp_mpath, flag, value, collection):
     if collection:
         params.extend(["--collection", collection])
     result = runner.invoke(eo, params)
-    assert result.exit_code == 0
+    if result.exit_code != 0:
+        raise result.exception
     assert out_path.ls()