firstname=u'u1', lastname=u'u1')

        Session().commit()

        username = usr.username

        id_, params = _build_data(self.apikey, 'delete_user',

                                  userid=username,)

        response = api_call(self, params)

        ret = 'failed to delete ID:%s %s' % (usr.user_id,

                                             usr.username)

        expected = ret

        self._compare_error(id_, expected, given=response.body)

    @parameterized.expand([('firstname', 'new_username'),

                           ('lastname', 'new_username'),

                           ('email', 'new_username'),

                           ('admin', True),

                           ('admin', False),

                           ('ldap_dn', 'test'),

                           ('ldap_dn', None),

                           ('active', False),

                           ('active', True),

                           ('password', 'newpass')

                           ])

    def test_api_update_user(self, name, expected):

        usr = UserModel().get_by_username(self.TEST_USER_LOGIN)

        kw = {name: expected,

              'userid': usr.user_id}

        id_, params = _build_data(self.apikey, 'update_user', **kw)

        response = api_call(self, params)

        ret = {

        'msg': 'updated user ID:%s %s' % (usr.user_id, self.TEST_USER_LOGIN),

        'user': jsonify(UserModel()\

                            .get_by_username(self.TEST_USER_LOGIN)\

                            .get_api_data())

        }

        expected = ret

        self._compare_ok(id_, expected, given=response.body)

    def test_api_update_user_no_changed_params(self):

        usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)

        ret = jsonify(usr.get_api_data())

        id_, params = _build_data(self.apikey, 'update_user',

                                  userid=TEST_USER_ADMIN_LOGIN)

        response = api_call(self, params)

        ret = {

        'msg': 'updated user ID:%s %s' % (usr.user_id, TEST_USER_ADMIN_LOGIN),

        'user': ret

        }

        expected = ret

        self._compare_ok(id_, expected, given=response.body)

    def test_api_update_user_by_user_id(self):

        usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)

        ret = jsonify(usr.get_api_data())

        id_, params = _build_data(self.apikey, 'update_user',

                                  userid=usr.user_id)

        response = api_call(self, params)

        ret = {

        'msg': 'updated user ID:%s %s' % (usr.user_id, TEST_USER_ADMIN_LOGIN),

        'user': ret

        }

        expected = ret

        self._compare_ok(id_, expected, given=response.body)

    @mock.patch.object(UserModel, 'update_user', crash)

    def test_api_update_user_when_exception_happens(self):

        usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)

        ret = jsonify(usr.get_api_data())

        id_, params = _build_data(self.apikey, 'update_user',

                                  userid=usr.user_id)

        response = api_call(self, params)

        ret = 'failed to update user `%s`' % usr.user_id

        expected = ret

        self._compare_error(id_, expected, given=response.body)

    def test_api_get_repo(self):

        new_group = 'some_new_group'

        make_users_group(new_group)

        RepoModel().grant_users_group_permission(repo=self.REPO,

                                                 group_name=new_group,

                                                 perm='repository.read')

        Session().commit()

        id_, params = _build_data(self.apikey, 'get_repo',

                                  repoid=self.REPO)

        response = api_call(self, params)

        repo = RepoModel().get_by_repo_name(self.REPO)

        ret = repo.get_api_data()

        members = []

        for user in repo.repo_to_perm:

            perm = user.permission.permission_name

            user = user.user

            user_data = user.get_api_data()

            user_data['type'] = "user"

            user_data['permission'] = perm

            members.append(user_data)

        for users_group in repo.users_group_to_perm:

            perm = users_group.permission.permission_name

            users_group = users_group.users_group

            users_group_data = users_group.get_api_data()

            users_group_data['type'] = "users_group"

            users_group_data['permission'] = perm

            members.append(users_group_data)

        ret['members'] = members

        expected = ret

        self._compare_ok(id_, expected, given=response.body)

        destroy_users_group(new_group)

    def test_api_get_repo_by_non_admin(self):

        id_, params = _build_data(self.apikey, 'get_repo',

                                  repoid=self.REPO)

        response = api_call(self, params)

        repo = RepoModel().get_by_repo_name(self.REPO)

        ret = repo.get_api_data()

        members = []

        for user in repo.repo_to_perm:

            perm = user.permission.permission_name

            user = user.user

            user_data = user.get_api_data()

            user_data['type'] = "user"

            user_data['permission'] = perm

            members.append(user_data)

        for users_group in repo.users_group_to_perm:

            perm = users_group.permission.permission_name

            users_group = users_group.users_group

            users_group_data = users_group.get_api_data()

            users_group_data['type'] = "users_group"

            users_group_data['permission'] = perm

            members.append(users_group_data)

        ret['members'] = members

        expected = ret

        self._compare_ok(id_, expected, given=response.body)

    def test_api_get_repo_by_non_admin_no_permission_to_repo(self):

        RepoModel().grant_user_permission(repo=self.REPO,

                                          user=self.TEST_USER_LOGIN,

                                          perm='repository.none')

        id_, params = _build_data(self.apikey_regular, 'get_repo',

                                  repoid=self.REPO)

        response = api_call(self, params)

        expected = 'repository `%s` does not exist' % (self.REPO)

        self._compare_error(id_, expected, given=response.body)

    def test_api_get_repo_that_doesn_not_exist(self):

        id_, params = _build_data(self.apikey, 'get_repo',

                                  repoid='no-such-repo')

        response = api_call(self, params)

        ret = 'repository `%s` does not exist' % 'no-such-repo'

        expected = ret

        self._compare_error(id_, expected, given=response.body)

    def test_api_get_repos(self):

        id_, params = _build_data(self.apikey, 'get_repos')

        response = api_call(self, params)

        result = []

        for repo in RepoModel().get_all():

            result.append(repo.get_api_data())

        ret = jsonify(result)

        expected = ret

        self._compare_ok(id_, expected, given=response.body)

    def test_api_get_repos_non_admin(self):

        id_, params = _build_data(self.apikey_regular, 'get_repos')

        response = api_call(self, params)

        result = []

        for repo in RepoModel().get_all_user_repos(self.TEST_USER_LOGIN):

            result.append(repo.get_api_data())

        ret = jsonify(result)

        expected = ret

        self._compare_ok(id_, expected, given=response.body)

    @parameterized.expand([('all', 'all'),

                           ('dirs', 'dirs'),

                           ('files', 'files'), ])

    def test_api_get_repo_nodes(self, name, ret_type):

        rev = 'tip'

        path = '/'

        id_, params = _build_data(self.apikey, 'get_repo_nodes',

                                  repoid=self.REPO, revision=rev,

                                  root_path=path,

                                  ret_type=ret_type)

        response = api_call(self, params)

        # we don't the actual return types here since it's tested somewhere

        # else

        expected = json.loads(response.body)['result']

        self._compare_ok(id_, expected, given=response.body)

    def test_api_get_repo_nodes_bad_revisions(self):

        rev = 'i-dont-exist'

        path = '/'

        id_, params = _build_data(self.apikey, 'get_repo_nodes',

                                  repoid=self.REPO, revision=rev,

                                  root_path=path,)

        response = api_call(self, params)

        expected = 'failed to get repo: `%s` nodes' % self.REPO

        self._compare_error(id_, expected, given=response.body)

    def test_api_get_repo_nodes_bad_path(self):

        rev = 'tip'

        path = '/idontexits'

        id_, params = _build_data(self.apikey, 'get_repo_nodes',

                                  repoid=self.REPO, revision=rev,

                                  root_path=path,)

        response = api_call(self, params)

        expected = 'failed to get repo: `%s` nodes' % self.REPO

        self._compare_error(id_, expected, given=response.body)

    def test_api_get_repo_nodes_bad_ret_type(self):

        rev = 'tip'

        path = '/'

        ret_type = 'error'

        id_, params = _build_data(self.apikey, 'get_repo_nodes',

                                  repoid=self.REPO, revision=rev,

                                  root_path=path,

                                  ret_type=ret_type)

import time

from rhodecode.tests import *

from rhodecode.model.meta import Session

from rhodecode.model.db import User, RhodeCodeSetting, Repository

from rhodecode.lib.utils import set_rhodecode_config

from rhodecode.tests.models.common import _make_repo, _make_group

from rhodecode.model.repo import RepoModel

from rhodecode.model.repos_group import ReposGroupModel

class TestHomeController(TestController):

    def test_index(self):

        self.log_user()

        response = self.app.get(url(controller='home', action='index'))

        #if global permission is set

        response.mustcontain('href="/%s/summary"' % HG_REPO)

        response.mustcontain("""<img class="icon" title="Mercurial repository" """

                        """alt="Mercurial repository" src="/images/icons/hg"""

                        """icon.png"/>""")

        response.mustcontain("""<img class="icon" title="public repository" """

                        """alt="public repository" src="/images/icons/lock_"""

                        """open.png"/>""")

        response.mustcontain(

"""<a title="Marcin Kuzminski &amp;lt;marcin@python-works.com&amp;gt;:\n

merge" class="tooltip" href="/vcs_test_hg/changeset/27cd5cce30c96924232"""

"""dffcd24178a07ffeb5dfc">r173:27cd5cce30c9</a>"""

)

    def test_repo_summary_with_anonymous_access_disabled(self):

        anon = User.get_by_username('default')

        anon.active = False

        Session().add(anon)

        Session().commit()

        time.sleep(1.5)  # must sleep for cache (1s to expire)

        try:

            response = self.app.get(url(controller='summary',

                                        action='index', repo_name=HG_REPO),

                                        status=302)

            assert 'login' in response.location

        finally:

            anon = User.get_by_username('default')

            anon.active = True

            Session().add(anon)

            Session().commit()

    def test_index_with_anonymous_access_disabled(self):

        anon = User.get_by_username('default')

        anon.active = False

        Session().add(anon)

        Session().commit()

        time.sleep(1.5)  # must sleep for cache (1s to expire)

        try:

            response = self.app.get(url(controller='home', action='index'),

                                    status=302)

            assert 'login' in response.location

        finally:

            anon = User.get_by_username('default')

            anon.active = True

            Session().add(anon)

            Session().commit()

    def _set_l_dash(self, set_to):

        self.app.post(url('admin_setting', setting_id='visual'),

                      params=dict(_method='put',

                                  rhodecode_lightweight_dashboard=set_to,))

    def test_index_with_lightweight_dashboard(self):

        self.log_user()

        self._set_l_dash(True)

        try:

            response = self.app.get(url(controller='home', action='index'))

            response.mustcontain("""var data = {"totalRecords": %s""" % len(Repository.getAll()))

        finally:

            self._set_l_dash(False)

    def test_index_page_on_groups(self):

        self.log_user()

        _make_repo(name='gr1/repo_in_group', repos_group=_make_group('gr1'))

        Session().commit()

        response = self.app.get(url('repos_group_home', group_name='gr1'))

        try:

            response.mustcontain("""gr1/repo_in_group""")

        finally:

            RepoModel().delete('gr1/repo_in_group')

            ReposGroupModel().delete(repos_group='gr1', force_delete=True)

            Session().commit()

    def test_index_page_on_groups_with_lightweight_dashboard(self):

        self.log_user()

        self._set_l_dash(True)

        _make_repo(name='gr1/repo_in_group', repos_group=_make_group('gr1'))

        Session().commit()

        response = self.app.get(url('repos_group_home', group_name='gr1'))

        try:

            response.mustcontain("""gr1/repo_in_group""")

        finally:

            self._set_l_dash(False)

            RepoModel().delete('gr1/repo_in_group')

            ReposGroupModel().delete(repos_group='gr1', force_delete=True)

            Session().commit()