From 90b45538f23edcee1f6cdcbcf3f75ca852715bef Mon Sep 17 00:00:00 2001 From: Sandro Bonazzola Date: Fri, 2 Sep 2022 11:32:40 +0200 Subject: [PATCH] pycodestyle fixes: kraken/cerberus/setup.py Signed-off-by: Sandro Bonazzola --- kraken/cerberus/setup.py | 79 +++++++++++++++++++++++++++++++--------- 1 file changed, 61 insertions(+), 18 deletions(-) diff --git a/kraken/cerberus/setup.py b/kraken/cerberus/setup.py index ae1cb7ac..26efc152 100644 --- a/kraken/cerberus/setup.py +++ b/kraken/cerberus/setup.py @@ -4,30 +4,45 @@ import sys import json -# Get cerberus status def get_status(config, start_time, end_time): + """ + Get cerberus status + """ cerberus_status = True check_application_routes = False application_routes_status = True if config["cerberus"]["cerberus_enabled"]: cerberus_url = config["cerberus"]["cerberus_url"] - check_application_routes = config["cerberus"]["check_applicaton_routes"] + check_application_routes = \ + config["cerberus"]["check_applicaton_routes"] if not cerberus_url: - logging.error("url where Cerberus publishes True/False signal is not provided.") + logging.error( + "url where Cerberus publishes True/False signal " + "is not provided." + ) sys.exit(1) cerberus_status = requests.get(cerberus_url, timeout=60).content cerberus_status = True if cerberus_status == b"True" else False - # Fail if the application routes monitored by cerberus experience downtime during the chaos + # Fail if the application routes monitored by cerberus + # experience downtime during the chaos if check_application_routes: - application_routes_status, unavailable_routes = application_status(cerberus_url, start_time, end_time) + application_routes_status, unavailable_routes = application_status( + cerberus_url, + start_time, + end_time + ) if not application_routes_status: logging.error( - "Application routes: %s monitored by cerberus encountered downtime during the run, failing" + "Application routes: %s monitored by cerberus " + "encountered downtime during the run, failing" % unavailable_routes ) else: - logging.info("Application routes being monitored didn't encounter any downtime during the run!") + logging.info( + "Application routes being monitored " + "didn't encounter any downtime during the run!" + ) if not cerberus_status: logging.error( @@ -39,42 +54,65 @@ def get_status(config, start_time, end_time): if not application_routes_status or not cerberus_status: sys.exit(1) else: - logging.info("Received a go signal from Ceberus, the cluster is healthy. " "Test passed.") + logging.info( + "Received a go signal from Ceberus, the cluster is healthy. " + "Test passed." + ) return cerberus_status -# Function to publish kraken status to cerberus def publish_kraken_status(config, failed_post_scenarios, start_time, end_time): + """ + Publish kraken status to cerberus + """ cerberus_status = get_status(config, start_time, end_time) if not cerberus_status: if failed_post_scenarios: if config["kraken"]["exit_on_failure"]: logging.info( - "Cerberus status is not healthy and post action scenarios " "are still failing, exiting kraken run" + "Cerberus status is not healthy and post action scenarios " + "are still failing, exiting kraken run" ) sys.exit(1) else: - logging.info("Cerberus status is not healthy and post action scenarios " "are still failing") + logging.info( + "Cerberus status is not healthy and post action scenarios " + "are still failing" + ) else: if failed_post_scenarios: if config["kraken"]["exit_on_failure"]: logging.info( - "Cerberus status is healthy but post action scenarios " "are still failing, exiting kraken run" + "Cerberus status is healthy but post action scenarios " + "are still failing, exiting kraken run" ) sys.exit(1) else: - logging.info("Cerberus status is healthy but post action scenarios " "are still failing") + logging.info( + "Cerberus status is healthy but post action scenarios " + "are still failing" + ) -# Check application availability def application_status(cerberus_url, start_time, end_time): + """ + Check application availability + """ if not cerberus_url: - logging.error("url where Cerberus publishes True/False signal is not provided.") + logging.error( + "url where Cerberus publishes True/False signal is not provided." + ) sys.exit(1) else: duration = (end_time - start_time) / 60 - url = cerberus_url + "/" + "history" + "?" + "loopback=" + str(duration) - logging.info("Scraping the metrics for the test duration from cerberus url: %s" % url) + url = "{baseurl}/history?loopback={duration}".format( + baseurl=cerberus_url, + duration=str(duration) + ) + logging.info( + "Scraping the metrics for the test " + "duration from cerberus url: %s" % url + ) try: failed_routes = [] status = True @@ -88,6 +126,11 @@ def application_status(cerberus_url, start_time, end_time): else: continue except Exception as e: - logging.error("Failed to scrape metrics from cerberus API at %s: %s" % (url, e)) + logging.error( + "Failed to scrape metrics from cerberus API at %s: %s" % ( + url, + e + ) + ) sys.exit(1) return status, set(failed_routes)