tools.testing: Add way to re-play recorded queries in gatekeeper

to simplify gatekeeper development add support for DEBUG_INPUT which can
be used to report content from files gathered in DEBUG run.

Signed-off-by: Lukáš Doktor <ldoktor@redhat.com>
This commit is contained in:
Lukáš Doktor 2025-05-15 10:31:30 +02:00
parent 1a15990ee1
commit 9f8c8ea851
No known key found for this signature in database
GPG Key ID: 26B362E47FCF22C1

View File

@ -43,6 +43,10 @@ if os.environ.get("DEBUG", "false") == "true":
os.makedirs(DEBUG_DIR) os.makedirs(DEBUG_DIR)
else: else:
DEBUG_DIR = None DEBUG_DIR = None
if os.environ.get("DEBUG_INPUT", "") == "":
DEBUG_INPUT = None
else:
DEBUG_INPUT = os.path.abspath(os.environ["DEBUG_INPUT"])
class Checker: class Checker:
@ -153,10 +157,15 @@ class Checker:
def fetch_json_from_url(self, url, task, params=None): def fetch_json_from_url(self, url, task, params=None):
"""Fetches URL and reports json output""" """Fetches URL and reports json output"""
print(url, file=sys.stderr) print(url, file=sys.stderr)
response = requests.get(url, headers=_GH_HEADERS, params=params, if DEBUG_INPUT:
timeout=60) with open(f"{os.path.join(DEBUG_INPUT, task)}.json", "r",
response.raise_for_status() encoding="utf8") as inp:
output = response.json() output = json.load(inp)
else:
response = requests.get(url, headers=_GH_HEADERS, params=params,
timeout=60)
response.raise_for_status()
output = response.json()
if DEBUG_DIR: if DEBUG_DIR:
with open(f"{os.path.join(DEBUG_DIR, task)}.json", "w", with open(f"{os.path.join(DEBUG_DIR, task)}.json", "w",
encoding="utf8") as out: encoding="utf8") as out: