95 lines
2.6 KiB
Python
95 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
|
|
from testcontainers import PostgresContainer
|
|
import requests
|
|
import yaml
|
|
import sys
|
|
|
|
pg_versions = [9.2, 9.6, 10, 11, 12, 13, 14, 15, 16, 17]
|
|
|
|
pgmon_port = 93849
|
|
|
|
tests = {}
|
|
|
|
container = None
|
|
|
|
|
|
def std_version(version):
|
|
if version[0] == "9":
|
|
return int(f"{version[0]}0{version[1]}00")
|
|
else:
|
|
return int(f"{version}0000")
|
|
|
|
|
|
def run_test(metric, params, status, check):
|
|
"""
|
|
Validate the return code and restults of a query
|
|
|
|
params:
|
|
metric: The name of the metric to test
|
|
params: A dictionary of query parameters to use when testing
|
|
status: The expected status code
|
|
check: A regular expression to validate the results (or None)
|
|
"""
|
|
result = requests.get(f"http://localhost:{pgmon_port}/{metric}", params=params)
|
|
|
|
if result.status_code != status:
|
|
print(
|
|
f"FAIL: {metric}[{params}] returned wrong status code: {result.status_code}"
|
|
)
|
|
return False
|
|
|
|
if re.match(check, result.text):
|
|
print(f"SUCCESS: {metric}[{params}]")
|
|
return True
|
|
else:
|
|
print(f"FAIL: {metric}[{params}] result is invalid, got:\n {result.text}")
|
|
return False
|
|
|
|
|
|
def run_all_tests(version):
|
|
"""
|
|
Run all defined tests against the current running instance
|
|
|
|
params:
|
|
version: The PostgreSQL version currently being tested (server_version_num format)
|
|
"""
|
|
errors = 0
|
|
|
|
# Convert versions like 12 to 120000
|
|
version_num = std_version(version)
|
|
|
|
# Loop through all of the metrics to test.
|
|
for metric in tests.keys():
|
|
params = metric.get("params", {})
|
|
status = 200
|
|
check = ""
|
|
|
|
# Find the correct version of the status and check parameters (assuming there are any).
|
|
# If there are any check conditions, find the highest version that does not exceed the version we're currently testing against.
|
|
# To do this, we order the keys (versions) in reverse, so we start with the highest.
|
|
for v in reversed(sorted(metric.get("expect", {}).keys())):
|
|
# If we've reached a version <= the one we're testing use it.
|
|
if int(v) <= version_num:
|
|
status = metric["expect"][v]["status"]
|
|
check = metric["expect"][v]["check"]
|
|
break
|
|
|
|
if not run_test(metric, metrics[metric].get(params, {}), status, check):
|
|
errors += 1
|
|
|
|
return errors
|
|
|
|
|
|
def start_test_db(version):
|
|
# container = PostgresContainer()
|
|
pass
|
|
|
|
|
|
# Read the test script
|
|
try:
|
|
with open("metric_tests.yml", "r") as f:
|
|
tests = yaml.safe_load(f)
|
|
except yaml.parser.ParserError as e:
|
|
sys.exit("Failed to parse metrics_test.yml: {e}")
|