diff --git a/sedr/__init__.py b/sedr/__init__.py index 5a63eb1..e6fee06 100644 --- a/sedr/__init__.py +++ b/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( diff --git a/sedr/edreq11.py b/sedr/edreq11.py index 728fb94..fea7ce5 100644 --- a/sedr/edreq11.py +++ b/sedr/edreq11.py @@ -15,35 +15,59 @@ 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" @@ -51,6 +75,7 @@ def requirementA11_1(jsondata: str) -> tuple[bool, str]: 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, "" diff --git a/sedr/preflight.py b/sedr/preflight.py new file mode 100644 index 0000000..080801d --- /dev/null +++ b/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 diff --git a/sedr/rodeoprofile10.py b/sedr/rodeoprofile10.py index b63bca5..c195060 100644 --- a/sedr/rodeoprofile10.py +++ b/sedr/rodeoprofile10.py @@ -1,6 +1,7 @@ """rodeo-edr-profile requirements. See .""" 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, "", diff --git a/sedr/schemat.py b/sedr/schemat.py index c86095e..96a4305 100644 --- a/sedr/schemat.py +++ b/sedr/schemat.py @@ -91,94 +91,6 @@ def after_call(context, case, response): ) -@schema.include( - path_regex="^" + os.path.join(util.args.base_path, "conformance") -).parametrize() -@settings(max_examples=util.args.iterations, deadline=None) -def test_edr_conformance(case): - """Test /conformance endpoint.""" - global use_rodeoprofile - response = case.call() - conformance_json = json.loads(response.text) - - if ( - util.args.rodeo_profile - or rodeoprofile.conformance_url in conformance_json["conformsTo"] - ): - use_rodeoprofile = True - util.logger.info( - "Including tests for Rodeo profile %s", rodeoprofile.conformance_url - ) - - if "conformsTo" not in conformance_json: - spec_ref = "https://docs.ogc.org/is/19-072/19-072.html#_4129e3d3-9428-4e91-9bfc-645405ed2369" - raise AssertionError( - f"Conformance page /conformance does not contain a conformsTo attribute. See {spec_ref} for more info." - ) - - resolves, resolves_message = util.test_conformance_links( - jsondata=conformance_json["conformsTo"] - ) - if not resolves: - raise AssertionError(resolves_message) - - requirementA2_2_A5, requirementA2_2_A5_message = edreq.requirementA2_2_A5( - jsondata=conformance_json["conformsTo"], siteurl=util.args.url - ) - if not requirementA2_2_A5: - util.logger.error(requirementA2_2_A5_message) - # raise AssertionError(requirementA2_2_A5_message) - - requirementA2_2_A7, requirementA2_2_A7_message = edreq.requirementA2_2_A7( - response.raw.version - ) - if not requirementA2_2_A7: - raise AssertionError(requirementA2_2_A7_message) - - requirementA11_1, requirementA11_1_message = edreq.requirementA11_1( - jsondata=conformance_json["conformsTo"] - ) - if not requirementA11_1: - raise AssertionError(requirementA11_1_message) - - if use_rodeoprofile: - requirement7_1, requirement7_1_message = rodeoprofile.requirement7_1( - jsondata=conformance_json["conformsTo"] - ) - if not requirement7_1: - raise AssertionError(requirement7_1_message) - - util.logger.debug("Conformance %s tested OK", response.url) - - -@schema.include(path_regex="^" + util.args.base_path + "$").parametrize() -@settings(max_examples=util.args.iterations, deadline=None) -def test_edr_landingpage(case): - """Test that the landing page contains required elements.""" - spec_ref = "https://docs.ogc.org/is/19-072/19-072.html#_7c772474-7037-41c9-88ca-5c7e95235389" - landingpage_json = None - response = case.call() - try: - landingpage_json = json.loads(response.text) - landing, landing_message = util.parse_landing_json(landingpage_json) - if not landing: - raise AssertionError( - f"Landing page is missing required elements. See <{spec_ref}> for more info. {landing_message}" - ) - - util.logger.debug("Landingpage %s tested OK", response.url) - except json.decoder.JSONDecodeError: - util.logger.warning("Landing page is not valid JSON.") - raise AssertionError("Landing page is not valid JSON") - - if use_rodeoprofile: - requirement7_2, requirement7_2_message = rodeoprofile.requirement7_2( - jsondata=landingpage_json - ) - if not requirement7_2: - raise AssertionError(requirement7_2_message) - - @schema.include( path_regex="^" + os.path.join(util.args.base_path, "collections$") ).parametrize() diff --git a/sedr/test_schemat.py b/sedr/test_schemat.py index c0b371c..ee83b38 100644 --- a/sedr/test_schemat.py +++ b/sedr/test_schemat.py @@ -2,7 +2,6 @@ import unittest import util -import pytest class TestInit(unittest.TestCase): diff --git a/sedr/test_util.py b/sedr/test_util.py deleted file mode 100644 index 412935f..0000000 --- a/sedr/test_util.py +++ /dev/null @@ -1,14 +0,0 @@ -"""Unit tests for util.py.""" - -import unittest -import json -import util - - -class TestUtil(unittest.TestCase): - def test_parse_landing_json(self): - """Test parsing a generic landing page in json.""" - with open("testdata/landingpage.json", "r", encoding="utf-8") as f: - landingpage_json = json.loads(f.read()) - landing, _ = util.parse_landing_json(landingpage_json) - self.assertTrue(landing) diff --git a/sedr/util.py b/sedr/util.py index 6e64f67..fad57ff 100644 --- a/sedr/util.py +++ b/sedr/util.py @@ -6,7 +6,6 @@ import json from urllib.parse import urlsplit - __author__ = "Lars Falk-Petersen" __license__ = "GPL-3.0" @@ -55,6 +54,12 @@ def parse_args(args, version: str = "") -> argparse.Namespace: default=False, help="Treat SHOULD in any profile as SHALL. Default False.", ) + parser.add_argument( + "--timeout", + type=int, + default=10, + help="Set timeout for requests. Default 10.", + ) args = parser.parse_args(args) # Parse out base_path for convenience @@ -64,7 +69,7 @@ def parse_args(args, version: str = "") -> argparse.Namespace: def set_up_logging(args, logfile=None, version: str = "") -> logging.Logger: """Set up logging.""" - loglevel = logging.WARNING + loglevel = logging.DEBUG logger = logging.getLogger(__file__) logger.setLevel(logging.DEBUG) @@ -129,38 +134,19 @@ def parse_locations(jsondata) -> None: def test_conformance_links(jsondata) -> tuple[bool, str]: # pylint: disable=unused-argument """Test that all conformance links are valid and resolves. - TODO: http://www.opengis.net/spec/ogcapi-common-2/1.0/conf/collections doesn't work, so postponing this. + TODO: http://www.opengis.net/spec/ogcapi-common-2/1.0/conf/collections doesn't work, so not erroring out. """ - # for link in conformance_json["conformsTo"]: - # resp = None - # try: - # resp = requests.head(url=link, timeout=10) - # except requests.exceptions.MissingSchema as error: - # raise AssertionError( - # f"Link <{link}> from /conformance is malformed: {error})." - # ) from error - # assert ( - # resp.status_code < 400 - # ), f"Link {link} from /conformance is broken (gives status code {resp.status_code})." - return True, "" - - -def parse_landing_json(jsondata) -> tuple[bool, str]: - """Parse landing page if it is valid JSON. TODO: move to edreq.py and link to standard.""" - # See https://github.com/metno/sedr/issues/6 - if "title" not in jsondata: - return False, "Landing page does not contain a title." - if "description" not in jsondata: - logger.warning("Landing page does not contain a description.") - if "links" not in jsondata: - return False, "Landing page does not contain links." - for link in jsondata["links"]: - if not isinstance(link, dict): - return False, f"Link {link} is not a dictionary." - if "href" not in link: - return False, f"Link {link} does not have a href attribute." - if "rel" not in link: - return False, f"Link {link} does not have a rel attribute." + msg = "" + for link in jsondata["conformsTo"]: + resp = None + try: + resp = requests.head(url=link, timeout=10) + except requests.exceptions.MissingSchema as error: + msg += f"test_conformance_links Link <{link}> from /conformance is malformed: {error}). " + if not resp.status_code < 400: + msg += f"test_conformance_links Link {link} from /conformance is broken (gives status code {resp.status_code}). " + if msg: + return False, msg return True, ""