diff --git a/cts/cts-regression.in b/cts/cts-regression.in
index 3199f73bf6..3f1119d3ed 100644
--- a/cts/cts-regression.in
+++ b/cts/cts-regression.in
@@ -1,298 +1,300 @@
#!@PYTHON@
"""Convenience wrapper for running Pacemaker regression tests.
Usage: cts-regression [-h] [-V] [-v] [COMPONENT ...]
"""
# pylint doesn't like the module name "cts-regression" which is an invalid complaint for this file
# but probably something we want to continue warning about elsewhere
# pylint: disable=invalid-name
# pacemaker imports need to come after we modify sys.path, which pylint will complain about.
# pylint: disable=wrong-import-position
__copyright__ = 'Copyright 2012-2025 the Pacemaker project contributors'
__license__ = 'GNU General Public License version 2 or later (GPLv2+) WITHOUT ANY WARRANTY'
import argparse
import os
import subprocess
import sys
import textwrap
# These imports allow running from a source checkout after running `make`.
# Note that while this doesn't necessarily mean it will successfully run tests,
# but being able to see --help output can be useful.
if os.path.exists("@abs_top_srcdir@/python"):
sys.path.insert(0, "@abs_top_srcdir@/python")
# pylint: disable=comparison-of-constants,comparison-with-itself,condition-evals-to-constant
if os.path.exists("@abs_top_builddir@/python") and "@abs_top_builddir@" != "@abs_top_srcdir@":
sys.path.insert(0, "@abs_top_builddir@/python")
from pacemaker.buildoptions import BuildOptions
from pacemaker.exitstatus import ExitStatus
+
class Component():
"""A class for running regression tests on a component.
"Component" refers to a Pacemaker component, such as the scheduler.
:attribute name: The name of the component.
:type name: str
:attribute description: The description of the component.
:type description: str
:attribute requires_root: Whether the component's tests must be run
as root.
:type requires_root: bool
:attribute supports_valgrind: Whether the component's tests support
running under valgrind.
:type supports_valgrind: bool
:attribute cmd: The command to run the component's tests, along with
any required options.
:type cmd: list[str]
:method run([verbose=False], [valgrind=False]): Run the component's
regression tests and return the result.
"""
def __init__(self, name, description, test_home, requires_root=False,
supports_valgrind=False):
- """Constructor for the :class:`Component` class.
+ """Create a new :class:`Component` instance.
:param name: The name of the component.
:type name: str
:param description: The description of the component.
:type description: str
:param test_home: The directory where the component's tests
reside.
:type test_home: str
:param requires_root: Whether the component's tests must be run
as root.
:type requires_root: bool
:param supports_valgrind: Whether the component's tests support
running under valgrind.
:type supports_valgrind: bool
"""
self.name = name
self.description = description
self.requires_root = requires_root
self.supports_valgrind = supports_valgrind
if self.name == 'pacemaker_remote':
self.cmd = [os.path.join(test_home, 'cts-exec'), '-R']
else:
self.cmd = [os.path.join(test_home, f"cts-{self.name}")]
def run(self, verbose=False, valgrind=False):
"""Run the component's regression tests and return the result.
:param verbose: Whether to increase test output verbosity.
:type verbose: bool
:param valgrind: Whether to run the test under valgrind.
:type valgrind: bool
:return: The exit code from the component's test suite.
:rtype: :class:`ExitStatus`
"""
print(f"Executing the {self.name} regression tests")
print('=' * 60)
cmd = self.cmd
if self.requires_root and os.geteuid() != 0:
print('Enter the sudo password if prompted')
cmd = ['sudo'] + self.cmd
if verbose:
cmd.append('--verbose')
if self.supports_valgrind and valgrind:
cmd.append('--valgrind')
try:
rc = ExitStatus(subprocess.call(cmd))
except OSError as err:
error_print(f"Failed to execute {self.name} tests: {err}")
rc = ExitStatus.NOT_INSTALLED
print('=' * 60 + '\n\n')
return rc
class ComponentsArgAction(argparse.Action):
"""A class to handle `components` arguments.
This class handles special cases and cleans up the `components`
list. Specifically, it does the following:
* Enforce a default value of ['cli', 'scheduler'].
* Replace the 'all' alias with the components that it represents.
* Get rid of duplicates.
The main motivation is that when the `choices` argument of
:meth:`parser.add_argument()` is specified, the `default` argument
must contain exactly one value (not `None` and not a list). We want
our default to be a list of components, namely `cli` and
`scheduler`.
"""
def __call__(self, parser, namespace, values, option_string=None):
+ """Process `components` arguments."""
all_components = ['attrd', 'cli', 'exec', 'fencing', 'scheduler']
default_components = ['cli', 'scheduler']
if not values:
setattr(namespace, self.dest, default_components)
return
# If no argument is specified, the default gets passed as a
# string 'default' instead of as a list ['default']. Probably
# a bug in argparse. The below gives us a list.
if not isinstance(values, list):
values = [values]
components = set(values)
# If 'all', is found, replace it with the components it represents.
try:
components.remove('all')
components.update(set(all_components))
except KeyError:
pass
# Same for 'default'
try:
components.remove('default')
components.update(set(default_components))
except KeyError:
pass
setattr(namespace, self.dest, sorted(list(components)))
def error_print(msg):
"""Print an error message.
:param msg: Message to print.
:type msg: str
"""
print(f" * ERROR: {msg}")
def run_components(components, verbose=False, valgrind=False):
"""Run components' regression tests and report results for each.
:param components: A list of names of components for which to run
tests.
:type components: list[:class:`Component`]
:return: :attr:`ExitStatus.OK` if all tests were successful,
:attr:`ExitStatus.ERROR` otherwise.
:rtype: :class:`ExitStatus`
"""
failed = []
for comp in components:
rc = comp.run(verbose, valgrind)
if rc != ExitStatus.OK:
error_print(f"{comp.name} regression tests failed ({rc})")
failed.append(comp.name)
if failed:
print('Failed regression tests:', end='')
for comp in failed:
print(f" {comp}", end='')
print()
return ExitStatus.ERROR
return ExitStatus.OK
def main():
"""Run Pacemaker regression tests as specified by arguments."""
try:
test_home = os.path.dirname(os.readlink(sys.argv[0]))
except OSError:
test_home = os.path.dirname(sys.argv[0])
# Available components
components = {
'attrd': Component(
'attrd',
'Attribute manager',
test_home,
requires_root=True,
supports_valgrind=False,
),
'cli': Component(
'cli',
'Command-line tools',
test_home,
requires_root=False,
supports_valgrind=True,
),
'exec': Component(
'exec',
'Local resource agent executor',
test_home,
requires_root=True,
supports_valgrind=False,
),
'fencing': Component(
'fencing',
'Fencer',
test_home,
requires_root=True,
supports_valgrind=False,
),
'scheduler': Component(
'scheduler',
'Action scheduler',
test_home,
requires_root=False,
supports_valgrind=True,
),
}
if BuildOptions.REMOTE_ENABLED:
components['pacemaker_remote'] = Component(
'pacemaker_remote',
'Resource agent executor in remote mode',
test_home,
requires_root=True,
supports_valgrind=False,
)
# Build up program description
description = textwrap.dedent('''\
Run Pacemaker regression tests.
Available components (default components are 'cli scheduler'):
''')
for name, comp in sorted(components.items()):
description += f"\n {name:<20} {comp.description}"
description += f'\n {"all":<20} Synonym for "cli exec fencing scheduler"'
# Parse the arguments
parser = argparse.ArgumentParser(
description=description,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
choices = sorted(components.keys()) + ['all', 'default']
parser.add_argument('-V', '--verbose', action='store_true',
help='Increase test verbosity')
parser.add_argument('-v', '--valgrind', action='store_true',
help='Run test commands under valgrind')
parser.add_argument('components', nargs='*', choices=choices,
default='default',
action=ComponentsArgAction, metavar='COMPONENT',
help="One of the components to test, or 'all'")
args = parser.parse_args()
# Run the tests
selected = [components[x] for x in args.components]
rc = run_components(selected, args.verbose, args.valgrind)
sys.exit(rc)
if __name__ == '__main__':
main()