def operations_are_retried_on_refresh():
    changes_store = ChangesStore()
    changes_store.try_or_save(note_store.updateNote, developer_token, note)
    note_store.updateNote.side_effect = None
    changes_store.retry_failed_operations()

    assert_not_in(operation, changes_store.saved_operations)

def test_numpy_reset_array_undec():
    "Test '%reset array' functionality"
    _ip.ex('import numpy as np')
    _ip.ex('a = np.empty(2)')
    nt.assert_in('a', _ip.user_ns)
    _ip.magic('reset -f array')
    nt.assert_not_in('a', _ip.user_ns)

 def assert_no_completion(**kwargs):
     _, matches = complete(**kwargs)
     nt.assert_not_in('abc', matches)
     nt.assert_not_in('abc\'', matches)
     nt.assert_not_in('abc\']', matches)
     nt.assert_not_in('\'abc\'', matches)
     nt.assert_not_in('\'abc\']', matches)

    def test_project_data_tools(self):
        # Deny anonymous to see 'private-bugs' tool
        role = M.ProjectRole.by_name("*anonymous")._id
        read_permission = M.ACE.allow(role, "read")
        app = M.Project.query.get(shortname="test").app_instance("private-bugs")
        if read_permission in app.config.acl:
            app.config.acl.remove(read_permission)

        # admin sees both 'Tickets' tools
        r = self.app.get("/rest/p/test?doap")
        p = r.xml.find(self.ns + "Project")
        tools = p.findall(self.ns_sf + "feature")
        tools = [
            (
                t.find(self.ns_sf + "Feature").find(self.ns + "name").text,
                t.find(self.ns_sf + "Feature").find(self.foaf + "page").items()[0][1],
            )
            for t in tools
        ]
        assert_in(("Tickets", "http://localhost/p/test/bugs/"), tools)
        assert_in(("Tickets", "http://localhost/p/test/private-bugs/"), tools)

        # anonymous sees only non-private tool
        r = self.app.get("/rest/p/test?doap", extra_environ={"username": "*anonymous"})
        p = r.xml.find(self.ns + "Project")
        tools = p.findall(self.ns_sf + "feature")
        tools = [
            (
                t.find(self.ns_sf + "Feature").find(self.ns + "name").text,
                t.find(self.ns_sf + "Feature").find(self.foaf + "page").items()[0][1],
            )
            for t in tools
        ]
        assert_in(("Tickets", "http://localhost/p/test/bugs/"), tools)
        assert_not_in(("Tickets", "http://localhost/p/test/private-bugs/"), tools)

def test_assert_cwd_unchanged_not_masking_exceptions():
    # Test that we are not masking out other "more important" exceptions

    orig_cwd = os.getcwd()

    @assert_cwd_unchanged
    def do_chdir_value_error():
        os.chdir(os.pardir)
        raise ValueError("error exception")

    with swallow_logs() as cml:
        with assert_raises(ValueError) as cm:
            do_chdir_value_error()
        # retrospect exception
        if PY2:
            # could not figure out how to make it legit for PY3
            # but on manual try -- works, and exception traceback is not masked out
            exc_info = sys.exc_info()
            assert_in('raise ValueError("error exception")', traceback.format_exception(*exc_info)[-2])

        eq_(orig_cwd, os.getcwd(),
            "assert_cwd_unchanged didn't return us back to %s" % orig_cwd)
        assert_in("Mitigating and changing back", cml.out)

    # and again but allowing to chdir
    @assert_cwd_unchanged(ok_to_chdir=True)
    def do_chdir_value_error():
        os.chdir(os.pardir)
        raise ValueError("error exception")

    with swallow_logs() as cml:
        assert_raises(ValueError, do_chdir_value_error)
        eq_(orig_cwd, os.getcwd(),
            "assert_cwd_unchanged didn't return us back to %s" % orig_cwd)
        assert_not_in("Mitigating and changing back", cml.out)

    def test_profile_sections(self):
        project = Project.query.get(shortname='u/test-user')
        app = project.app_instance('profile')

        def ep(n):
            m = mock.Mock()
            m.name = n
            m.load()().display.return_value = 'Section %s' % n
            return m
        eps = map(ep, ['a', 'b', 'c', 'd'])
        order = {'user_profile_sections.order': 'b, d,c , f '}
        if hasattr(type(app), '_sections'):
            delattr(type(app), '_sections')
        with mock.patch('allura.lib.helpers.iter_entry_points') as iep:
            with mock.patch.dict(tg.config, order):
                iep.return_value = eps
                sections = app.profile_sections
                assert_equal(sections, [
                    eps[1].load(),
                    eps[3].load(),
                    eps[2].load(),
                    eps[0].load()])
        r = self.app.get('/u/test-user/profile')
        assert_in('Section a', r.body)
        assert_in('Section b', r.body)
        assert_in('Section c', r.body)
        assert_in('Section d', r.body)
        assert_not_in('Section f', r.body)

def test_subtests_with_slash():
    """ Version 1: Subtest names with /'s are handled correctly """

    expected = 'group2/groupA/test/subtest 1'
    nt.assert_not_in(
        expected, RESULT.tests.iterkeys(),
        msg='{0} found in result, when it should not be'.format(expected))

    def test_project_data_tools(self):
        # Deny anonymous to see 'private-bugs' tool
        role = M.ProjectRole.by_name('*anonymous')._id
        read_permission = M.ACE.allow(role, 'read')
        app = M.Project.query.get(
            shortname='test').app_instance('private-bugs')
        if read_permission in app.config.acl:
            app.config.acl.remove(read_permission)

        # admin sees both 'Tickets' tools
        r = self.app.get('/rest/p/test?doap')
        p = r.xml.find(self.ns + 'Project')
        tools = p.findall(self.ns_sf + 'feature')
        tools = [(t.find(self.ns_sf + 'Feature').find(self.ns + 'name').text,
                  t.find(self.ns_sf + 'Feature').find(self.foaf + 'page').items()[0][1])
                 for t in tools]
        assert_in(('Tickets', 'http://localhost/p/test/bugs/'), tools)
        assert_in(('Tickets', 'http://localhost/p/test/private-bugs/'), tools)

        # anonymous sees only non-private tool
        r = self.app.get('/rest/p/test?doap',
                         extra_environ={'username': '*anonymous'})
        p = r.xml.find(self.ns + 'Project')
        tools = p.findall(self.ns_sf + 'feature')
        tools = [(t.find(self.ns_sf + 'Feature').find(self.ns + 'name').text,
                  t.find(self.ns_sf + 'Feature').find(self.foaf + 'page').items()[0][1])
                 for t in tools]
        assert_in(('Tickets', 'http://localhost/p/test/bugs/'), tools)
        assert_not_in(('Tickets', 'http://localhost/p/test/private-bugs/'), tools)

def test_assert_cwd_unchanged_not_masking_exceptions():
    # Test that we are not masking out other "more important" exceptions

    orig_dir = os.getcwd()

    @assert_cwd_unchanged
    def do_chdir_value_error():
        os.chdir(os.pardir)
        raise ValueError("error exception")

    with swallow_logs() as cml:
        assert_raises(ValueError, do_chdir_value_error)
        eq_(orig_dir, os.getcwd(),
            "assert_cwd_unchanged didn't return us back to %s" % orig_dir)
        assert_in("Mitigating and changing back", cml.out)


    # and again but allowing to chdir
    @assert_cwd_unchanged(ok_to_chdir=True)
    def do_chdir_value_error():
        os.chdir(os.pardir)
        raise ValueError("error exception")

    with swallow_logs() as cml:
        assert_raises(ValueError, do_chdir_value_error)
        eq_(orig_dir, os.getcwd(),
            "assert_cwd_unchanged didn't return us back to %s" % orig_dir)
        assert_not_in("Mitigating and changing back", cml.out)

    def test_subtests_with_slash(self):
        """backends.json.update_results (0 -> 1): Subtest names with /'s are handled correctly"""

        expected = 'group2/groupA/test/subtest 1'
        nt.assert_not_in(
            expected, six.iterkeys(self.RESULT.tests),
            msg='{0} found in result, when it should not be'.format(expected))

def test_alias_lifecycle():
    name = 'test_alias1'
    cmd = 'echo "Hello"'
    am = _ip.alias_manager
    am.clear_aliases()
    am.define_alias(name, cmd)
    assert am.is_alias(name)
    nt.assert_equal(am.retrieve_alias(name), cmd)
    nt.assert_in((name, cmd), am.aliases)
    
    # Test running the alias
    orig_system = _ip.system
    result = []
    _ip.system = result.append
    try:
        _ip.run_cell('%{}'.format(name))
        result = [c.strip() for c in result]
        nt.assert_equal(result, [cmd])
    finally:
        _ip.system = orig_system
    
    # Test removing the alias
    am.undefine_alias(name)
    assert not am.is_alias(name)
    with nt.assert_raises(ValueError):
        am.retrieve_alias(name)
    nt.assert_not_in((name, cmd), am.aliases)

def rm_method(method):
    radu = RadulaProxy(connection=boto.connect_s3())
    radu.make_bucket(subject=TEST_BUCKET)

    # give something to rm
    args = vars(_parse_args(['up']))
    expected = []
    for i in xrange(3):
        remote_file = REMOTE_FILE + str(i)
        expected.append(remote_file)
        args.update({
            "subject": TEST_FILE,
            "target": remote_file
        })
        radu.upload(**args)

    while len(expected):
        remove_file = expected.pop()
        sys.stdout.truncate(0)
        getattr(radu, method)(subject=remove_file)

        radu.keys(subject=TEST_BUCKET)
        keys = [k.strip() for k in sys.stdout.getvalue().strip().split("\n")]

        absent_key = os.path.basename(remove_file)
        assert_not_in(absent_key, keys, msg="Expecting absence of key mention '{0}'".format(absent_key))
        for expected_key in expected:
            expected_key = os.path.basename(expected_key)
            assert_in(expected_key, keys, msg="Expecting output containing '{0}'".format(expected_key))

def test_add_handle_by_names(hurl, hpath, cpath, lcpath):

    class mocked_dirs:
        user_data_dir = lcpath

    with patch('datalad.cmdline.helpers.dirs', mocked_dirs), \
            swallow_logs() as cml:

        # get testrepos and make them known to datalad:
        handle = install_handle(hurl, hpath)
        collection = register_collection(cpath)
        assert_not_in(handle.name, collection)

        return_value = add_handle(handle.name, collection.name)

        # now handle is listed by collection:
        collection._reload()
        assert_in(handle.name, collection)

        # test collection repo:
        ok_clean_git(cpath, annex=False)
        ok_(isdir(opj(cpath, handle.name)))
        ok_(exists(opj(cpath, handle.name, REPO_CONFIG_FILE)))
        ok_(exists(opj(cpath, handle.name, REPO_STD_META_FILE)))

        # evaluate return value:
        assert_is_instance(return_value, Handle,
                           "install_handle() returns object of "
                           "incorrect class: %s" % type(return_value))
        eq_(return_value.name, handle.name)
        eq_(urlparse(return_value.url).path, urlparse(handle.url).path)

    def test_remove_groups(self):
        """
        Test the user is removed from all the groups in the list
        """

        calculon = {
            'email': '[email protected]',
        }
        fry = {
            'email': '[email protected]',
        }

        # Create the groups by adding users to groups
        self.mock_ps.add_groups(calculon, ['group1', 'group2'])
        self.mock_ps.add_groups(fry, ['group1'])

        # Remove user from a group he doesn't belong to
        self.mock_ps.remove_groups(fry, ['group1', 'group2'])

        users = self.mock_ps.get_group('group1')
        assert_not_in('[email protected]',
                      [user['email'] for user in users])

        users = self.mock_ps.get_group('group2')
        assert_not_in('[email protected]',
                      [user['email'] for user in users])

    def _test_post_create_service(self, if_error_expected):
        with self.app.test_client():
            self.login()

        with mock.patch("app.main.views.services.data_api_client") as data_api_client:

            data_api_client.create_new_draft_service.return_value = {
                "services": {
                    "id": 1,
                    "supplierId": 1234,
                    "supplierName": "supplierName",
                    "lot": "SCS",
                    "status": "not-submitted",
                    "frameworkName": "frameworkName",
                    "links": {},
                    "updatedAt": "2015-06-29T15:26:07.650368Z",
                }
            }

            res = self.client.post("/suppliers/submission/g-cloud-7/create")

            assert_equal(res.status_code, 302)

            error_message = "?error={}".format(self._format_for_request(self._answer_required))

            if if_error_expected:
                assert_in(error_message, res.location)
            else:
                assert_not_in(error_message, res.location)

def test_warn_match():
    if not hasattr(nt, 'assert_logs'):
        raise SkipTest("Test requires nose.tests.assert_logs")
    class A(LoggingConfigurable):
        foo = Integer(config=True)
        bar = Integer(config=True)
        baz = Integer(config=True)
    
    logger = logging.getLogger('test_warn_match')
    
    cfg = Config({'A': {'bat': 5}})
    with nt.assert_logs(logger, logging.WARNING) as captured:
        a = A(config=cfg, log=logger)
    
    output = '\n'.join(captured.output)
    nt.assert_in('Did you mean one of: `bar, baz`?', output)
    nt.assert_in('Config option `bat` not recognized by `A`.', output)

    cfg = Config({'A': {'fool': 5}})
    with nt.assert_logs(logger, logging.WARNING) as captured:
        a = A(config=cfg, log=logger)
    
    output = '\n'.join(captured.output)
    nt.assert_in('Config option `fool` not recognized by `A`.', output)
    nt.assert_in('Did you mean `foo`?', output)

    cfg = Config({'A': {'totally_wrong': 5}})
    with nt.assert_logs(logger, logging.WARNING) as captured:
        a = A(config=cfg, log=logger)

    output = '\n'.join(captured.output)
    nt.assert_in('Config option `totally_wrong` not recognized by `A`.', output)
    nt.assert_not_in('Did you mean', output)

    def test_inheritable_attribute(self):
        # days_early_for_beta isn't a basic attribute of Sequence
        assert_false(hasattr(SequenceDescriptor, 'days_early_for_beta'))

        # days_early_for_beta is added by InheritanceMixin
        assert_true(hasattr(InheritanceMixin, 'days_early_for_beta'))

        root = SequenceFactory.build(policy={'days_early_for_beta': '2'})
        ProblemFactory.build(parent=root)

        # InheritanceMixin will be used when processing the XML
        assert_in(InheritanceMixin, root.xblock_mixins)

        seq = self.process_xml(root)

        assert_equals(seq.unmixed_class, SequenceDescriptor)
        assert_not_equals(type(seq), SequenceDescriptor)

        # days_early_for_beta is added to the constructed sequence, because
        # it's in the InheritanceMixin
        assert_equals(2, seq.days_early_for_beta)

        # days_early_for_beta is a known attribute, so we shouldn't include it
        # in xml_attributes
        assert_not_in('days_early_for_beta', seq.xml_attributes)

    def click_through_all_pages_using_previous_button(self, response,
                                                      per_page):
        num_issues = len(self.issues)
        pages = number_of_pages(num_issues, per_page)
        page = pages

        while True:
            soup = BeautifulSoup(response.body)
            issues_html = soup.find(id='issue-list').text.strip()

            # x is the index of the issue expected at the top of this page
            x = (page - 1) * per_page

            for i in self.issues[x:x+per_page]:
                assert_in(i['title'], issues_html)

            for i in self.issues[:x]:
                assert_not_in(i['title'], issues_html)

            for i in self.issues[x+per_page:]:
                assert_not_in(i['title'], issues_html)

            page -= 1
            if page == 0:
                break

            response = response.click(linkid='pagination-previous-link')

    def test_pickle_dump_load(self):
        # Wipe current cache
        DescriptorMemoryElement.MEMORY_CACHE = {}

        # Make a couple descriptors
        v1 = numpy.array([1, 2, 3])
        d1 = DescriptorMemoryElement('test', 0)
        d1.set_vector(v1)

        v2 = numpy.array([4, 5, 6])
        d2 = DescriptorMemoryElement('test', 1)
        d2.set_vector(v2)

        ntools.assert_in(('test', 0), DescriptorMemoryElement.MEMORY_CACHE)
        ntools.assert_in(('test', 1), DescriptorMemoryElement.MEMORY_CACHE)

        d1_s = cPickle.dumps(d1)
        d2_s = cPickle.dumps(d2)

        # Wipe cache again
        DescriptorMemoryElement.MEMORY_CACHE = {}
        ntools.assert_not_in(('test', 0), DescriptorMemoryElement.MEMORY_CACHE)
        ntools.assert_not_in(('test', 1), DescriptorMemoryElement.MEMORY_CACHE)

        # Attempt reconstitution
        d1_r = cPickle.loads(d1_s)
        d2_r = cPickle.loads(d2_s)

        numpy.testing.assert_array_equal(v1, d1_r.vector())
        numpy.testing.assert_array_equal(v2, d2_r.vector())

        # Cache should now have those entries back in it
        ntools.assert_in(('test', 0), DescriptorMemoryElement.MEMORY_CACHE)
        ntools.assert_in(('test', 1), DescriptorMemoryElement.MEMORY_CACHE)

def test_macro_include_permissions():
    p_nbhd = M.Neighborhood.query.get(name='Projects')
    p_test = M.Project.query.get(shortname='test', neighborhood_id=p_nbhd._id)
    wiki = p_test.app_instance('wiki')
    wiki2 = p_test.app_instance('wiki2')
    with h.push_context(p_test._id, app_config_id=wiki.config._id):
        p = WM.Page.upsert(title='CanRead')
        p.text = 'Can see this!'
        p.commit()
        ThreadLocalORMSession.flush_all()

    with h.push_context(p_test._id, app_config_id=wiki2.config._id):
        role = M.ProjectRole.by_name('*anonymous')._id
        read_perm = M.ACE.allow(role, 'read')
        acl = c.app.config.acl
        if read_perm in acl:
            acl.remove(read_perm)
        p = WM.Page.upsert(title='CanNotRead')
        p.text = 'Can not see this!'
        p.commit()
        ThreadLocalORMSession.flush_all()

    with h.push_context(p_test._id, app_config_id=wiki.config._id):
        c.user = M.User.anonymous()
        md = '[[include ref=CanRead]]\n[[include ref=wiki2:CanNotRead]]'
        html = g.markdown_wiki.convert(md)
        assert_in('Can see this!', html)
        assert_not_in('Can not see this!', html)
        assert_in("[[include: you don't have a read permission for wiki2:CanNotRead]]", html)