Python nose.tools 模块，assert_false() 实例源码

def test_upgrade_from_sha_with_unicode_password(self):
        user = factories.User()
        password = u'testpassword\xc2\xa0'
        user_obj = model.User.by_name(user['name'])

        # setup our user with an old password hash
        old_hash = self._set_password(password)
        user_obj._password = old_hash
        user_obj.save()

        nt.assert_true(user_obj.validate_password(password))
        nt.assert_not_equals(old_hash, user_obj.password)
        nt.assert_true(pbkdf2_sha512.identify(user_obj.password))
        nt.assert_true(pbkdf2_sha512.verify(password, user_obj.password))

        # check that we now allow unicode characters
        nt.assert_false(pbkdf2_sha512.verify('testpassword',
                                             user_obj.password))

def test_histmanager_disabled():
    """Ensure that disabling the history manager doesn't create a database."""
    cfg = Config()
    cfg.HistoryAccessor.enabled = False

    ip = get_ipython()
    with TemporaryDirectory() as tmpdir:
        hist_manager_ori = ip.history_manager
        hist_file = os.path.join(tmpdir, 'history.sqlite')
        cfg.HistoryManager.hist_file = hist_file
        try:
            ip.history_manager = HistoryManager(shell=ip, config=cfg)
            hist = [u'a=1', u'def f():\n    test = 1\n    return test', u"b='€Æ¾÷ß'"]
            for i, h in enumerate(hist, start=1):
                ip.history_manager.store_inputs(i, h)
            nt.assert_equal(ip.history_manager.input_hist_raw, [''] + hist)
            ip.history_manager.reset()
            ip.history_manager.end_session()
        finally:
            ip.history_manager = hist_manager_ori

    # hist_file should not be created
    nt.assert_false(os.path.exists(hist_file))

def test_command_chain_dispatcher_fofo():
    """Test a mixture of failing and succeeding hooks."""
    fail1 = Fail(u'fail1')
    fail2 = Fail(u'fail2')
    okay1 = Okay(u'okay1')
    okay2 = Okay(u'okay2')

    dp = CommandChainDispatcher([(0, fail1),
                                 # (5, okay1), # add this later
                                 (10, fail2),
                                 (15, okay2)])
    dp.add(okay1, 5)

    nt.assert_equal(dp(), u'okay1')

    nt.assert_true(fail1.called)
    nt.assert_true(okay1.called)
    nt.assert_false(fail2.called)
    nt.assert_false(okay2.called)

def test_last_two_blanks():
    nt.assert_false(isp.last_two_blanks(''))
    nt.assert_false(isp.last_two_blanks('abc'))
    nt.assert_false(isp.last_two_blanks('abc\n'))
    nt.assert_false(isp.last_two_blanks('abc\n\na'))
    nt.assert_false(isp.last_two_blanks('abc\n \n'))
    nt.assert_false(isp.last_two_blanks('abc\n\n'))

    nt.assert_true(isp.last_two_blanks('\n\n'))
    nt.assert_true(isp.last_two_blanks('\n\n '))
    nt.assert_true(isp.last_two_blanks('\n \n'))
    nt.assert_true(isp.last_two_blanks('abc\n\n '))
    nt.assert_true(isp.last_two_blanks('abc\n\n\n'))
    nt.assert_true(isp.last_two_blanks('abc\n\n \n'))
    nt.assert_true(isp.last_two_blanks('abc\n\n \n '))
    nt.assert_true(isp.last_two_blanks('abc\n\n \n \n'))
    nt.assert_true(isp.last_two_blanks('abc\nd\n\n\n'))
    nt.assert_true(isp.last_two_blanks('abc\nd\ne\nf\n\n\n'))

def test_deepreload():
    "Test that dreload does deep reloads and skips excluded modules."
    with TemporaryDirectory() as tmpdir:
        with prepended_to_syspath(tmpdir):
            with open(os.path.join(tmpdir, 'A.py'), 'w') as f:
                f.write("class Object(object):\n    pass\n")
            with open(os.path.join(tmpdir, 'B.py'), 'w') as f:
                f.write("import A\n")
            import A
            import B

            # Test that A is not reloaded.
            obj = A.Object()
            dreload(B, exclude=['A'])
            nt.assert_true(isinstance(obj, A.Object))

            # Test that A is reloaded.
            obj = A.Object()
            dreload(B)
            nt.assert_false(isinstance(obj, A.Object))

def test_greedy_completions():
    ip = get_ipython()
    ip.ex('a=list(range(5))')
    _,c = ip.complete('.',line='a[0].')
    nt.assert_false('.real' in c,
                    "Shouldn't have completed on a[0]: %s"%c)
    with greedy_completion():
        def _(line, cursor_pos, expect, message):
            _,c = ip.complete('.', line=line, cursor_pos=cursor_pos)
            nt.assert_in(expect, c, message%c)

        yield _, 'a[0].', 5, 'a[0].real', "Should have completed on a[0].: %s"
        yield _, 'a[0].r', 6, 'a[0].real', "Should have completed on a[0].r: %s"

        if sys.version_info > (3,4):
            yield _, 'a[0].from_', 10, 'a[0].from_bytes', "Should have completed on a[0].from_: %s"

def test_histmanager_disabled():
    """Ensure that disabling the history manager doesn't create a database."""
    cfg = Config()
    cfg.HistoryAccessor.enabled = False

    ip = get_ipython()
    with TemporaryDirectory() as tmpdir:
        hist_manager_ori = ip.history_manager
        hist_file = os.path.join(tmpdir, 'history.sqlite')
        cfg.HistoryManager.hist_file = hist_file
        try:
            ip.history_manager = HistoryManager(shell=ip, config=cfg)
            hist = [u'a=1', u'def f():\n    test = 1\n    return test', u"b='€Æ¾÷ß'"]
            for i, h in enumerate(hist, start=1):
                ip.history_manager.store_inputs(i, h)
            nt.assert_equal(ip.history_manager.input_hist_raw, [''] + hist)
            ip.history_manager.reset()
            ip.history_manager.end_session()
        finally:
            ip.history_manager = hist_manager_ori

    # hist_file should not be created
    nt.assert_false(os.path.exists(hist_file))

def test_command_chain_dispatcher_fofo():
    """Test a mixture of failing and succeeding hooks."""
    fail1 = Fail(u'fail1')
    fail2 = Fail(u'fail2')
    okay1 = Okay(u'okay1')
    okay2 = Okay(u'okay2')

    dp = CommandChainDispatcher([(0, fail1),
                                 # (5, okay1), # add this later
                                 (10, fail2),
                                 (15, okay2)])
    dp.add(okay1, 5)

    nt.assert_equal(dp(), u'okay1')

    nt.assert_true(fail1.called)
    nt.assert_true(okay1.called)
    nt.assert_false(fail2.called)
    nt.assert_false(okay2.called)

def test_last_two_blanks():
    nt.assert_false(isp.last_two_blanks(''))
    nt.assert_false(isp.last_two_blanks('abc'))
    nt.assert_false(isp.last_two_blanks('abc\n'))
    nt.assert_false(isp.last_two_blanks('abc\n\na'))
    nt.assert_false(isp.last_two_blanks('abc\n \n'))
    nt.assert_false(isp.last_two_blanks('abc\n\n'))

    nt.assert_true(isp.last_two_blanks('\n\n'))
    nt.assert_true(isp.last_two_blanks('\n\n '))
    nt.assert_true(isp.last_two_blanks('\n \n'))
    nt.assert_true(isp.last_two_blanks('abc\n\n '))
    nt.assert_true(isp.last_two_blanks('abc\n\n\n'))
    nt.assert_true(isp.last_two_blanks('abc\n\n \n'))
    nt.assert_true(isp.last_two_blanks('abc\n\n \n '))
    nt.assert_true(isp.last_two_blanks('abc\n\n \n \n'))
    nt.assert_true(isp.last_two_blanks('abc\nd\n\n\n'))
    nt.assert_true(isp.last_two_blanks('abc\nd\ne\nf\n\n\n'))

def test_deepreload():
    "Test that dreload does deep reloads and skips excluded modules."
    with TemporaryDirectory() as tmpdir:
        with prepended_to_syspath(tmpdir):
            with open(os.path.join(tmpdir, 'A.py'), 'w') as f:
                f.write("class Object(object):\n    pass\n")
            with open(os.path.join(tmpdir, 'B.py'), 'w') as f:
                f.write("import A\n")
            import A
            import B

            # Test that A is not reloaded.
            obj = A.Object()
            dreload(B, exclude=['A'])
            nt.assert_true(isinstance(obj, A.Object))

            # Test that A is reloaded.
            obj = A.Object()
            dreload(B)
            nt.assert_false(isinstance(obj, A.Object))

def test_get_phi_topic_from_memory():
    sampler2 = GibbsSampler(corpus, nTopics=3, nIter=5)
    sampler2._initialize()
    sampler2.run()

    f = sampler2.get_phi_topic_file_name(0)
    yield assert_false, os.path.isfile(f)

    phi = sampler2.get_phi_topic()
    yield assert_equal, phi.shape, (sampler2.nTopics, sampler2.VT)

    phi = sampler2.get_phi_topic(index=2)
    yield assert_equal, phi.shape, (sampler2.nTopics, sampler2.VT)

    phi = sampler2.get_phi_topic(start=0, end=5)
    yield assert_equal, phi.shape, (sampler2.nTopics, sampler2.VT)

def test_get_theta_from_memory():
    sampler2 = GibbsSampler(corpus, nTopics=3, nIter=5)
    sampler2._initialize()
    sampler2.run()

    f = sampler2.get_theta_file_name(0)
    yield assert_false, os.path.isfile(f)

    r = sampler2.get_theta()
    yield assert_equal, r.shape, (sampler2.DT, sampler2.nTopics)

    r = sampler2.get_theta(index=2)
    yield assert_equal, r.shape, (sampler2.DT, sampler2.nTopics)

    r = sampler2.get_theta(start=0, end=5)
    yield assert_equal, r.shape, (sampler2.DT, sampler2.nTopics)

def test_get_phi_opinion_from_memory():
    sampler2 = GibbsSampler(corpus, nTopics=3, nIter=5)
    sampler2._initialize()
    sampler2.run()

    f = sampler2.get_phi_opinion_file_name(0, 0)
    yield assert_false, os.path.isfile(f)

    phi = sampler2.get_phi_opinion()
    yield assert_equal, len(phi), sampler2.nPerspectives
    yield assert_equal, phi[0].shape, (sampler2.nTopics, sampler2.VO)

    phi = sampler2.get_phi_opinion(index=2)
    yield assert_equal, len(phi), sampler2.nPerspectives
    yield assert_equal, phi[0].shape, (sampler2.nTopics, sampler2.VO)

    phi = sampler2.get_phi_opinion(start=0, end=5)
    yield assert_equal, len(phi), sampler2.nPerspectives
    yield assert_equal, phi[0].shape, (sampler2.nTopics, sampler2.VO)

def remove_bonding(api):
    engine = api.system_service()

    def _remove_bonding(host):
        host_service = engine.hosts_service().host_service(id=host.id)
        network_utils_v4.detach_network_from_host(
            engine, host_service, MIGRATION_NETWORK, BOND_NAME)

    network_utils_v4.set_network_required_in_cluster(engine, MIGRATION_NETWORK,
                                                     CLUSTER_NAME, False)
    utils.invoke_in_parallel(
        _remove_bonding, test_utils.hosts_in_cluster_v4(engine, CLUSTER_NAME))

    for host in test_utils.hosts_in_cluster_v4(engine, CLUSTER_NAME):
        host_service = engine.hosts_service().host_service(id=host.id)
        nt.assert_false(_host_is_attached_to_network(engine, host_service,
                                                     MIGRATION_NETWORK))

def remove_bonding(api):
    engine = api.system_service()

    def _remove_bonding(host):
        host_service = engine.hosts_service().host_service(id=host.id)
        network_utils_v4.detach_network_from_host(
            engine, host_service, MIGRATION_NETWORK, BOND_NAME)

    network_utils_v4.set_network_required_in_cluster(engine, MIGRATION_NETWORK,
                                                     CLUSTER_NAME, False)
    utils.invoke_in_parallel(
        _remove_bonding, test_utils.hosts_in_cluster_v4(engine, CLUSTER_NAME))

    for host in test_utils.hosts_in_cluster_v4(engine, CLUSTER_NAME):
        host_service = engine.hosts_service().host_service(id=host.id)
        nt.assert_false(_host_is_attached_to_network(engine, host_service,
                                                     MIGRATION_NETWORK))

def test_greedy_completions():
    ip = get_ipython()
    ip.ex('a=list(range(5))')
    _,c = ip.complete('.',line='a[0].')
    nt.assert_false('.real' in c,
                    "Shouldn't have completed on a[0]: %s"%c)
    with greedy_completion():
        def _(line, cursor_pos, expect, message):
            _,c = ip.complete('.', line=line, cursor_pos=cursor_pos)
            nt.assert_in(expect, c, message%c)

        yield _, 'a[0].', 5, 'a[0].real', "Should have completed on a[0].: %s"
        yield _, 'a[0].r', 6, 'a[0].real', "Should have completed on a[0].r: %s"

        if sys.version_info > (3,4):
            yield _, 'a[0].from_', 10, 'a[0].from_bytes', "Should have completed on a[0].from_: %s"

def test_histmanager_disabled():
    """Ensure that disabling the history manager doesn't create a database."""
    cfg = Config()
    cfg.HistoryAccessor.enabled = False

    ip = get_ipython()
    with TemporaryDirectory() as tmpdir:
        hist_manager_ori = ip.history_manager
        hist_file = os.path.join(tmpdir, 'history.sqlite')
        cfg.HistoryManager.hist_file = hist_file
        try:
            ip.history_manager = HistoryManager(shell=ip, config=cfg)
            hist = [u'a=1', u'def f():\n    test = 1\n    return test', u"b='€Æ¾÷ß'"]
            for i, h in enumerate(hist, start=1):
                ip.history_manager.store_inputs(i, h)
            nt.assert_equal(ip.history_manager.input_hist_raw, [''] + hist)
            ip.history_manager.reset()
            ip.history_manager.end_session()
        finally:
            ip.history_manager = hist_manager_ori

    # hist_file should not be created
    nt.assert_false(os.path.exists(hist_file))

def test_last_two_blanks():
    nt.assert_false(isp.last_two_blanks(''))
    nt.assert_false(isp.last_two_blanks('abc'))
    nt.assert_false(isp.last_two_blanks('abc\n'))
    nt.assert_false(isp.last_two_blanks('abc\n\na'))
    nt.assert_false(isp.last_two_blanks('abc\n \n'))
    nt.assert_false(isp.last_two_blanks('abc\n\n'))

    nt.assert_true(isp.last_two_blanks('\n\n'))
    nt.assert_true(isp.last_two_blanks('\n\n '))
    nt.assert_true(isp.last_two_blanks('\n \n'))
    nt.assert_true(isp.last_two_blanks('abc\n\n '))
    nt.assert_true(isp.last_two_blanks('abc\n\n\n'))
    nt.assert_true(isp.last_two_blanks('abc\n\n \n'))
    nt.assert_true(isp.last_two_blanks('abc\n\n \n '))
    nt.assert_true(isp.last_two_blanks('abc\n\n \n \n'))
    nt.assert_true(isp.last_two_blanks('abc\nd\n\n\n'))
    nt.assert_true(isp.last_two_blanks('abc\nd\ne\nf\n\n\n'))

def test_deepreload():
    "Test that dreload does deep reloads and skips excluded modules."
    with TemporaryDirectory() as tmpdir:
        with prepended_to_syspath(tmpdir):
            with open(os.path.join(tmpdir, 'A.py'), 'w') as f:
                f.write("class Object(object):\n    pass\n")
            with open(os.path.join(tmpdir, 'B.py'), 'w') as f:
                f.write("import A\n")
            import A
            import B

            # Test that A is not reloaded.
            obj = A.Object()
            dreload(B, exclude=['A'])
            nt.assert_true(isinstance(obj, A.Object))

            # Test that A is reloaded.
            obj = A.Object()
            dreload(B)
            nt.assert_false(isinstance(obj, A.Object))

def test_histmanager_disabled():
    """Ensure that disabling the history manager doesn't create a database."""
    cfg = Config()
    cfg.HistoryAccessor.enabled = False

    ip = get_ipython()
    with TemporaryDirectory() as tmpdir:
        hist_manager_ori = ip.history_manager
        hist_file = os.path.join(tmpdir, 'history.sqlite')
        cfg.HistoryManager.hist_file = hist_file
        try:
            ip.history_manager = HistoryManager(shell=ip, config=cfg)
            hist = [u'a=1', u'def f():\n    test = 1\n    return test', u"b='€Æ¾÷ß'"]
            for i, h in enumerate(hist, start=1):
                ip.history_manager.store_inputs(i, h)
            nt.assert_equal(ip.history_manager.input_hist_raw, [''] + hist)
            ip.history_manager.reset()
            ip.history_manager.end_session()
        finally:
            ip.history_manager = hist_manager_ori

    # hist_file should not be created
    nt.assert_false(os.path.exists(hist_file))

def test_command_chain_dispatcher_fofo():
    """Test a mixture of failing and succeeding hooks."""
    fail1 = Fail(u'fail1')
    fail2 = Fail(u'fail2')
    okay1 = Okay(u'okay1')
    okay2 = Okay(u'okay2')

    dp = CommandChainDispatcher([(0, fail1),
                                 # (5, okay1), # add this later
                                 (10, fail2),
                                 (15, okay2)])
    dp.add(okay1, 5)

    nt.assert_equal(dp(), u'okay1')

    nt.assert_true(fail1.called)
    nt.assert_true(okay1.called)
    nt.assert_false(fail2.called)
    nt.assert_false(okay2.called)

def test_last_two_blanks():
    nt.assert_false(isp.last_two_blanks(''))
    nt.assert_false(isp.last_two_blanks('abc'))
    nt.assert_false(isp.last_two_blanks('abc\n'))
    nt.assert_false(isp.last_two_blanks('abc\n\na'))
    nt.assert_false(isp.last_two_blanks('abc\n \n'))
    nt.assert_false(isp.last_two_blanks('abc\n\n'))

    nt.assert_true(isp.last_two_blanks('\n\n'))
    nt.assert_true(isp.last_two_blanks('\n\n '))
    nt.assert_true(isp.last_two_blanks('\n \n'))
    nt.assert_true(isp.last_two_blanks('abc\n\n '))
    nt.assert_true(isp.last_two_blanks('abc\n\n\n'))
    nt.assert_true(isp.last_two_blanks('abc\n\n \n'))
    nt.assert_true(isp.last_two_blanks('abc\n\n \n '))
    nt.assert_true(isp.last_two_blanks('abc\n\n \n \n'))
    nt.assert_true(isp.last_two_blanks('abc\nd\n\n\n'))
    nt.assert_true(isp.last_two_blanks('abc\nd\ne\nf\n\n\n'))

def test_deepreload():
    "Test that dreload does deep reloads and skips excluded modules."
    with TemporaryDirectory() as tmpdir:
        with prepended_to_syspath(tmpdir):
            with open(os.path.join(tmpdir, 'A.py'), 'w') as f:
                f.write("class Object(object):\n    pass\n")
            with open(os.path.join(tmpdir, 'B.py'), 'w') as f:
                f.write("import A\n")
            import A
            import B

            # Test that A is not reloaded.
            obj = A.Object()
            dreload(B, exclude=['A'])
            nt.assert_true(isinstance(obj, A.Object))

            # Test that A is reloaded.
            obj = A.Object()
            dreload(B)
            nt.assert_false(isinstance(obj, A.Object))

def assert_is_not_none(obj, msg=None):
    """Same as assert_false(obj is None), with a nicer default message."""
    if not msg:
        msg = '{orig!r} is None'.format(orig=obj)
    assert_false(obj is None, msg=msg)

def test_powerview_create_private_defaults_to_false(self):
        '''
        PowerViews are public by default, if private option isn't specified.
        '''
        sysadmin = Sysadmin()

        create_dict = self._make_create_data_dict()
        del create_dict['private']

        powerview_result = toolkit.get_action('powerview_create')(
            context={'user': sysadmin['name']},
            data_dict=create_dict
        )

        nosetools.assert_false(powerview_result['private'])

def test_upgrade_from_sha_with_wrong_password_fails_to_upgrade(self):
        user = factories.User()
        password = u'testpassword'
        user_obj = model.User.by_name(user['name'])

        old_hash = self._set_password(password)
        user_obj._password = old_hash
        user_obj.save()

        nt.assert_false(user_obj.validate_password('wrongpass'))
        nt.assert_equals(old_hash, user_obj.password)
        nt.assert_false(pbkdf2_sha512.identify(user_obj.password))

def test_upgrade_from_pbkdf2_fails_with_wrong_password(self):
        user = factories.User()
        password = u'testpassword'
        user_obj = model.User.by_name(user['name'])

        # setup hash with salt/rounds less than the default

        old_hash = pbkdf2_sha512.encrypt(password, salt_size=2, rounds=10)
        user_obj._password = old_hash
        user_obj.save()

        nt.assert_false(user_obj.validate_password('wrong_pass'))
        # check that the hash has _not_ been updated
        nt.assert_equals(old_hash, user_obj.password)

def test_has_open_quotes3():
    for s in ["''", "''' '''", "'hi' 'ipython'"]:
        nt.assert_false(completer.has_open_quotes(s))

def test_has_open_quotes4():
    for s in ['""', '""" """', '"hi" "ipython"']:
        nt.assert_false(completer.has_open_quotes(s))

def test_greedy_completions():
    """
    Test the capability of the Greedy completer. 

    Most of the test here do not really show off the greedy completer, for proof
    each of the text bellow now pass with Jedi. The greedy completer is capable of more. 

    See the :any:`test_dict_key_completion_contexts`

    """
    ip = get_ipython()
    ip.ex('a=list(range(5))')
    _,c = ip.complete('.',line='a[0].')
    nt.assert_false('.real' in c,
                    "Shouldn't have completed on a[0]: %s"%c)
    with greedy_completion(), provisionalcompleter():
        def _(line, cursor_pos, expect, message, completion):
            _,c = ip.complete('.', line=line, cursor_pos=cursor_pos)
            with provisionalcompleter():
                completions = ip.Completer.completions(line, cursor_pos)
            nt.assert_in(expect, c, message%c)
            nt.assert_in(completion, completions)

        yield _, 'a[0].', 5, 'a[0].real', "Should have completed on a[0].: %s", Completion(5,5, 'real')
        yield _, 'a[0].r', 6, 'a[0].real', "Should have completed on a[0].r: %s", Completion(5,6, 'real')

        if sys.version_info > (3, 4):
            yield _, 'a[0].from_', 10, 'a[0].from_bytes', "Should have completed on a[0].from_: %s", Completion(5, 10, 'from_bytes')

def test_paste_py_multi_r(self):
        "Now, test that self.paste -r works"
        self.test_paste_py_multi()
        nt.assert_equal(ip.user_ns.pop('x'), [1,2,3])
        nt.assert_equal(ip.user_ns.pop('y'), [1,4,9])
        nt.assert_false('x' in ip.user_ns)
        ip.magic('paste -r')
        nt.assert_equal(ip.user_ns['x'], [1,2,3])
        nt.assert_equal(ip.user_ns['y'], [1,4,9])

def test_reset_in():
    "Test '%reset in' magic"
    # test '%reset -f in'
    _ip.run_cell("parrot", store_history=True)
    nt.assert_true('parrot' in [_ip.user_ns[x] for x in ('_i','_ii','_iii')])
    _ip.magic('%reset -f in')
    nt.assert_false('parrot' in [_ip.user_ns[x] for x in ('_i','_ii','_iii')])
    nt.assert_equal(len(set(_ip.user_ns['In'])), 1)

def test_last_blank():
    nt.assert_false(isp.last_blank(''))
    nt.assert_false(isp.last_blank('abc'))
    nt.assert_false(isp.last_blank('abc\n'))
    nt.assert_false(isp.last_blank('abc\na'))

    nt.assert_true(isp.last_blank('\n'))
    nt.assert_true(isp.last_blank('\n '))
    nt.assert_true(isp.last_blank('abc\n '))
    nt.assert_true(isp.last_blank('abc\n\n'))
    nt.assert_true(isp.last_blank('abc\nd\n\n'))
    nt.assert_true(isp.last_blank('abc\nd\ne\n\n'))
    nt.assert_true(isp.last_blank('abc \n \n \n\n'))

def test_cellmagic_help(self):
        self.sp.push('%%cellm?')
        nt.assert_false(self.sp.push_accepts_more())

def test_linux():
    nt.assert_false(sys.platform.startswith('linux'),"This test can't run under linux")

def test_shutdown():
    """Kernel exits after polite shutdown_request"""
    with new_kernel() as kc:
        km = kc.parent
        execute(u'a = 1', kc=kc)
        wait_for_idle(kc)
        kc.shutdown()
        for i in range(100): # 10s timeout
            if km.is_alive():
                time.sleep(.1)
            else:
                break
        nt.assert_false(km.is_alive())

def test_embed_kernel_namespace():
    """IPython.embed_kernel() inherits calling namespace"""
    cmd = '\n'.join([
        'from IPython import embed_kernel',
        'def go():',
        '    a=5',
        '    b="hi there"',
        '    embed_kernel()',
        'go()',
        '',
    ])

    with setup_kernel(cmd) as client:
        # oinfo a (int)
        msg_id = client.inspect('a')
        msg = client.get_shell_msg(block=True, timeout=TIMEOUT)
        content = msg['content']
        nt.assert_true(content['found'])
        text = content['data']['text/plain']
        nt.assert_in(u'5', text)

        # oinfo b (str)
        msg_id = client.inspect('b')
        msg = client.get_shell_msg(block=True, timeout=TIMEOUT)
        content = msg['content']
        nt.assert_true(content['found'])
        text = content['data']['text/plain']
        nt.assert_in(u'hi there', text)

        # oinfo c (undefined)
        msg_id = client.inspect('c')
        msg = client.get_shell_msg(block=True, timeout=TIMEOUT)
        content = msg['content']
        nt.assert_false(content['found'])

def test_oinfo_not_found():
    flush_channels()

    msg_id = KC.inspect('dne')
    reply = KC.get_shell_msg(timeout=TIMEOUT)
    validate_message(reply, 'inspect_reply', msg_id)
    content = reply['content']
    nt.assert_false(content['found'])

def test_has_open_quotes3():
    for s in ["''", "''' '''", "'hi' 'ipython'"]:
        nt.assert_false(completer.has_open_quotes(s))

def test_has_open_quotes4():
    for s in ['""', '""" """', '"hi" "ipython"']:
        nt.assert_false(completer.has_open_quotes(s))

def test_paste_py_multi_r(self):
        "Now, test that self.paste -r works"
        self.test_paste_py_multi()
        nt.assert_equal(ip.user_ns.pop('x'), [1,2,3])
        nt.assert_equal(ip.user_ns.pop('y'), [1,4,9])
        nt.assert_false('x' in ip.user_ns)
        ip.magic('paste -r')
        nt.assert_equal(ip.user_ns['x'], [1,2,3])
        nt.assert_equal(ip.user_ns['y'], [1,4,9])

def test_reset_in():
    "Test '%reset in' magic"
    # test '%reset -f in'
    _ip.run_cell("parrot", store_history=True)
    nt.assert_true('parrot' in [_ip.user_ns[x] for x in ('_i','_ii','_iii')])
    _ip.magic('%reset -f in')
    nt.assert_false('parrot' in [_ip.user_ns[x] for x in ('_i','_ii','_iii')])
    nt.assert_equal(len(set(_ip.user_ns['In'])), 1)

def test_last_blank():
    nt.assert_false(isp.last_blank(''))
    nt.assert_false(isp.last_blank('abc'))
    nt.assert_false(isp.last_blank('abc\n'))
    nt.assert_false(isp.last_blank('abc\na'))

    nt.assert_true(isp.last_blank('\n'))
    nt.assert_true(isp.last_blank('\n '))
    nt.assert_true(isp.last_blank('abc\n '))
    nt.assert_true(isp.last_blank('abc\n\n'))
    nt.assert_true(isp.last_blank('abc\nd\n\n'))
    nt.assert_true(isp.last_blank('abc\nd\ne\n\n'))
    nt.assert_true(isp.last_blank('abc \n \n \n\n'))

def test_cellmagic_help(self):
        self.sp.push('%%cellm?')
        nt.assert_false(self.sp.push_accepts_more())

def test_linux():
    nt.assert_false(sys.platform.startswith('linux'),"This test can't run under linux")

def test_add_parameter():
    pName = 'nTopics'

    yield assert_false, hasattr(config, pName)

    add_parameter(pName, nTopics, jsonFile)
    config2 = load_config(jsonFile)

    yield assert_equal, config2[pName], nTopics

def detach_vm_network_from_host_0(api):
    engine = api.system_service()

    host = test_utils.hosts_in_cluster_v4(engine, CLUSTER_NAME)[0]
    host_service = engine.hosts_service().host_service(id=host.id)

    network_utils_v4.set_network_required_in_cluster(
        engine, VM_NETWORK, CLUSTER_NAME, False)
    network_utils_v4.detach_network_from_host(engine, host_service, VM_NETWORK)

    nt.assert_false(_host_is_attached_to_network(engine, host_service,
                                                 VM_NETWORK))

def _validate_db_empty(token_id, engine_ip):
    networks = _get_networks(token_id, engine_ip)['networks']
    nt.assert_false(networks)
    ports = _get_ports(token_id, engine_ip)['ports']
    nt.assert_false(ports)
    subnets = _get_subnets(token_id, engine_ip)['subnets']
    nt.assert_false(subnets)

def detach_vm_network_from_host(api):
    host = test_utils.hosts_in_cluster_v3(api, CLUSTER_NAME)[0]

    network_utils_v3.set_network_required_in_cluster(
        api, VM_NETWORK, CLUSTER_NAME, False)
    network_utils_v3.detach_network_from_host(api, host, VM_NETWORK)

    nt.assert_false(_host_is_attached_to_network(api, host, VM_NETWORK))

def remove_bonding(api):
    def _remove_bonding(host):
        network_utils_v3.detach_network_from_host(api, host, MIGRATION_NETWORK,
                                                  BOND_NAME)

    network_utils_v3.set_network_required_in_cluster(api, MIGRATION_NETWORK,
                                                     CLUSTER_NAME, False)
    utils.invoke_in_parallel(_remove_bonding,
                             test_utils.hosts_in_cluster_v3(api, CLUSTER_NAME))

    for host in test_utils.hosts_in_cluster_v3(api, CLUSTER_NAME):
        nt.assert_false(_host_is_attached_to_network(api, host,
                                                     MIGRATION_NETWORK))