Rework how functions from edr, ogcapi and profile is collected and ru…

…n, to make code easier to follow. Use Rich for better output.

requirements.txt

@@ -2,3 +2,4 @@ schemathesis~=3.34
 pytest~=8.3
 shapely~=2.0
 requests~=2.32.0
+rich~=13.8.0

sedr/__init__.py

@@ -2,12 +2,16 @@
 
 __author__ = "Lars Falk-Petersen"
 __license__ = "GPL-3.0"
-__version__ = "v0.7.9"
+__version__ = "v0.8.0"
 
 import sys
 import pytest
+
 import util
 import preflight
+import edreq12 as edreq
+import ogcapi10 as ogcapi
+import rodeoprofile10 as rodeoprofile
 
 
 def run_schemat() -> None:
@@ -18,6 +22,27 @@ def run_schemat() -> None:
 
 def main() -> None:
     """Run the main program."""
+
+    # Collect tests to run
+    util.test_functions["landing"] += edreq.tests_landing + ogcapi.tests_landing
+    util.test_functions["conformance"] += (
+        edreq.tests_conformance + ogcapi.tests_conformance
+    )
+    util.test_functions["collection"] += (
+        edreq.tests_collection + ogcapi.tests_collections
+    )
+    if util.args.rodeo_profile:
+        util.logger.info(
+            "Including tests for Rodeo profile %s", rodeoprofile.conformance_url
+        )
+        util.test_functions["landing"] += rodeoprofile.tests_landing
+        util.test_functions["conformance"] += rodeoprofile.tests_conformance
+        util.test_functions["collection"] += rodeoprofile.tests_collection
+
+    # TODO: include profile tests based on conformance_url
+    # if rodeoprofile.conformance_url in conformance_json["conformsTo"]:
+    #     util.args.rodeo_profile = True
+
     if preflight.main():
         run_schemat()
     else:

sedr/edreq12.py

@@ -1,5 +1,6 @@
 """EDR requirements."""
 
+from collections.abc import Callable
 import util
 
 edr_version = "1.2"
@@ -16,7 +17,7 @@
 ]
 
 
-def requirementA2_2_A5(jsondata: dict, siteurl="") -> tuple[bool, str]:
+def requirementA2_2_A5(jsondata: dict) -> tuple[bool, str]:
     """
     OGC API - Environmental Data Retrieval Standard
     Version: 1.2
@@ -26,40 +27,21 @@ def requirementA2_2_A5(jsondata: dict, siteurl="") -> tuple[bool, str]:
     jsondata should be the "conformsTo"-part of the conformance page.
     """
     spec_url = f"{edr_root_url}#req_core_conformance"
+
     if "conformsTo" not in jsondata:
-        return (
-            False,
-            f"Conformance page <{siteurl}conformance> does not contain a "
-            f"conformsTo attribute. See <{spec_url}> for more info.",
+        return False, (
+            f"Conformance page does not contain a "
+            f"conformsTo attribute. See <{spec_url}> for more info."
         )
+
     for url in conformance_urls:
         if url not in jsondata["conformsTo"]:
-            return (
-                False,
-                f"Conformance page <{siteurl}conformance> does not contain "
-                f"the core edr class {url}. See <{spec_url}> for more info.",
+            return False, (
+                f"Conformance page does not contain "
+                f"the core edr class {url}. See <{spec_url}> for more info."
             )
 
-    util.logger.debug(
-        "requirementA2_2_A5: conformance page contains the required EDR classes."
-    )
-    return True, ""
-
-
-def requirementA2_2_A7(version: int) -> tuple[bool, str]:
-    """
-    OGC API - Environmental Data Retrieval Standard
-    Version: 1.2
-    Requirement Annex A2.2 A7
-
-    Check if HTTP1.1 was used.
-    """
-    spec_url = f"{edr_root_url}#_req_core_http"
-    if version == 11:
-        util.logger.debug("requirementA2_2_A7 HTTP version 1.1 was used.")
-        return True, ""
-
-    return False, f"HTTP version 1.1 was not used. See <{spec_url}> for more info."
+    return True, "Conformance page contains the required EDR classes."
 
 
 def requirementA11_1(jsondata: dict) -> tuple[bool, str]:
@@ -77,8 +59,7 @@ def requirementA11_1(jsondata: dict) -> tuple[bool, str]:
             if (
                 "oas31" in url or "oas30" in url  # TODO: oas30 should be removed
             ):
-                util.logger.debug("requirementA11_1 Found openapi class <%s>", url)
-                return True, url
+                return True, f"Found openapi class <{url}>. "
             return (
                 False,
                 f"OpenAPI version {util.args.openapi_version} and version in "
@@ -90,3 +71,8 @@ def requirementA11_1(jsondata: dict) -> tuple[bool, str]:
         f"Conformance page /conformance does not contain an openapi class. "
         f"See <{spec_url}> for more info.",
     )
+
+
+tests_landing: list[Callable[[dict], tuple[bool, str]]] = []
+tests_conformance = [requirementA2_2_A5, requirementA11_1]
+tests_collection: list[Callable[[dict], tuple[bool, str]]] = []

sedr/ogcapi10.py

@@ -1,6 +1,7 @@
 """OGC API Common requirements."""
 
-import util
+from collections.abc import Callable
+import requests
 
 ogc_api_common_version = "1.0"
 ogc_api_common_url = "https://docs.ogc.org/is/19-072/19-072.html"
@@ -58,5 +59,34 @@ def requirement9_1(jsondata: dict) -> tuple[bool, str]:
             False,
             f"Landing page does not contain a service-desc link. See <{spec_ref}> for more info.",
         )
-    util.logger.debug("requirement9_1 Landing page contains required elements.")
-    return True, ""
+    return True, "Landing page contains required elements. "
+
+
+def test_conformance_links(jsondata: dict, timeout: int = 10) -> tuple[bool, str]:
+    """Test that all conformance links are valid and resolves."""
+    msg = ""
+    valid = True
+    for link in jsondata["conformsTo"]:
+        if link in [
+            "http://www.opengis.net/spec/ogcapi-common-2/1.0/conf/conformance",
+            "http://www.opengis.net/spec/ogcapi-common-2/1.0/conf/collections",
+            "http://www.opengis.net/spec/ogcapi-edr-1/1.2/req/oas31",
+        ]:
+            # TODO: These links are part of the standard but doesn't work, so skipping for now.
+            msg += f"test_conformance_links Link {link} doesn't resolv, but that is a known issue. "
+            continue
+        response = requests.Response()
+        try:
+            response = requests.head(url=link, timeout=timeout)
+        except requests.exceptions.MissingSchema as error:
+            valid = False
+            msg += f"test_conformance_links Link <{link}> from /conformance is malformed: {error}). "
+        if not response.status_code < 400:
+            valid = False
+            msg += f"test_conformance_links Link {link} from /conformance is broken (gives status code {response.status_code}). "
+    return valid, msg
+
+
+tests_landing = [requirement9_1]
+tests_conformance = [test_conformance_links]
+tests_collections: list[Callable[[dict], tuple[bool, str]]] = []