Refactor the code of osccore-qa-testing-tools to comply with PEP8.
Related-prod: PRODX-42195
Change-Id: Id05e7584e0d024127ce1bd5042cfe681a1b52e2d
diff --git a/log_helper/log_helper.py b/log_helper/log_helper.py
index 48190c2..8e47a9a 100755
--- a/log_helper/log_helper.py
+++ b/log_helper/log_helper.py
@@ -1,88 +1,123 @@
#!/usr/bin/env python3
+import os
import subprocess
import sys
-import os
-import yaml
from os import path
-import config
+import config
+import yaml
param_is_yaml = False
if len(sys.argv) == 1:
param_is_yaml = path.isfile(config.TEMPEST_REPORT_YAML)
if param_is_yaml is False:
- print('TEMPEST_REPORT_YAML config parameter is not a file')
- raise Exception('TEMPEST_REPORT_YAML config parameter is not a file')
+ print("TEMPEST_REPORT_YAML config parameter is not a file")
+ raise Exception("TEMPEST_REPORT_YAML config parameter is not a file")
def log_gather(resource_id, sub_resource, log_level=None):
- """ Get all log lines related to resource-id
- :param resource_id: ID resource, e.g. request-id, server-id
- :param sub_resource: name of sub_resource log file, e.g subnet.log for neutron resource
- :param log_level: substring for resource_id log: e.g. get only ERROR log level messages, optional
- """
+ """Get all log lines related to resource-id
+ :param resource_id: ID resource, e.g. request-id, server-id
+ :param sub_resource: name of sub_resource log file,
+ e.g subnet.log for neutron resource
+ :param log_level: substring for resource_id log: e.g. get only
+ ERROR log level messages, optional
+ """
try:
directory = os.walk(config.LOG_DIR)
except IndexError:
- print('Parameter <LOG_DIR> is not provided')
- raise ValueError('Parameter <LOG_DIR> is not provided')
+ print("Parameter <LOG_DIR> is not provided")
+ raise ValueError("Parameter <LOG_DIR> is not provided")
if param_is_yaml:
for dirs in directory:
- run_cmd = f"grep -a {resource_id} {dirs[0]}/* >> {config.RESULTS_DIR}/{sub_resource}"
+ run_cmd = (
+ f"grep -a {resource_id} {dirs[0]}/* >> "
+ f"{config.RESULTS_DIR}/{sub_resource}"
+ )
subprocess.run(run_cmd, shell=True)
else:
for dirs in directory:
if log_level:
- run_cmd = f"grep -lE '{resource_id}.*{log_level}|{log_level}.*{resource_id}' {dirs[0]}/* >> '{config.RESULTS_DIR}/tmp.log'"
+ run_cmd = (
+ f"grep -lE '{resource_id}.*{log_level}|{log_level}"
+ f".*{resource_id}' {dirs[0]}/* >> "
+ f"'{config.RESULTS_DIR}/tmp.log'"
+ )
else:
- run_cmd = f"grep -l {resource_id} {dirs[0]}/* >> '{config.RESULTS_DIR}/tmp.log'"
+ run_cmd = (
+ f"grep -l {resource_id} {dirs[0]}/* >> "
+ f"'{config.RESULTS_DIR}/tmp.log'"
+ )
subprocess.run(run_cmd, shell=True)
- with open(config.RESULTS_DIR + '/tmp.log') as f:
+ with open(f"{config.RESULTS_DIR}/tmp.log") as f:
files = f.readlines()
for file in files:
subd = file.split("/")
- log_dir = subd[-4] + "." + subd[-3] + "." + subd[-2]
- log_name = subd[-1].replace('\n', '')
- os.makedirs(os.path.join(config.RESULTS_DIR, sys.argv[1], log_dir), exist_ok=True)
- path = os.path.join(config.RESULTS_DIR, sys.argv[1], log_dir, log_name)
+ log_dir = f"{subd[-4]}.{subd[-3]}.{subd[-2]}"
+ log_name = subd[-1].replace("\n", "")
+ os.makedirs(
+ os.path.join(config.RESULTS_DIR, sys.argv[1], log_dir),
+ exist_ok=True,
+ )
+ path = os.path.join(
+ config.RESULTS_DIR, sys.argv[1], log_dir, log_name
+ )
if log_level:
- run_cmd = f"grep -aE '{resource_id}.*{log_level}|{log_level}.*{resource_id}' {file} >> {path}"
+ run_cmd = (
+ f"grep -aE '{resource_id}.*{log_level}|{log_level}"
+ f".*{resource_id}' {file} >> {path}"
+ )
else:
run_cmd = f"grep -a {resource_id} {file} >> {path}"
- subprocess.run(run_cmd.replace('\n', ''), shell=True)
+ subprocess.run(run_cmd.replace("\n", ""), shell=True)
- os.remove(config.RESULTS_DIR + '/tmp.log')
+ os.remove(f"{config.RESULTS_DIR}/tmp.log")
if param_is_yaml:
- print('Find all the failed tempest tests from YAML file')
+ print("Find all the failed tempest tests from YAML file")
with open(config.TEMPEST_REPORT_YAML) as f:
test_resources = yaml.safe_load(f)
for test in test_resources.items():
# Find all the failed tempest tests from YAML file and gather logs for
# related resources in corresponded folders
- if test[1]['status'] == 'failure':
- print('Collecting logs for ' + test[0])
- os.makedirs(os.path.join(config.RESULTS_DIR, test[0]), exist_ok=True)
- for resource in test[1]['resources']:
- os.makedirs(os.path.join(config.RESULTS_DIR, test[0], resource), exist_ok=True)
- for sub_resource in test[1]['resources'][resource]:
- log_gather(list(test[1]['resources'][resource][sub_resource])[0],
- os.path.join(test[0], resource, sub_resource + '.' + 'log'))
+ if test[1]["status"] == "failure":
+ print(f"Collecting logs for {test[0]}")
+ os.makedirs(
+ os.path.join(config.RESULTS_DIR, test[0]), exist_ok=True
+ )
+ for resource in test[1]["resources"]:
+ os.makedirs(
+ os.path.join(config.RESULTS_DIR, test[0], resource),
+ exist_ok=True,
+ )
+ for sub_resource in test[1]["resources"][resource]:
+ log_gather(
+ list(test[1]["resources"][resource][sub_resource])[0],
+ os.path.join(
+ test[0], resource, sub_resource + "." + "log"
+ ),
+ )
else:
- print('Find all the related log for one specific test or id with error')
+ print("Find all the related log for one specific test or id with error")
os.makedirs(os.path.join(config.RESULTS_DIR, sys.argv[1]), exist_ok=True)
if len(sys.argv) == 3:
- log_gather(sys.argv[1], os.path.join(sys.argv[1], 'test' + '.' + 'log'), log_level=sys.argv[2])
+ log_gather(
+ sys.argv[1],
+ os.path.join(sys.argv[1], "test" + "." + "log"),
+ log_level=sys.argv[2],
+ )
else:
- log_gather(sys.argv[1], os.path.join(sys.argv[1], 'test' + '.' + 'log'))
+ log_gather(
+ sys.argv[1], os.path.join(sys.argv[1], "test" + "." + "log")
+ )
-print('The logger is finished')
+print("The logger is finished")