10 improve error message for bad service desc

sedr/edreq11.py

@@ -14,7 +14,7 @@
 ]
 
 
-def requirementA2_2_A5(jsondata: str, siteurl="") -> tuple[bool, str]:
+def requirementA2_2_A5(jsondata: dict, siteurl="") -> tuple[bool, str]:
     """
     OGC API - Environmental Data Retrieval Standard
     Version: 1.1
@@ -58,7 +58,7 @@ def requirementA2_2_A7(version: int) -> tuple[bool, str]:
     return False, f"HTTP version 1.1 was not used. See <{spec_url}> for more info."
 
 
-def requirementA11_1(jsondata: str) -> tuple[bool, str]:
+def requirementA11_1(jsondata: dict) -> tuple[bool, str]:
     """
     OGC API - Environmental Data Retrieval Standard
     Version: 1.1
@@ -88,7 +88,7 @@ def requirementA11_1(jsondata: str) -> tuple[bool, str]:
     )
 
 
-def requirement9_1(jsondata) -> tuple[bool, str]:
+def requirement9_1(jsondata: dict) -> tuple[bool, str]:
     """
     OGC API - Common - Part 1: Core
     Version: 1.0.0
@@ -115,6 +115,8 @@ def requirement9_1(jsondata) -> tuple[bool, str]:
             False,
             "Landing page does not contain links. See <{spec_ref}> for more info.",
         )
+
+    service_desc = ""
     for link in jsondata["links"]:
         if not isinstance(link, dict):
             return (
@@ -131,5 +133,12 @@ def requirement9_1(jsondata) -> tuple[bool, str]:
                 False,
                 f"Link {link} does not have a rel attribute. See <{spec_ref}> for more info.",
             )
+        if link["rel"] == "service-desc":
+            service_desc = link["href"]
+    if not service_desc:
+        return (
+            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, ""

sedr/preflight.py

@@ -20,35 +20,35 @@ def test_site_response(url: str, timeout=10) -> bool:
     return True
 
 
-def parse_landing(url, timeout=10) -> bool:
+def parse_landing(url, timeout=10) -> tuple[bool, dict]:
     """Test that the landing page contains required elements."""
-    landing_json = None
+    landing_json = {}
     response = requests.get(url, timeout=timeout)
 
     try:
         landing_json = response.json()
     except json.decoder.JSONDecodeError:
         util.logger.warning("Landing page <%s> is not valid JSON.", url)
-        return False
+        return False, landing_json
 
     landing, requirement9_1_message = edreq.requirement9_1(landing_json)
     if not landing:
         util.logger.error(requirement9_1_message)
-        return False
+        return False, landing_json
 
     requirementA2_2_A7, requirementA2_2_A7_message = edreq.requirementA2_2_A7(
         response.raw.version
     )
     if not requirementA2_2_A7:
         util.logger.error(requirementA2_2_A7_message)
-        return False
+        return False, landing_json
 
     return True, landing_json
 
 
 def parse_conformance(url: str, timeout: int, landing_json) -> bool:
     """Test that the conformance page contains required elements."""
-    conformance_json = None
+    conformance_json = {}
     response = requests.get(url, timeout=timeout)
 
     try:
@@ -57,25 +57,27 @@ def parse_conformance(url: str, timeout: int, landing_json) -> bool:
         util.logger.warning("Conformance page <%s> is not valid JSON.", url)
         return False
 
-    resolves, resolves_message = util.test_conformance_links(jsondata=conformance_json)
-    util.logger.error(resolves_message)
-    # TODO: reenable when all conformance links resolves
-    # if not resolves and util.args.strict:
-    #     return False
+    resolves, resolves_message = util.test_conformance_links(jsondata=conformance_json, timeout=util.args.timeout)
+    if not resolves and util.args.strict:
+        util.logger.error(resolves_message)
+        if util.args.strict:
+            return False
 
     requirementA2_2_A5, requirementA2_2_A5_message = edreq.requirementA2_2_A5(
         jsondata=conformance_json, siteurl=util.args.url
     )
     if not requirementA2_2_A5:
         util.logger.error(requirementA2_2_A5_message)
-        return False
+        if util.args.strict:
+            return False
 
     requirementA11_1, requirementA11_1_message = edreq.requirementA11_1(
         jsondata=conformance_json
     )
     if not requirementA11_1:
         util.logger.error(requirementA11_1_message)
-        return False
+        if util.args.strict:
+            return False
 
     # Rodeo profile
 
@@ -87,19 +89,21 @@ def parse_conformance(url: str, timeout: int, landing_json) -> bool:
             "Including tests for Rodeo profile %s", rodeoprofile.conformance_url
         )
 
-        requirement7_2, requirement7_2_message = rodeoprofile.requirement7_2(
-            jsondata=landing_json
-        )
-        if not requirement7_2:
-            util.logger.error(requirement7_2_message)
-            return False
-
         requirement7_1, requirement7_1_message = rodeoprofile.requirement7_1(
             jsondata=conformance_json
         )
         if not requirement7_1:
             util.logger.error(requirement7_1_message)
-            return False
+            if util.args.strict:
+                return False
+
+        requirement7_2, requirement7_2_message = rodeoprofile.requirement7_2(
+            jsondata=landing_json, timeout=util.args.timeout
+        )
+        if not requirement7_2:
+            util.logger.error(requirement7_2_message)
+            if util.args.strict:
+                return False
 
     return True
 

sedr/rodeoprofile10.py

@@ -1,6 +1,7 @@
 """rodeo-edr-profile requirements. See <http://rodeo-project.eu/rodeo-edr-profile>."""
 
 import json
+import requests
 import util
 
 conformance_url = "http://rodeo-project.eu/spec/rodeo-edr-profile/1/req/core"
@@ -9,7 +10,7 @@
 )
 
 
-def requirement7_1(jsondata: str) -> tuple[bool, str]:
+def requirement7_1(jsondata: dict) -> tuple[bool, str]:
     """Check if the conformance page contains the required EDR classes."""
     spec_url = f"{spec_base_url}#_requirements_class_core"
     if conformance_url not in jsondata["conformsTo"]:
@@ -21,34 +22,69 @@ def requirement7_1(jsondata: str) -> tuple[bool, str]:
     return True, ""
 
 
-def requirement7_2(jsondata: str) -> tuple[bool, str]:
-    """Check OpenAPI."""
+def requirement7_2(jsondata: dict, timeout: int) -> tuple[bool, str]:
+    """
+    RODEO EDR Profile
+    Version: 0.1.0
+
+    7.2. OpenAPI
+
+    jsondata should be a valid landing page json dict.
+    """
     spec_url = f"{spec_base_url}#_openapi"
     openapi_type = "application/vnd.oai.openapi+json;version="  # 3.0"
     servicedoc_type = "text/html"
 
-    # A, B, C
+    service_desc_link = ""
+    service_desc_type = ""
     for link in jsondata["links"]:
         if link["rel"] == "service-desc":
-            if openapi_type not in link["type"]:
-                return (
-                    False,
-                    f"OpenAPI link service-desc should identify the content as "
-                    "openAPI and include version. Example "
-                    "<application/vnd.oai.openapi+json;version=3.0>. Found: "
-                    f"<{link['type']}> See <{spec_url}> and <{spec_base_url}"
-                    "#_openapi_2> for more info.",
-                )
+            service_desc_link = link["href"]
+            service_desc_type = link["type"]
             break
-    else:
+
+    if not service_desc_link:
         return (
             False,
             f"No service-desc link found. See <{spec_url}> for more info.",
         )
 
-    # D
+    # C - relation type
+    if openapi_type not in service_desc_type:
+        return (
+            False,
+            f"OpenAPI link service-desc should identify the content as "
+            "openAPI and include version. Example "
+            "<application/vnd.oai.openapi+json;version=3.0>. Found: "
+            f"<{service_desc_type}> See <{spec_url}> and <{spec_base_url}"
+            "#_openapi_2> for more info.",
+        )
+
+    # A - described using an OpenAPI document
+    response = requests.get(service_desc_link, timeout=timeout)
+    if not response.status_code < 400:
+        return (
+            False,
+            f"OpenAPI link service-desc <{service_desc_link}> doesn't respond properly. "
+            f"Status code: {response.status_code}.",
+        )
+
+    # B - encoded as JSON
+    try:
+        _ = response.json()
+    except (json.JSONDecodeError, TypeError) as err:
+        return (
+            False,
+            f"OpenAPI link service-desc <{service_desc_link}> does not contain valid JSON.\n"
+            f"Error: {err}",
+        )
+
+    # D API documentation
+    service_doc_link = ""
     for link in jsondata["links"]:
         if link["rel"] == "service-doc":
+            service_doc_link = link["href"]
+
             if servicedoc_type not in link["type"]:
                 return (
                     False,
@@ -58,13 +94,22 @@ def requirement7_2(jsondata: str) -> tuple[bool, str]:
     else:
         return (
             False,
-            f"Landing page should linkt to service-doc, with type {servicedoc_type}. See <{spec_url}> for more info.",
+            f"Landing page should link to service-doc. See <{spec_url}> for more info.",
+        )
+
+    response = requests.get(service_doc_link, timeout=timeout)
+    if not response.status_code < 400:
+        return (
+            False,
+            f"OpenAPI link service-desc <{link["href"]}> doesn't respond properly. "
+            f"Status code: {response.status_code}. See <{spec_url}> for more info.",
         )
+
     util.logger.debug("Rodeoprofile Requirement 7.2 OK")
     return True, ""
 
 
-def requirement7_3(jsondata) -> tuple[bool, str]:
+def requirement7_3(jsondata: dict) -> tuple[bool, str]:
     """Check collection identifier. Can only test B, C.
     Should only be tested if --strict is set."""
     spec_url = f"{spec_base_url}#_collection_identifier"
@@ -101,7 +146,7 @@ def requirement7_3(jsondata) -> tuple[bool, str]:
     )
 
 
-def requirement7_4(jsondata: str) -> tuple[bool, str]:
+def requirement7_4(jsondata: dict) -> tuple[bool, str]:
     """Check collection title. Can only test A, B."""
     spec_url = f"{spec_base_url}#_collection_title"
 
@@ -125,7 +170,7 @@ def requirement7_4(jsondata: str) -> tuple[bool, str]:
     )
 
 
-def requirement7_5(jsondata: str) -> tuple[bool, str]:
+def requirement7_5(jsondata: dict) -> tuple[bool, str]:
     """Check collection license. Can't test D."""
     spec_url = f"{spec_base_url}#_collection_license"
     # A, B

sedr/schemat.py

@@ -32,7 +32,7 @@ def set_up_schemathesis(args) -> BaseOpenAPISchema:
 
     if args.openapi == "":
         # Attempt to find schema URL automatically
-        args.openapi = util.locate_openapi_url(args.url)
+        args.openapi = util.locate_openapi_url(args.url, timeout=util.args.timeout)
         if len(args.openapi) == 0:
             raise AssertionError(
                 "Unable to find openapi spec for API. Please supply manually with --openapi <url>"

sedr/test_rodeoprofile10.py

@@ -0,0 +1,86 @@
+"""Unit tests for rodeoprofile10.py."""
+
+import unittest
+import util
+import rodeoprofile10 as profile
+
+
+class TestRodeoprofile(unittest.TestCase):
+    __version__ = "testversion"
+    util.args = util.parse_args([], __version__)
+    util.logger = util.set_up_logging(
+        args=util.args, logfile=util.args.log_file, version=__version__
+    )
+
+    def test_requirement7_2(self):
+        landing_json_good = {
+            "title": "EDR isobaric from Grib",
+            "description": "An EDR API for isobaric data from Grib files",
+            "links": [
+                {
+                    "href": "https://edrisobaric.k8s.met.no/",
+                    "rel": "self",
+                    "type": "application/json",
+                    "title": "Landing Page",
+                },
+                {
+                    "href": "https://edrisobaric.k8s.met.no/api",
+                    "rel": "service-desc",
+                    "type": "application/vnd.oai.openapi+json;version=3.1",
+                    "title": "OpenAPI document",
+                },
+                {
+                    "href": "https://edrisobaric.k8s.met.no/docs",
+                    "rel": "service-doc",
+                    "type": "text/html",
+                    "title": "OpenAPI document",
+                },
+                {
+                    "href": "https://edrisobaric.k8s.met.no/conformance",
+                    "rel": "conformance",
+                    "type": "application/json",
+                    "title": "Conformance document",
+                },
+                {
+                    "href": "https://edrisobaric.k8s.met.no/collections",
+                    "rel": "data",
+                    "type": "application/json",
+                    "title": "Collections metadata in JSON",
+                },
+            ],
+            "provider": {
+                "name": "Meteorologisk institutt / The Norwegian Meteorological Institute",
+                "url": "https://api.met.no/",
+            },
+            "contact": {
+                "email": "[email protected]",
+                "phone": "+47.22963000",
+                "address": "Henrik Mohns plass 1",
+                "postalCode": "0313",
+                "city": "Oslo",
+                "country": "Norway",
+            },
+        }
+
+        ok, _ = profile.requirement7_2(landing_json_good, timeout=10)
+        self.assertTrue(ok)
+
+        landing_json_bad = {
+            "title": "EDR isobaric from Grib",
+            "description": "An EDR API for isobaric data from Grib files",
+            "links": [
+                {
+                    "href": "https://edrisobaric.k8s.met.no/",
+                    "rel": "self",
+                    "type": "application/json",
+                    "title": "Landing Page",
+                },
+            ],
+        }
+
+        ok, _ = profile.requirement7_2(landing_json_bad, timeout=10)
+        self.assertFalse(ok)
+
+
+if __name__ == "__main__":
+    unittest.main()