Python unittest 模块，SkipTest() 实例源码

def download_test_files_if_not_present(self):
        '''
        Download %s file at G-node for testing
        url_for_tests is global at beginning of this file.

        ''' % self.ioclass.__name__
        if not self.use_network:
            raise unittest.SkipTest("Requires download of data from the web")

        url = url_for_tests+self.shortname
        try:
            make_all_directories(self.files_to_download, self.local_test_dir)
            download_test_file(self.files_to_download,
                               self.local_test_dir, url)
        except IOError as exc:
            raise unittest.SkipTest(exc)

def setUp(self):
        """
        Test is skipped if run with native protocol version <4
        """
        self.support_v5 = True
        if PROTOCOL_VERSION < 4:
            raise unittest.SkipTest(
                "Native protocol 4,0+ is required for custom payloads, currently using %r"
                % (PROTOCOL_VERSION,))
        try:
            self.cluster = Cluster(protocol_version=ProtocolVersion.MAX_SUPPORTED, allow_beta_protocol_version=True)
            self.session = self.cluster.connect()
        except NoHostAvailable:
            log.info("Protocol Version 5 not supported,")
            self.cluster = Cluster(protocol_version=PROTOCOL_VERSION)
            self.session = self.cluster.connect()
            self.support_v5 = False

        self.nodes_currently_failing = []
        self.node1, self.node2, self.node3 = get_cluster().nodes.values()

def _deferredSkip(condition, reason):
    def decorator(test_func):
        if not (isinstance(test_func, type) and
                issubclass(test_func, unittest.TestCase)):
            @wraps(test_func)
            def skip_wrapper(*args, **kwargs):
                if condition():
                    raise unittest.SkipTest(reason)
                return test_func(*args, **kwargs)
            test_item = skip_wrapper
        else:
            # Assume a class is decorated
            test_item = test_func
            test_item.__unittest_skip__ = CheckCondition(condition)
        test_item.__unittest_skip_why__ = reason
        return test_item
    return decorator

def compare_cpu_gpu(tensor_constructor, arg_constructor, fn, t, precision=1e-5):
    def tmp(self):
        cpu_tensor = tensor_constructor(t)
        gpu_tensor = to_gpu(cpu_tensor)
        cpu_args = arg_constructor(t)
        gpu_args = [to_gpu(arg) for arg in cpu_args]
        cpu_result = getattr(cpu_tensor, fn)(*cpu_args)
        try:
            gpu_result = getattr(gpu_tensor, fn)(*gpu_args)
        except RuntimeError as e:
            reason = e.args[0]
            if 'unimplemented data type' in reason:
                raise unittest.SkipTest('unimplemented data type')
            raise
        except AttributeError as e:
            reason = e.args[0]
            if 'object has no attribute' in reason:
                raise unittest.SkipTest('unimplemented data type')
            raise
        # If one changes, another should change as well
        self.assertEqual(cpu_tensor, gpu_tensor, precision)
        self.assertEqual(cpu_args, gpu_args, precision)
        # Compare results
        self.assertEqual(cpu_result, gpu_result, precision)
    return tmp

def test_cuda(self, test_case):
        if not TEST_CUDA or not self.should_test_cuda:
            raise unittest.SkipTest('Excluded from CUDA tests')
        try:
            cpu_input = self._get_input()
            type_map = {
                torch.DoubleTensor: torch.cuda.FloatTensor,
            }
            gpu_input = to_gpu(cpu_input, type_map=type_map)

            cpu_target = self.target
            gpu_target = to_gpu(self.target, type_map=type_map)

            cpu_module = self.constructor(*self.constructor_args)
            gpu_module = self.constructor(*self.constructor_args).float().cuda()

            cpu_output = test_case._forward_criterion(cpu_module, cpu_input, cpu_target)
            gpu_output = test_case._forward_criterion(gpu_module, gpu_input, gpu_target)
            test_case.assertEqual(cpu_output, gpu_output, 2e-4)

            cpu_gradInput = test_case._backward_criterion(cpu_module, cpu_input, cpu_target)
            gpu_gradInput = test_case._backward_criterion(gpu_module, gpu_input, gpu_target)
            test_case.assertEqual(cpu_gradInput, gpu_gradInput, 2e-4)
        except NotImplementedError:
            pass

def test_skiptest_in_setupclass(self):
        class Test(unittest.TestCase):
            @classmethod
            def setUpClass(cls):
                raise unittest.SkipTest('foo')
            def test_one(self):
                pass
            def test_two(self):
                pass

        result = self.runTests(Test)
        self.assertEqual(result.testsRun, 0)
        self.assertEqual(len(result.errors), 0)
        self.assertEqual(len(result.skipped), 1)
        skipped = result.skipped[0][0]
        self.assertEqual(str(skipped), 'setUpClass (%s.Test)' % __name__)

def test_skiptest_in_setupmodule(self):
        class Test(unittest.TestCase):
            def test_one(self):
                pass
            def test_two(self):
                pass

        class Module(object):
            @staticmethod
            def setUpModule():
                raise unittest.SkipTest('foo')

        Test.__module__ = 'Module'
        sys.modules['Module'] = Module

        result = self.runTests(Test)
        self.assertEqual(result.testsRun, 0)
        self.assertEqual(len(result.errors), 0)
        self.assertEqual(len(result.skipped), 1)
        skipped = result.skipped[0][0]
        self.assertEqual(str(skipped), 'setUpModule (Module)')

def copy_xxmodule_c(directory):
    """Helper for tests that need the xxmodule.c source file.

    Example use:

        def test_compile(self):
            copy_xxmodule_c(self.tmpdir)
            self.assertIn('xxmodule.c', os.listdir(self.tmpdir))

    If the source file can be found, it will be copied to *directory*.  If not,
    the test will be skipped.  Errors during copy are not caught.
    """
    filename = _get_xxmodule_path()
    if filename is None:
        raise unittest.SkipTest('cannot find xxmodule.c (test must run in '
                                'the python build dir)')
    shutil.copy(filename, directory)

def get_tests(package, mask, verbosity, exclude=()):
    """Return a list of skipped test modules, and a list of test cases."""
    tests = []
    skipped = []
    for modname in find_package_modules(package, mask):
        if modname.split(".")[-1] in exclude:
            skipped.append(modname)
            if verbosity > 1:
                print >> sys.stderr, "Skipped %s: excluded" % modname
            continue
        try:
            mod = __import__(modname, globals(), locals(), ['*'])
        except (ResourceDenied, unittest.SkipTest) as detail:
            skipped.append(modname)
            if verbosity > 1:
                print >> sys.stderr, "Skipped %s: %s" % (modname, detail)
            continue
        for name in dir(mod):
            if name.startswith("_"):
                continue
            o = getattr(mod, name)
            if type(o) is type(unittest.TestCase) and issubclass(o, unittest.TestCase):
                tests.append(o)
    return skipped, tests

def test_attributes(self):
        p = self.thetype(capture, 1, 2, a=10, b=20)
        # attributes should be readable
        self.assertEqual(p.func, capture)
        self.assertEqual(p.args, (1, 2))
        self.assertEqual(p.keywords, dict(a=10, b=20))
        # attributes should not be writable
        if not isinstance(self.thetype, type):
            return
        if "__pypy__" in sys.modules:
            raise unittest.SkipTest("In the PyPy execution environment")
        self.assertRaises(TypeError, setattr, p, 'func', map)
        self.assertRaises(TypeError, setattr, p, 'args', (1, 2))
        self.assertRaises(TypeError, setattr, p, 'keywords', dict(a=1, b=2))

        p = self.thetype(hex)
        try:
            del p.__dict__
        except TypeError:
            pass
        else:
            self.fail('partial object allowed __dict__ to be deleted')

def __getattr__(self, item):
        if self._pandas_only:
            raise SkipTest("empyrical.%s expects pandas-only inputs that have "
                           "dt indices/labels" % item)

        func = super(ConvertPandasEmpyricalProxy, self).__getattr__(item)

        @wraps(func)
        def convert_args(*args, **kwargs):
            args = [self._convert(arg) if isinstance(arg, NDFrame) else arg
                    for arg in args]
            kwargs = {
                k: self._convert(v) if isinstance(v, NDFrame) else v
                for k, v in iteritems(kwargs)
            }
            return func(*args, **kwargs)

        return convert_args

def get_e2e_configuration():
    config = Configuration()
    config.host = None
    if os.path.exists(
            os.path.expanduser(kube_config.KUBE_CONFIG_DEFAULT_LOCATION)):
        kube_config.load_kube_config(client_configuration=config)
    else:
        print('Unable to load config from %s' %
              kube_config.KUBE_CONFIG_DEFAULT_LOCATION)
        for url in ['https://%s:8443' % DEFAULT_E2E_HOST,
                    'http://%s:8080' % DEFAULT_E2E_HOST]:
            try:
                urllib3.PoolManager().request('GET', url)
                config.host = url
                config.verify_ssl = False
                urllib3.disable_warnings()
                break
            except urllib3.exceptions.HTTPError:
                pass
    if config.host is None:
        raise unittest.SkipTest('Unable to find a running Kubernetes instance')
    print('Running test against : %s' % config.host)
    config.assert_hostname = False
    return config

def test_08_create_adf(self):
        # Get number of EF_DIR records
        status, data = self.shell.read("/2F00")
        self.shell.assertOk(status, data)
        numOfRecords = len(data.split(';')) - 1
        # Use the next free Id
        dirPath = "/ADF%d/" % numOfRecords
        try:
            self.shell.delete(dirPath)
        except:
            pass
        status, out = self.shell.create(dirPath)
        if status == "status NOK":
            raise unittest.SkipTest(
                """Known issue: ADF creation doesn't work for some SIM cards
                 (INCORRECT_PARAMETER_IN_DATA_FIELD is returned)""")
        #self.shell.assertOk(status, out)
        status, out = self.shell.delete(dirPath)
        self.shell.assertOk(status, out)

def gc_collect():
    """Run enough GC collections to make object finalizers run."""

    # XXX Tests of GC and cleanup behavior are generally flaky and icky,
    # especially when you target all of Python 2.7, 3.3+ and PyPy. Their result
    # quickly depends on other tests, the arguments to the test runner and the
    # computer running the tests. This skips them all for now.
    raise unittest.SkipTest

    if platform.python_implementation() == 'PyPy':
        # Since PyPy use garbage collection instead of reference counting
        # objects are not finalized before the next major GC collection.
        # Currently, the best way we have to ensure a major GC collection has
        # run is to call gc.collect() a number of times.
        [gc.collect() for _ in range(10)]
    else:
        gc.collect()

def test_get_environment(self):
        try:
            environment = self.device.get_environment()
        except NotImplementedError:
            raise SkipTest()
        result = len(environment) > 0

        for fan, fan_data in environment['fans'].items():
            result = result and self._test_model(models.fan, fan_data)

        for power, power_data in environment['power'].items():
            result = result and self._test_model(models.power, power_data)

        for temperature, temperature_data in environment['temperature'].items():
            result = result and self._test_model(models.temperature, temperature_data)

        for cpu, cpu_data in environment['cpu'].items():
            result = result and self._test_model(models.cpu, cpu_data)

        result = result and self._test_model(models.memory, environment['memory'])

        self.assertTrue(result)

def test_get_bgp_neighbors(self):
        try:
            get_bgp_neighbors = self.device.get_bgp_neighbors()
        except NotImplementedError:
            raise SkipTest()
        result = 'global' in get_bgp_neighbors.keys()

        if not result:
            print('global is not part of the returned vrfs')
        else:
            for vrf, vrf_data in get_bgp_neighbors.items():
                result = result and isinstance(vrf_data['router_id'], text_type)
                if not result:
                    print('router_id is not {}'.format(text_type))

                for peer, peer_data in vrf_data['peers'].items():
                    result = result and self._test_model(models.peer, peer_data)

                    for af, af_data in peer_data['address_family'].items():
                        result = result and self._test_model(models.af, af_data)

            self.assertTrue(result)

def test_get_bgp_neighbors_detail(self):
        try:
            get_bgp_neighbors_detail = self.device.get_bgp_neighbors_detail()
        except NotImplementedError:
            raise SkipTest()

        result = len(get_bgp_neighbors_detail) > 0

        for vrf, vrf_ases in get_bgp_neighbors_detail.items():
            result = result and isinstance(vrf, text_type)
            for remote_as, neighbor_list in vrf_ases.items():
                result = result and isinstance(remote_as, int)
                for neighbor in neighbor_list:
                    result = result and self._test_model(models.peer_details, neighbor)

        self.assertTrue(result)

def test_get_optics(self):

        try:
            get_optics = self.device.get_optics()
        except NotImplementedError:
            raise SkipTest()

        assert isinstance(get_optics, dict)

        for iface, iface_data in get_optics.items():
            assert isinstance(iface, text_type)
            for channel in iface_data['physical_channels']['channel']:
                assert len(channel) == 2
                assert isinstance(channel['index'], int)
                for field in ['input_power', 'output_power',
                              'laser_bias_current']:

                    assert len(channel['state'][field]) == 4
                    assert isinstance(channel['state'][field]['instant'],
                                      float)
                    assert isinstance(channel['state'][field]['avg'], float)
                    assert isinstance(channel['state'][field]['min'], float)
                    assert isinstance(channel['state'][field]['max'], float)

def test_importlib_import_relative_pkg(self):
            """Verify that package is importable relatively"""
            print_importers()
            assert __package__
            # need globals to handle relative imports
            # __import__ checks sys.modules by itself
            # but the test is not reflecting anything if we use the already loaded module.
            if sys.modules.get(__package__ + '.pkg'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.pkg'))
            else:
                pkg = importlib.__import__('pkg', globals=globals(), level=1)
                test_pkg = pkg

                self.assertTrue(test_pkg is not None)
                self.assertTrue(test_pkg.TestClassInSubPkg is not None)
                self.assertTrue(callable(test_pkg.TestClassInSubPkg))

                # TODO : implement some differences and check we get them...
                if hasattr(importlib, 'reload'):  # recent version of importlib
                    # attempting to reload
                    importlib.reload(test_pkg)
                else:
                    pass

def test_importlib_import_relative_pkg_submodule(self):
            """Verify that package is importable relatively"""
            print_importers()
            assert __package__
            # need globals to handle relative imports
            # __import__ checks sys.modules by itself
            # but the test is not reflecting anything if we use the already loaded module.
            if sys.modules.get(__package__ + '.pkg.submodule'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.pkg.submodule'))
            else:
                pkg = importlib.__import__('pkg.submodule', globals=globals(), level=1)
                test_mod = pkg.submodule

                self.assertTrue(test_mod is not None)
                self.assertTrue(test_mod.TestClassInSubModule is not None)
                self.assertTrue(callable(test_mod.TestClassInSubModule))

                # TODO : implement some differences and check we get them...
                if hasattr(importlib, 'reload'):  # recent version of importlib
                    # attempting to reload
                    importlib.reload(test_mod)
                else:
                    pass

def test_importlib_import_relative_pkg_bytecode(self):
            """Verify that package is importable relatively"""
            print_importers()
            assert __package__
            # need globals to handle relative imports
            # __import__ checks sys.modules by itself
            # but the test is not reflecting anything if we use the already loaded module.
            if sys.modules.get(__package__ + '.pkg.bytecode'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.pkg.bytecode'))
            else:
                pkg = importlib.__import__('pkg.bytecode', globals=globals(), level=1)
                test_mod = pkg.bytecode

                self.assertTrue(test_mod is not None)
                self.assertTrue(test_mod.TestClassInBytecode is not None)
                self.assertTrue(callable(test_mod.TestClassInBytecode))

                # TODO : implement some differences and check we get them...
                if hasattr(importlib, 'reload'):  # recent version of importlib
                    # attempting to reload
                    importlib.reload(test_mod)
                else:
                    pass

def test_importlib_import_class_from_relative_pkg(self):
            """Verify that message class is importable relatively"""
            print_importers()
            assert __package__
            # need globals to handle relative imports
            # __import__ checks sys.modules by itself
            # but the test is not reflecting anything if we use the already loaded module.
            if sys.modules.get(__package__ + '.pkg'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.pkg'))
            else:
                pkg = importlib.__import__('pkg', globals=globals(), level=1)
                test_class_in_subpkg = pkg.TestClassInSubPkg

                self.assertTrue(test_class_in_subpkg is not None)
                self.assertTrue(callable(test_class_in_subpkg))

                # TODO : implement some differences and check we get them...
                if hasattr(importlib, 'reload'):  # recent version of importlib
                    # attempting to reload
                    importlib.reload(pkg)
                else:
                    pass

def test_importlib_import_class_from_relative_pkg_submodule(self):
            """Verify that package is importable relatively"""
            print_importers()
            assert __package__
            # need globals to handle relative imports
            # __import__ checks sys.modules by itself
            # but the test is not reflecting anything if we use the already loaded module.
            if sys.modules.get(__package__ + '.pkg.submodule'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.pkg.submodule'))
            else:
                pkg = importlib.__import__('pkg.submodule', globals=globals(), level=1)
                test_class_in_submodule = pkg.submodule.TestClassInSubModule

                self.assertTrue(test_class_in_submodule is not None)
                self.assertTrue(callable(test_class_in_submodule))

                # TODO : implement some differences and check we get them...
                if hasattr(importlib, 'reload'):  # recent version of importlib
                    # attempting to reload
                    importlib.reload(pkg)
                else:
                    pass

def test_importlib_import_relative_ns_subpkg(self):
            """Verify that package is importable relatively"""
            print_importers()
            assert __package__

            # __import__ checks sys.modules by itself
            # but the test is not reflecting anything if we use the already loaded module.
            if sys.modules.get(__package__ + '.nspkg.subpkg'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.nspkg.subpkg'))
            else:
                nspkg = importlib.__import__('nspkg.subpkg', globals=globals(),
                                             level=1)  # need globals to handle relative imports
                test_pkg = nspkg.subpkg

                self.assertTrue(test_pkg is not None)
                self.assertTrue(test_pkg.TestClassInSubPkg is not None)
                self.assertTrue(callable(test_pkg.TestClassInSubPkg))

                # TODO : implement some differences and check we get them...
                if hasattr(importlib, 'reload'):  # recent version of importlib
                    # attempting to reload
                    importlib.reload(test_pkg)
                else:
                    pass

def test_importlib_import_relative_ns_subpkg_submodule(self):
            """Verify that package is importable relatively"""
            print_importers()
            assert __package__

            # __import__ checks sys.modules by itself
            # but the test is not reflecting anything if we use the already loaded module.
            if sys.modules.get(__package__ + '.nspkg.subpkg.submodule'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.nspkg.subpkg.submodule'))
            else:
                nspkg = importlib.__import__('nspkg.subpkg.submodule', globals=globals(),
                                             level=1)  # need globals to handle relative imports
                test_mod = nspkg.subpkg.submodule

                self.assertTrue(test_mod is not None)
                self.assertTrue(test_mod.TestClassInSubModule is not None)
                self.assertTrue(callable(test_mod.TestClassInSubModule))

                # TODO : implement some differences and check we get them...
                if hasattr(importlib, 'reload'):  # recent version of importlib
                    # attempting to reload
                    importlib.reload(test_mod)
                else:
                    pass

def test_importlib_import_relative_ns_subpkg_bytecode(self):
            """Verify that package is importable relatively"""
            print_importers()
            assert __package__

            # __import__ checks sys.modules by itself
            # but the test is not reflecting anything if we use the already loaded module.
            if sys.modules.get(__package__ + '.nspkg.subpkg.bytecode'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.nspkg.subpkg.bytecode'))
            else:
                nspkg = importlib.__import__('nspkg.subpkg.bytecode', globals=globals(),
                                             level=1)  # need globals to handle relative imports
                test_mod = nspkg.subpkg.bytecode

                self.assertTrue(test_mod is not None)
                self.assertTrue(test_mod.TestClassInBytecode is not None)
                self.assertTrue(callable(test_mod.TestClassInBytecode))

                # TODO : implement some differences and check we get them...
                if hasattr(importlib, 'reload'):  # recent version of importlib
                    # attempting to reload
                    importlib.reload(test_mod)
                else:
                    pass

def test_importlib_import_class_from_relative_ns_subpkg(self):
            """Verify that message class is importable relatively"""
            print_importers()
            assert __package__

            # __import__ checks sys.modules by itself
            # but the test is not reflecting anything if we use the already loaded module.
            if sys.modules.get(__package__ + '.nspkg.subpkg'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.nspkg.subpkg'))
            else:
                nspkg = importlib.__import__('nspkg.subpkg', globals=globals(),
                                             level=1)  # need globals to handle relative imports
                test_class_in_subpkg = nspkg.subpkg.TestClassInSubPkg

                self.assertTrue(test_class_in_subpkg is not None)
                self.assertTrue(callable(test_class_in_subpkg))

                # TODO : implement some differences and check we get them...
                if hasattr(importlib, 'reload'):  # recent version of importlib
                    # attempting to reload
                    importlib.reload(nspkg)
                else:
                    pass

def test_importlib_import_class_from_relative_ns_subpkg_bytecode(self):
            """Verify that package is importable relatively"""
            print_importers()
            assert __package__

            # __import__ checks sys.modules by itself
            # but the test is not reflecting anything if we use the already loaded module.
            if sys.modules.get(__package__ + '.nspkg.subpkg.bytecode'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.nspkg.subpkg.bytecode'))
            else:
                nspkg = importlib.__import__('nspkg.subpkg.bytecode', globals=globals(),
                                             level=1)  # need globals to handle relative imports
                test_class_in_bytecode = nspkg.subpkg.bytecode.TestClassInBytecode

                self.assertTrue(test_class_in_bytecode is not None)
                self.assertTrue(callable(test_class_in_bytecode))

                # TODO : implement some differences and check we get them...
                if hasattr(importlib, 'reload'):  # recent version of importlib
                    # attempting to reload
                    importlib.reload(nspkg)
                else:
                    pass

def test_importlib_loadmodule_relative_pkg(self):
            """Verify that package is importable relatively"""
            print_importers()
            assert __package__
            if sys.modules.get(__package__ + '.pkg'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.pkg'))
            else:
                # load_module returns existing modules from sys.modules by specification
                # see https://docs.python.org/3.3/library/importlib.html#importlib.abc.Loader)
                pkg_loader = importlib.find_loader(__package__ + '.pkg', [os.path.dirname(__file__)])
                pkg = pkg_loader.load_module(__package__ + '.pkg')
                # safely adding to sysmodules to be able to perform relative imports in there
                #sys.modules.setdefault(nspkg.__name__, nspkg)

            self.assertTrue(pkg is not None)
            self.assertTrue(pkg.TestClassInSubPkg is not None)
            self.assertTrue(callable(pkg.TestClassInSubPkg))

            # Note : apparently reload is broken with find_loader (at least on python 3.5)
            # _find_spec in reload() apparently returns None...
            #
            # => not testing reload in that case (this API is obsolete anyway)

def test_importlib_loadmodule_relative_pkg_submodule(self):
            """Verify that package is importable relatively"""
            print_importers()
            assert __package__
            if sys.modules.get(__package__ + '.pkg'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.pkg'))
            else:
                # load_module returns existing modules from sys.modules by specification
                # see https://docs.python.org/3.3/library/importlib.html#importlib.abc.Loader)
                pkg_loader = importlib.find_loader(__package__ + '.pkg', [os.path.dirname(__file__)])
                pkg = pkg_loader.load_module(__package__ + '.pkg')
                # safely adding to sysmodules to be able to perform relative imports in there
                #sys.modules.setdefault(nspkg.__name__, nspkg)

            # here we should get the module that has already be loaded while executing subpkg
            submodule = sys.modules.get(__package__ + '.pkg.submodule')

            self.assertTrue(submodule is not None)
            self.assertTrue(submodule.TestClassInSubModule is not None)
            self.assertTrue(callable(submodule.TestClassInSubModule))

            # Note : apparently reload is broken with find_loader (at least on python 3.5)
            # _find_spec in reload() apparently returns None...
            #
            # => not testing reload in that case (this API is obsolete anyway)

def test_importlib_importmodule_relative_ns_subpkg(self):
            """Verify that package is importable relatively"""
            print_importers()
            assert __package__

            # import_module checks sys.modules by itself
            # but the test is not reflecting anything if we use the already loaded module.
            if sys.modules.get(__package__ + '.nspkg.subpkg'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.nspkg.subpkg'))
            else:
                test_pkg = importlib.import_module('.nspkg.subpkg', package=__package__)

                self.assertTrue(test_pkg is not None)
                self.assertTrue(test_pkg.TestClassInSubPkg is not None)
                self.assertTrue(callable(test_pkg.TestClassInSubPkg))

                # TODO : implement some differences and check we get them...
                if hasattr(importlib, 'reload'):  # recent version of importlib
                    # attempting to reload
                    importlib.reload(test_pkg)
                else:
                    pass

def test_importlib_importmodule_relative_ns_subpkg_bytecode(self):
            """Verify that package is importable relatively"""
            print_importers()
            assert __package__

            # import_module checks sys.modules by itself
            # but the test is not reflecting anything if we use the already loaded module.
            if sys.modules.get(__package__ + '.nspkg.subpkg.bytecode'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.nspkg.subpkg.bytecode'))
            else:
                test_mod = importlib.import_module('.nspkg.subpkg.bytecode', package=__package__)

                self.assertTrue(test_mod is not None)
                self.assertTrue(test_mod.TestClassInBytecode is not None)
                self.assertTrue(callable(test_mod.TestClassInBytecode))

                # TODO : implement some differences and check we get them...
                if hasattr(importlib, 'reload'):  # recent version of importlib
                    # attempting to reload
                    importlib.reload(test_mod)
                else:
                    pass

def test_importlib_importmodule_class_from_relative_ns_subpkg(self):
            """Verify that test class is importable relatively"""
            print_importers()
            assert __package__

            # import_module checks sys.modules by itself
            # but the test is not reflecting anything if we use the already loaded module.
            if sys.modules.get(__package__ + '.nspkg.subpkg'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.nspkg.subpkg'))
            else:
                nspkg_subpkg = importlib.import_module('.nspkg.subpkg', package=__package__)
                test_class_in_subpkg = nspkg_subpkg.TestClassInSubPkg

                self.assertTrue(test_class_in_subpkg is not None)
                self.assertTrue(callable(test_class_in_subpkg))

                # TODO : implement some differences and check we get them...
                if hasattr(importlib, 'reload'):  # recent version of importlib
                    # attempting to reload
                    importlib.reload(nspkg_subpkg)
                else:
                    pass

def test_importlib_importmodule_class_from_relative_ns_subpkg_submodule(self):
            """Verify that test class is importable relatively"""
            print_importers()
            assert __package__

            # import_module checks sys.modules by itself
            # but the test is not reflecting anything if we use the already loaded module.
            if sys.modules.get(__package__ + '.nspkg.subpkg.submodule'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.nspkg.subpkg.submodule'))
            else:
                nspkg_subpkg_submodule = importlib.import_module('.nspkg.subpkg.submodule', package=__package__)
                test_class_in_submodule = nspkg_subpkg_submodule.TestClassInSubModule

                self.assertTrue(test_class_in_submodule is not None)
                self.assertTrue(callable(test_class_in_submodule))

                # TODO : implement some differences and check we get them...
                if hasattr(importlib, 'reload'):  # recent version of importlib
                    # attempting to reload
                    importlib.reload(nspkg_subpkg_submodule)
                else:
                    pass

def test_importlib_importmodule_class_from_relative_ns_subpkg_bytecode(self):
            """Verify that test class is importable relatively"""
            print_importers()
            assert __package__

            # import_module checks sys.modules by itself
            # but the test is not reflecting anything if we use the already loaded module.
            if sys.modules.get(__package__ + '.nspkg.subpkg.bytecode'):
                raise unittest.SkipTest("module previously loaded".format(__package__ + '.nspkg.subpkg.bytecode'))
            else:
                nspkg_subpkg_bytecode = importlib.import_module('.nspkg.subpkg.bytecode', package=__package__)
                test_class_in_bytecode = nspkg_subpkg_bytecode.TestClassInBytecode

                self.assertTrue(test_class_in_bytecode is not None)
                self.assertTrue(callable(test_class_in_bytecode))

                # TODO : implement some differences and check we get them...
                if hasattr(importlib, 'reload'):  # recent version of importlib
                    # attempting to reload
                    importlib.reload(nspkg_subpkg_bytecode)
                else:
                    pass

def __init__(self,
                 conf_template,
                 udp_enabled=False):

        if os.environ.get("INFLUXDB_PYTHON_SKIP_SERVER_TESTS", None) == 'True':
            raise unittest.SkipTest(
                "Skipping server test (INFLUXDB_PYTHON_SKIP_SERVER_TESTS)"
            )

        self.influxd_path = self.find_influxd_path()

        errors = 0
        while True:
            try:
                self._start_server(conf_template, udp_enabled)
                break
            # Happens when the ports are already in use.
            except RuntimeError as e:
                errors += 1
                if errors > 2:
                    raise e

def find_influxd_path(self):
        influxdb_bin_path = os.environ.get(
            'INFLUXDB_PYTHON_INFLUXD_PATH',
            None
        )

        if influxdb_bin_path is None:
            influxdb_bin_path = distutils.spawn.find_executable('influxd')
            if not influxdb_bin_path:
                try:
                    influxdb_bin_path = subprocess.check_output(
                        ['which', 'influxd']
                    ).strip()
                except subprocess.CalledProcessError:
                    # fallback on :
                    influxdb_bin_path = '/opt/influxdb/influxd'

        if not os.path.isfile(influxdb_bin_path):
            raise unittest.SkipTest("Could not find influxd binary")

        version = subprocess.check_output([influxdb_bin_path, 'version'])
        print("InfluxDB version: %s" % version, file=sys.stderr)

        return influxdb_bin_path

def test_is_ancestor(self):
        git = self.rorepo.git
        if git.version_info[:3] < (1, 8, 0):
            raise SkipTest("git merge-base --is-ancestor feature unsupported")

        repo = self.rorepo
        c1 = 'f6aa8d1'
        c2 = '763ef75'
        self.assertTrue(repo.is_ancestor(c1, c1))
        self.assertTrue(repo.is_ancestor("master", "master"))
        self.assertTrue(repo.is_ancestor(c1, c2))
        self.assertTrue(repo.is_ancestor(c1, "master"))
        self.assertFalse(repo.is_ancestor(c2, c1))
        self.assertFalse(repo.is_ancestor("master", c1))
        for i, j in itertools.permutations([c1, 'ffffff', ''], r=2):
            self.assertRaises(GitCommandError, repo.is_ancestor, i, j)

def test_work_tree_unsupported(self, rw_dir):
        git = Git(rw_dir)
        if git.version_info[:3] < (2, 5, 1):
            raise SkipTest("worktree feature unsupported")

        rw_master = self.rorepo.clone(join_path_native(rw_dir, 'master_repo'))
        rw_master.git.checkout('HEAD~10')
        worktree_path = join_path_native(rw_dir, 'worktree_repo')
        if Git.is_cygwin():
            worktree_path = cygpath(worktree_path)
        try:
            rw_master.git.worktree('add', worktree_path, 'master')
        except Exception as ex:
            raise AssertionError(ex, "It's ok if TC not running from `master`.")

        self.failUnlessRaises(InvalidGitRepositoryError, Repo, worktree_path)

def setUpClass(cls):
        """ Initialise the test class.
        """
        warnings.simplefilter('ignore', ResourceWarning)

        # Get a client to the local docker engine.
        client = docker.from_env()

        # Skip the tests in this class if not running from a manager node.
        if not client.info()['Swarm']['ControlAvailable']:
            raise unittest.SkipTest('This test must be run from a swarm '
                                    'manager node.')
            # client.swarm.init()

        # Create a logging server
        # FIXME(BM): This should not be needed to test heartbeat!
        paas = Paas()
        cls.logger = paas.run_service(
            'logging_server',
            'sip',
            [logging.handlers.DEFAULT_TCP_LOGGING_PORT],
            ['python3', 'sip/common/logging_server.py'])

        # Wait for the logging server to come online.
        time.sleep(3)

def setUpClass(cls):
        """ Set up test fixture.

        Skip the tests if it is not possible to connect to the Spark Master.
        """
        paas = SparkPaaS()
        try:
            master_url = "http://{}:{}".format(
                paas.spark_master['url'], paas.spark_master['master_port'])
            req = requests.get(master_url)
            if not req.ok:
                raise unittest.SkipTest('No 200 response from Spark Master '
                                        '@[{}], Skipping tests.'.
                                        format(master_url))
        except requests.exceptions.RequestException:
            print('Exception thrown..')
            raise unittest.SkipTest("Cannot connect to Spark Master @ [{}]. "
                                    "Skipping tests.". format(master_url))
        cls.paas = paas

def test_181_execute_gte_v10(self):
        if self.client.server_version < pkg_resources.parse_version('10.0'):
            raise unittest.SkipTest(
                'Not applicable to Odoo version less then 10.0')

        res = self.client.execute('res.partner', 'read', 1)
        self.assertIsInstance(res, list)
        self.assertEqual(len(res), 1)
        self.assertIsInstance(res[0], dict)
        self.assertEqual(res[0]['id'], 1)

        res = self.client.execute('res.partner', 'read', [1])
        self.assertIsInstance(res, list)
        self.assertEqual(len(res), 1)
        self.assertIsInstance(res[0], dict)
        self.assertEqual(res[0]['id'], 1)

def requires(resource, msg=None):
    """Raise ResourceDenied if the specified resource is not available.

    If the caller's module is __main__ then automatically return True.  The
    possibility of False being returned occurs when regrtest.py is
    executing.
    """
    if resource == 'gui' and not _is_gui_available():
        raise unittest.SkipTest("Cannot use the 'gui' resource")
    # see if the caller's module is __main__ - if so, treat as if
    # the resource was set
    if sys._getframe(1).f_globals.get("__name__") == "__main__":
        return
    if not is_resource_enabled(resource):
        if msg is None:
            msg = "Use of the %r resource not enabled" % resource
        raise ResourceDenied(msg)

def bigaddrspacetest(f):
    """Decorator for tests that fill the address space."""
    def wrapper(self):
        if max_memuse < MAX_Py_ssize_t:
            if MAX_Py_ssize_t >= 2**63 - 1 and max_memuse >= 2**31:
                raise unittest.SkipTest(
                    "not enough memory: try a 32-bit build instead")
            else:
                raise unittest.SkipTest(
                    "not enough memory: %.1fG minimum needed"
                    % (MAX_Py_ssize_t / (1024 ** 3)))
        else:
            return f(self)
    return wrapper

#=======================================================================
# unittest integration.

def test_setup_class_install_environment_predefined_no_dir(self):
        from calmjs.cli import PackageManagerDriver
        from calmjs import cli

        utils.stub_os_environ(self)
        utils.stub_mod_call(self, cli)
        cwd = mkdtemp(self)
        # we have the mock_tempfile context...
        self.assertEqual(self.mock_tempfile.count, 1)
        os.chdir(cwd)

        # a very common use case
        os.environ['CALMJS_TEST_ENV'] = '.'
        TestCase = type('TestCase', (unittest.TestCase,), {})
        # the directory not there.
        with self.assertRaises(unittest.SkipTest):
            utils.setup_class_install_environment(
                TestCase, PackageManagerDriver, [])
        # temporary directory should not be created as the skip will
        # also stop the teardown from running
        self.assertEqual(self.mock_tempfile.count, 1)
        # this is still set, but irrelevant.
        self.assertEqual(TestCase._env_root, cwd)
        # tmpdir not set.
        self.assertFalse(hasattr(TestCase, '_cls_tmpdir'))

def test_linear_regression_numpy(self):
        """ Test the linear regression using numpy (if installed). """
        # test that numpy is installed
        try:
            import numpy
            numpy.__name__
        except ImportError:
            raise unittest.SkipTest('cannot test as optional numpy is not installed')
        # perform the test with numpy
        from supvisors.utils import get_linear_regression, get_simple_linear_regression
        xdata = [2, 4, 6, 8, 10, 12]
        ydata = [3, 4, 5, 6, 7, 8]
        # test linear regression
        a, b = get_linear_regression(xdata, ydata)
        self.assertAlmostEqual(0.5, a)
        self.assertAlmostEqual(2.0, b)
        # test simple linear regression
        a, b = get_simple_linear_regression(ydata)
        self.assertAlmostEqual(1.0, a)
        self.assertAlmostEqual(3.0, b)

def test_ipv4(self):
        """ Test the ipv4 method. """
        # complex to test as it depends on the network configuration of the operating system
        # check that there is at least one entry looking like an IP address
        from supvisors.addressmapper import AddressMapper
        # test that netifaces is installed
        try:
            import netifaces
            netifaces.__name__
        except ImportError:
            raise unittest.SkipTest('cannot test as optional netifaces is not installed')
        # test function
        ip_list = AddressMapper.ipv4()
        self.assertTrue(ip_list)
        for ip in ip_list:
            self.assertRegexpMatches(ip, r'^\d{1,3}(.\d{1,3}){3}$')

def skip(reason):
    """A decorator to unconditionally skip a test:

    .. code-block:: python

        @datatest.skip('Not finished collecting raw data.')
        class TestSumTotals(datatest.DataTestCase):
            def test_totals(self):
                ...
    """
    def decorator(test_item):
        if not isinstance(test_item, type):
            orig_item = test_item           # <- Not in unittest.skip()
            @functools.wraps(test_item)
            def skip_wrapper(*args, **kwargs):
                raise unittest.SkipTest(reason)
            test_item = skip_wrapper
            test_item._wrapped = orig_item  # <- Not in unittest.skip()

        test_item.__unittest_skip__ = True
        test_item.__unittest_skip_why__ = reason
        return test_item
    return decorator

def test_skiptest_in_setupclass(self):
        class Test(unittest.TestCase):
            @classmethod
            def setUpClass(cls):
                raise unittest.SkipTest('foo')
            def test_one(self):
                pass
            def test_two(self):
                pass

        result = self.runTests(Test)
        self.assertEqual(result.testsRun, 0)
        self.assertEqual(len(result.errors), 0)
        self.assertEqual(len(result.skipped), 1)
        skipped = result.skipped[0][0]
        self.assertEqual(str(skipped), 'setUpClass (%s.Test)' % __name__)

def test_skiptest_in_setupmodule(self):
        class Test(unittest.TestCase):
            def test_one(self):
                pass
            def test_two(self):
                pass

        class Module(object):
            @staticmethod
            def setUpModule():
                raise unittest.SkipTest('foo')

        Test.__module__ = 'Module'
        sys.modules['Module'] = Module

        result = self.runTests(Test)
        self.assertEqual(result.testsRun, 0)
        self.assertEqual(len(result.errors), 0)
        self.assertEqual(len(result.skipped), 1)
        skipped = result.skipped[0][0]
        self.assertEqual(str(skipped), 'setUpModule (Module)')