Page MenuHomeClusterLabs Projects
Files F3687270

No OneTemporary
Actions

View Options

diff --git a/python/pacemaker/_cts/tests/ctstest.py b/python/pacemaker/_cts/tests/ctstest.py
index 76174699a3..4c372b14af 100644
--- a/python/pacemaker/_cts/tests/ctstest.py
+++ b/python/pacemaker/_cts/tests/ctstest.py
@@ -1,249 +1,246 @@
"""Base classes for CTS tests."""
__all__ = ["CTSTest"]
__copyright__ = "Copyright 2000-2024 the Pacemaker project contributors"
__license__ = "GNU General Public License version 2 or later (GPLv2+) WITHOUT ANY WARRANTY"
import re
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):
"""Set up 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.
This object can be used to search log files for matching patterns
during this test's run.
Arguments:
patterns -- A list of regular expressions to match against the log
timeout -- Default number of seconds to watch a log file at a time;
this can be overridden by the timeout= parameter to
self.look on an as-needed basis
name -- A unique name to use when logging about this watch
"""
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):
"""
Search through log files for messages.
Arguments:
prefix -- The string to look for at the beginning of lines,
or "LocalBadNews:" if None.
watch -- The LogWatcher object to use for searching.
local_ignore -- 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
@property
def errors_to_ignore(self):
"""Return a list of errors which should be ignored."""
return []

File Metadata

Mime Type
text/x-diff
Expires
Mon, Apr 21, 7:09 PM (10 h, 49 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1665401
Default Alt Text
(7 KB)

Event Timeline