Refactor code

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"]

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
 

pytest.ini

@@ -6,4 +6,4 @@ addopts = -rA
 log_cli = true
 log_cli_level = WARNING
 testpaths =
-    sedr.py
+    ./sedr/*.py

requirements-dev.txt

@@ -1,4 +1,5 @@
+setuptools # prospector requirement
 tox
 black
 prospector
-bandit
+bandit

requirements.txt

@@ -1,4 +1,4 @@
 schemathesis~=3.34
 pytest~=8.3
 shapely~=2.0
-requests
+requests~=2.32.0

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"])

sedr.py → sedr/schemat.py

@@ -1,104 +1,30 @@
 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":
         schemathesis.experimental.OPEN_API_3_1.enable()
     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,16 +40,16 @@ 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,
             case.text[0:150],
         )
 
 
-@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,22 +151,22 @@ 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":
                 raise AssertionError(
                     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,68 +190,36 @@ 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
         assert (
             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__])