[PATCH v3 3/9] KVM: selftests: Add timeout option in selftests runner

Vipin Sharma
2025-09-30T16:36:54Z

Add a command line argument in KVM selftest runner to limit amount of
time (seconds) given to a test for execution. Kill the test if it
exceeds the given timeout. Define a new SelftestStatus.TIMED_OUT to
denote a selftest final result. Add terminal color for status messages
of timed out tests.

Set the default value of 120 seconds for all tests.

Signed-off-by: Vipin Sharma <vipinsh@google.com>
---
.../testing/selftests/kvm/runner/__main__.py | 9 ++++-
.../testing/selftests/kvm/runner/selftest.py | 33 ++++++++++++-------
.../selftests/kvm/runner/test_runner.py | 2 +-
3 files changed, 31 insertions(+), 13 deletions(-)

diff --git a/tools/testing/selftests/kvm/runner/__main__.py b/tools/testing/selftests/kvm/runner/__main__.py
index 943c3bfe2eb6..5cedc5098a54 100644
--- a/tools/testing/selftests/kvm/runner/__main__.py
+++ b/tools/testing/selftests/kvm/runner/__main__.py
@@ -37,6 +37,11 @@ def cli():
default=".",
help="Finds the test executables in the given path. Default is the current directory.")

+ parser.add_argument("--timeout",
+ default=120,
+ type=int,
+ help="Timeout, in seconds, before runner kills the running test. (Default: 120 seconds)")
+
return parser.parse_args()

@@ -44,6 +49,7 @@ def setup_logging():
class TerminalColorFormatter(logging.Formatter):
reset = "\033[0m"
red_bold = "\033[31;1m"
+ red = "\033[31;1m"
green = "\033[32m"
yellow = "\033[33m"
blue = "\033[34m"
@@ -52,7 +58,8 @@ def setup_logging():
SelftestStatus.PASSED: green,
SelftestStatus.NO_RUN: blue,
SelftestStatus.SKIPPED: yellow,
- SelftestStatus.FAILED: red_bold
+ SelftestStatus.FAILED: red_bold,
+ SelftestStatus.TIMED_OUT: red
}

def __init__(self, fmt=None, datefmt=None):
diff --git a/tools/testing/selftests/kvm/runner/selftest.py b/tools/testing/selftests/kvm/runner/selftest.py
index a94b6d4cda05..4783785ca230 100644
--- a/tools/testing/selftests/kvm/runner/selftest.py
+++ b/tools/testing/selftests/kvm/runner/selftest.py
@@ -17,6 +17,7 @@ class SelftestStatus(enum.IntEnum):
NO_RUN = 22
SKIPPED = 23
FAILED = 24
+ TIMED_OUT = 25

def __str__(self):
return str.__str__(self.name)
@@ -28,7 +29,7 @@ class Selftest:
Extract the test execution command from test file and executes it.
"""

- def __init__(self, test_path, path):
+ def __init__(self, test_path, path, timeout):
test_command = pathlib.Path(test_path).read_text().strip()
if not test_command:
raise ValueError("Empty test command in " + test_path)
@@ -37,6 +38,7 @@ class Selftest:
self.exists = os.path.isfile(test_command.split(maxsplit=1)[0])
self.test_path = test_path
self.command = test_command
+ self.timeout = timeout
self.status = SelftestStatus.NO_RUN
self.stdout = ""
self.stderr = ""
@@ -50,15 +52,24 @@ class Selftest:
"universal_newlines": True,
"shell": True,
"stdout": subprocess.PIPE,
- "stderr": subprocess.PIPE
+ "stderr": subprocess.PIPE,
+ "timeout": self.timeout,
}
- proc = subprocess.run(self.command, **run_args)
- self.stdout = proc.stdout
- self.stderr = proc.stderr

- if proc.returncode == 0:
- self.status = SelftestStatus.PASSED
- elif proc.returncode == 4:
- self.status = SelftestStatus.SKIPPED
- else:
- self.status = SelftestStatus.FAILED
+ try:
+ proc = subprocess.run(self.command, **run_args)
+ self.stdout = proc.stdout
+ self.stderr = proc.stderr
+
+ if proc.returncode == 0:
+ self.status = SelftestStatus.PASSED
+ elif proc.returncode == 4:
+ self.status = SelftestStatus.SKIPPED
+ else:
+ self.status = SelftestStatus.FAILED
+ except subprocess.TimeoutExpired as e:
+ self.status = SelftestStatus.TIMED_OUT
+ if e.stdout is not None:
+ self.stdout = e.stdout
+ if e.stderr is not None:
+ self.stderr = e.stderr
diff --git a/tools/testing/selftests/kvm/runner/test_runner.py b/tools/testing/selftests/kvm/runner/test_runner.py
index acc9fb3dabde..bea82c6239cd 100644
--- a/tools/testing/selftests/kvm/runner/test_runner.py
+++ b/tools/testing/selftests/kvm/runner/test_runner.py
@@ -15,7 +15,7 @@ class TestRunner:
self.tests = []

for testcase in testcases:
- self.tests.append(Selftest(testcase, args.path))
+ self.tests.append(Selftest(testcase, args.path, args.timeout))

def _log_result(self, test_result):
logger.info("*** stdout ***\n" + test_result.stdout)
--
2.51.0.618.g983fd99d29-goog