path: root/cts/cts-regression.in

#!@PYTHON@
"""Convenience wrapper for running Pacemaker regression tests.

Usage: cts-regression [-h] [-V] [-v] [COMPONENT ...]
"""

__copyright__ = 'Copyright 2012-2023 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")

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.

        :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, 'cts-%s' % 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('Executing the %s regression tests' % self.name)
        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('Failed to execute %s tests: %s' % (self.name, 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):
        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('      * ERROR:   %s' % 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('%s regression tests failed (%s)' % (comp.name, rc))
            failed.append(comp.name)

    if failed:
        print('Failed regression tests:', end='')
        for comp in failed:
            print(' %s' % 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 += '\n {:<20} {}'.format(name, comp.description)

    description += (
        '\n {:<20} Synonym for "cli exec fencing scheduler"'.format('all')
    )

    # 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()