Add tests for STAC Collections #2, add missing files

tests/conftest.py

@@ -0,0 +1,18 @@
+from pytest import Parser
+
+
+def pytest_addoption(parser: Parser) -> None:
+    parser.addoption(
+        "--withcog",
+        action="store_true",
+        default=False,
+        help="also test cog conversion (slow)",
+    )
+
+
+def pytest_generate_tests(metafunc):
+    # This is called for every test. Only get/set command line arguments
+    # if the argument is specified in the list of test "fixturenames".
+    option_value = metafunc.config.option.withcog
+    if "withcog" in metafunc.fixturenames and option_value is True:
+        metafunc.parametrize("withcog", [option_value])

tests/constants.py

@@ -0,0 +1,55 @@
+SRC_FOLDER = "./tests/data-files/"
+TEST_FILES = [
+    "C3S-LC-L4-LCCS-Map-300m-P1Y-2020-v2.1.1",
+    "C3S-LC-L4-LCCS-Map-300m-P1Y-2016-v2.1.1",
+    "ESACCI-LC-L4-LCCS-Map-300m-P1Y-2015-v2.0.7cds",
+    "ESACCI-LC-L4-LCCS-Map-300m-P1Y-1992-v2.0.7cds",
+]
+
+# Collection
+TITLE = "ESA Climate Change Initiative Land Cover"
+START_DATETIME = "1992-01-01T00:00:00Z"
+END_DATETIME = "2020-12-31T23:59:59Z"
+BBOX = [-180.0, -90.0, 180.0, 90.0]
+
+# Item
+GEOMETRY = {
+    "type": "Polygon",
+    "coordinates": [[[-180, -90], [180, -90], [180, 90], [-180, 90], [-180, -90]]],
+}
+
+# Common
+GSD = 300
+V1 = "2.0.7cds"
+V2 = "2.1.1"
+VERSIONS = [V1, V2]
+
+# Projection
+EPSG_CODE = 4326
+
+# Scientific
+DOI = "10.24381/cds.006f2c9a"
+
+# Assets
+COG_MEDIA_TYPE = "image/tiff; application=geotiff; profile=cloud-optimized"
+COG_ROLES_DATA = ["data", "cloud-optimized"]
+COG_ROLES_QUALITY = ["quality", "cloud-optimized"]
+
+NETCDF_TITLE = "Original netCDF 4 file"
+NETCDF_MEDIA_TYPE = "application/netcdf"
+NETCDF_ROLES = ["data", "quality", "source"]
+NETCDF_KEY = "netcdf"
+
+DATA_VARIABLES = [
+    "change_count",
+    "current_pixel_state",
+    "lccs_class",
+    "observation_count",
+    "processed_flag",
+]
+
+TABLES = [
+    "current_pixel_state",
+    "processed_flag",
+    "lccs_class",
+]

tests/test_commands.py

@@ -10,22 +10,7 @@
 
 from stactools.esa_cci_lc.commands import create_esaccilc_command
 
-SRC_FOLDER = "./tests/data-files/"
-
-TEST_FILES = [
-    "C3S-LC-L4-LCCS-Map-300m-P1Y-2020-v2.1.1",
-    "C3S-LC-L4-LCCS-Map-300m-P1Y-2016-v2.1.1",
-    "ESACCI-LC-L4-LCCS-Map-300m-P1Y-2015-v2.0.7cds",
-    "ESACCI-LC-L4-LCCS-Map-300m-P1Y-1992-v2.0.7cds",
-]
-
-COG_KEYS = [
-    "change_count",
-    "current_pixel_state",
-    "lccs_class",
-    "observation_count",
-    "processed_flag",
-]
+from . import constants
 
 
 class CommandsTest(CliTestCase):
@@ -34,7 +19,7 @@ def create_subcommand_functions(self) -> List[Callable[[Group], Command]]:
 
     def test_create_collection(self) -> None:
         with TemporaryDirectory() as tmp_dir:
-            src_file = os.path.join(SRC_FOLDER, "collection.json")
+            src_file = os.path.join(constants.SRC_FOLDER, "collection.json")
             destination = os.path.join(tmp_dir, "collection.json")
 
             result = self.run_command(
@@ -68,18 +53,22 @@ def test_create_collection(self) -> None:
             self.assertEqual(diff, {})
 
     def test_create_item(self, withcog: bool = False) -> None:
-        for id in TEST_FILES:
+        for id in constants.TEST_FILES:
             with self.subTest(id=id):
                 with TemporaryDirectory() as tmp_dir:
                     src_data_filename = f"{id}.nc"
                     stac_filename = f"{id}.json"
 
-                    src_collection = os.path.join(SRC_FOLDER, "collection.json")
-                    src_data_file = os.path.join(SRC_FOLDER, src_data_filename)
+                    src_collection = os.path.join(
+                        constants.SRC_FOLDER, "collection.json"
+                    )
+                    src_data_file = os.path.join(
+                        constants.SRC_FOLDER, src_data_filename
+                    )
                     dest_data_file = os.path.join(tmp_dir, src_data_filename)
                     shutil.copyfile(src_data_file, dest_data_file)
 
-                    src_stac = os.path.join(SRC_FOLDER, stac_filename)
+                    src_stac = os.path.join(constants.SRC_FOLDER, stac_filename)
                     dest_stac = os.path.join(tmp_dir, stac_filename)
 
                     cmd = (
@@ -110,7 +99,7 @@ def test_create_item(self, withcog: bool = False) -> None:
                     if withcog:
                         del truth_item["properties"]["classification:classes"]
                     else:
-                        for key in COG_KEYS:
+                        for key in constants.DATA_VARIABLES:
                             del truth_item["assets"][key]
 
                     diff = DeepDiff(

tests/test_stac.py

@@ -1,21 +1,91 @@
 import unittest
 
+# from datetime import datetime, timezone
+from typing import Any, Dict, List
+
 from stactools.esa_cci_lc import stac
 
+from . import constants
+
+THUMBNAIL = "https://example.com/thumb.png"
+
+TEST_COLLECTIONS: List[Dict[str, Any]] = [
+    {
+        "id": "my-id",
+        "thumbnail": THUMBNAIL,
+        "start_time": "2017-01-01T00:00:00.000Z",
+        "end_time": "2019-12-31T23:59:59.999Z",
+    },
+    {
+        "nocog": True,
+    },
+    {
+        "nonetcdf": True,
+    },
+]
+
 
 class StacTest(unittest.TestCase):
     def test_create_collection(self) -> None:
-        # Write tests for each for the creation of a STAC Collection
-        # Create the STAC Collection...
-        collection = stac.create_collection()
-        collection.set_self_href("")
+        for test_data in TEST_COLLECTIONS:
+            with self.subTest(test_data=test_data):
+                id: str = test_data["id"] if "id" in test_data else "esa-cci-lc"
+                nocog: bool = test_data["nocog"] if "nocog" in test_data else False
+                nonetcdf: bool = (
+                    test_data["nonetcdf"] if "nonetcdf" in test_data else False
+                )
 
-        # Check that it has some required attributes
-        self.assertEqual(collection.id, "esa-cci-lc")
-        # self.assertEqual(collection.other_attr...
+                collection = stac.create_collection(**test_data)
+                collection.set_self_href("")
+                collection.validate()
+                collection_dict = collection.to_dict()
 
-        # Validate
-        collection.validate()
+                self.assertEqual(collection.id, id)
+
+                self.assertEqual(collection_dict["sci:doi"], constants.DOI)
+
+                self.assertTrue("summaries" in collection_dict)
+                summaries = collection_dict["summaries"]
+                self.assertEqual(summaries["gsd"], [constants.GSD])
+                self.assertEqual(summaries["esa_cci_lc:version"], constants.VERSIONS)
+                self.assertIn("classification:classes", summaries)
+                self.assertEqual(summaries["proj:epsg"], [constants.EPSG_CODE])
+
+                self.assertTrue("item_assets" in collection_dict)
+                assets: Dict[str, Dict[str, Any]] = collection_dict["item_assets"]
+
+                asset_count = 6
+                if nocog:
+                    asset_count -= 5
+                if nonetcdf:
+                    asset_count -= 1
+                self.assertEqual(len(assets), asset_count)
+
+                # Check COG assets
+                for key in constants.DATA_VARIABLES:
+                    self.assertEqual(key in assets, not nocog)
+                    if not nocog:
+                        asset = assets[key]
+                        self.assertNotIn("href", asset)
+                        self.assertIn("description", asset)
+                        self.assertEqual(asset["type"], constants.COG_MEDIA_TYPE)
+                        self.assertIn("cloud-optimized", asset["roles"])
+                        if key in constants.TABLES:
+                            self.assertIn("classification:classes", asset)
+                        else:
+                            self.assertNotIn("classification:classes", asset)
+
+                # Check netCDF asset
+                if nonetcdf:
+                    self.assertFalse("netcdf" in assets)
+                else:
+                    self.assertTrue("netcdf" in assets)
+                    asset = assets["netcdf"]
+                    self.assertIn("title", asset)
+                    self.assertNotIn("href", asset)
+                    self.assertEqual(asset["type"], constants.NETCDF_MEDIA_TYPE)
+                    self.assertIn("source", asset["roles"])
+                    self.assertNotIn("classification:classes", asset)
 
     def test_create_item(self) -> None:
         self.assertTrue(True)