Page MenuHomeClusterLabs Projects
Files F3686940

No OneTemporary
Actions

View Options

diff --git a/python/pacemaker/_cts/tests/ctstest.py b/python/pacemaker/_cts/tests/ctstest.py
index fd63b914a1..cf9eaa5e4d 100644
--- a/python/pacemaker/_cts/tests/ctstest.py
+++ b/python/pacemaker/_cts/tests/ctstest.py
@@ -1,261 +1,260 @@
""" Base classes for CTS tests """
__all__ = ["CTSTest"]
__copyright__ = "Copyright 2000-2023 the Pacemaker project contributors"
__license__ = "GNU General Public License version 2 or later (GPLv2+) WITHOUT ANY WARRANTY"
import re
-from pacemaker._cts.audits import AuditConstraint, AuditResource
from pacemaker._cts.environment import EnvFactory
from pacemaker._cts.logging import LogFactory
from pacemaker._cts.patterns import PatternSelector
from pacemaker._cts.remote import RemoteFactory
from pacemaker._cts.timer import Timer
from pacemaker._cts.watcher import LogWatcher
# Disable various pylint warnings that occur in so many places throughout this
# file it's easiest to just take care of them globally. This does introduce the
# possibility that we'll miss some other cause of the same warning, but we'll
# just have to be careful.
# pylint doesn't understand that self._rsh is callable.
# pylint: disable=not-callable
class CTSTest:
""" The base class for all cluster tests. This implements a basic set of
properties and behaviors like setup, tear down, time keeping, and
statistics tracking. It is up to specific tests to implement their own
specialized behavior on top of this class.
"""
def __init__(self, cm):
""" Create a new CTSTest instance
Arguments:
cm -- A ClusterManager instance
"""
# pylint: disable=invalid-name
self.audits = []
self.name = None
self.templates = PatternSelector(cm["Name"])
self.stats = {
"auditfail": 0,
"calls": 0,
"failure": 0,
"skipped": 0,
"success": 0
}
self._cm = cm
self._env = EnvFactory().getInstance()
self._r_o2cb = None
self._r_ocfs2 = []
self._rsh = RemoteFactory().getInstance()
self._logger = LogFactory()
self._timers = {}
self.benchmark = True # which tests to benchmark
self.failed = False
self.is_experimental = False
self.is_loop = False
self.is_unsafe = False
self.is_valgrind = False
self.passed = True
def log(self, args):
""" Log a message """
self._logger.log(args)
def debug(self, args):
""" Log a debug message """
self._logger.debug(args)
def get_timer(self, key="test"):
""" Get the start time of the given timer """
try:
return self._timers[key].start_time
except KeyError:
return 0
def set_timer(self, key="test"):
""" Set the start time of the given timer to now, and return
that time
"""
if key not in self._timers:
self._timers[key] = Timer(self._logger, self.name, key)
self._timers[key].start()
return self._timers[key].start_time
def log_timer(self, key="test"):
""" Log the elapsed time of the given timer """
if key not in self._timers:
return
elapsed = self._timers[key].elapsed
self.debug("%s:%s runtime: %.2f" % (self.name, key, elapsed))
del self._timers[key]
def incr(self, name):
""" Increment the given stats key """
if name not in self.stats:
self.stats[name] = 0
self.stats[name] += 1
# Reset the test passed boolean
if name == "calls":
self.passed = True
def failure(self, reason="none"):
""" Increment the failure count, with an optional failure reason """
self.passed = False
self.incr("failure")
self._logger.log(("Test %s" % self.name).ljust(35) + " FAILED: %s" % reason)
return False
def success(self):
""" Increment the success count """
self.incr("success")
return True
def skipped(self):
""" Increment the skipped count """
self.incr("skipped")
return True
def __call__(self, node):
""" Perform this test """
raise NotImplementedError
def audit(self):
""" Perform all the relevant audits (see ClusterAudit), returning
whether or not they all passed.
"""
passed = True
for audit in self.audits:
if not audit():
self._logger.log("Internal %s Audit %s FAILED." % (self.name, audit.name))
self.incr("auditfail")
passed = False
return passed
def setup(self, node):
""" Setup this test """
# node is used in subclasses
# pylint: disable=unused-argument
return self.success()
def teardown(self, node):
""" Tear down this test """
# node is used in subclasses
# pylint: disable=unused-argument
return self.success()
def create_watch(self, patterns, timeout, name=None):
""" Create a new LogWatcher object with the given patterns, timeout,
and optional name. This object can be used to search log files
for matching patterns during this test's run.
"""
if not name:
name = self.name
return LogWatcher(self._env["LogFileName"], patterns, self._env["nodes"], self._env["LogWatcher"], name, timeout)
def local_badnews(self, prefix, watch, local_ignore=None):
""" Use the given watch object to search through log files for messages
starting with the given prefix. If no prefix is given, use
"LocalBadNews:" by default. The optional local_ignore list should
be a list of regexes that, if found in a line, will cause that line
to be ignored.
Return the number of matches found.
"""
errcount = 0
if not prefix:
prefix = "LocalBadNews:"
ignorelist = [" CTS: ", prefix]
if local_ignore:
ignorelist += local_ignore
while errcount < 100:
match = watch.look(0)
if match:
add_err = True
for ignore in ignorelist:
if add_err and re.search(ignore, match):
add_err = False
if add_err:
self._logger.log("%s %s" % (prefix, match))
errcount += 1
else:
break
else:
self._logger.log("Too many errors!")
watch.end()
return errcount
def is_applicable(self):
""" Return True if this test is applicable in the current test configuration.
This method must be implemented by all subclasses.
"""
if self.is_loop and not self._env["loop-tests"]:
return False
if self.is_unsafe and not self._env["unsafe-tests"]:
return False
if self.is_valgrind and not self._env["valgrind-tests"]:
return False
if self.is_experimental and not self._env["experimental-tests"]:
return False
if self._env["benchmark"] and not self.benchmark:
return False
return True
def can_run_now(self, node):
""" Return True if we can meaningfully run right now """
# node is used in subclasses
# pylint: disable=unused-argument
return True
@property
def errors_to_ignore(self):
""" Return list of errors which should be ignored """
return []

File Metadata

Mime Type
text/x-diff
Expires
Mon, Apr 21, 6:13 PM (1 d, 42 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1660158
Default Alt Text
(7 KB)

Event Timeline