python类SkipTest()的实例源码

def testSocketAuthInstallPlugin(self):
        # needs plugin. lets install it.
        cur = self.connections[0].cursor()
        try:
            cur.execute("install plugin auth_socket soname 'auth_socket.so'")
            TestAuthentication.socket_found = True
            self.socket_plugin_name = 'auth_socket'
            self.realtestSocketAuth()
        except pymysql.err.InternalError:
            try:
                cur.execute("install soname 'auth_socket'")
                TestAuthentication.socket_found = True
                self.socket_plugin_name = 'unix_socket'
                self.realtestSocketAuth()
            except pymysql.err.InternalError:
                TestAuthentication.socket_found = False
                raise unittest2.SkipTest('we couldn\'t install the socket plugin')
        finally:
            if TestAuthentication.socket_found:
                cur.execute("uninstall plugin %s" % self.socket_plugin_name)

def test_json(self):
        args = self.databases[0].copy()
        args["charset"] = "utf8mb4"
        conn = pymysql.connect(**args)
        if not self.mysql_server_is(conn, (5, 7, 0)):
            raise SkipTest("JSON type is not supported on MySQL <= 5.6")

        self.safe_create_table(conn, "test_json", """\
create table test_json (
    id int not null,
    json JSON not null,
    primary key (id)
);""")
        cur = conn.cursor()

        json_str = u'{"hello": "?????"}'
        cur.execute("INSERT INTO test_json (id, `json`) values (42, %s)", (json_str,))
        cur.execute("SELECT `json` from `test_json` WHERE `id`=42")
        res = cur.fetchone()[0]
        self.assertEqual(json.loads(res), json.loads(json_str))

        cur.execute("SELECT CAST(%s AS JSON) AS x", (json_str,))
        res = cur.fetchone()[0]
        self.assertEqual(json.loads(res), json.loads(json_str))

def testSocketAuthInstallPlugin(self):
        # needs plugin. lets install it.
        cur = self.connections[0].cursor()
        try:
            cur.execute("install plugin auth_socket soname 'auth_socket.so'")
            TestAuthentication.socket_found = True
            self.socket_plugin_name = 'auth_socket'
            self.realtestSocketAuth()
        except pymysql.err.InternalError:
            try:
                cur.execute("install soname 'auth_socket'")
                TestAuthentication.socket_found = True
                self.socket_plugin_name = 'unix_socket'
                self.realtestSocketAuth()
            except pymysql.err.InternalError:
                TestAuthentication.socket_found = False
                raise unittest2.SkipTest('we couldn\'t install the socket plugin')
        finally:
            if TestAuthentication.socket_found:
                cur.execute("uninstall plugin %s" % self.socket_plugin_name)

def test_json(self):
        args = self.databases[0].copy()
        args["charset"] = "utf8mb4"
        conn = pymysql.connect(**args)
        if not self.mysql_server_is(conn, (5, 7, 0)):
            raise SkipTest("JSON type is not supported on MySQL <= 5.6")

        self.safe_create_table(conn, "test_json", """\
create table test_json (
    id int not null,
    json JSON not null,
    primary key (id)
);""")
        cur = conn.cursor()

        json_str = u'{"hello": "?????"}'
        cur.execute("INSERT INTO test_json (id, `json`) values (42, %s)", (json_str,))
        cur.execute("SELECT `json` from `test_json` WHERE `id`=42")
        res = cur.fetchone()[0]
        self.assertEqual(json.loads(res), json.loads(json_str))

        cur.execute("SELECT CAST(%s AS JSON) AS x", (json_str,))
        res = cur.fetchone()[0]
        self.assertEqual(json.loads(res), json.loads(json_str))

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 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 setUpClass(cls):
        ReusedPySparkTestCase.setUpClass()
        cls.tempdir = tempfile.NamedTemporaryFile(delete=False)
        try:
            cls.sc._jvm.org.apache.hadoop.hive.conf.HiveConf()
        except py4j.protocol.Py4JError:
            cls.tearDownClass()
            raise unittest.SkipTest("Hive is not available")
        except TypeError:
            cls.tearDownClass()
            raise unittest.SkipTest("Hive is not available")
        os.unlink(cls.tempdir.name)
        _scala_HiveContext =\
            cls.sc._jvm.org.apache.spark.sql.hive.test.TestHiveContext(cls.sc._jsc.sc())
        cls.sqlCtx = HiveContext(cls.sc, _scala_HiveContext)
        cls.testData = [Row(key=i, value=str(i)) for i in range(100)]
        cls.df = cls.sc.parallelize(cls.testData).toDF()

def quiet_run(self, result, func, *args, **kwargs):
        try:
            func(*args, **kwargs)
        except (KeyboardInterrupt, SystemExit):
            raise
        except unittest.SkipTest as e:
            if hasattr(result, 'addSkip'):
                result.addSkip(self, str(e))
            else:
                warnings.warn("TestResult has no addSkip method, skips not reported",
                              RuntimeWarning, 2)
                result.addSuccess(self)
            return False
        except:
            result.addError(self, self.__exc_info())
            return False
        return True

def testSocketAuthInstallPlugin(self):
        # needs plugin. lets install it.
        cur = self.connections[0].cursor()
        try:
            cur.execute("install plugin auth_socket soname 'auth_socket.so'")
            TestAuthentication.socket_found = True
            self.socket_plugin_name = 'auth_socket'
            self.realtestSocketAuth()
        except pymysql.err.InternalError:
            try:
                cur.execute("install soname 'auth_socket'")
                TestAuthentication.socket_found = True
                self.socket_plugin_name = 'unix_socket'
                self.realtestSocketAuth()
            except pymysql.err.InternalError:
                TestAuthentication.socket_found = False
                raise unittest2.SkipTest('we couldn\'t install the socket plugin')
        finally:
            if TestAuthentication.socket_found:
                cur.execute("uninstall plugin %s" % self.socket_plugin_name)

def setUp(self):
        self.dirname = os.path.join(tempfile.gettempdir(),
                                    'files_for_testing_neo',
                                    'klustakwik/test1')
        if not os.path.exists(self.dirname):
            raise unittest.SkipTest('data directory does not exist: ' +
                                    self.dirname)

def test1(self):
        """Tests that files can be loaded by basename"""
        kio = KlustaKwikIO(filename=os.path.join(self.dirname, 'basename'))
        if not BaseTestIO.use_network:
            raise unittest.SkipTest("Requires download of data from the web")
        fetfiles = kio._fp.read_filenames('fet')

        self.assertEqual(len(fetfiles), 2)
        self.assertEqual(os.path.abspath(fetfiles[0]),
                         os.path.abspath(os.path.join(self.dirname,
                                                      'basename.fet.0')))
        self.assertEqual(os.path.abspath(fetfiles[1]),
                         os.path.abspath(os.path.join(self.dirname,
                                                      'basename.fet.1')))

def test3(self):
        """Tests that files can be loaded by basename2"""
        kio = KlustaKwikIO(filename=os.path.join(self.dirname, 'basename2'))
        if not BaseTestIO.use_network:
            raise unittest.SkipTest("Requires download of data from the web")
        clufiles = kio._fp.read_filenames('clu')

        self.assertEqual(len(clufiles), 1)
        self.assertEqual(os.path.abspath(clufiles[1]),
                         os.path.abspath(os.path.join(self.dirname,
                                                      'basename2.clu.1')))

def setUp(self):
        self.dirname = os.path.join(tempfile.gettempdir(),
                                    'files_for_testing_neo',
                                    'klustakwik/test2')
        if not os.path.exists(self.dirname):
            raise unittest.SkipTest('data directory does not exist: ' +
                                    self.dirname)

def setUp(self):
        self.dirname = os.path.join(tempfile.gettempdir(),
                                    'files_for_testing_neo',
                                    'klustakwik/test3')
        if not os.path.exists(self.dirname):
            raise unittest.SkipTest('data directory does not exist: ' +
                                    self.dirname)

def setUp(self):
        super(CommonTests, self).setUp()
        data_dir = os.path.join(self.local_test_dir,
                                'Cheetah_v{}'.format(self.cheetah_version))

        self.sn = os.path.join(data_dir, 'original_data')
        self.pd = os.path.join(data_dir, 'plain_data')
        if not os.path.exists(self.sn):
            raise unittest.SkipTest('data file does not exist:' + self.sn)

def setUp(self):
        BaseTestIO.setUp(self)
        if sys.platform.startswith('win'):
            distantfile = 'http://download.multichannelsystems.com/download_data/software/neuroshare/nsMCDLibrary_3.7b.zip'
            localfile = os.path.join(tempfile.gettempdir(),'nsMCDLibrary_3.7b.zip')
            if not os.path.exists(localfile):
                urlretrieve(distantfile, localfile)
            if platform.architecture()[0].startswith('64'):
                self.dllname = os.path.join(tempfile.gettempdir(),'Matlab/Matlab-Import-Filter/Matlab_Interface/nsMCDLibrary64.dll')
                if not os.path.exists(self.dllname):
                    zip = zipfile.ZipFile(localfile)
                    zip.extract('Matlab/Matlab-Import-Filter/Matlab_Interface/nsMCDLibrary64.dll', path = tempfile.gettempdir())
            else:
                self.dllname = os.path.join(tempfile.gettempdir(),'Matlab/Matlab-Import-Filter/Matlab_Interface/nsMCDLibrary.dll')
                if not os.path.exists(self.dllname):
                    zip = zipfile.ZipFile(localfile)
                    zip.extract('Matlab/Matlab-Import-Filter/Matlab_Interface/nsMCDLibrary.dll', path = tempfile.gettempdir())

        elif sys.platform.startswith('linux'):
            if platform.architecture()[0].startswith('64'):
                distantfile = 'http://download.multichannelsystems.com/download_data/software/neuroshare/nsMCDLibrary_Linux64_3.7b.tar.gz'
                localfile = os.path.join(tempfile.gettempdir(),'nsMCDLibrary_Linux64_3.7b.tar.gz')
            else:
                distantfile = 'http://download.multichannelsystems.com/download_data/software/neuroshare/nsMCDLibrary_Linux32_3.7b.tar.gz'
                localfile = os.path.join(tempfile.gettempdir(),'nsMCDLibrary_Linux32_3.7b.tar.gz')
            if not os.path.exists(localfile):
                urlretrieve(distantfile, localfile)
            self.dllname = os.path.join(tempfile.gettempdir(),'nsMCDLibrary/nsMCDLibrary.so')
            if not os.path.exists(self.dllname):
                tar = tarfile.open(localfile)
                tar.extract('nsMCDLibrary/nsMCDLibrary.so', path = tempfile.gettempdir())
        else:
            raise unittest.SkipTest("Not currently supported on OS X")

def setUp(self):
        self.dirname = os.path.join(tempfile.gettempdir(),
                                    'files_for_testing_neo',
                                    'klustakwik/test1')
        if not os.path.exists(self.dirname):
            raise unittest.SkipTest('data directory does not exist: ' +
                                    self.dirname)