Merge pull request #27 from metno/16-add-preflight-check

16 add preflight check

sedr/__init__.py

@@ -2,18 +2,27 @@
 
 __author__ = "Lars Falk-Petersen"
 __license__ = "GPL-3.0"
-__version__ = "v0.7.5"
+__version__ = "v0.8.0"
 
 import sys
 import pytest
 import util
+import preflight
 
 
-def main() -> None:
+def run_schemat() -> None:
     # show-capture=no means hide stdout/stderr. Should change debug output instead.
     pytest.main(["-rA", "--show-capture=no", "./sedr/schemat.py"])
 
 
+def main() -> None:
+    """Run the main program."""
+    if preflight.main():
+        run_schemat()
+    else:
+        sys.exit(1)
+
+
 # Handle --version and --help
 util.args = util.parse_args(sys.argv[1:], __version__)
 util.logger = util.set_up_logging(

sedr/edreq11.py

@@ -15,42 +15,67 @@
 
 
 def requirementA2_2_A5(jsondata: str, siteurl="") -> tuple[bool, str]:
-    """Check if the conformance page contains the required EDR classes.
+    """
+    OGC API - Environmental Data Retrieval Standard
+    Version: 1.1
+    Requirement Annex A2.2 A5
 
+    Check if the conformance page contains the required EDR classes.
     jsondata should be the "conformsTo"-part of the conformance page.
     """
     spec_url = "https://docs.ogc.org/is/19-086r6/19-086r6.html#_c9401fee-54b9-d116-8365-af0f85a8243d"
+    if "conformsTo" not in jsondata:
+        return (
+            False,
+            f"Conformance page <{siteurl}conformance> does not contain a conformsTo attribute. See <{spec_url}> for more info.",
+        )
     for url in conformance_urls:
-        if url not in jsondata:
+        if url not in jsondata["conformsTo"]:
             return (
                 False,
                 f"Conformance page <{siteurl}conformance> does not contain 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]:
-    """Check if HTTP1.1 was used."""
+    """
+    OGC API - Environmental Data Retrieval Standard
+    Version: 1.1
+    Requirement Annex A2.2 A7
+
+    Check if HTTP1.1 was used.
+    """
     spec_url = "https://docs.ogc.org/is/19-086r6/19-086r6.html#_0d0c25a0-850f-2aa5-9acb-06efcc04d452"
     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."
 
 
 def requirementA11_1(jsondata: str) -> tuple[bool, str]:
-    """Check if the conformance page contains openapi classes, and that they match our version."""
+    """
+    OGC API - Environmental Data Retrieval Standard
+    Version: 1.1
+    Requirement A11.1
+
+    Check if the conformance page contains openapi classes, and that they match our version."""
     spec_url = "https://docs.ogc.org/is/19-086r6/19-086r6.html#_cc7dd5e3-1d54-41ff-b5ba-c5fcb99fa663"
 
-    for url in jsondata:
+    for url in jsondata["conformsTo"]:
         if url in openapi_conformance_urls:
             if (
                 util.args.openapi_version == "3.1"
                 and "oas31" in url
                 or util.args.openapi_version == "3.0"
                 and "oas30" in url
             ):
+                util.logger.debug("requirementA11_1 Found openapi class <%s>", url)
                 return True, url
             return (
                 False,
@@ -61,3 +86,50 @@ def requirementA11_1(jsondata: str) -> tuple[bool, str]:
         False,
         f"Conformance page /conformance does not contain an openapi class. See <{spec_url}> for more info.",
     )
+
+
+def requirement9_1(jsondata) -> tuple[bool, str]:
+    """
+    OGC API - Common - Part 1: Core
+    Version: 1.0.0
+    Requirement 9.1
+
+    Test that the landing page contains required elements.
+
+    TODO: See https://github.com/metno/sedr/issues/6
+    """
+    spec_ref = "https://docs.ogc.org/is/19-072/19-072.html#_7c772474-7037-41c9-88ca-5c7e95235389"
+
+    if "title" not in jsondata:
+        return (
+            False,
+            "Landing page does not contain a title. See <{spec_ref}> for more info.",
+        )
+    if "description" not in jsondata:
+        return (
+            False,
+            "Landing page does not contain a description. See <{spec_ref}> for more info.",
+        )
+    if "links" not in jsondata:
+        return (
+            False,
+            "Landing page does not contain links. See <{spec_ref}> for more info.",
+        )
+    for link in jsondata["links"]:
+        if not isinstance(link, dict):
+            return (
+                False,
+                f"Link {link} is not a dictionary. See <{spec_ref}> for more info.",
+            )
+        if "href" not in link:
+            return (
+                False,
+                f"Link {link} does not have a href attribute. See <{spec_ref}> for more info.",
+            )
+        if "rel" not in link:
+            return (
+                False,
+                f"Link {link} does not have a rel attribute. See <{spec_ref}> for more info.",
+            )
+    util.logger.debug("requirement9_1 Landing page contains required elements.")
+    return True, ""

sedr/preflight.py

@@ -0,0 +1,121 @@
+"""Run a series of simple preflight checks before invoking schemathesis."""
+
+import util
+import requests
+import json
+from urllib.parse import urljoin
+import edreq11 as edreq
+import rodeoprofile10 as rodeoprofile
+
+
+def test_site_response(url: str, timeout=10) -> bool:
+    """Check basic http response."""
+    response = requests.get(url, timeout=timeout)
+    if not response.status_code < 400:
+        util.logger.error(
+            "Landing page doesn't respond correctly: status code: %s",
+            response.status_code,
+        )
+        return False
+    return True
+
+
+def parse_landing(url, timeout=10) -> bool:
+    """Test that the landing page contains required elements."""
+    landing_json = None
+    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
+
+    landing, requirement9_1_message = edreq.requirement9_1(landing_json)
+    if not landing:
+        util.logger.error(requirement9_1_message)
+        return False
+
+    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 True, landing_json
+
+
+def parse_conformance(url: str, timeout: int, landing_json) -> bool:
+    """Test that the conformance page contains required elements."""
+    conformance_json = None
+    response = requests.get(url, timeout=timeout)
+
+    try:
+        conformance_json = response.json()
+    except json.decoder.JSONDecodeError:
+        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
+
+    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
+
+    requirementA11_1, requirementA11_1_message = edreq.requirementA11_1(
+        jsondata=conformance_json
+    )
+    if not requirementA11_1:
+        util.logger.error(requirementA11_1_message)
+        return False
+
+    # Rodeo profile
+
+    if (
+        util.args.rodeo_profile
+        or rodeoprofile.conformance_url in conformance_json["conformsTo"]
+    ):
+        util.logger.info(
+            "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
+
+    return True
+
+
+def main():
+    conformance_url = urljoin(util.args.url, "/conformance")
+
+    if not test_site_response(util.args.url, util.args.timeout):
+        return False
+
+    landing_ok, landing_json = parse_landing(util.args.url, util.args.timeout)
+    if not landing_ok:
+        return False
+
+    if not parse_conformance(conformance_url, util.args.timeout, landing_json):
+        return False
+
+    util.logger.info("Preflight checks passed.")
+    return True

sedr/rodeoprofile10.py

@@ -1,6 +1,7 @@
 """rodeo-edr-profile requirements. See <http://rodeo-project.eu/rodeo-edr-profile>."""
 
 import json
+import util
 
 conformance_url = "http://rodeo-project.eu/spec/rodeo-edr-profile/1/req/core"
 spec_base_url = (
@@ -9,17 +10,14 @@
 
 
 def requirement7_1(jsondata: str) -> tuple[bool, str]:
-    """Check if the conformance page contains the required EDR classes.
-
-    jsondata should be the "conformsTo"-part of the conformance page.
-    """
+    """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:
+    if conformance_url not in jsondata["conformsTo"]:
         return (
             False,
             f"Conformance page /conformance does not contain the profile class {conformance_url}. See <{spec_url}> for more info.",
         )
-
+    util.logger.debug("Rodeoprofile Requirement 7.1 OK")
     return True, ""
 
 
@@ -62,6 +60,7 @@ def requirement7_2(jsondata: str) -> tuple[bool, str]:
             False,
             f"Landing page should linkt to service-doc, with type {servicedoc_type}. See <{spec_url}> for more info.",
         )
+    util.logger.debug("Rodeoprofile Requirement 7.2 OK")
     return True, ""
 
 
@@ -95,6 +94,7 @@ def requirement7_3(jsondata) -> tuple[bool, str]:
             f"Collection must have an id. None found in collection <{jsondata}>."
             f"Error {err}.",
         )
+    util.logger.debug("Rodeoprofile Requirement 7.3 OK")
     return (
         True,
         "",
@@ -112,12 +112,13 @@ def requirement7_4(jsondata: str) -> tuple[bool, str]:
                 False,
                 f"Collection title should not exceed 50 chars. See <{spec_url}> for more info.",
             )
-    except (json.JSONDecodeError, KeyError) as err:
+    except (json.JSONDecodeError, KeyError):
         # A
         return (
             False,
             f"Collection must have a title, but it seems to be missing. See <{spec_url}> and {spec_base_url}#_collection_title_2 for more info.",
         )
+    util.logger.debug("Rodeoprofile Requirement 7.4 OK")
     return (
         True,
         "",
@@ -141,6 +142,7 @@ def requirement7_5(jsondata: str) -> tuple[bool, str]:
             False,
             f"Collection <{jsondata['id']}> is missing a license link with rel='license'. See <{spec_url}> A, B for more info.",
         )
+    util.logger.debug("Rodeoprofile Requirement 7.5 OK")
     return (
         True,
         "",