scripts/get_unit_test_report.py - openbmc/openbmc-build-scripts - Gitiles

 #!/usr/bin/python

 # This script generates the unit test coverage report for openbmc project.
 #
 # Usage:
 # get_unit_test_report.py target_dir [url_file]
 #
 # Positional arguments:
 # target_dir  Target directory in pwd to place all cloned repos and logs.
 # url_file    Text file containing url of repositories. Optional.
 #             By using this argument, the user can get a report only for
 #             specific repositories given in the file.
 #             Refer ./scripts/repositories.txt
 #
 # Examples:
 #     get_unit_test_report.py target_dir
 #     get_unit_test_report.py target_dir repositories.txt
 #
 # Output format:
 #
 # ***********************************OUTPUT***********************************
 # https://github.com/openbmc/phosphor-dbus-monitor.git               NO
 # https://github.com/openbmc/phosphor-sel-logger.git;protocol=git    NO
 # ***********************************OUTPUT***********************************
 #
 # Other outputs and errors are redirected to output.log and debug.log in target_dir.

 import argparse
 import logging
 import os
 import re
 import requests
 import shutil
 import sys
 import subprocess

 # Repo list not expected to contain UT. Will be moved to a file in future.
 skip_list = ["openbmc-tools", "inarp", "openbmc", "openbmc.github.io",
              "phosphor-ecc", "phosphor-pcie-presence", "phosphor-u-boot-env-mgr",
              "rrd-ipmi-blob", "librrdplus", "openpower-inventory-upload",
              "openpower-logging", "openpower-power-control", "docs",
              "openbmc-test-automation", "openbmc-build-scripts", "skeleton",
              "linux",
              # Not active, expected to be archived soon.
              "ibm-pldm-oem"]


 # Create parser.
 text = '''%(prog)s target_dir [url_file]

 Example usages:
 get_unit_test_report.py target_dir
 get_unit_test_report.py target_dir repositories.txt'''

 parser = argparse.ArgumentParser(usage=text,
                                  description="Script generates the unit test coverage report")
 parser.add_argument("target_dir", type=str,
                     help='''Name of a non-existing directory in pwd to store all
                             cloned repos, logs and UT reports''')
 parser.add_argument("url_file", type=str, nargs='?',
                     help='''Text file containing url of repositories.
                             By using this argument, the user can get a report only for
                             specific repositories given in the file.
                             Refer ./scripts/repositories.txt''')
 args = parser.parse_args()

 input_urls = []
 if args.url_file:
     try:
         # Get URLs from the file.
         with open(args.url_file) as reader:
             file_content = reader.read().splitlines()
             input_urls = list(filter(lambda x:x, file_content))
         if not(input_urls):
             print("Input file {} is empty. Quitting...".format(args.url_file))
             quit()
     except IOError as e:
         print("Issue in reading file '{}'. Reason: {}".format(args.url_file,
                                                                   str(e)))
         quit()


 # Create target working directory.
 pwd = os.getcwd()
 working_dir = os.path.join(pwd, args.target_dir)
 try:
     os.mkdir(working_dir)
 except OSError as e:
     answer = input("Target directory " + working_dir + " already exists. "
                        + "Do you want to delete [Y/N]: ")
     if answer == "Y":
         try:
             shutil.rmtree(working_dir)
             os.mkdir(working_dir)
         except OSError as e:
             print(str(e))
             quit()
     else:
         print("Exiting....")
         quit()

 # Create log directory.
 log_dir = os.path.join(working_dir, "logs")
 try:
     os.mkdir(log_dir)
 except OSError as e:
     print("Unable to create log directory: " + log_dir)
     print(str(e))
     quit()


 # Log files
 debug_file = os.path.join(log_dir, "debug.log")
 output_file = os.path.join(log_dir, "output.log")
 logging.basicConfig(format='%(levelname)s - %(message)s', level=logging.DEBUG,
                     filename=debug_file)
 logger = logging.getLogger(__name__)

 # Create handlers
 console_handler = logging.StreamHandler()
 file_handler = logging.FileHandler(output_file)
 console_handler.setLevel(logging.INFO)
 file_handler.setLevel(logging.INFO)

 # Create formatters and add it to handlers
 log_format = logging.Formatter('%(message)s')
 console_handler.setFormatter(log_format)
 file_handler.setFormatter(log_format)

 # Add handlers to the logger
 logger.addHandler(console_handler)
 logger.addHandler(file_handler)


 # Create report directory.
 report_dir = os.path.join(working_dir, "reports")
 try:
     os.mkdir(report_dir)
 except OSError as e:
     logger.error("Unable to create report directory: " + report_dir)
     logger.error(str(e))
     quit()

 # Clone OpenBmc build scripts.
 try:
     output = subprocess.check_output("git clone https://github.com/openbmc/openbmc-build-scripts.git",
                                      shell=True, cwd=working_dir, stderr=subprocess.STDOUT)
     logger.debug(output)
 except subprocess.CalledProcessError as e:
     logger.error(e.output)
     logger.error(e.cmd)
     logger.error("Unable to clone openbmc-build-scripts")
     quit()

 repo_data = []
 if input_urls:
     api_url = "https://api.github.com/repos/openbmc/"
     for url in input_urls:
         try:
             repo_name = url.strip().split('/')[-1].split(";")[0].split(".")[0]
         except IndexError as e:
             logger.error("ERROR: Unable to get sandbox name for url " + url)
             logger.error("Reason: " + str(e))
             continue

         try:
             resp = requests.get(api_url + repo_name)
             if resp.status_code != 200:
                 logger.info(api_url + repo_name + " ==> " + resp.reason)
                 continue
             repo_data.extend([resp.json()])
         except ValueError as e:
             logger.error("ERROR: Failed to get response for " + repo_name)
             logger.error(resp)
             continue

 else:
     # Get number of pages.
     resp = requests.head('https://api.github.com/users/openbmc/repos')
     if resp.status_code != 200:
         logger.error("Error! Unable to get repositories")
         logger.error(resp.status_code)
         logger.error(resp.reason)
         quit()
     num_of_pages = int(resp.links['last']['url'].split('page=')[-1])
     logger.debug("No. of pages: " + str(num_of_pages))

     # Fetch data from all pages.
     for page in range(1, num_of_pages+1):
         resp = requests.get('https://api.github.com/users/openbmc/repos?page='
                             + str(page))
         data = resp.json()
         repo_data.extend(data)


 # Get URLs and their archive status from response.
 url_info = {}
 for repo in repo_data:
     try:
         url_info[repo["clone_url"]] = repo["archived"]
     except KeyError as e:
         logger.error("Failed to get archived status of {}".format(repo))
         url_info[repo["clone_url"]] = False
         continue
 logger.debug(url_info)
 repo_count = len(url_info)
 logger.info("Number of repositories (Including archived): " + str(repo_count))

 # Clone repository and run unit test.
 coverage_report = []
 counter = 0
 tested_report_count = 0
 coverage_count = 0
 unit_test_count = 0
 no_report_count = 0
 error_count = 0
 skip_count = 0
 archive_count = 0
 url_list = sorted(url_info)
 for url in url_list:
     ut_status = "NO"
     skip = False
     if url_info[url]:
         ut_status = "ARCHIVED"
         skip = True
     else:
         try:
             # Eg: url = "https://github.com/openbmc/u-boot.git"
             #     sandbox_name = "u-boot"
             sandbox_name = url.strip().split('/')[-1].split(";")[0].split(".")[0]
         except IndexError as e:
             logger.error("ERROR: Unable to get sandbox name for url " + url)
             logger.error("Reason: " + str(e))
             continue

         if (sandbox_name in skip_list or
             re.match(r'meta-', sandbox_name)):
             logger.debug("SKIPPING: " + sandbox_name)
             skip = True
             ut_status = "SKIPPED"
         else:
             checkout_cmd = "rm -rf " + sandbox_name + ";git clone " + url
             try:
                 subprocess.check_output(checkout_cmd, shell=True, cwd=working_dir,
                                                  stderr=subprocess.STDOUT)
             except subprocess.CalledProcessError as e:
                 logger.debug(e.output)
                 logger.debug(e.cmd)
                 logger.debug("Failed to clone " + sandbox_name)
                 ut_status = "ERROR"
                 skip = True
     if not(skip):
         docker_cmd = "WORKSPACE=$(pwd) UNIT_TEST_PKG=" + sandbox_name + " " + \
                      "./openbmc-build-scripts/run-unit-test-docker.sh"
         try:
             result = subprocess.check_output(docker_cmd, cwd=working_dir, shell=True,
                                              stderr=subprocess.STDOUT)
             logger.debug(result)
             logger.debug("UT BUILD COMPLETED FOR: " + sandbox_name)

         except subprocess.CalledProcessError as e:
             logger.debug(e.output)
             logger.debug(e.cmd)
             logger.debug("UT BUILD EXITED FOR: " + sandbox_name)
             ut_status = "ERROR"

         folder_name = os.path.join(working_dir, sandbox_name)
         repo_report_dir = os.path.join(report_dir, sandbox_name)

         report_names = ("coveragereport", "test-suite.log", "LastTest.log")
         find_cmd = "".join("find " + folder_name + " -name " + report + ";"
                            for report in report_names)
         try:
             result = subprocess.check_output(find_cmd, shell=True)
             result = result.decode("utf-8")
         except subprocess.CalledProcessError as e:
             logger.debug(e.output)
             logger.debug(e.cmd)
             logger.debug("CMD TO FIND REPORT FAILED FOR: " + sandbox_name)
             ut_status = "ERROR"

         if ut_status != "ERROR":
             if result:
                 if result.__contains__("coveragereport"):
                     ut_status = "YES, COVERAGE"
                     coverage_count += 1
                 elif "test-suite.log" in result:
                     ut_status = "YES, UNIT TEST"
                     unit_test_count += 1
                 elif "LastTest.log" in result:
                     file_names = result.splitlines()
                     for file in file_names:
                         pattern_count_cmd = "sed -n '/Start testing/,/End testing/p;' " + \
                               file + "|wc -l"
                         try:
                             num_of_lines = subprocess.check_output(pattern_count_cmd,
                                                                    shell=True)
                         except subprocess.CalledProcessError as e:
                             logger.debug(e.output)
                             logger.debug(e.cmd)
                             logger.debug("CONTENT CHECK FAILED FOR: " + sandbox_name)
                             ut_status = "ERROR"

                         if int(num_of_lines.strip()) > 5:
                             ut_status = "YES, UNIT TEST"
                             unit_test_count += 1

         if "YES" in ut_status:
             tested_report_count += 1
             result = result.splitlines()
             for file_path in result:
                 destination = os.path.dirname(os.path.join(report_dir,
                                                            os.path.relpath(file_path,
                                                                            working_dir)))
                 copy_cmd = "mkdir -p " + destination + ";cp -rf " + \
                            file_path.strip() + " " + destination
                 try:
                     subprocess.check_output(copy_cmd, shell=True)
                 except subprocess.CalledProcessError as e:
                     logger.debug(e.output)
                     logger.debug(e.cmd)
                     logger.info("FAILED TO COPY REPORTS FOR: " + sandbox_name)

     if ut_status == "ERROR":
         error_count += 1
     elif ut_status == "NO":
         no_report_count += 1
     elif ut_status == "SKIPPED":
         skip_count += 1
     elif ut_status == "ARCHIVED":
         archive_count += 1

     coverage_report.append("{:<65}{:<10}".format(url.strip(), ut_status))
     counter += 1
     logger.info(str(counter) + " in " + str(repo_count) + " completed")

 logger.info("*" * 30 + "UNIT TEST COVERAGE REPORT" + "*" * 30)
 for res in coverage_report:
     logger.info(res)
 logger.info("*" * 30 + "UNIT TEST COVERAGE REPORT" + "*" * 30)

 logger.info("REPORTS: " + report_dir)
 logger.info("LOGS: " + log_dir)
 logger.info("*" * 85)
 logger.info("SUMMARY: ")
 logger.info("TOTAL REPOSITORIES     : " + str(repo_count))
 logger.info("TESTED REPOSITORIES    : " + str(tested_report_count))
 logger.info("ERROR                  : " + str(error_count))
 logger.info("COVERAGE REPORT        : " + str(coverage_count))
 logger.info("UNIT TEST REPORT       : " + str(unit_test_count))
 logger.info("NO REPORT              : " + str(no_report_count))
 logger.info("SKIPPED                : " + str(skip_count))
 logger.info("ARCHIVED               : " + str(archive_count))
 logger.info("*" * 85)
	#!/usr/bin/python

	# This script generates the unit test coverage report for openbmc project.
	#
	# Usage:
	# get_unit_test_report.py target_dir [url_file]
	#
	# Positional arguments:
	# target_dir Target directory in pwd to place all cloned repos and logs.
	# url_file Text file containing url of repositories. Optional.
	# By using this argument, the user can get a report only for
	# specific repositories given in the file.
	# Refer ./scripts/repositories.txt
	#
	# Examples:
	# get_unit_test_report.py target_dir
	# get_unit_test_report.py target_dir repositories.txt
	#
	# Output format:
	#
	# *********************************OUTPUT*********************************
	# https://github.com/openbmc/phosphor-dbus-monitor.git NO
	# https://github.com/openbmc/phosphor-sel-logger.git;protocol=git NO
	# *********************************OUTPUT*********************************
	#
	# Other outputs and errors are redirected to output.log and debug.log in target_dir.

	import argparse
	import logging
	import os
	import re
	import requests
	import shutil
	import sys
	import subprocess

	# Repo list not expected to contain UT. Will be moved to a file in future.
	skip_list = ["openbmc-tools", "inarp", "openbmc", "openbmc.github.io",
	"phosphor-ecc", "phosphor-pcie-presence", "phosphor-u-boot-env-mgr",
	"rrd-ipmi-blob", "librrdplus", "openpower-inventory-upload",
	"openpower-logging", "openpower-power-control", "docs",
	"openbmc-test-automation", "openbmc-build-scripts", "skeleton",
	"linux",
	# Not active, expected to be archived soon.
	"ibm-pldm-oem"]


	# Create parser.
	text = '''%(prog)s target_dir [url_file]

	Example usages:
	get_unit_test_report.py target_dir
	get_unit_test_report.py target_dir repositories.txt'''

	parser = argparse.ArgumentParser(usage=text,
	description="Script generates the unit test coverage report")
	parser.add_argument("target_dir", type=str,
	help='''Name of a non-existing directory in pwd to store all
	cloned repos, logs and UT reports''')
	parser.add_argument("url_file", type=str, nargs='?',
	help='''Text file containing url of repositories.
	By using this argument, the user can get a report only for
	specific repositories given in the file.
	Refer ./scripts/repositories.txt''')
	args = parser.parse_args()

	input_urls = []
	if args.url_file:
	try:
	# Get URLs from the file.
	with open(args.url_file) as reader:
	file_content = reader.read().splitlines()
	input_urls = list(filter(lambda x:x, file_content))
	if not(input_urls):
	print("Input file {} is empty. Quitting...".format(args.url_file))
	quit()
	except IOError as e:
	print("Issue in reading file '{}'. Reason: {}".format(args.url_file,
	str(e)))
	quit()


	# Create target working directory.
	pwd = os.getcwd()
	working_dir = os.path.join(pwd, args.target_dir)
	try:
	os.mkdir(working_dir)
	except OSError as e:
	answer = input("Target directory " + working_dir + " already exists. "
	+ "Do you want to delete [Y/N]: ")
	if answer == "Y":
	try:
	shutil.rmtree(working_dir)
	os.mkdir(working_dir)
	except OSError as e:
	print(str(e))
	quit()
	else:
	print("Exiting....")
	quit()

	# Create log directory.
	log_dir = os.path.join(working_dir, "logs")
	try:
	os.mkdir(log_dir)
	except OSError as e:
	print("Unable to create log directory: " + log_dir)
	print(str(e))
	quit()


	# Log files
	debug_file = os.path.join(log_dir, "debug.log")
	output_file = os.path.join(log_dir, "output.log")
	logging.basicConfig(format='%(levelname)s - %(message)s', level=logging.DEBUG,
	filename=debug_file)
	logger = logging.getLogger(__name__)

	# Create handlers
	console_handler = logging.StreamHandler()
	file_handler = logging.FileHandler(output_file)
	console_handler.setLevel(logging.INFO)
	file_handler.setLevel(logging.INFO)

	# Create formatters and add it to handlers
	log_format = logging.Formatter('%(message)s')
	console_handler.setFormatter(log_format)
	file_handler.setFormatter(log_format)

	# Add handlers to the logger
	logger.addHandler(console_handler)
	logger.addHandler(file_handler)


	# Create report directory.
	report_dir = os.path.join(working_dir, "reports")
	try:
	os.mkdir(report_dir)
	except OSError as e:
	logger.error("Unable to create report directory: " + report_dir)
	logger.error(str(e))
	quit()

	# Clone OpenBmc build scripts.
	try:
	output = subprocess.check_output("git clone https://github.com/openbmc/openbmc-build-scripts.git",
	shell=True, cwd=working_dir, stderr=subprocess.STDOUT)
	logger.debug(output)
	except subprocess.CalledProcessError as e:
	logger.error(e.output)
	logger.error(e.cmd)
	logger.error("Unable to clone openbmc-build-scripts")
	quit()

	repo_data = []
	if input_urls:
	api_url = "https://api.github.com/repos/openbmc/"
	for url in input_urls:
	try:
	repo_name = url.strip().split('/')[-1].split(";")[0].split(".")[0]
	except IndexError as e:
	logger.error("ERROR: Unable to get sandbox name for url " + url)
	logger.error("Reason: " + str(e))
	continue

	try:
	resp = requests.get(api_url + repo_name)
	if resp.status_code != 200:
	logger.info(api_url + repo_name + " ==> " + resp.reason)
	continue
	repo_data.extend([resp.json()])
	except ValueError as e:
	logger.error("ERROR: Failed to get response for " + repo_name)
	logger.error(resp)
	continue

	else:
	# Get number of pages.
	resp = requests.head('https://api.github.com/users/openbmc/repos')
	if resp.status_code != 200:
	logger.error("Error! Unable to get repositories")
	logger.error(resp.status_code)
	logger.error(resp.reason)
	quit()
	num_of_pages = int(resp.links['last']['url'].split('page=')[-1])
	logger.debug("No. of pages: " + str(num_of_pages))

	# Fetch data from all pages.
	for page in range(1, num_of_pages+1):
	resp = requests.get('https://api.github.com/users/openbmc/repos?page='
	+ str(page))
	data = resp.json()
	repo_data.extend(data)


	# Get URLs and their archive status from response.
	url_info = {}
	for repo in repo_data:
	try:
	url_info[repo["clone_url"]] = repo["archived"]
	except KeyError as e:
	logger.error("Failed to get archived status of {}".format(repo))
	url_info[repo["clone_url"]] = False
	continue
	logger.debug(url_info)
	repo_count = len(url_info)
	logger.info("Number of repositories (Including archived): " + str(repo_count))

	# Clone repository and run unit test.
	coverage_report = []
	counter = 0
	tested_report_count = 0
	coverage_count = 0
	unit_test_count = 0
	no_report_count = 0
	error_count = 0
	skip_count = 0
	archive_count = 0
	url_list = sorted(url_info)
	for url in url_list:
	ut_status = "NO"
	skip = False
	if url_info[url]:
	ut_status = "ARCHIVED"
	skip = True
	else:
	try:
	# Eg: url = "https://github.com/openbmc/u-boot.git"
	# sandbox_name = "u-boot"
	sandbox_name = url.strip().split('/')[-1].split(";")[0].split(".")[0]
	except IndexError as e:
	logger.error("ERROR: Unable to get sandbox name for url " + url)
	logger.error("Reason: " + str(e))
	continue

	if (sandbox_name in skip_list or
	re.match(r'meta-', sandbox_name)):
	logger.debug("SKIPPING: " + sandbox_name)
	skip = True
	ut_status = "SKIPPED"
	else:
	checkout_cmd = "rm -rf " + sandbox_name + ";git clone " + url
	try:
	subprocess.check_output(checkout_cmd, shell=True, cwd=working_dir,
	stderr=subprocess.STDOUT)
	except subprocess.CalledProcessError as e:
	logger.debug(e.output)
	logger.debug(e.cmd)
	logger.debug("Failed to clone " + sandbox_name)
	ut_status = "ERROR"
	skip = True
	if not(skip):
	docker_cmd = "WORKSPACE=$(pwd) UNIT_TEST_PKG=" + sandbox_name + " " + \
	"./openbmc-build-scripts/run-unit-test-docker.sh"
	try:
	result = subprocess.check_output(docker_cmd, cwd=working_dir, shell=True,
	stderr=subprocess.STDOUT)
	logger.debug(result)
	logger.debug("UT BUILD COMPLETED FOR: " + sandbox_name)

	except subprocess.CalledProcessError as e:
	logger.debug(e.output)
	logger.debug(e.cmd)
	logger.debug("UT BUILD EXITED FOR: " + sandbox_name)
	ut_status = "ERROR"

	folder_name = os.path.join(working_dir, sandbox_name)
	repo_report_dir = os.path.join(report_dir, sandbox_name)

	report_names = ("coveragereport", "test-suite.log", "LastTest.log")
	find_cmd = "".join("find " + folder_name + " -name " + report + ";"
	for report in report_names)
	try:
	result = subprocess.check_output(find_cmd, shell=True)
	result = result.decode("utf-8")
	except subprocess.CalledProcessError as e:
	logger.debug(e.output)
	logger.debug(e.cmd)
	logger.debug("CMD TO FIND REPORT FAILED FOR: " + sandbox_name)
	ut_status = "ERROR"

	if ut_status != "ERROR":
	if result:
	if result.__contains__("coveragereport"):
	ut_status = "YES, COVERAGE"
	coverage_count += 1
	elif "test-suite.log" in result:
	ut_status = "YES, UNIT TEST"
	unit_test_count += 1
	elif "LastTest.log" in result:
	file_names = result.splitlines()
	for file in file_names:
	pattern_count_cmd = "sed -n '/Start testing/,/End testing/p;' " + \
	file + "\|wc -l"
	try:
	num_of_lines = subprocess.check_output(pattern_count_cmd,
	shell=True)
	except subprocess.CalledProcessError as e:
	logger.debug(e.output)
	logger.debug(e.cmd)
	logger.debug("CONTENT CHECK FAILED FOR: " + sandbox_name)
	ut_status = "ERROR"

	if int(num_of_lines.strip()) > 5:
	ut_status = "YES, UNIT TEST"
	unit_test_count += 1

	if "YES" in ut_status:
	tested_report_count += 1
	result = result.splitlines()
	for file_path in result:
	destination = os.path.dirname(os.path.join(report_dir,
	os.path.relpath(file_path,
	working_dir)))
	copy_cmd = "mkdir -p " + destination + ";cp -rf " + \
	file_path.strip() + " " + destination
	try:
	subprocess.check_output(copy_cmd, shell=True)
	except subprocess.CalledProcessError as e:
	logger.debug(e.output)
	logger.debug(e.cmd)
	logger.info("FAILED TO COPY REPORTS FOR: " + sandbox_name)

	if ut_status == "ERROR":
	error_count += 1
	elif ut_status == "NO":
	no_report_count += 1
	elif ut_status == "SKIPPED":
	skip_count += 1
	elif ut_status == "ARCHIVED":
	archive_count += 1

	coverage_report.append("{:<65}{:<10}".format(url.strip(), ut_status))
	counter += 1
	logger.info(str(counter) + " in " + str(repo_count) + " completed")

	logger.info("" 30 + "UNIT TEST COVERAGE REPORT" + "" 30)
	for res in coverage_report:
	logger.info(res)
	logger.info("" 30 + "UNIT TEST COVERAGE REPORT" + "" 30)

	logger.info("REPORTS: " + report_dir)
	logger.info("LOGS: " + log_dir)
	logger.info("" 85)
	logger.info("SUMMARY: ")
	logger.info("TOTAL REPOSITORIES : " + str(repo_count))
	logger.info("TESTED REPOSITORIES : " + str(tested_report_count))
	logger.info("ERROR : " + str(error_count))
	logger.info("COVERAGE REPORT : " + str(coverage_count))
	logger.info("UNIT TEST REPORT : " + str(unit_test_count))
	logger.info("NO REPORT : " + str(no_report_count))
	logger.info("SKIPPED : " + str(skip_count))
	logger.info("ARCHIVED : " + str(archive_count))
	logger.info("" 85)