diff --git a/.env_sample b/.env_sample new file mode 100644 index 0000000..01d3fc8 --- /dev/null +++ b/.env_sample @@ -0,0 +1,4 @@ +export FTEST_AUTH="foobar" +export FTEST_HOST="http://ftest.example.org" +export FTEST_TARGET_HOST="http://localhost:3000" +export FTEST_USER="john@example.org" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..73e0417 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,24 @@ +attrs==23.1.0 +certifi==2023.7.22 +charset-normalizer==3.3.0 +idna==3.4 +jsonschema==4.19.1 +jsonschema-specifications==2023.7.1 +referencing==0.30.2 +requests==2.31.0 +rpds-py==0.10.3 +urllib3==2.0.5 +attrs==23.1.0 +cachetools==5.3.1 +certifi==2023.7.22 +charset-normalizer==3.3.0 +frozendict==2.3.8 +idna==3.4 +jsonschema==4.19.1 +jsonschema-specifications==2023.7.1 +lxml==4.9.3 +PyLD==2.0.3 +referencing==0.30.2 +requests==2.31.0 +rpds-py==0.10.3 +urllib3==2.0.5 diff --git a/run.py b/run.py new file mode 100644 index 0000000..4db11d1 --- /dev/null +++ b/run.py @@ -0,0 +1,146 @@ +import os +from urllib.parse import urlparse, urlunparse + +import requests +import jsonschema +import logging + +LOG_FILE = "nodeinfo.log" + + +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_USER = get_env("FTEST_USER") # actor ex: john@example.org +print(f"TEST USER: {TEST_USER}") +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): + self_link = None + profile_link = None + + 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") + + +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__": + max_score = 5 + score = 0 + resp = query_nodeinfo() + json = resp.json() + score += 1 + + success = [] + failures = {} + + try: + test_links(json) + score += 1 + success.append("test_links") + except Exception as e: + logger.error(e) + failures["test_links"] = e + + print("\n\n===============") + if score == max_score: + print("All tests passed") + elif score > 0: + print(f"Partial success. {score} out of {max_score} tests passed") + + print("Summary:\n") + + logs = "" + + if success: + print(f"Successful tests:\n") + for s in success: + log = f"[OK] {s}\n" + print(log) + logs += log + + if failures: + print(f"\n\nFailed tests:\n") + for _, (test, error) in enumerate(failures.items()): + log = f"[FAIL] {test} failed with error:\n{error}\n-----\n" + print(log) + logs += log + +# upload_logs_to_ftest(len(failures) == 0, logs)