feat: import code from ForgeFlux/ap-test/common

ftest_common/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2023 Aravinth Manivannan <realaravinth@batsense.net>
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later

ftest_common/env.py

@@ -0,0 +1,22 @@
+# SPDX-FileCopyrightText: 2023 Aravinth Manivannan <realaravinth@batsense.net>
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+import os
+
+from .logger import 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
+
+FTEST_AUTH = get_env("FTEST_AUTH")
+FTEST_HOST = get_env("FTEST_HOST")
+FTEST_USER = get_env("FTEST_USER")
+TARGET_HOST = get_env("FTEST_TARGET_HOST")

ftest_common/logger.py

@@ -0,0 +1,54 @@
+# SPDX-FileCopyrightText: 2023 Aravinth Manivannan <realaravinth@batsense.net>
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+from urllib.parse import urlparse, urlunparse
+import logging
+
+import requests
+
+LOG_FILE = "ap.log"
+
+
+def configure_logger():
+    logger = logging.getLogger("ap")
+    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
+
+
+logger = configure_logger()
+
+
+def upload_logs_to_ftest(success: bool, logs: str):
+    from .env import FTEST_HOST, FTEST_AUTH
+
+    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)

ftest_common/obj.py

@@ -0,0 +1,18 @@
+# SPDX-FileCopyrightText: 2023 Aravinth Manivannan <realaravinth@batsense.net>
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+import os
+
+import requests
+
+from .logger import logger
+
+
+def get_ap_obj(url):
+    logger.info(f"Getting object: {url}")
+    res = requests.get(url, headers={"Accept": "application/activity+json"})
+    msg = f"AP server response:\n\nSTATUS: {res.status_code}\n\nHEADERS:\n {res.headers}\n\nRESPONSE PAYLOAD:\n{res.text}"
+    logger.debug(msg)
+    assert res.status_code == 200, f"Request FAILURE. {msg}"
+    return res.json()

ftest_common/webfinger.py

@@ -0,0 +1,33 @@
+# SPDX-FileCopyrightText: 2023 Aravinth Manivannan <realaravinth@batsense.net>
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+from urllib.parse import urlparse, urlunparse
+
+import requests
+
+from .logger import logger
+
+
+def get_webfinger():
+    from .env import TARGET_HOST, FTEST_USER
+
+    parsed_target_host = urlparse(TARGET_HOST)
+    webfinger = urlunparse(
+        (
+            parsed_target_host.scheme,
+            parsed_target_host.netloc,
+            "/.well-known/webfinger",
+            "",
+            f"resource=acct:{FTEST_USER}",
+            "",
+        )
+    )
+    logger.info(f"Query webfinger: {webfinger}")
+    res = requests.get(webfinger)
+    logger.debug(
+        f"webfinger 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] webfinger query response is HTTP 200")
+    return res.json()