def test_simple_sleep(self):
     if IS_TRAVIS:
         skip()
     start = time.time()
     sleep(0.02)
     delay = time.time() - start
     assert 0.02 - 0.004 <= delay < 0.02 + 0.02, delay

def test_creation_fail2():
    """
    Try to create two Managers with the same position.
    This should fail without leaving any partial records in
    the database.

    """
    setupClass([DIManager, DIEmployee, DIPerson])

    kwargs = {'firstName': 'John', 'lastName': 'Doe',
              'position': 'Project Manager'}
    DIManager(**kwargs)
    persons = DIEmployee.select(DIPerson.q.firstName == 'John')
    assert persons.count() == 1

    kwargs = {'firstName': 'John', 'lastName': 'Doe II',
              'position': 'Project Manager'}
    raises(Exception, DIManager, **kwargs)
    persons = DIPerson.select(DIPerson.q.firstName == 'John')
    assert persons.count() == 1

    if not supports('transactions'):
        skip("Transactions aren't supported")
    transaction = DIPerson._connection.transaction()
    kwargs = {'firstName': 'John', 'lastName': 'Doe III',
              'position': 'Project Manager'}
    raises(Exception, DIManager, connection=transaction, **kwargs)
    transaction.rollback()
    transaction.begin()
    persons = DIPerson.select(DIPerson.q.firstName == 'John',
                              connection=transaction)
    assert persons.count() == 1

    def test_jit_link_module(self):
        if drv.Context.get_device().compute_capability() < (3, 5):
            from pytest import skip
            skip("need compute capability 3.5 or higher for dynamic parallelism")

        test_outer_cu = '''#include <cstdio>
        __global__ void test_kernel() {
            extern __global__ void test_kernel_inner();
            printf("Hello outer world!\\n");
            test_kernel_inner<<<2, 1>>>();
        }'''

        test_inner_cu = '''#include <cstdio>
        __global__ void test_kernel_inner() {
            printf("  Hello inner world!\\n");
        }'''

        from pycuda.compiler import DynamicModule
        mod = DynamicModule()
        mod.add_source(test_outer_cu, nvcc_options=['-rdc=true', '-lcudadevrt'])
        mod.add_source(test_inner_cu, nvcc_options=['-rdc=true', '-lcudadevrt'])
        mod.add_stdlib('cudadevrt')
        mod.link()

        test_kernel = mod.get_function('test_kernel')
        test_kernel(grid=(2,1), block=(1,1,1))

 def test_cjk(self):
     import sys
     if sys.maxunicode < 0x10ffff:
         skip("requires a 'wide' python build.")
     import unicodedata
     cases = ((0x3400, 0x4DB5),
              (0x4E00, 0x9FA5))
     if unicodedata.unidata_version >= "4.1":
         cases = ((0x3400, 0x4DB5),
                  (0x4E00, 0x9FBB),
                  (0x20000, 0x2A6D6))
     for first, last in cases:
         # Test at and inside the boundary
         for i in (first, first + 1, last - 1, last):
             charname = 'CJK UNIFIED IDEOGRAPH-%X'%i
             assert unicodedata.name(unichr(i)) == charname
             assert unicodedata.lookup(charname) == unichr(i)
         # Test outside the boundary
         for i in first - 1, last + 1:
             charname = 'CJK UNIFIED IDEOGRAPH-%X'%i
             try:
                 unicodedata.name(unichr(i))
             except ValueError:
                 pass
             raises(KeyError, unicodedata.lookup, charname)

def check_netconf(netconf):
    """
        Check netconf is valid
    """
    if type(netconf) != type ({}):
        test.skip("found %s while expecting %s" % (type(netconf), type({})))

    for keyword in ("name", "network", "service"):
        error = "You must put '%s' in your net configuration" % keyword
        if not netconf.has_key(keyword):
            test.skip(error)

    name_type = type(netconf["name"])
    if name_type != type(""):
        test.skip("examining name, we found %s while expecting %s" %
                  (name_type, type("")))

    network_instance = netconf["network"]
    if not isinstance(network_instance, IP):
        test.skip("examining network, we found %s while expecting %s" %
                  (netconf["network"], "an IP object"))

    service_type = type(netconf["service"])
    if service_type != type(set()):
        test.skip("examining service, we found %s while expecting %s" %
                  (service_type, type(set())))

def test_package_keep_rules():
    if not have_dpkg():
        pytest.skip("unsupported configuration")
    rc = ScopedFile('cmake/.mirbuildrc', RC_PKG)
    scon = ScopedFile('debian/control', CONTROL, BPY.path)
    sins = ScopedFile('debian/libtest-dev.install', INSTALL_etc, BPY.path)
    sgit = ScopedFile('.git/foo', 'nothing', BPY.path)
    debdir = posixpath.split(BPY.path)[0]
    tw = TreeWatcher(debdir)
    bpy = BPY(BPY_pkg, 'package', '--debian-pkg-keep-rules')
    assert bpy.exitcode == 0
    (ed, ef, md, mf) = tw.diff(relative = True)
    assert not (md or mf or ed)
    deb = 'libtest-dev_2.0.18-1-git.f76b429-1_' + BPY.buildarch() + '.deb'
    src = 'test_2.0.18-1-git.f76b429-1.tar.gz'
    assert ef == set(['project/build.py', 'project/debian/rules', deb, 'test_2.0.18-1-git.f76b429-1_' + BPY.buildarch() + '.changes',
                      'test_2.0.18-1-git.f76b429-1.dsc', 'test_2.0.18-1-git.f76b429-1.tar.gz'])
    (dirs, files) = BPY.debcontents(posixpath.join(debdir, deb), filter = lambda x: not re.match('./usr/share/doc', x))
    print dirs, files
    assert set(files) == set(['./usr/lib/libtest.a', './usr/etc/bla', './usr/include/test/test.h'])
    (dirs, files) = BPY.srccontents(posixpath.join(debdir, src), filter = lambda x: re.search('/\.git/', x))
    print dirs, files
    if tar_ignore_supported():
        assert set(files) == set() and set(dirs) == set()
    del bpy     # force cleanup
    tw.clean()

 def test_is_zombie_del_with_frame(self):
     try:
         import _stackless # are we on pypy with a stackless build?
     except ImportError:
         skip("only works on pypy-c-stackless")
     import gc
     res = []
     class MyCoroutine(coroutine):
         def __del__(self):
             res.append(self.is_zombie)
     main = coroutine.getcurrent()
     def f():
         print 'in coro'
         main.switch()
     co = MyCoroutine()
     co.bind(f)
     co.switch()
     del co
     for i in range(10):
         gc.collect()
         if res:
             break
     co = coroutine()
     co.bind(f)
     co.switch()
     assert res[0], "is_zombie was False in __del__"

    def test_category_loc_word(self):
        s = self.s
        import locale

        try:
            s.void_locale()
            opcodes1 = s.encode_literal("b") + [
                s.OPCODES["category"],
                s.CHCODES["category_loc_word"],
                s.OPCODES["success"],
            ]
            opcodes2 = s.encode_literal("b") + [
                s.OPCODES["category"],
                s.CHCODES["category_loc_not_word"],
                s.OPCODES["success"],
            ]
            s.assert_no_match(opcodes1, "b\xFC")
            s.assert_no_match(opcodes1, u"b\u00FC")
            s.assert_match(opcodes2, "b\xFC")
            locale.setlocale(locale.LC_ALL, "de_DE")
            s.assert_match(opcodes1, "b\xFC")
            s.assert_no_match(opcodes1, u"b\u00FC")
            s.assert_no_match(opcodes2, "b\xFC")
            s.void_locale()
        except locale.Error:
            # skip test
            skip("locale error")

    def setup_class(cls, with_ha=False):
        """
            This method is called once when the class is first instanciated. It
            sets up the testing conditions, while saving the current configuration.
        """
        Test.setup_class()

        cls.test_path = cls.results_path+cls.date+"/TestConfig/"
        if not isdir(cls.test_path):
            makedirs(cls.test_path)

        cls.with_ha = with_ha

        cls.local_ha_state = cls.client.call("ha", "getState")[0]
        if cls.local_ha_state != "CONNECTED" and with_ha:
            test.skip("!! Are you mad ? You're trying to do a HA test without being actually connected (local state is %s) !!" % cls.local_ha_state)

        if with_ha:
            cls.remote_client = cls.createClient(cust_host="5.0.0.2")
            # Check HA states locally and remotly

            cls.remote_ha_state = cls.remote_client.call("ha", "getState")[0]
            if cls.remote_ha_state != "CONNECTED":
                test.skip("!! Are you mad ? You're trying to do a HA test without being actually connected (remote state is %s) !!" % cls.remote_ha_state)

            cls.ssh_client = cls.createSSHClient(host="5.0.0.2")

 def test_scanner_zero_width_match(self):
     import re, sys
     if sys.version_info[:2] == (2, 3):
         skip("2.3 is different here")
     p = re.compile(".*").scanner("bla")
     assert ("bla", "") == (p.search().group(0), p.search().group(0))
     assert None == p.search()

 def test_group_takes_long(self):
     import re
     import sys
     if sys.version_info < (2, 7, 9):
         skip()
     assert re.match("(foo)", "foo").group(1) == "foo"
     exc = raises(IndexError, re.match("", "").group, sys.maxsize + 1)
     assert str(exc.value) == "no such group"

    def test_compress_function_huge_data(self):
        if not self.HUGE_OK:
            skip("skipping test requiring lots of memory")
        from bz2 import compress
    
        HUGE_DATA = self.TEXT * 10000

        data = compress(HUGE_DATA)
        assert self.decompress(data) == HUGE_DATA

def test_tuple_iter():
    skip("CDefinedInt implementation")
    def tuple_double(number):
        tuple = (number,)
        for item in tuple:
            number += item
        return number
    cl_double = make_cl_func(tuple_double, [int])
    assert cl_double(1) == 2

    def test_nonempty_supported_image_formats(self, device, ctx_getter):
        context = ctx_getter()

        if device.image_support:
            assert len(cl.get_supported_image_formats(
                    context, cl.mem_flags.READ_ONLY, cl.mem_object_type.IMAGE2D)) > 0
        else:
            from py.test import skip
            skip("images not supported on %s" % device.name)

 def test_large_flag(self):
     import sys
     if sys.platform == "darwin":
         skip("Mac OS doesn't have any large flag in fcntl.h")
     import fcntl, sys
     if sys.maxint == 2147483647:
         assert fcntl.DN_MULTISHOT == -2147483648
     else:
         assert fcntl.DN_MULTISHOT == 2147483648
     fcntl.fcntl(0, fcntl.F_NOTIFY, fcntl.DN_MULTISHOT)

def check_ifaces(ifaces):
    """
        Check ifaces is valid
    """
    if type(ifaces) != type(()):
        test.skip("found %s while expecting %s" % (type(ifaces), type({})))

    for eth in ifaces:
        if type(eth) != type(""):
            test.skip("found %s while expecting %s" % (type(eth), type({})))

 def test_compress_huge_data(self):
     if not self.HUGE_OK:
         skip("skipping test requiring lots of memory")
     from bz2 import BZ2Compressor            
     
     HUGE_DATA = self.TEXT * 10000
     bz2c = BZ2Compressor()
     raises(TypeError, bz2c.compress)
     data = bz2c.compress(HUGE_DATA)
     data = "%s%s" % (data, bz2c.flush())
     assert self.decompress(data) == HUGE_DATA

 def test_factorialrec(self):
     skip("does not make sense yet")        
     self.run_source('''
         def main(n):
             if n > 1:
                 return n * main(n-1)
             else:
                 return 1
     ''',
                ([5], 120),
                 ([20], 2432902008176640000L))

    def test_register_host_memory(self):
        if drv.get_version() < (4,):
            from py.test import skip
            skip("register_host_memory only exists on CUDA 4.0 and later")

        import sys
        if sys.platform == "darwin":
            from py.test import skip
            skip("register_host_memory is not supported on OS X")

        a = drv.aligned_empty((2**20,), np.float64, alignment=4096)
        drv.register_host_memory(a)

 def test_getlower_locale(self):
     import locale, sre_constants
     UPPER_AE = "\xc4"
     LOWER_AE = "\xe4"
     UPPER_PI = u"\u03a0"
     try:
         locale.setlocale(locale.LC_ALL, "de_DE")
         s.assert_lower_equal([("a", "a"), ("A", "a"), (UPPER_AE, LOWER_AE),
             (u"\u00c4", u"\u00e4"), (UPPER_PI, UPPER_PI)],
             sre_constants.SRE_FLAG_LOCALE)
     except locale.Error:
         # skip test
         skip("unsupported locale de_DE")