From 160db2b4357c6e18174e76e7bef8568c87246c28 Mon Sep 17 00:00:00 2001 From: Lars Falk-Petersen Date: Mon, 23 Sep 2024 10:38:52 +0200 Subject: [PATCH] Refactor code --- Dockerfile | 5 +- README.md | 12 ++- pytest.ini | 2 +- requirements-dev.txt | 3 +- requirements.txt | 2 +- sedr/__init__.py | 7 ++ sedr.py => sedr/schemat.py | 160 +++++++------------------------------ sedr/utils.py | 115 ++++++++++++++++++++++++++ tox.ini | 10 +-- 9 files changed, 171 insertions(+), 145 deletions(-) create mode 100644 sedr/__init__.py rename sedr.py => sedr/schemat.py (58%) create mode 100644 sedr/utils.py diff --git a/Dockerfile b/Dockerfile index e1d34ff..d2f03d4 100644 --- a/Dockerfile +++ b/Dockerfile @@ -23,11 +23,12 @@ RUN apt-get update && \ # Set workdir WORKDIR /app -COPY *.py requirements.txt pytest.ini ./ +COPY requirements.txt pytest.ini ./ +COPY sedr/ ./sedr/ RUN python3 -m venv ./venv && \ ./venv/bin/pip install -r /app/requirements.txt # # Run as nonroot user # USER sedr -ENTRYPOINT ["/app/venv/bin/python", "sedr.py"] +ENTRYPOINT ["/app/venv/bin/python", "./sedr/__init__.py"] diff --git a/README.md b/README.md index 649ce49..14890f8 100644 --- a/README.md +++ b/README.md @@ -41,8 +41,16 @@ Not production ready. Schemathesis and pytest are the main components. Stateful tests are not used, as they require openapi links in the spec. -- Python -- [Schemathesis](https://schemathesis.readthedocs.io/en/stable/) +### Components + +- Python 3 +- [Schemathesis](https://schemathesis.readthedocs.io/en/stable/), which use pytest to run + +### Limitations + +- Openapi 3.1 +- OGC EDR API version 1.1 +- Very basic tests for now ## How to contribute diff --git a/pytest.ini b/pytest.ini index 92f3b02..4292a88 100644 --- a/pytest.ini +++ b/pytest.ini @@ -6,4 +6,4 @@ addopts = -rA log_cli = true log_cli_level = WARNING testpaths = - sedr.py + ./sedr/*.py diff --git a/requirements-dev.txt b/requirements-dev.txt index fc8ad03..389fd78 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -1,4 +1,5 @@ +setuptools # prospector requirement tox black prospector -bandit \ No newline at end of file +bandit diff --git a/requirements.txt b/requirements.txt index 2f9732a..0312601 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ schemathesis~=3.34 pytest~=8.3 shapely~=2.0 -requests +requests~=2.32.0 diff --git a/sedr/__init__.py b/sedr/__init__.py new file mode 100644 index 0000000..46dc7d5 --- /dev/null +++ b/sedr/__init__.py @@ -0,0 +1,7 @@ +"""Main init file for sedr. Schemathesis related in schemat.py, others in utils.py.""" + +import pytest + + +if __name__ == "__main__": + pytest.main(["-rA", "--show-capture=no", "./sedr/schemat.py"]) diff --git a/sedr.py b/sedr/schemat.py similarity index 58% rename from sedr.py rename to sedr/schemat.py index b7a467c..93d9272 100644 --- a/sedr.py +++ b/sedr/schemat.py @@ -1,61 +1,18 @@ import requests +import json import schemathesis from hypothesis import settings -import pytest -import json -import sys -import argparse -import logging import shapely from shapely.wkt import loads as wkt_loads -from urllib.parse import urlsplit -sedr_version = "v0.7.0" -edr_version = "1.1" +import utils + # Globals extents = {} collection_ids = {} -def parse_args() -> argparse.Namespace: - """Parse arguments.""" - parser = argparse.ArgumentParser(description="schemathesis-edr") - parser.add_argument( - "--openapi", - type=str, - help="URL to openapi spec for API", - default="https://edrisobaric.k8s.met.no/api", - ) - parser.add_argument( - "--url", type=str, help="URL to API", default="https://edrisobaric.k8s.met.no" - ) - parser.add_argument( - "--iterations", - type=int, - default=50, - help="Amount of examples to generate, per test", - ) - parser.add_argument( - "--log-file", - type=str, - default=None, - help="Log output", - ) - parser.add_argument( - "--openapi-version", - choices=["3.0", "3.1"], - default="3.1", - help="Choose openapi version used in API. Default 3.1. Options are: 3.0, 3.1.", - ) - - args = parser.parse_args() - # Parse out base_path for conveience - args.base_path = urlsplit(args.url).path or "/" - - return args - - def set_up_schemathesis(args): """Set up schemathesis.""" if args.openapi_version == "3.1": @@ -63,42 +20,11 @@ def set_up_schemathesis(args): return schemathesis.from_uri(args.openapi, base_url=args.url) -args = parse_args() -schema = set_up_schemathesis(args) - - -def set_up_logging(args, logfile=None) -> logging.Logger: - """Set up logging.""" - loglevel = logging.WARNING - - logger = logging.getLogger(__file__) - logger.setLevel(logging.DEBUG) - - # File - if logfile is not None: - try: - with open(file=logfile, mode="w", encoding="utf-8") as f: - f.write( - f"SEDR version {sedr_version} on python {sys.version}, schemathesis {schemathesis.__version__} \nTesting url {args.url}, openapi {args.openapi}, openapi-version {args.openapi_version}.\n\n" - ) - except PermissionError as err: - print(f"Could not write to logfile {logfile}: {err}\nIf you're running this as a docker container, make sure you mount the log dir (docker run -v host-dir:container-dir) and give log option to sedr using the container-dir (--log-file /container-dir/debug.log).") - sys.exit(1) - - fh = logging.FileHandler(mode="a", filename=logfile) - fh.setLevel(logging.DEBUG) - logger.addHandler(fh) - - # Console - stdout_handler = logging.StreamHandler() - stdout_handler.setLevel(loglevel) - logger.addHandler(stdout_handler) - - return logger +schema = set_up_schemathesis(utils.args) @schema.parametrize() # parametrize => Create tests for all operations in schema -@settings(max_examples=args.iterations) +@settings(max_examples=utils.args.iterations) def test_api(case): """Run schemathesis standard tests.""" try: @@ -114,7 +40,7 @@ def after_call(self, response, case): """Hook runs after any call to the API.""" if case.request: # Log calls with status - logger.debug( + utils.logger.debug( "after_call URL %s gave %s - %s", case.request.path_url, case.status_code, @@ -122,8 +48,8 @@ def after_call(self, response, case): ) -@schema.include(path_regex="^" + args.base_path + "$").parametrize() -@settings(max_examples=args.iterations) +@schema.include(path_regex="^" + utils.args.base_path + "$").parametrize() +@settings(max_examples=utils.args.iterations) def test_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" @@ -137,15 +63,15 @@ def test_landingpage(case): try: landingpage_json = json.loads(response.text) except json.decoder.JSONDecodeError as e: - logger.error("Landing page is not valid JSON.") + utils.logger.error("Landing page is not valid JSON.") raise AssertionError( f"Expected valid JSON but got {e}. See {spec_ref} for more info." ) from e if "title" not in landingpage_json: - logger.warning("Landing page does not contain a title.") + utils.logger.warning("Landing page does not contain a title.") if "description" not in landingpage_json: - logger.warning("Landing page does not contain a description.") + utils.logger.warning("Landing page does not contain a description.") assert ( "links" in landingpage_json ), "Landing page does not contain links. See {spec_ref} for more info." @@ -160,11 +86,11 @@ def test_landingpage(case): "rel" in link ), f"Link {link} does not have a rel attribute. See {spec_ref} for more info." - logger.debug("Landingpage %s tested OK", response.url) + utils.logger.debug("Landingpage %s tested OK", response.url) -@schema.include(path_regex="^" + args.base_path + "conformance").parametrize() -@settings(max_examples=args.iterations) +@schema.include(path_regex="^" + utils.args.base_path + "conformance").parametrize() +@settings(max_examples=utils.args.iterations) def test_conformance(case): """Test that the conformance links are valid.""" spec_ref = "https://docs.ogc.org/is/19-072/19-072.html#_4129e3d3-9428-4e91-9bfc-645405ed2369" @@ -187,11 +113,11 @@ def test_conformance(case): resp.status_code < 400 ), f"Link {link} from /conformance is broken (gives status code {resp.status_code})." - logger.debug("Conformance %s tested OK", response.url) + utils.logger.debug("Conformance %s tested OK", response.url) @schema.parametrize() -@settings(max_examples=args.iterations) +@settings(max_examples=utils.args.iterations) def test_collections(case): """The default testing in function test_api() will fuzz the collections. This function will test that collections contain EDR spesifics.""" global collection_ids, extents @@ -210,7 +136,7 @@ def test_collections(case): collection_url = collection["links"][0]["href"] collection_ids[collection_url] = collection["id"] - logger.debug("test_collections found collection id %s", collection["id"]) + utils.logger.debug("test_collections found collection id %s", collection["id"]) extent = None try: @@ -225,10 +151,10 @@ def test_collections(case): Example [[1, 2, 3, 4], [5, 6, 7, 8]]. Was <{collection['extent']['spatial']['bbox']}>. See {spec_ref} for more info." extents[collection_url] = tuple(extent) - logger.debug( + utils.logger.debug( "test_collections found extent for %s: %s", collection_url, extent ) - except AttributeError as err: + except AttributeError: pass except KeyError as err: if err.args[0] == "extent": @@ -236,11 +162,11 @@ def test_collections(case): f"Unable to find extent for collection ID {collection['id']}. Found [{', '.join(collection.keys())}]. See {spec_ref} for more info." ) from err - logger.debug("Collections %s tested OK", response.url) + utils.logger.debug("Collections %s tested OK", response.url) @schema.parametrize() -@settings(max_examples=args.iterations) +@settings(max_examples=utils.args.iterations) def test_positions(case): """The default test in function test_api() will fuzz the coordinates. This function will test response given by coords inside and outside of the extent.""" if not case.path.endswith("/position"): @@ -264,7 +190,7 @@ def test_positions(case): ] ) if extent.contains(point): - logger.debug( + utils.logger.debug( "test_WKT Testing value INSIDE of extent, %s", case.query["coords"] ) # Assert that the HTTP status code is 200 @@ -272,60 +198,28 @@ def test_positions(case): response.status_code == 200 ), f"Expected status code 200 but got {response.status_code}" else: - logger.debug( + utils.logger.debug( "test_WKT Testing value OUTSIDE of extent, %s", case.query["coords"] ) assert ( response.status_code != 200 ), f"Expected status code 422 but got {response.status_code}" - logger.debug("Positions %s tested OK", response.url) + utils.logger.debug("Positions %s tested OK", response.url) @schema.parametrize() -@settings(max_examples=args.iterations) +@settings(max_examples=utils.args.iterations) def test_locations(case): """The default test in function test_api() will fuzz parameters. This function can test .../locations for EDR speicifics.""" if not case.path.endswith("/locations"): return response = case.call() - parse_locations(response.text) - - logger.debug("Locations %s tested OK", response.url) + utils.parse_locations(response.text) + utils.logger.debug("Locations %s tested OK", response.url) -def parse_locations(jsondata): - """Parse locations from JSON, test geometries.""" - try: - _ = json.loads(jsondata) - except json.JSONDecodeError as e: - raise AssertionError(f"parse_locations: Invalid JSON from {jsondata}") from e - - assert "name" in json.loads( - jsondata - ), 'Expected "name": "locations" in /locations, didn\'t find "name".' - assert ( - "locations" in json.loads(jsondata)["name"] - ), 'Expected "name": "locations" key in /locations, didn\'t find "locations".' - assert "features" in json.loads(jsondata), 'Expected "features" in /locations.' - - for feature in json.loads(jsondata)["features"]: - if feature["geometry"]["type"] == "Point": - _ = shapely.Point(feature["geometry"]["coordinates"]) - elif feature["geometry"]["type"] == "Polygon": - _ = shapely.Polygon(feature["geometry"]["coordinates"]) - else: - raise AssertionError( - f"Unable to create geometry type {feature['geometry']['type']} from coords {feature['geometry']['coordinates']}" - ) - - -logger = set_up_logging(args=args, logfile=args.log_file) # /collections is tested by default, if it exists. Adding this to require it to exist get_collections = schema["/collections"]["GET"] - - -if __name__ == "__main__": - pytest.main(["-rA", "--show-capture=no", __file__]) diff --git a/sedr/utils.py b/sedr/utils.py new file mode 100644 index 0000000..daa5b1b --- /dev/null +++ b/sedr/utils.py @@ -0,0 +1,115 @@ +import sys +import logging +import schemathesis +import argparse +import json +from urllib.parse import urlsplit +import shapely + + +__version__ = "v0.7.1" +edr_version = "1.1" + + +def parse_args() -> argparse.Namespace: + """Parse arguments.""" + parser = argparse.ArgumentParser(description="schemathesis-edr") + parser.add_argument( + "--openapi", + type=str, + help="URL to openapi spec for API", + default="https://edrisobaric.k8s.met.no/api", + ) + parser.add_argument( + "--url", type=str, help="URL to API", default="https://edrisobaric.k8s.met.no" + ) + parser.add_argument( + "--iterations", + type=int, + default=50, + help="Amount of examples to generate, per test", + ) + parser.add_argument( + "--log-file", + type=str, + default=None, + help="Log output", + ) + parser.add_argument( + "--openapi-version", + choices=["3.0", "3.1"], + default="3.1", + help="Choose openapi version used in API. Default 3.1. Options are: 3.0, 3.1.", + ) + + args = parser.parse_args() + # Parse out base_path for conveience + args.base_path = urlsplit(args.url).path or "/" + + return args + + +def set_up_logging(args, logfile=None) -> logging.Logger: + """Set up logging.""" + loglevel = logging.WARNING + + logger = logging.getLogger(__file__) + logger.setLevel(logging.DEBUG) + + # File + if logfile is not None: + try: + with open(file=logfile, mode="w", encoding="utf-8") as f: + f.write( + f"SEDR version {__version__} on python {sys.version}, schemathesis " + \ + f"{schemathesis.__version__} \nTesting url {args.url}, openapi {args.openapi}, " + \ + f"openapi-version {args.openapi_version}.\n\n" + ) + except PermissionError as err: + print( + f"Could not write to logfile {logfile}: {err}\nIf you're running this as a docker " + \ + "container, make sure you mount the log dir (docker run -v host-dir:container-dir) " + \ + "and give log option to sedr using the container-dir (--log-file /container-dir/debug.log)." + ) + sys.exit(1) + + fh = logging.FileHandler(mode="a", filename=logfile) + fh.setLevel(logging.DEBUG) + logger.addHandler(fh) + + # Console + stdout_handler = logging.StreamHandler() + stdout_handler.setLevel(loglevel) + logger.addHandler(stdout_handler) + + return logger + + +def parse_locations(jsondata): + """Parse locations from JSON, test geometries.""" + try: + _ = json.loads(jsondata) + except json.JSONDecodeError as e: + raise AssertionError(f"parse_locations: Invalid JSON from {jsondata}") from e + + assert "name" in json.loads( + jsondata + ), 'Expected "name": "locations" in /locations, didn\'t find "name".' + assert ( + "locations" in json.loads(jsondata)["name"] + ), 'Expected "name": "locations" key in /locations, didn\'t find "locations".' + assert "features" in json.loads(jsondata), 'Expected "features" in /locations.' + + for feature in json.loads(jsondata)["features"]: + if feature["geometry"]["type"] == "Point": + _ = shapely.Point(feature["geometry"]["coordinates"]) + elif feature["geometry"]["type"] == "Polygon": + _ = shapely.Polygon(feature["geometry"]["coordinates"]) + else: + raise AssertionError( + f"Unable to create geometry type {feature['geometry']['type']} from coords {feature['geometry']['coordinates']}" + ) + + +args = parse_args() +logger = set_up_logging(args=args, logfile=args.log_file) diff --git a/tox.ini b/tox.ini index aae058a..64d547c 100644 --- a/tox.ini +++ b/tox.ini @@ -1,7 +1,7 @@ [tox] envlist = black, prospector, bandit -[testenv] ; This is a live test against edrisobaric, and should not be run automatically +[testenv] ; This is a live test against the service edrisobaric, and should not be run automatically # See also config file pytest.ini deps = -r{toxinidir}/requirements.txt @@ -18,25 +18,25 @@ commands = prospector --no-autodetect \ --test-warnings \ --die-on-tool-error \ --zero-exit \ - {toxinidir}/07-edr-spesifics.py + {toxinidir}/sedr/ [testenv:black] # Check code style deps = -r{toxinidir}/requirements.txt -r{toxinidir}/requirements-dev.txt -commands = black --check {toxinidir}/07-edr-spesifics.py +commands = black --check {toxinidir}/sedr/ ; [testenv:mypy] ; # Check typing ; deps = ; -r{toxinidir}/requirements.txt ; -r{toxinidir}/requirements-dev.txt -; commands = mypy --follow-imports skip {toxinidir}/07-edr-spesifics.py +; commands = mypy --follow-imports skip {toxinidir}/*.py [testenv:bandit] # Check for security issues deps = -r{toxinidir}/requirements.txt -r{toxinidir}/requirements-dev.txt -commands = bandit --ini .bandit {toxinidir}/07-edr-spesifics.py +commands = bandit -r --ini .bandit {toxinidir}/sedr/