@@ -7,6 +7,8 @@ import argparse
import logging
import os
import sys
+import datetime
+import pathlib
from test_runner import TestRunner
from selftest import SelftestStatus
@@ -42,10 +44,20 @@ def cli():
type=int,
help="Timeout, in seconds, before runner kills the running test. (Default: 120 seconds)")
+ parser.add_argument("-o",
+ "--output",
+ nargs='?',
+ help="Dumps test runner output which includes each test execution result, their stdouts and stderrs hierarchically in the given directory.")
+
+ parser.add_argument("--append-output-time",
+ action="store_true",
+ default=False,
+ help="Appends timestamp to the output directory.")
+
return parser.parse_args()
-def setup_logging():
+def setup_logging(args):
class TerminalColorFormatter(logging.Formatter):
reset = "\033[0m"
red_bold = "\033[31;1m"
@@ -72,12 +84,26 @@ def setup_logging():
logger = logging.getLogger("runner")
logger.setLevel(logging.INFO)
+ formatter_args = {
+ "fmt": "%(asctime)s | %(message)s",
+ "datefmt": "%H:%M:%S"
+ }
+
ch = logging.StreamHandler()
- ch_formatter = TerminalColorFormatter(fmt="%(asctime)s | %(message)s",
- datefmt="%H:%M:%S")
+ ch_formatter = TerminalColorFormatter(**formatter_args)
ch.setFormatter(ch_formatter)
logger.addHandler(ch)
+ if args.output != None:
+ if (args.append_output_time):
+ args.output += datetime.datetime.now().strftime(".%Y.%m.%d.%H.%M.%S")
+ pathlib.Path(args.output).mkdir(parents=True, exist_ok=True)
+ logging_file = os.path.join(args.output, "log")
+ fh = logging.FileHandler(logging_file)
+ fh_formatter = logging.Formatter(**formatter_args)
+ fh.setFormatter(fh_formatter)
+ logger.addHandler(fh)
+
def fetch_testcases_in_dirs(dirs):
testcases = []
@@ -98,7 +124,7 @@ def fetch_testcases(args):
def main():
args = cli()
- setup_logging()
+ setup_logging(args)
testcases = fetch_testcases(args)
return TestRunner(testcases, args).start()
@@ -7,6 +7,7 @@ import pathlib
import enum
import os
import subprocess
+import contextlib
class SelftestStatus(enum.IntEnum):
"""
@@ -29,7 +30,7 @@ class Selftest:
Extract the test execution command from test file and executes it.
"""
- def __init__(self, test_path, path, timeout):
+ def __init__(self, test_path, path, timeout, output_dir):
test_command = pathlib.Path(test_path).read_text().strip()
if not test_command:
raise ValueError("Empty test command in " + test_path)
@@ -39,15 +40,14 @@ class Selftest:
self.test_path = test_path
self.command = test_command
self.timeout = timeout
+ if output_dir is not None:
+ output_dir = os.path.join(output_dir, test_path.lstrip("./"))
+ self.output_dir = output_dir
self.status = SelftestStatus.NO_RUN
self.stdout = ""
self.stderr = ""
- def run(self):
- if not self.exists:
- self.stderr = "File doesn't exist."
- return
-
+ def _run(self, output=None, error=None):
run_args = {
"universal_newlines": True,
"shell": True,
@@ -70,4 +70,30 @@ class Selftest:
out, err = e.stdout, e.stderr
self.stdout = out.decode("utf-8", "replace") if isinstance(out, bytes) else (out or "")
+ if output is not None:
+ output.write(self.stdout)
+
self.stderr = err.decode("utf-8", "replace") if isinstance(err, bytes) else (err or "")
+ if error is not None:
+ error.write(self.stderr)
+
+ def run(self):
+ if not self.exists:
+ self.stderr = "File doesn't exist."
+ return
+
+ if self.output_dir is not None:
+ pathlib.Path(self.output_dir).mkdir(parents=True, exist_ok=True)
+
+ output = None
+ error = None
+ with contextlib.ExitStack() as stack:
+ if self.output_dir is not None:
+ output_path = os.path.join(self.output_dir, "stdout")
+ output = stack.enter_context(
+ open(output_path, encoding="utf-8", mode="w"))
+
+ error_path = os.path.join(self.output_dir, "stderr")
+ error = stack.enter_context(
+ open(error_path, encoding="utf-8", mode="w"))
+ return self._run(output, error)
@@ -13,9 +13,11 @@ logger = logging.getLogger("runner")
class TestRunner:
def __init__(self, testcases, args):
self.tests = []
+ self.output_dir = args.output
for testcase in testcases:
- self.tests.append(Selftest(testcase, args.path, args.timeout))
+ self.tests.append(Selftest(testcase, args.path, args.timeout,
+ args.output))
def _log_result(self, test_result):
logger.info("*** stdout ***\n" + test_result.stdout)
Add a command line flag, -o/--output, to selftest runner which enables it to save individual tests output (stdout & stderr) stream to a directory in a hierarchical way. Create folder hierarchy same as tests hieararcy given by --testcases and --dirs. Also, add a command line flag, --append-output-time, which will append timestamp (format YYYY.M.DD.HH.MM.SS) to the directory name given in --output flag. Example: python3 runner --dirs test -o test_result --append_output_time This will create test_result.2025.06.06.08.45.57 directory. Signed-off-by: Vipin Sharma <vipinsh@google.com> --- .../testing/selftests/kvm/runner/__main__.py | 34 +++++++++++++++-- .../testing/selftests/kvm/runner/selftest.py | 38 ++++++++++++++++--- .../selftests/kvm/runner/test_runner.py | 4 +- 3 files changed, 65 insertions(+), 11 deletions(-)