Merge pull request #796 from MetaCell/feature/enable-schemathesis-v3

Enable experimental OPEN_API_3_1 support for schematesis; Update openapi.json look up logic

Author: filippomc

applications/jupyterhub/deploy/resources/hub/jupyterhub_config.py

@@ -24,7 +24,7 @@
 configuration_directory = os.path.dirname(os.path.realpath(__file__))
 sys.path.insert(0, configuration_directory)
 
-from z2jh import ( # noqa
+from z2jh import (  # noqa
     get_config,
     get_name,
     get_name_env,

applications/jupyterhub/deploy/resources/hub/z2jh.py

@@ -142,4 +142,4 @@ def set_config_if_not_none(cparent, name, key):
     """
     data = get_config(key)
     if data is not None:
-        setattr(cparent, name, data)
+        setattr(cparent, name, data)

tools/cloudharness-test/cloudharness_test/apitest_init.py

@@ -1,43 +1,45 @@
 import os
 import logging
-import requests
-
 import schemathesis as st
+from schemathesis.hooks import HookContext
 
 from cloudharness.auth import get_token
 
+st.experimental.OPEN_API_3_1.enable()
+
 if "APP_URL" or "APP_SCHEMA_FILE" in os.environ:
     app_schema = os.environ.get("APP_SCHEMA_FILE", None)
     app_url = os.environ.get("APP_URL", "http://samples.ch.local/api")
     logging.info("Start schemathesis tests on %s", app_url)
+
     schema = None
+
+    # First, attempt to load the local file if provided
     if app_schema:
-        # Test locally with harness-test -- use local schema for convenience during test development
-        openapi_uri = app_schema
         try:
-            schema = st.from_file(openapi_uri)
+            schema = st.from_file(app_schema)
+            logging.info("Successfully loaded schema from local file: %s", app_schema)
         except st.exceptions.SchemaError:
-            logging.exception("The local schema file %s cannot be loaded. Attempting loading from URL", openapi_uri)
+            logging.exception("The local schema file %s cannot be loaded. Attempting loading from URL", app_schema)
 
+    # If no schema from file, then loop over URL candidates
     if not schema:
-        # remote testing: might be /api/openapi.json or /openapi.json
-        try:
-            openapi_uri = openapi_uri = app_url + "/openapi.json"
-            logging.info("Using openapi spec at %s", openapi_uri)
-            schema = st.from_uri(openapi_uri)
-        except st.exceptions.SchemaError:
-            # Use alternative configuration
+        candidates = [
+            app_url.rstrip("/") + "/openapi.json",
+            app_url.rstrip("/") + "/api/openapi.json",
+        ]
+        for candidate in candidates:
             try:
-                openapi_uri = app_url.replace("/api", "") + "/openapi.json"
-                print(requests.get(openapi_uri))
-                schema = st.from_uri(openapi_uri)
+                logging.info("Attempting to load schema from URI: %s", candidate)
+                schema = st.from_uri(candidate)
+                logging.info("Successfully loaded schema from %s", candidate)
+                break  # Exit loop on successful load
             except st.exceptions.SchemaError as e:
-                raise Exception(
-                    f"Cannot setup api tests: {openapi_uri} not valid. Check your deployment is up and configuration") from e
-
-        except Exception as e:
-            raise Exception(
-                f"Cannot setup api tests: {openapi_uri}: {e}") from e
+                logging.warning("Failed to load schema from %s: %s", candidate, e)
+            except Exception as e:
+                logging.error("Unexpected error when loading schema from %s: %s", candidate, e)
+        if not schema:
+            raise Exception("Cannot setup API tests: No valid schema found. Check your deployment and configuration.")
 
     if "USERNAME" in os.environ and "PASSWORD" in os.environ:
         logging.info("Setting token from username and password")
@@ -66,3 +68,21 @@ def set(self, case, data, context):
                 case.headers = case.headers or {}
                 case.headers["Authorization"] = f"Bearer {data}"
                 case.headers["Cookie"] = f"kc-access={data}"
+
+    UNSAFE_VALUES = ("%", )
+
+    @st.hook
+    def filter_path_parameters(context: HookContext, x):
+        # Extract the candidate value.
+        param = x["key"] if isinstance(x, dict) and "key" in x else x
+
+        if param is None or param == "":
+            return True
+
+        param_str = str(param)
+
+        # Reject if any unsafe substring is present.
+        if any(unsafe in param_str for unsafe in UNSAFE_VALUES):
+            return False
+
+        return True