summaryrefslogtreecommitdiffstats
path: root/qa/tasks/ceph_test_case.py
diff options
context:
space:
mode:
Diffstat (limited to 'qa/tasks/ceph_test_case.py')
-rw-r--r--qa/tasks/ceph_test_case.py133
1 files changed, 129 insertions, 4 deletions
diff --git a/qa/tasks/ceph_test_case.py b/qa/tasks/ceph_test_case.py
index 3f8a152d7..649c0e53c 100644
--- a/qa/tasks/ceph_test_case.py
+++ b/qa/tasks/ceph_test_case.py
@@ -2,6 +2,7 @@ from typing import Optional, TYPE_CHECKING
import unittest
import time
import logging
+from io import StringIO
from teuthology.exceptions import CommandFailedError
@@ -13,7 +14,106 @@ log = logging.getLogger(__name__)
class TestTimeoutError(RuntimeError):
pass
-class CephTestCase(unittest.TestCase):
+
+class RunCephCmd:
+
+ def run_ceph_cmd(self, *args, **kwargs):
+ """
+ *args and **kwargs must contain arguments that are accepted by
+ vstart_runner.LocalRemote._do_run() or teuhology.orchestra.run.run()
+ methods.
+ """
+ if kwargs.get('args') is None and args:
+ if len(args) == 1:
+ args = args[0]
+ kwargs['args'] = args
+ return self.mon_manager.run_cluster_cmd(**kwargs)
+
+ def get_ceph_cmd_result(self, *args, **kwargs):
+ """
+ *args and **kwargs must contain arguments that are accepted by
+ vstart_runner.LocalRemote._do_run() or teuhology.orchestra.run.run()
+ methods.
+ """
+ if kwargs.get('args') is None and args:
+ if len(args) == 1:
+ args = args[0]
+ kwargs['args'] = args
+ return self.run_ceph_cmd(**kwargs).exitstatus
+
+ def get_ceph_cmd_stdout(self, *args, **kwargs):
+ """
+ *args and **kwargs must contain arguments that are accepted by
+ vstart_runner.LocalRemote._do_run() or teuhology.orchestra.run.run()
+ methods.
+ """
+ if kwargs.get('args') is None and args:
+ if len(args) == 1:
+ args = args[0]
+ kwargs['args'] = args
+ kwargs['stdout'] = kwargs.pop('stdout', StringIO())
+ return self.run_ceph_cmd(**kwargs).stdout.getvalue()
+
+ def assert_retval(self, proc_retval, exp_retval):
+ msg = (f'expected return value: {exp_retval}\n'
+ f'received return value: {proc_retval}\n')
+ assert proc_retval == exp_retval, msg
+
+ def _verify(self, proc, exp_retval=None, exp_errmsgs=None):
+ if exp_retval is None and exp_errmsgs is None:
+ raise RuntimeError('Method didn\'t get enough parameters. Pass '
+ 'return value or error message expected from '
+ 'the command/process.')
+
+ if exp_retval is not None:
+ self.assert_retval(proc.returncode, exp_retval)
+ if exp_errmsgs is None:
+ return
+
+ if isinstance(exp_errmsgs, str):
+ exp_errmsgs = (exp_errmsgs, )
+ exp_errmsgs = tuple([e.lower() for e in exp_errmsgs])
+
+ proc_stderr = proc.stderr.getvalue().lower()
+ msg = ('didn\'t find any of the expected string in stderr.\n'
+ f'expected string: {exp_errmsgs}\n'
+ f'received error message: {proc_stderr}\n'
+ 'note: received error message is converted to lowercase')
+ for e in exp_errmsgs:
+ if e in proc_stderr:
+ break
+ # this else is meant for the for loop above.
+ else:
+ assert False, msg
+
+ def negtest_ceph_cmd(self, args, retval=None, errmsgs=None, **kwargs):
+ """
+ Conduct a negative test for the given Ceph command.
+
+ retval and errmsgs are parameters to confirm the cause of command
+ failure.
+
+ *args and **kwargs must contain arguments that are accepted by
+ vstart_runner.LocalRemote._do_run() or teuhology.orchestra.run.run()
+ methods.
+
+ NOTE: errmsgs is expected to be a tuple, but in case there's only one
+ error message, it can also be a string. This method will add the string
+ to a tuple internally.
+ """
+ kwargs['args'] = args
+ # execution is needed to not halt on command failure because we are
+ # conducting negative testing
+ kwargs['check_status'] = False
+ # stderr is needed to check for expected error messages.
+ kwargs['stderr'] = StringIO()
+
+ proc = self.run_ceph_cmd(**kwargs)
+ self._verify(proc, retval, errmsgs)
+ return proc
+
+
+class CephTestCase(unittest.TestCase, RunCephCmd):
"""
For test tasks that want to define a structured set of
tests implemented in python. Subclass this with appropriate
@@ -36,9 +136,23 @@ class CephTestCase(unittest.TestCase):
# their special needs. If not met, tests will be skipped.
REQUIRE_MEMSTORE = False
+ def _init_mon_manager(self):
+ # if vstart_runner.py has invoked this code
+ if 'Local' in str(type(self.ceph_cluster)):
+ from tasks.vstart_runner import LocalCephManager
+ self.mon_manager = LocalCephManager(ctx=self.ctx)
+ # else teuthology has invoked this code
+ else:
+ from tasks.ceph_manager import CephManager
+ self.mon_manager = CephManager(self.ceph_cluster.admin_remote,
+ ctx=self.ctx, logger=log.getChild('ceph_manager'))
+
def setUp(self):
self._mon_configs_set = set()
+ self._init_mon_manager()
+ self.admin_remote = self.ceph_cluster.admin_remote
+
self.ceph_cluster.mon_manager.raw_cluster_cmd("log",
"Starting test {0}".format(self.id()))
@@ -148,12 +262,14 @@ class CephTestCase(unittest.TestCase):
return ContextManager()
- def wait_for_health(self, pattern, timeout):
+ def wait_for_health(self, pattern, timeout, check_in_detail=None):
"""
Wait until 'ceph health' contains messages matching the pattern
+ Also check if @check_in_detail matches detailed health messages
+ only when @pattern is a code string.
"""
def seen_health_warning():
- health = self.ceph_cluster.mon_manager.get_mon_health()
+ health = self.ceph_cluster.mon_manager.get_mon_health(debug=False, detail=bool(check_in_detail))
codes = [s for s in health['checks']]
summary_strings = [s[1]['summary']['message'] for s in health['checks'].items()]
if len(summary_strings) == 0:
@@ -164,7 +280,16 @@ class CephTestCase(unittest.TestCase):
if pattern in ss:
return True
if pattern in codes:
- return True
+ if not check_in_detail:
+ return True
+ # check if the string is in detail list if asked
+ detail_strings = [ss['message'] for ss in \
+ [s for s in health['checks'][pattern]['detail']]]
+ log.debug(f'detail_strings: {detail_strings}')
+ for ds in detail_strings:
+ if check_in_detail in ds:
+ return True
+ log.debug(f'detail string "{check_in_detail}" not found')
log.debug("Not found expected summary strings yet ({0})".format(summary_strings))
return False