#!/usr/bin/env python3
import argparse
import logging
import os
import shutil
import subprocess
import sys
import time
import git
import junitparser
from lava_create_testreport import lava_create_testreport
TESTS_GIT_URL = "git@gitlab.com:garz-fricke/tests/development-tests.git"
def call(cmd, stdout=None):
logging.debug(f"Call: {cmd}")
try:
if stdout is None:
result = subprocess.run(cmd, capture_output=True, check=True)
else:
result = subprocess.run(cmd, stdout=stdout, check=True)
except subprocess.CalledProcessError as e:
out = e.stdout.decode() if e.stdout is not None else ""
err = e.stderr.decode() if e.stderr is not None else ""
logging.error(f"Command failed {cmd}: {out} {err}")
exit(1)
if result is not None and result.stdout is not None:
res = result.stdout.decode().strip()
logging.debug(f"Command returned: {res}")
else:
res = ""
return res
def main():
# FIXME: Remove the sourcery check deactivation below and refactor this method in
# order to enhance code quality and make the check pass.
# sourcery skip: low-code-quality
parser = argparse.ArgumentParser()
parser.add_argument(
"--verbose",
help="""More verbose output.""",
action="store_true",
default=False,
)
parser.add_argument(
"--fng-install",
help="""Url to the download link of the install script.""",
dest="fnginstall",
type=str,
required=True,
)
parser.add_argument(
"--name",
help="""Base name for the submitted job.""",
default="Testjob for {platform}",
)
parser.add_argument(
"--dry",
help="""Don't really submit anything""",
action="store_true",
default=False,
)
parser.add_argument(
"--test-repo",
help="""Repository where the tests and access scripts are stored.""",
default=TESTS_GIT_URL,
dest="testrepo",
)
parser.add_argument(
"--test-repo-branch",
help="""Branch of the repository to check out.""",
default=None,
dest="testrepo_branch",
)
parser.add_argument(
"--checkout-path",
help="""When set the script expects the tests to be checkout out in the given directory.""",
default=None,
dest="checkout_path",
)
parser.add_argument(
"--results-path",
help="""Subfolder where the result files are stored.""",
default="results",
dest="results_path",
)
parser.add_argument(
"--all-devices",
help="""Submit the test to all devices of a given device type, platforms parameter is treated as devicetype in this case.""",
default=False,
action="store_true",
dest="all_devices",
)
parser.add_argument(
"--test-plan",
help="""Name pattern of the test plan to use, file needs to be
found in the 'tests' subfolder of the test repo and may use
{platform} to be replaced by the actual platform name.""",
default="{platform}.jinja2",
dest="test_suite",
)
parser.add_argument(
"--report-name",
help="""Name of the generated markdown result. If omnitted, report is generation is skipped.""",
default=None,
dest="report_name",
)
# TODO add parameters to specify branch or revision,
# but this is needed to be implemented in the tests also.
parser.add_argument(
"platforms",
help="""Platforms to submit the tests for.""",
nargs="+",
)
args, _ = parser.parse_known_args()
if args.verbose:
logging.basicConfig(level=10)
else:
logging.basicConfig(level=40)
if args.checkout_path is None:
checkout_path = "test"
if os.path.exists(checkout_path):
shutil.rmtree(checkout_path)
logging.debug(f"Cloning {args.testrepo}")
try:
if args.testrepo_branch is not None:
testrepo = git.Repo.clone_from(
args.testrepo, checkout_path, branch=args.testrepo_branch
)
else:
testrepo = git.Repo.clone_from(args.testrepo, checkout_path)
except git.GitError as error:
print("Failed to clone the test repo:", error)
sys.exit(1)
else:
checkout_path = args.checkout_path
testrepo = git.Repo(checkout_path)
logging.debug(f"Using test repo at {checkout_path}")
testrepo_revision = testrepo.head.commit.hexsha
if os.path.exists(args.results_path):
shutil.rmtree(args.results_path)
os.mkdir(args.results_path)
jobs = {}
cmd_submit = os.path.join(checkout_path, "scripts", "submit.py")
cmd_query = os.path.join(checkout_path, "scripts", "query.py")
cmd_submitall = os.path.join(checkout_path, "scripts", "submit_all.py")
cmd_generate = os.path.join(checkout_path, "scripts", "generate_lava_job.py")
logging.debug(f"Test suite {args.test_suite}")
for platform in args.platforms:
test_suite = os.path.join(
checkout_path, "tests", args.test_suite.format(platform=platform)
)
logging.debug(f"Test suite {test_suite}")
if os.path.splitext(test_suite)[1] == ".jinja2":
cmd = [
cmd_generate,
"--test-plan",
test_suite,
"--test-repo",
args.testrepo,
"--testrepo-revision",
testrepo_revision,
"--fng-install",
args.fnginstall,
"--job-name",
args.name.format(platform=platform),
]
logging.debug(f"Generate job: {cmd}")
jobfile = os.path.join(args.results_path, f"{platform}.yaml")
with open(jobfile, "w", encoding="utf-8") as jobfile_handle:
call(cmd, stdout=jobfile_handle)
else:
jobfile = test_suite
if args.all_devices:
cmd = [cmd_submitall, "--device-type", platform, jobfile]
else:
cmd = [cmd_submit, jobfile]
if args.dry:
print(f"Skipping submit because of dry run: {cmd}")
continue
result = call(cmd)
for line in result.splitlines():
url = line.strip().replace("\n", "")
print(f"Started testjob {platform}: {url}")
jobid = url.split("/")[-1]
jobs[jobid] = [platform, ""]
logging.debug(f"Queued jobs: {jobs}")
if not jobs:
logging.info("No jobs queued.")
return 0
# Wait for the results
pending = True
while pending:
time.sleep(10)
print(".", end="", flush=True)
pending = False
for jobid in jobs:
result = call([cmd_query, jobid]).replace("\n", "")
jobs[jobid][1] = result
if result == "Pending":
pending = True
# Check the results
passed = True
print("")
for jobid in jobs:
jobplatform = jobs[jobid][0]
jobresult = jobs[jobid][1]
# Get results as yaml
resultfile = os.path.join(
args.results_path, f"results-{jobid}-{jobplatform}.yaml"
)
with open(resultfile, "w", encoding="utf-8") as resultfile_handle:
call([cmd_query, "--get-results", jobid], stdout=resultfile_handle)
# Get results as junit xml
resultfile = os.path.join(
args.results_path, f"results-{jobid}-{jobplatform}.xml"
)
with open(resultfile, "w", encoding="utf-8") as resultfile_handle:
call(
[
cmd_query,
"--get-results",
"--result-format",
"junit",
jobid,
],
stdout=resultfile_handle,
)
# Read the number of errors from the results
resultsxml = junitparser.JUnitXml.fromfile(resultfile)
joberrors = resultsxml.errors
if jobresult != "Complete" or joberrors > 0:
passed = False
print(f"Test result for {jobplatform}: {joberrors} Errors, {jobresult}")
# Create report as MarkDown
if args.report_name is not None:
filename = os.path.join(args.results_path, args.report_name)
os.makedirs(os.path.dirname(filename), exist_ok=True)
with open(filename, "w", encoding="utf-8") as f:
f.write(lava_create_testreport(jobs))
if passed:
sys.exit(0)
sys.exit(1)
if __name__ == "__main__":
main()