nodeinfo-test/run.py

171 lines
4.6 KiB
Python
Raw Permalink Normal View History

import os
from urllib.parse import urlparse, urlunparse
import requests
import jsonschema
import logging
LOG_FILE = "nodeinfo.log"
2023-10-01 14:14:32 +05:30
NODEINFO_VERSIONS = {
1.0: "http://nodeinfo.diaspora.software/ns/schema/1.0",
1.1: "http://nodeinfo.diaspora.software/ns/schema/1.1",
2.0: "http://nodeinfo.diaspora.software/ns/schema/2.0",
2.1: "http://nodeinfo.diaspora.software/ns/schema/2.1",
}
def configure_logger():
logger = logging.getLogger("nodeinfo")
logger.setLevel(logging.DEBUG)
fh = logging.FileHandler(LOG_FILE)
fh.setLevel(logging.DEBUG)
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
formatter = logging.Formatter(
"%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
fh.setFormatter(formatter)
ch.setFormatter(formatter)
logger.addHandler(fh)
logger.addHandler(ch)
return logger
def get_env(name) -> str:
env = FTEST_AUTH = os.environ.get(name)
logger.info(f"Environment: {name}: {env}")
if env is None:
raise Exception(
f"Please set environment variable {name}. See https://git.batsense.net/ForgeFlux/nodeinfo-test#environment-variables"
)
return env
logger = configure_logger()
FTEST_AUTH = get_env("FTEST_AUTH")
FTEST_HOST = get_env("FTEST_HOST")
TARGET_HOST = get_env("FTEST_TARGET_HOST")
TEST_HOST = urlparse(TARGET_HOST).netloc
def query_nodeinfo():
parsed_target_host = urlparse(TARGET_HOST)
nodeinfo = urlunparse(
(
parsed_target_host.scheme,
parsed_target_host.netloc,
"/.well-known/nodeinfo",
"",
"",
"",
)
)
logger.info(f"Query nodeinfo: {nodeinfo}")
res = requests.get(nodeinfo)
logger.debug(
f"nodeinfo response:\n\nSTATUS: {res.status_code}\n\nHEADERS:\n {res.headers}\n\nRESPONSE PAYLOAD:\n{res.json()}"
)
assert res.status_code == 200
logger.info("[SUCCESS] nodeinfo query response is HTTP 200")
return res
def test_links(resp):
for link in resp["links"]:
assert "href" in link, "'href' present in link item"
assert "rel" in link, "'rel' present in link item"
logger.info("[SUCESS] links passed schema validation")
2023-10-01 14:14:32 +05:30
def test_schema(source, version):
schema = requests.get(NODEINFO_VERSIONS[version]).json()
resp = requests.get(source).json()
jsonschema.validate(resp, schema)
logger.info(f"[SUCESS] passed Nodeinfo {version} schema validation")
def upload_logs_to_ftest(success: bool, logs: str):
parsed_ftest_host = urlparse(FTEST_HOST)
ftest = urlunparse(
(
parsed_ftest_host.scheme,
parsed_ftest_host.netloc,
f"/api/v1/{FTEST_AUTH}/results",
"",
"",
"",
)
)
logger.info(f"Uploading logs to ftest server {ftest}")
payload = {"success": success, "logs": logs}
res = requests.post(ftest, json=payload, headers={"Origin": "http://example.org"})
if res.status_code == 200:
logger.info("Upload successful")
else:
print(res)
if __name__ == "__main__":
2023-10-01 14:14:32 +05:30
max_score = "NaN"
score = 0
resp = query_nodeinfo()
json = resp.json()
success = []
failures = {}
try:
test_links(json)
score += 1
success.append("test_links")
except Exception as e:
logger.error(e)
failures["test_links"] = e
2023-10-01 14:14:32 +05:30
max_score = 1 + len(json["links"])
for link in json["links"]:
for version in NODEINFO_VERSIONS:
if NODEINFO_VERSIONS[version] == link["rel"]:
test_name = f"test_schema_Nodeinfo_{version}"
try:
test_schema(link["href"], version)
score += 1
success.append(test_name)
except Exception as e:
logger.error(e)
failures[test_name] = e
print("\n\n===============")
if score == max_score:
2023-10-01 14:14:32 +05:30
logger.info("All tests passed")
elif score > 0:
2023-10-01 14:14:32 +05:30
logger.info(f"Partial success. {score} out of {max_score} tests passed")
2023-10-01 14:14:32 +05:30
logger.info("Summary:\n")
logs = ""
if success:
2023-10-01 14:14:32 +05:30
logger.info(f"Successful tests:\n")
for s in success:
log = f"[OK] {s}\n"
2023-10-01 14:14:32 +05:30
logger.info(log)
logs += log
if failures:
2023-10-01 14:14:32 +05:30
logger.error(f"\n\nFailed tests:\n")
for _, (test, error) in enumerate(failures.items()):
log = f"[FAIL] {test} failed with error:\n{error}\n-----\n"
2023-10-01 14:14:32 +05:30
logger.error(log)
logs += log
2023-10-01 14:15:06 +05:30
upload_logs_to_ftest(len(failures) == 0, logs)