autocomplete: allow to pass in a GET flag to control if non-active users should be returned in

autocomplete data.

rhodecode/controllers/home.py

@@ -1,284 +1,289 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 Home controller for RhodeCode Enterprise
 """
 
 import logging
 import time
 import re
 
 from pylons import tmpl_context as c, request, url, config
 from pylons.i18n.translation import _
 from sqlalchemy.sql import func
 
 from rhodecode.lib.auth import (
     LoginRequired, HasPermissionAllDecorator, AuthUser,
     HasRepoGroupPermissionAnyDecorator, XHRRequired)
 from rhodecode.lib.base import BaseController, render
 from rhodecode.lib.index import searcher_from_config
 from rhodecode.lib.ext_json import json
 from rhodecode.lib.utils import jsonify
-from rhodecode.lib.utils2 import safe_unicode
+from rhodecode.lib.utils2 import safe_unicode, str2bool
 from rhodecode.model.db import Repository, RepoGroup
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.scm import RepoList, RepoGroupList
 
 
 log = logging.getLogger(__name__)
 
 
 class HomeController(BaseController):
     def __before__(self):
         super(HomeController, self).__before__()
 
     def ping(self):
         """
         Ping, doesn't require login, good for checking out the platform
         """
         instance_id = getattr(c, 'rhodecode_instanceid', '')
         return 'pong[%s] => %s' % (instance_id, self.ip_addr,)
 
     @LoginRequired()
     @HasPermissionAllDecorator('hg.admin')
     def error_test(self):
         """
         Test exception handling and emails on errors
         """
         class TestException(Exception):
             pass
 
         msg = ('RhodeCode Enterprise %s test exception. Generation time: %s'
                % (c.rhodecode_name, time.time()))
         raise TestException(msg)
 
     def _get_groups_and_repos(self, repo_group_id=None):
         # repo groups groups
         repo_group_list = RepoGroup.get_all_repo_groups(group_id=repo_group_id)
         _perms = ['group.read', 'group.write', 'group.admin']
         repo_group_list_acl = RepoGroupList(repo_group_list, perm_set=_perms)
         repo_group_data = RepoGroupModel().get_repo_groups_as_dict(
             repo_group_list=repo_group_list_acl, admin=False)
 
         # repositories
         repo_list = Repository.get_all_repos(group_id=repo_group_id)
         _perms = ['repository.read', 'repository.write', 'repository.admin']
         repo_list_acl = RepoList(repo_list, perm_set=_perms)
         repo_data = RepoModel().get_repos_as_dict(
             repo_list=repo_list_acl, admin=False)
 
         return repo_data, repo_group_data
 
     @LoginRequired()
     def index(self):
         c.repo_group = None
 
         repo_data, repo_group_data = self._get_groups_and_repos()
         # json used to render the grids
         c.repos_data = json.dumps(repo_data)
         c.repo_groups_data = json.dumps(repo_group_data)
 
         return render('/index.html')
 
     @LoginRequired()
     @HasRepoGroupPermissionAnyDecorator('group.read', 'group.write',
                                         'group.admin')
     def index_repo_group(self, group_name):
         """GET /repo_group_name: Show a specific item"""
         c.repo_group = RepoGroupModel()._get_repo_group(group_name)
         repo_data, repo_group_data = self._get_groups_and_repos(
             c.repo_group.group_id)
 
         # json used to render the grids
         c.repos_data = json.dumps(repo_data)
         c.repo_groups_data = json.dumps(repo_group_data)
 
         return render('index_repo_group.html')
 
     def _get_repo_list(self, name_contains=None, repo_type=None, limit=20):
         query = Repository.query()\
             .order_by(func.length(Repository.repo_name))\
             .order_by(Repository.repo_name)
 
         if repo_type:
             query = query.filter(Repository.repo_type == repo_type)
 
         if name_contains:
             ilike_expression = u'%{}%'.format(safe_unicode(name_contains))
             query = query.filter(
                 Repository.repo_name.ilike(ilike_expression))
             query = query.limit(limit)
 
         all_repos = query.all()
         repo_iter = self.scm_model.get_repos(all_repos)
         return [
             {
                 'id': obj['name'],
                 'text': obj['name'],
                 'type': 'repo',
                 'obj': obj['dbrepo'],
                 'url': url('summary_home', repo_name=obj['name'])
             }
             for obj in repo_iter]
 
     def _get_repo_group_list(self, name_contains=None, limit=20):
         query = RepoGroup.query()\
             .order_by(func.length(RepoGroup.group_name))\
             .order_by(RepoGroup.group_name)
 
         if name_contains:
             ilike_expression = u'%{}%'.format(safe_unicode(name_contains))
             query = query.filter(
                 RepoGroup.group_name.ilike(ilike_expression))
             query = query.limit(limit)
 
         all_groups = query.all()
         repo_groups_iter = self.scm_model.get_repo_groups(all_groups)
         return [
             {
                 'id': obj.group_name,
                 'text': obj.group_name,
                 'type': 'group',
                 'obj': {},
                 'url': url('repo_group_home', group_name=obj.group_name)
             }
             for obj in repo_groups_iter]
 
     def _get_hash_commit_list(self, hash_starts_with=None, limit=20):
         if not hash_starts_with or len(hash_starts_with) < 3:
             return []
 
         commit_hashes = re.compile('([0-9a-f]{2,40})').findall(hash_starts_with)
 
         if len(commit_hashes) != 1:
             return []
 
         commit_hash_prefix = commit_hashes[0]
 
         auth_user = AuthUser(
             user_id=c.rhodecode_user.user_id, ip_addr=self.ip_addr)
         searcher = searcher_from_config(config)
         result = searcher.search(
             'commit_id:%s*' % commit_hash_prefix, 'commit', auth_user)
 
         return [
             {
                 'id': entry['commit_id'],
                 'text': entry['commit_id'],
                 'type': 'commit',
                 'obj': {'repo': entry['repository']},
                 'url': url('changeset_home',
                     repo_name=entry['repository'], revision=entry['commit_id'])
             }
             for entry in result['results']]
 
     @LoginRequired()
     @XHRRequired()
     @jsonify
     def goto_switcher_data(self):
         query = request.GET.get('query')
         log.debug('generating goto switcher list, query %s', query)
 
         res = []
         repo_groups = self._get_repo_group_list(query)
         if repo_groups:
             res.append({
                 'text': _('Groups'),
                 'children': repo_groups
             })
 
         repos = self._get_repo_list(query)
         if repos:
             res.append({
                 'text': _('Repositories'),
                 'children': repos
             })
 
         commits = self._get_hash_commit_list(query)
         if commits:
             unique_repos = {}
             for commit in commits:
                 unique_repos.setdefault(commit['obj']['repo'], []
                     ).append(commit)
 
             for repo in unique_repos:
                 res.append({
                     'text': _('Commits in %(repo)s') % {'repo': repo},
                     'children': unique_repos[repo]
                 })
 
         data = {
             'more': False,
             'results': res
         }
         return data
 
     @LoginRequired()
     @XHRRequired()
     @jsonify
     def repo_list_data(self):
         query = request.GET.get('query')
         repo_type = request.GET.get('repo_type')
         log.debug('generating repo list, query:%s', query)
 
         res = []
         repos = self._get_repo_list(query, repo_type=repo_type)
         if repos:
             res.append({
                 'text': _('Repositories'),
                 'children': repos
             })
 
         data = {
             'more': False,
             'results': res
         }
         return data
 
     @LoginRequired()
     @XHRRequired()
     @jsonify
     def user_autocomplete_data(self):
         query = request.GET.get('query')
+        active = str2bool(request.GET.get('active') or True)
 
         repo_model = RepoModel()
-        _users = repo_model.get_users(name_contains=query)
+        _users = repo_model.get_users(
+            name_contains=query, only_active=active)
 
         if request.GET.get('user_groups'):
             # extend with user groups
-            _user_groups = repo_model.get_user_groups(name_contains=query)
+            _user_groups = repo_model.get_user_groups(
+                name_contains=query, only_active=active)
             _users = _users + _user_groups
 
         return {'suggestions': _users}
 
     @LoginRequired()
     @XHRRequired()
     @jsonify
     def user_group_autocomplete_data(self):
         query = request.GET.get('query')
+        active = str2bool(request.GET.get('active') or True)
 
         repo_model = RepoModel()
-        _user_groups = repo_model.get_user_groups(name_contains=query)
+        _user_groups = repo_model.get_user_groups(
+            name_contains=query, only_active=active)
         _user_groups = _user_groups
 
         return {'suggestions': _user_groups}
 

rhodecode/tests/functional/test_home.py

@@ -1,365 +1,379 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import json
 
 from mock import patch
 import pytest
 
 import rhodecode
 from rhodecode.lib.utils import map_groups
 from rhodecode.model.db import Repository, User, RepoGroup
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.settings import SettingsModel
 from rhodecode.tests import TestController, url, TEST_USER_ADMIN_LOGIN
 from rhodecode.tests.fixture import Fixture
 
 
 fixture = Fixture()
 
 
 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('Add Repository')
 
         # search for objects inside the JavaScript JSON
         for repo in Repository.getAll():
             response.mustcontain('"name_raw": "%s"' % repo.repo_name)
 
     def test_index_contains_backend_specific_details(self, backend):
         self.log_user()
         response = self.app.get(url(controller='home', action='index'))
         tip = backend.repo.get_commit().raw_id
 
         # html in javascript variable:
         response.mustcontain(r'<i class=\"icon-%s\"' % (backend.alias, ))
         response.mustcontain(r'href=\"/%s\"' % (backend.repo_name, ))
 
         response.mustcontain("""/%s/changeset/%s""" % (backend.repo_name, tip))
         response.mustcontain("""Added a symlink""")
 
     def test_index_with_anonymous_access_disabled(self):
         with fixture.anon_access(False):
             response = self.app.get(url(controller='home', action='index'),
                                     status=302)
             assert 'login' in response.location
 
     def test_index_page_on_groups(self, autologin_user, repo_group):
         response = self.app.get(url('repo_group_home', group_name='gr1'))
         response.mustcontain("gr1/repo_in_group")
 
     def test_index_page_on_group_with_trailing_slash(
             self, autologin_user, repo_group):
         response = self.app.get(url('repo_group_home', group_name='gr1') + '/')
         response.mustcontain("gr1/repo_in_group")
 
     @pytest.fixture(scope='class')
     def repo_group(self, request):
         gr = fixture.create_repo_group('gr1')
         fixture.create_repo(name='gr1/repo_in_group', repo_group=gr)
 
         @request.addfinalizer
         def cleanup():
             RepoModel().delete('gr1/repo_in_group')
             RepoGroupModel().delete(repo_group='gr1', force_delete=True)
             Session().commit()
 
     def test_index_with_name_with_tags(self, autologin_user):
         user = User.get_by_username('test_admin')
         user.name = '<img src="/image1" onload="alert(\'Hello, World!\');">'
         user.lastname = (
             '<img src="/image2" onload="alert(\'Hello, World!\');">')
         Session().add(user)
         Session().commit()
 
         response = self.app.get(url(controller='home', action='index'))
         response.mustcontain(
             '&lt;img src=&#34;/image1&#34; onload=&#34;'
             'alert(&#39;Hello, World!&#39;);&#34;&gt;')
         response.mustcontain(
             '&lt;img src=&#34;/image2&#34; onload=&#34;'
             'alert(&#39;Hello, World!&#39;);&#34;&gt;')
 
     @pytest.mark.parametrize("name, state", [
         ('Disabled', False),
         ('Enabled', True),
     ])
     def test_index_show_version(self, autologin_user, name, state):
         version_string = 'RhodeCode Enterprise %s' % rhodecode.__version__
 
         show = SettingsModel().get_setting_by_name('show_version')
         show.app_settings_value = state
         Session().add(show)
         Session().commit()
         response = self.app.get(url(controller='home', action='index'))
         if state is True:
             response.mustcontain(version_string)
         if state is False:
             response.mustcontain(no=[version_string])
 
 
 class TestUserAutocompleteData(TestController):
     def test_returns_list_of_users(self, user_util):
         self.log_user()
         user = user_util.create_user(is_active=True)
         user_name = user.username
         response = self.app.get(
             url(controller='home', action='user_autocomplete_data'),
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=200)
         result = json.loads(response.body)
         values = [suggestion['value'] for suggestion in result['suggestions']]
         assert user_name in values
 
+    def test_returns_inactive_users_when_active_flag_sent(self, user_util):
+        self.log_user()
+        user = user_util.create_user(is_active=False)
+        user_name = user.username
+        response = self.app.get(
+            url(controller='home', action='user_autocomplete_data',
+                user_groups='true', active='0'),
+            headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=200)
+        result = json.loads(response.body)
+        values = [suggestion['value'] for suggestion in result['suggestions']]
+        assert user_name in values
+
     def test_returns_groups_when_user_groups_sent(self, user_util):
         self.log_user()
         group = user_util.create_user_group(user_groups_active=True)
         group_name = group.users_group_name
         response = self.app.get(
             url(controller='home', action='user_autocomplete_data',
                 user_groups='true'),
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=200)
         result = json.loads(response.body)
         values = [suggestion['value'] for suggestion in result['suggestions']]
         assert group_name in values
 
     def test_result_is_limited_when_query_is_sent(self):
         self.log_user()
         fake_result = [
             {
                 'first_name': 'John',
                 'value_display': 'hello{} (John Smith)'.format(i),
                 'icon_link': '/images/user14.png',
                 'value': 'hello{}'.format(i),
                 'last_name': 'Smith',
                 'username': 'hello{}'.format(i),
                 'id': i,
                 'value_type': u'user'
             }
             for i in range(10)
         ]
         users_patcher = patch.object(
             RepoModel, 'get_users', return_value=fake_result)
         groups_patcher = patch.object(
             RepoModel, 'get_user_groups', return_value=fake_result)
 
         query = 'hello'
         with users_patcher as users_mock, groups_patcher as groups_mock:
             response = self.app.get(
                 url(controller='home', action='user_autocomplete_data',
                     user_groups='true', query=query),
                 headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=200)
 
         result = json.loads(response.body)
-        users_mock.assert_called_once_with(name_contains=query)
-        groups_mock.assert_called_once_with(name_contains=query)
+        users_mock.assert_called_once_with(
+            name_contains=query, only_active=True)
+        groups_mock.assert_called_once_with(
+            name_contains=query, only_active=True)
         assert len(result['suggestions']) == 20
 
 
 def assert_and_get_content(result):
     repos = []
     groups = []
     commits = []
     for data in result:
         for data_item in data['children']:
             assert data_item['id']
             assert data_item['text']
             assert data_item['url']
             if data_item['type'] == 'repo':
                 repos.append(data_item)
             elif data_item['type'] == 'group':
                 groups.append(data_item)
             elif data_item['type'] == 'commit':
                 commits.append(data_item)
             else:
                 raise Exception('invalid type %s' % data_item['type'])
 
     return repos, groups, commits
 
 
 class TestGotoSwitcherData(TestController):
     required_repos_with_groups = [
         'abc',
         'abc-fork',
         'forks/abcd',
         'abcd',
         'abcde',
         'a/abc',
         'aa/abc',
         'aaa/abc',
         'aaaa/abc',
         'repos_abc/aaa/abc',
         'abc_repos/abc',
         'abc_repos/abcd',
         'xxx/xyz',
         'forked-abc/a/abc'
     ]
 
     @pytest.fixture(autouse=True, scope='class')
     def prepare(self, request, pylonsapp):
         for repo_and_group in self.required_repos_with_groups:
             # create structure of groups and return the last group
 
             repo_group = map_groups(repo_and_group)
 
             RepoModel()._create_repo(
                 repo_and_group, 'hg', 'test-ac', TEST_USER_ADMIN_LOGIN,
                 repo_group=getattr(repo_group, 'group_id', None))
 
             Session().commit()
 
         request.addfinalizer(self.cleanup)
 
     def cleanup(self):
         # first delete all repos
         for repo_and_groups in self.required_repos_with_groups:
             repo = Repository.get_by_repo_name(repo_and_groups)
             if repo:
                 RepoModel().delete(repo)
                 Session().commit()
 
         # then delete all empty groups
         for repo_and_groups in self.required_repos_with_groups:
             if '/' in repo_and_groups:
                 r_group = repo_and_groups.rsplit('/', 1)[0]
                 repo_group = RepoGroup.get_by_group_name(r_group)
                 if not repo_group:
                     continue
                 parents = repo_group.parents
                 RepoGroupModel().delete(repo_group, force_delete=True)
                 Session().commit()
 
                 for el in reversed(parents):
                     RepoGroupModel().delete(el, force_delete=True)
                     Session().commit()
 
     def test_returns_list_of_repos_and_groups(self):
         self.log_user()
 
         response = self.app.get(
             url(controller='home', action='goto_switcher_data'),
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=200)
         result = json.loads(response.body)['results']
 
         repos, groups, commits = assert_and_get_content(result)
 
         assert len(repos) == len(Repository.get_all())
         assert len(groups) == len(RepoGroup.get_all())
         assert len(commits) == 0
 
     def test_returns_list_of_repos_and_groups_filtered(self):
         self.log_user()
 
         response = self.app.get(
             url(controller='home', action='goto_switcher_data'),
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest', },
             params={'query': 'abc'}, status=200)
         result = json.loads(response.body)['results']
 
         repos, groups, commits = assert_and_get_content(result)
 
         assert len(repos) == 13
         assert len(groups) == 5
         assert len(commits) == 0
 
     def test_returns_list_of_properly_sorted_and_filtered(self):
         self.log_user()
 
         response = self.app.get(
             url(controller='home', action='goto_switcher_data'),
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest', },
             params={'query': 'abc'}, status=200)
         result = json.loads(response.body)['results']
 
         repos, groups, commits = assert_and_get_content(result)
 
         test_repos = [x['text'] for x in repos[:4]]
         assert ['abc', 'abcd', 'a/abc', 'abcde'] == test_repos
 
         test_groups = [x['text'] for x in groups[:4]]
         assert ['abc_repos', 'repos_abc',
                 'forked-abc', 'forked-abc/a'] == test_groups
 
 
 class TestRepoListData(TestController):
     def test_returns_list_of_repos_and_groups(self, user_util):
         self.log_user()
 
         response = self.app.get(
             url(controller='home', action='repo_list_data'),
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=200)
         result = json.loads(response.body)['results']
 
         repos, groups, commits = assert_and_get_content(result)
 
         assert len(repos) == len(Repository.get_all())
         assert len(groups) == 0
         assert len(commits) == 0
 
     def test_returns_list_of_repos_and_groups_filtered(self):
         self.log_user()
 
         response = self.app.get(
             url(controller='home', action='repo_list_data'),
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest', },
             params={'query': 'vcs_test_git'}, status=200)
         result = json.loads(response.body)['results']
 
         repos, groups, commits = assert_and_get_content(result)
 
         assert len(repos) == len(Repository.query().filter(
             Repository.repo_name.ilike('%vcs_test_git%')).all())
         assert len(groups) == 0
         assert len(commits) == 0
 
     def test_returns_list_of_repos_and_groups_filtered_with_type(self):
         self.log_user()
 
         response = self.app.get(
             url(controller='home', action='repo_list_data'),
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest', },
             params={'query': 'vcs_test_git', 'repo_type': 'git'}, status=200)
         result = json.loads(response.body)['results']
 
         repos, groups, commits = assert_and_get_content(result)
 
         assert len(repos) == len(Repository.query().filter(
             Repository.repo_name.ilike('%vcs_test_git%')).all())
         assert len(groups) == 0
         assert len(commits) == 0
 
     def test_returns_list_of_repos_non_ascii_query(self):
         self.log_user()
         response = self.app.get(
             url(controller='home', action='repo_list_data'),
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest', },
             params={'query': 'ć_vcs_test_ą', 'repo_type': 'git'}, status=200)
         result = json.loads(response.body)['results']
 
         repos, groups, commits = assert_and_get_content(result)
 
         assert len(repos) == 0
         assert len(groups) == 0
         assert len(commits) == 0