Also verify endpoint structure: Query endpoints from FTL and check if all properties mentioned in the docs are present (and of correct type) and that there are no extra properties we forgot to document. Furthermore, also verify that the provided examples are of correct type, too.

Signed-off-by: DL6ER <dl6er@dl6er.de>
This commit is contained in:
DL6ER
2023-01-03 21:51:13 +01:00
parent 5019cc7ac7
commit f2d68f20d7
16 changed files with 527 additions and 259 deletions

View File

@@ -1,139 +1,75 @@
#!/bin/python3
# Pi-hole: A black hole for Internet advertisements
# (c) 2023 Pi-hole, LLC (https://pi-hole.net)
# Network-wide ad blocking via your own hardware.
#
# FTL Engine - auxiliary files
# API test script
#
# This file is copyright under the latest version of the EUPL.
# Please see LICENSE file for your rights under this license.
import yaml
import json
import urllib.request
from libs.openAPI import openApi
from libs.FTLAPI import FTLAPI
from libs.responseVerifyer import ResponseVerifyer
# OpenAPI specs are split into multiple files, this script extracts the endpoints from them
base_path = "src/api/docs/content/specs/"
# List of methods we want to extract
methods = ["get", "post", "put", "delete"]
API_ROOT = "/api"
# Prepare list of YAML endpoints
YAMLendpoints = {}
for method in methods:
YAMLendpoints[method] = []
# Cache for already read files
yamls = {}
def read_yaml_maybe_cache(file: str) -> dict:
if file not in yamls:
# Read the file
with open(file, "r") as stream:
try:
yamls[file] = yaml.safe_load(stream)
except Exception as e:
print("Exception when reading " + file + ": " + str(e))
exit(1)
return yamls[file]
# Read and parse the main file
try:
# Get the paths
paths = read_yaml_maybe_cache(base_path + "main.yaml")["paths"]
except Exception as e:
print(e)
exit(1)
# Resolve a reference
def resolveSingleReference(ref_str: str, k: str):
# Read and parse the referenced file
ref = ref_str.partition("#")
if len(ref[0]) == 0:
# Empty references are not allowed
raise Exception("Empty reference, always specify a file in the API specification")
# If the file link is empty, we refer to the current file
file = base_path + ref[0]
# Read the referenced file
try:
# Extract the YAML
refYML_full = read_yaml_maybe_cache(file)
refYML = refYML_full.copy()
# Reduce to what we want to import
for x in ref[2].split("/"):
if len(x) > 0:
#if x not in refYML:
refYML = refYML[x]
return refYML
except Exception as e:
print("Exception when reading " + file + ": " + str(e))
print("Tried to resolve " + ref_str + " in:\n" + json.dumps(refYML, indent=2))
if __name__ == "__main__":
# OpenAPI specs are split into multiple files, this script extracts the endpoints from them
openapi = openApi(base_path = "src/api/docs/content/specs/", api_root = "/api")
if not openapi.parse("main.yaml"):
exit(1)
# Recursively resolve references, this can take a few seconds
def recurseRef(dict_in: dict, dict_key: str):
for a in dict_in.keys():
next_dict_key = dict_key + "/" + a if len(dict_key) > 0 else a
if isinstance(dict_in[a], dict):
if "$ref" in dict_in[a]:
dict_in[a] = resolveSingleReference(dict_in[a]["$ref"], next_dict_key)
# Recurse into the new reference
recurseRef(dict_in[a], next_dict_key)
else:
# Recurse into the dictionary
recurseRef(dict_in[a], next_dict_key)
# Get endpoints from FTL
ftl = FTLAPI("http://127.0.0.1:8080")
ftl.get_endpoints()
# Recursively resolve references in the paths
# We do this in a separate step to avoid resolving references multiple times
# References are resolved in-place
print("Resolving references...")
recurseRef(paths, "")
print("...done\n")
errs = [0, 0, 0]
print("Endpoints in OpenAPI specs but not in FTL:")
# Check for endpoints in OpenAPI specs that are not defined in FTL
for path in openapi.endpoints["get"]:
if path not in ftl.endpoints:
print(" Missing GET endpoint in FTL: " + path)
errs[0] += 1
if errs[0] == 0:
print(" No missing endpoints\n")
# Build and sort the list of endpoints
for method in methods:
for path in paths:
if method in paths[path]:
# Strip possible variables from path
clean_path = API_ROOT + path.partition("/{")[0]
YAMLendpoints[method].append(clean_path)
YAMLendpoints[method] = sorted(YAMLendpoints[method])
# Check for endpoints in FTL that are not in the OpenAPI specs
print("Endpoints in FTL but not in OpenAPI specs:")
for path in ftl.endpoints:
if path not in openapi.endpoints["get"]:
# Ignore the docs endpoint
if path in ["/api/docs"]:
continue
print(" Missing GET endpoint in OpenAPI specs: " + path)
errs[1] += 1
if errs[1] == 0:
print(" No missing endpoints\n")
# Query the endpoints from FTL for comparison with the OpenAPI specs
try:
with urllib.request.urlopen("http://127.0.0.1:8080/api/ftl/endpoints") as url:
FTLendpoints = sorted(json.load(url)["endpoints"])
except Exception as e:
print("Exception: " + e)
exit(1)
print("Verifying endpoints...")
for path in openapi.endpoints["get"]:
verifyer = ResponseVerifyer(ftl, openapi)
errors = verifyer.verify_endpoint(path)
if len(errors) == 0:
print(" " + path + ": OK")
else:
print(" " + path + ":")
for error in errors:
print(" - " + error)
errs[2] += len(errors)
print("")
errs = [0, 0]
print("Endpoints in OpenAPI specs but not in FTL:")
# Check for endpoints in OpenAPI specs that are not defined in FTL
for path in YAMLendpoints["get"]:
if path not in FTLendpoints:
print(" Missing GET endpoint in FTL: " + path)
errs[0] += 1
if errs[0] == 0:
print(" No missing endpoints\n")
# Print the number error (if any)
if errs[0] > 0:
print("Found " + str(errs[0]) + " non-implemented endpoints")
if errs[1] > 0:
print("Found " + str(errs[1]) + " undocumented endpoints")
if errs[2] > 0:
print("Found " + str(errs[2]) + " endpoints not matching specs")
# Check for endpoints in FTL that are not in the OpenAPI specs
print("Endpoints in FTL but not in OpenAPI specs:")
for path in FTLendpoints:
if path not in YAMLendpoints["get"]:
# Ignore the docs endpoint
if path in ["/api/docs"]:
continue
print(" Missing GET endpoint in OpenAPI specs: " + path)
errs[1] += 1
if errs[1] == 0:
print(" No missing endpoints\n")
# Exit with an error if there are missing endpoints
if sum(errs) > 0:
exit(1)
# Print the number of missing endpoints (if there are any)
if errs[0] > 0:
print("Found " + str(errs[0]) + " non-implemented endpoints")
if errs[1] > 0:
print("Found " + str(errs[1]) + " undocumented endpoints")
# Exit with an error if there are missing endpoints
if sum(errs) > 0:
exit(1)
# If there are no missing endpoints, exit with success
print("No missing endpoints")
exit(0)
# If there are no missing endpoints, exit with success
print("No missing endpoints")
exit(0)