feat: import from ff/nodeinfo-test and ff/webfinger-test

This commit is contained in:
Aravinth Manivannan 2023-10-01 19:58:34 +05:30
parent e5a9ef0d7c
commit b20252f3e1
Signed by: realaravinth
GPG key ID: F8F50389936984FF
5 changed files with 129 additions and 0 deletions

5
common/__init__.py Normal file
View file

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

23
common/env.py Normal file
View file

@ -0,0 +1,23 @@
# 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")

50
common/logger.py Normal file
View file

@ -0,0 +1,50 @@
# SPDX-FileCopyrightText: 2023 Aravinth Manivannan <realaravinth@batsense.net>
#
# SPDX-License-Identifier: AGPL-3.0-or-later
import logging
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):
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)

18
common/obj.py Normal file
View file

@ -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()

33
common/webfinger.py Normal file
View file

@ -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()