<reponame>rpratap-bot/ceph
import contextlib
import logging
import os
import unittest
from unittest import suite, loader, case
from teuthology.task import interactive
from teuthology import misc
from tasks.cephfs.filesystem import Filesystem, MDSCluster, CephCluster
from tasks.mgr.mgr_test_case import MgrCluster

log = logging.getLogger(__name__)


class DecoratingLoader(loader.TestLoader):
    """
    A specialization of TestLoader that tags some extra attributes
    onto test classes as they are loaded.
    """
    def __init__(self, params):
        self._params = params
        super(DecoratingLoader, self).__init__()

    def _apply_params(self, obj):
        for k, v in self._params.items():
            setattr(obj, k, v)

    def loadTestsFromTestCase(self, testCaseClass):
        self._apply_params(testCaseClass)
        return super(DecoratingLoader, self).loadTestsFromTestCase(testCaseClass)

    def loadTestsFromName(self, name, module=None):
        result = super(DecoratingLoader, self).loadTestsFromName(name, module)

        # Special case for when we were called with the name of a method, we get
        # a suite with one TestCase
        tests_in_result = list(result)
        if len(tests_in_result) == 1 and isinstance(tests_in_result[0], case.TestCase):
            self._apply_params(tests_in_result[0])

        return result


class LogStream(object):
    def __init__(self):
        self.buffer = ""

    def write(self, data):
        self.buffer += data
        if "\n" in self.buffer:
            lines = self.buffer.split("\n")
            for line in lines[:-1]:
                log.info(line)
            self.buffer = lines[-1]

    def flush(self):
        pass


class InteractiveFailureResult(unittest.TextTestResult):
    """
    Specialization that implements interactive-on-error style
    behavior.
    """
    ctx = None

    def addFailure(self, test, err):
        log.error(self._exc_info_to_string(err, test))
        log.error("Failure in test '{0}', going interactive".format(
            self.getDescription(test)
        ))
        interactive.task(ctx=self.ctx, config=None)

    def addError(self, test, err):
        log.error(self._exc_info_to_string(err, test))
        log.error("Error in test '{0}', going interactive".format(
            self.getDescription(test)
        ))
        interactive.task(ctx=self.ctx, config=None)


@contextlib.contextmanager
def task(ctx, config):
    """
    Run the CephFS test cases.

    Run everything in tasks/cephfs/test_*.py:

    ::

        tasks:
          - install:
          - ceph:
          - ceph-fuse:
          - cephfs_test_runner:

    `modules` argument allows running only some specific modules:

    ::

        tasks:
            ...
          - cephfs_test_runner:
              modules:
                - tasks.cephfs.test_sessionmap
                - tasks.cephfs.test_auto_repair

    By default, any cases that can't be run on the current cluster configuration
    will generate a failure.  When the optional `fail_on_skip` argument is set
    to false, any tests that can't be run on the current configuration will
    simply be skipped:

    ::
        tasks:
            ...
         - cephfs_test_runner:
           fail_on_skip: false

    """

    ceph_cluster = CephCluster(ctx)

    if len(list(misc.all_roles_of_type(ctx.cluster, 'mds'))):
        mds_cluster = MDSCluster(ctx)
        fs = Filesystem(ctx)
    else:
        mds_cluster = None
        fs = None

    if len(list(misc.all_roles_of_type(ctx.cluster, 'mgr'))):
        mgr_cluster = MgrCluster(ctx)
    else:
        mgr_cluster = None

    # Mount objects, sorted by ID
    if hasattr(ctx, 'mounts'):
        mounts = [v for k, v in sorted(ctx.mounts.items(), key=lambda mount: mount[0])]
    else:
        # The test configuration has a filesystem but no fuse/kclient mounts
        mounts = []

    decorating_loader = DecoratingLoader({
        "ctx": ctx,
        "mounts": mounts,
        "fs": fs,
        "ceph_cluster": ceph_cluster,
        "mds_cluster": mds_cluster,
        "mgr_cluster": mgr_cluster,
    })

    fail_on_skip = config.get('fail_on_skip', True)

    # Put useful things onto ctx for interactive debugging
    ctx.fs = fs
    ctx.mds_cluster = mds_cluster
    ctx.mgr_cluster = mgr_cluster

    # Depending on config, either load specific modules, or scan for moduless
    if config and 'modules' in config and config['modules']:
        module_suites = []
        for mod_name in config['modules']:
            # Test names like cephfs.test_auto_repair
            module_suites.append(decorating_loader.loadTestsFromName(mod_name))
        overall_suite = suite.TestSuite(module_suites)
    else:
        # Default, run all tests
        overall_suite = decorating_loader.discover(
            os.path.join(
                os.path.dirname(os.path.abspath(__file__)),
                "cephfs/"
            )
        )

    if ctx.config.get("interactive-on-error", False):
        InteractiveFailureResult.ctx = ctx
        result_class = InteractiveFailureResult
    else:
        result_class = unittest.TextTestResult

    class LoggingResult(result_class):
        def startTest(self, test):
            log.info("Starting test: {0}".format(self.getDescription(test)))
            return super(LoggingResult, self).startTest(test)

        def addSkip(self, test, reason):
            if fail_on_skip:
                # Don't just call addFailure because that requires a traceback
                self.failures.append((test, reason))
            else:
                super(LoggingResult, self).addSkip(test, reason)

    # Execute!
    result = unittest.TextTestRunner(
        stream=LogStream(),
        resultclass=LoggingResult,
        verbosity=2,
        failfast=True).run(overall_suite)

    if not result.wasSuccessful():
        result.printErrors()  # duplicate output at end for convenience

        bad_tests = []
        for test, error in result.errors:
            bad_tests.append(str(test))
        for test, failure in result.failures:
            bad_tests.append(str(test))

        raise RuntimeError("Test failure: {0}".format(", ".join(bad_tests)))

    yield

import contextlib
import logging
import os
import unittest
from unittest import suite, loader, case
from teuthology.task import interactive
from teuthology import misc
from tasks.cephfs.filesystem import Filesystem, MDSCluster, CephCluster
from tasks.mgr.mgr_test_case import MgrCluster

log = logging.getLogger(__name__)


class DecoratingLoader(loader.TestLoader):
    """
    A specialization of TestLoader that tags some extra attributes
    onto test classes as they are loaded.
    """
    def __init__(self, params):
        self._params = params
        super(DecoratingLoader, self).__init__()

    def _apply_params(self, obj):
        for k, v in self._params.items():
            setattr(obj, k, v)

    def loadTestsFromTestCase(self, testCaseClass):
        self._apply_params(testCaseClass)
        return super(DecoratingLoader, self).loadTestsFromTestCase(testCaseClass)

    def loadTestsFromName(self, name, module=None):
        result = super(DecoratingLoader, self).loadTestsFromName(name, module)

        # Special case for when we were called with the name of a method, we get
        # a suite with one TestCase
        tests_in_result = list(result)
        if len(tests_in_result) == 1 and isinstance(tests_in_result[0], case.TestCase):
            self._apply_params(tests_in_result[0])

        return result


class LogStream(object):
    def __init__(self):
        self.buffer = ""

    def write(self, data):
        self.buffer += data
        if "\n" in self.buffer:
            lines = self.buffer.split("\n")
            for line in lines[:-1]:
                log.info(line)
            self.buffer = lines[-1]

    def flush(self):
        pass


class InteractiveFailureResult(unittest.TextTestResult):
    """
    Specialization that implements interactive-on-error style
    behavior.
    """
    ctx = None

    def addFailure(self, test, err):
        log.error(self._exc_info_to_string(err, test))
        log.error("Failure in test '{0}', going interactive".format(
            self.getDescription(test)
        ))
        interactive.task(ctx=self.ctx, config=None)

    def addError(self, test, err):
        log.error(self._exc_info_to_string(err, test))
        log.error("Error in test '{0}', going interactive".format(
            self.getDescription(test)
        ))
        interactive.task(ctx=self.ctx, config=None)


@contextlib.contextmanager
def task(ctx, config):
    """
    Run the CephFS test cases.

    Run everything in tasks/cephfs/test_*.py:

    ::

        tasks:
          - install:
          - ceph:
          - ceph-fuse:
          - cephfs_test_runner:

    `modules` argument allows running only some specific modules:

    ::

        tasks:
            ...
          - cephfs_test_runner:
              modules:
                - tasks.cephfs.test_sessionmap
                - tasks.cephfs.test_auto_repair

    By default, any cases that can't be run on the current cluster configuration
    will generate a failure.  When the optional `fail_on_skip` argument is set
    to false, any tests that can't be run on the current configuration will
    simply be skipped:

    ::
        tasks:
            ...
         - cephfs_test_runner:
           fail_on_skip: false

    """

    ceph_cluster = CephCluster(ctx)

    if len(list(misc.all_roles_of_type(ctx.cluster, 'mds'))):
        mds_cluster = MDSCluster(ctx)
        fs = Filesystem(ctx)
    else:
        mds_cluster = None
        fs = None

    if len(list(misc.all_roles_of_type(ctx.cluster, 'mgr'))):
        mgr_cluster = MgrCluster(ctx)
    else:
        mgr_cluster = None

    # Mount objects, sorted by ID
    if hasattr(ctx, 'mounts'):
        mounts = [v for k, v in sorted(ctx.mounts.items(), key=lambda mount: mount[0])]
    else:
        # The test configuration has a filesystem but no fuse/kclient mounts
        mounts = []

    decorating_loader = DecoratingLoader({
        "ctx": ctx,
        "mounts": mounts,
        "fs": fs,
        "ceph_cluster": ceph_cluster,
        "mds_cluster": mds_cluster,
        "mgr_cluster": mgr_cluster,
    })

    fail_on_skip = config.get('fail_on_skip', True)

    # Put useful things onto ctx for interactive debugging
    ctx.fs = fs
    ctx.mds_cluster = mds_cluster
    ctx.mgr_cluster = mgr_cluster

    # Depending on config, either load specific modules, or scan for moduless
    if config and 'modules' in config and config['modules']:
        module_suites = []
        for mod_name in config['modules']:
            # Test names like cephfs.test_auto_repair
            module_suites.append(decorating_loader.loadTestsFromName(mod_name))
        overall_suite = suite.TestSuite(module_suites)
    else:
        # Default, run all tests
        overall_suite = decorating_loader.discover(
            os.path.join(
                os.path.dirname(os.path.abspath(__file__)),
                "cephfs/"
            )
        )

    if ctx.config.get("interactive-on-error", False):
        InteractiveFailureResult.ctx = ctx
        result_class = InteractiveFailureResult
    else:
        result_class = unittest.TextTestResult

    class LoggingResult(result_class):
        def startTest(self, test):
            log.info("Starting test: {0}".format(self.getDescription(test)))
            return super(LoggingResult, self).startTest(test)

        def addSkip(self, test, reason):
            if fail_on_skip:
                # Don't just call addFailure because that requires a traceback
                self.failures.append((test, reason))
            else:
                super(LoggingResult, self).addSkip(test, reason)

    # Execute!
    result = unittest.TextTestRunner(
        stream=LogStream(),
        resultclass=LoggingResult,
        verbosity=2,
        failfast=True).run(overall_suite)

    if not result.wasSuccessful():
        result.printErrors()  # duplicate output at end for convenience

        bad_tests = []
        for test, error in result.errors:
            bad_tests.append(str(test))
        for test, failure in result.failures:
            bad_tests.append(str(test))

        raise RuntimeError("Test failure: {0}".format(", ".join(bad_tests)))

    yield
A specialization of TestLoader that tags some extra attributes     onto test classes as they are loaded. # Special case for when we were called with the name of a method, we get # a suite with one TestCase Specialization that implements interactive-on-error style     behavior. Run the CephFS test cases.      Run everything in tasks/cephfs/test_*.py:      ::          tasks:           - install:           - ceph:           - ceph-fuse:           - cephfs_test_runner:      `modules` argument allows running only some specific modules:      ::          tasks:             ...           - cephfs_test_runner:               modules:                 - tasks.cephfs.test_sessionmap                 - tasks.cephfs.test_auto_repair      By default, any cases that can't be run on the current cluster configuration     will generate a failure.  When the optional `fail_on_skip` argument is set     to false, any tests that can't be run on the current configuration will     simply be skipped:      ::         tasks:             ...          - cephfs_test_runner:            fail_on_skip: false # Mount objects, sorted by ID # The test configuration has a filesystem but no fuse/kclient mounts # Put useful things onto ctx for interactive debugging # Depending on config, either load specific modules, or scan for moduless # Test names like cephfs.test_auto_repair # Default, run all tests # Don't just call addFailure because that requires a traceback # Execute! # duplicate output at end for convenience