rmap.connect('public_journal_atom',

                 '%s/public_journal/atom' % ADMIN_PREFIX, controller='journal',

                 action="public_journal_atom")

    rmap.connect('public_journal_atom_old',

                 '%s/public_journal_atom' % ADMIN_PREFIX, controller='journal',

                 action="public_journal_atom")

    rmap.connect('toggle_following', '%s/toggle_following' % ADMIN_PREFIX,

                 controller='journal', action='toggle_following',

                 conditions=dict(method=["POST"]))

    # SEARCH

    rmap.connect('search', '%s/search' % ADMIN_PREFIX, controller='search',)

    rmap.connect('search_repo_admin', '%s/search/{repo_name:.*}' % ADMIN_PREFIX,

                 controller='search',

                 conditions=dict(function=check_repo))

    rmap.connect('search_repo', '/{repo_name:.*?}/search',

                 controller='search',

                 conditions=dict(function=check_repo),

                 )

    # LOGIN/LOGOUT/REGISTER/SIGN IN

    rmap.connect('login_home', '%s/login' % ADMIN_PREFIX, controller='login')

    rmap.connect('logout_home', '%s/logout' % ADMIN_PREFIX, controller='login',

                 action='logout')

    rmap.connect('register', '%s/register' % ADMIN_PREFIX, controller='login',

                 action='register')

    rmap.connect('reset_password', '%s/password_reset' % ADMIN_PREFIX,

                 controller='login', action='password_reset')

    rmap.connect('reset_password_confirmation',

                 '%s/password_reset_confirmation' % ADMIN_PREFIX,

                 controller='login', action='password_reset_confirmation')

    # FEEDS

    rmap.connect('rss_feed_home', '/{repo_name:.*?}/feed/rss',

                controller='feed', action='rss',

                conditions=dict(function=check_repo))

    rmap.connect('atom_feed_home', '/{repo_name:.*?}/feed/atom',

                controller='feed', action='atom',

                conditions=dict(function=check_repo))

    #==========================================================================

                prefix_error=False,

                encoding='UTF-8')

        if not UserModel().verify_reset_password_token(

            form_result['email'],

            form_result['timestamp'],

            form_result['token'],

        ):

            return htmlfill.render(

                render('/password_reset_confirmation.html'),

                defaults=form_result,

                errors={'token': _('Invalid password reset token')},

                prefix_error=False,

                encoding='UTF-8')

        UserModel().reset_password(form_result['email'], form_result['password'])

        h.flash(_('Successfully updated password'), category='success')

        raise HTTPFound(location=url('login_home'))

    def logout(self):

        session.delete()

        log.info('Logging out and deleting session for user')

        raise HTTPFound(location=url('home'))

        """Return the CSRF protection token for the session - just like it

        could have been screen scraped from a page with a form.

        Only intended for testing but might also be useful for other kinds

        of automation.

        """

        return h.session_csrf_secret_token()

        Generate wildcards using '*' character:

            "repository:vcs*" - search everything starting with 'vcs'

            "repository:*vcs*" - search for repository containing 'vcs'

        Optional AND / OR operators in queries

            "repository:vcs OR repository:test"

            "username:test AND repository:test*"

    '''))

def not_mapped_error(repo_name):

    flash(_('%s repository is not mapped to db perhaps'

            ' it was created or renamed from the filesystem'

            ' please run the application again'

            ' in order to rescan repositories') % repo_name, category='error')

def ip_range(ip_addr):

    from kallithea.model.db import UserIpMap

    s, e = UserIpMap._get_ip_range(ip_addr)

    return '%s - %s' % (s, e)

def session_csrf_secret_token():

    """Return (and create) the current session's CSRF protection token."""

    from tg import session

    if not session_csrf_secret_name in session:

        session[session_csrf_secret_name] = str(random.getrandbits(128))

        session.save()

    return session[session_csrf_secret_name]

def form(url, method="post", **attrs):

    """Like webhelpers.html.tags.form , but automatically adding

    session_csrf_secret_token for POST. The secret is thus never leaked in GET

    URLs.

    """

    form = insecure_form(url, method, **attrs)

    if method.lower() == 'get':

        return form

    return form + HTML.div(hidden(session_csrf_secret_name, session_csrf_secret_token()), style="display: none;")

                }

            })

        .fail(function(jqXHR, textStatus, errorThrown) {

                if (textStatus == "abort")

                    return;

                $target.html('<span class="bg-danger">ERROR: {0}</span>'.format(textStatus));

                $target.css('opacity','1.0');

            })

        ;

};

var ajaxGET = function(url, success, failure) {

    if(failure === undefined) {

        failure = function(jqXHR, textStatus, errorThrown) {

                if (textStatus != "abort")

                    alert("Ajax GET error: " + textStatus);

            };

    }

    return $.ajax({url: url, headers: {'X-PARTIAL-XHR': '1'}, cache: false})

        .done(success)

        .fail(failure);

};

var ajaxPOST = function(url, postData, success, failure) {

    var postData = _toQueryString(postData);

    if(failure === undefined) {

        failure = function(jqXHR, textStatus, errorThrown) {

                if (textStatus != "abort")

                    alert("Error posting to server: " + textStatus);

            };

    }

    return $.ajax({url: url, data: postData, type: 'POST', headers: {'X-PARTIAL-XHR': '1'}, cache: false})

        .done(success)

        .fail(failure);

};

/**

 * activate .show_more links

 * the .show_more must have an id that is the the id of an element to hide prefixed with _

 * the parentnode will be displayed

 */

var show_more_event = function(){

    $('.show_more').click(function(e){

        var el = e.currentTarget;

        $('#' + el.id.substring(1)).hide();

        $(el.parentNode).show();

    });

var _onSuccessFollow = function(target){

    var $target = $(target);

    var $f_cnt = $('#current_followers_count');

    if ($target.hasClass('follow')) {

        $target.removeClass('follow').addClass('following');

        $target.prop('title', _TM['Stop following this repository']);

        if ($f_cnt.html()) {

            var cnt = Number($f_cnt.html())+1;

            $f_cnt.html(cnt);

        }

    } else {

        $target.removeClass('following').addClass('follow');

        $target.prop('title', _TM['Start following this repository']);

        if ($f_cnt.html()) {

            var cnt = Number($f_cnt.html())-1;

            $f_cnt.html(cnt);

        }

    }

}

var toggleFollowingRepo = function(target, follows_repository_id){

    var args = 'follows_repository_id=' + follows_repository_id;

    $.post(TOGGLE_FOLLOW_URL, args, function(data){

            _onSuccessFollow(target);

        });

    return false;

};

var showRepoSize = function(target, repo_name){

    if(!$("#" + target).hasClass('loaded')){

        $("#" + target).html(_TM['Loading ...']);

        var url = pyroutes.url('repo_size', {"repo_name":repo_name});

        $.post(url, args, function(data) {

            $("#" + target).html(data);

            $("#" + target).addClass('loaded');

        });

    }

    return false;

};

/**

 * load tooltips dynamically based on data attributes, used for .lazy-cs changeset links

 */

var get_changeset_tooltip = function() {

    var $target = $(this);

    var tooltip = $target.data('tooltip');

    if (!tooltip) {

        var raw_id = $target.data('raw_id');

        var repo_name = $target.data('repo_name');

        var url = pyroutes.url('changeset_info', {"repo_name": repo_name, "revision": raw_id});

        $.ajax(url, {

                        // set mode on page load

                        var detected_mode = CodeMirror.findModeByExtension(${h.js(file.extension)});

                        if (detected_mode){

                            setCodeMirrorMode(myCodeMirror, detected_mode);

                            $mimetype_select.val(detected_mode.mime);

                        }

                    });

                </script>

            %endfor

            <div>

            ${h.submit('update',_('Update Gist'),class_="btn btn-success")}

            <a class="btn btn-default" href="${h.url('gist', gist_id=c.gist.gist_access_id)}">${_('Cancel')}</a>

            </div>

          ${h.end_form()}

          <script>

              $('#update').on('click', function(e){

                  e.preventDefault();

                  // check for newer version.

                  $.ajax({

                    url: ${h.js(h.url('edit_gist_check_revision', gist_id=c.gist.gist_access_id))},

                    dataType: 'json',

                    type: 'POST',

                    success: function(data) {

                      if(data.success == false){

                          $('#edit_error').show();

                      }

                      else{

                        $('#eform').submit();

                      }

                    }

                  });

              });

          </script>

        </div>

    </div>

</div>

</%def>

class NullHandler(logging.Handler):

    def emit(self, record):

        pass

class TestController(object):

    """Pytest-style test controller"""

    # Note: pytest base classes cannot have an __init__ method

    @pytest.fixture(autouse=True)

    def app_fixture(self):

        h = NullHandler()

        logging.getLogger("kallithea").addHandler(h)

        self.app = TestApp(testapp)

        return self.app

    def log_user(self, username=TEST_USER_ADMIN_LOGIN,

                 password=TEST_USER_ADMIN_PASS):

        self._logged_username = username

        response = self.app.post(url(controller='login', action='index'),

                                 {'username': username,

                                  'password': password,

        if 'Invalid username or password' in response.body:

            pytest.fail('could not login using %s %s' % (username, password))

        assert response.status == '302 Found'

        self.assert_authenticated_user(response, username)

        response = response.follow()

        return response.session['authuser']

    def _get_logged_user(self):

        return User.get_by_username(self._logged_username)

    def assert_authenticated_user(self, response, expected_username):

        cookie = response.session.get('authuser')

        user = cookie and cookie.get('user_id')

        user = user and User.get(user)

        user = user and user.username

        assert user == expected_username

    def checkSessionFlash(self, response, msg=None, skip=0, _matcher=lambda msg, m: msg in m):

        if 'flash' not in response.session:

            pytest.fail(safe_str(u'msg `%s` not found - session has no flash:\n%s' % (msg, response)))

        try:

            level, m = response.session['flash'][-1 - skip]

            if _matcher(msg, m):

                return

        except IndexError:

            pass

        pytest.fail(safe_str(u'msg `%s` not found in session flash (skipping %s): %s' %

                           (msg, skip,

                            ', '.join('`%s`' % m for level, m in response.session['flash']))))

    def checkSessionFlashRegex(self, response, regex, skip=0):

        self.checkSessionFlash(response, regex, skip=skip, _matcher=re.search)

from kallithea.tests.base import *

from kallithea.model.db import Setting

class TestAuthSettingsController(TestController):

    def _enable_plugins(self, plugins_list):

        test_url = url(controller='admin/auth_settings',

                       action='auth_settings')

        for plugin in plugins_list.split(','):

            enable = plugin.partition('kallithea.lib.auth_modules.')[-1]

            params.update({'%s_enabled' % enable: True})

        response = self.app.post(url=test_url, params=params)

        return params

        #self.checkSessionFlash(response, 'Auth settings updated successfully')

    def test_index(self):

        self.log_user()

        response = self.app.get(url(controller='admin/auth_settings',

                                    action='index'))

        response.mustcontain('Authentication Plugins')

    @skipif(not ldap_lib_installed, reason='skipping due to missing ldap lib')

    def test_ldap_save_settings(self):

        self.log_user()

        params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_ldap')

        params.update({'auth_ldap_host': u'dc.example.com',

                       'auth_ldap_port': '999',

                       'auth_ldap_tls_kind': 'PLAIN',

                       'auth_ldap_tls_reqcert': 'NEVER',

                       'auth_ldap_cacertdir': '',

from kallithea.tests.base import *

from kallithea.model.db import Setting

class TestDefaultsController(TestController):

    def test_index(self):

        self.log_user()

        response = self.app.get(url('defaults'))

        response.mustcontain('default_repo_private')

        response.mustcontain('default_repo_enable_statistics')

        response.mustcontain('default_repo_enable_downloads')

    def test_update_params_true_hg(self):

        self.log_user()

        params = {

            'default_repo_enable_downloads': True,

            'default_repo_enable_statistics': True,

            'default_repo_private': True,

            'default_repo_type': 'hg',

        }

        response = self.app.post(url('defaults_update', id='default'), params=params)

        self.checkSessionFlash(response, 'Default settings updated successfully')

        defs = Setting.get_default_repo_settings()

        assert params == defs

    def test_update_params_false_git(self):

        self.log_user()

        params = {

            'default_repo_enable_downloads': False,

            'default_repo_enable_statistics': False,

            'default_repo_private': False,

            'default_repo_type': 'git',

        }

        response = self.app.post(url('defaults_update', id='default'), params=params)

        self.checkSessionFlash(response, 'Default settings updated successfully')

        defs = Setting.get_default_repo_settings()

        assert params == defs

        g2 = _create_gist('gist2', lifetime=1400).gist_access_id

        g3 = _create_gist('gist3', description=u'gist3-desc').gist_access_id

        g4 = _create_gist('gist4', gist_type='private').gist_access_id

        response = self.app.get(url('gists'))

        # Test response...

        response.mustcontain('gist: %s' % g1)

        response.mustcontain('gist: %s' % g2)

        response.mustcontain('Expires: in 23 hours')  # we don't care about the end

        response.mustcontain('gist: %s' % g3)

        response.mustcontain('gist3-desc')

        response.mustcontain(no=['gist: %s' % g4])

    def test_index_private_gists(self):

        self.log_user()

        gist = _create_gist('gist5', gist_type='private')

        response = self.app.get(url('gists', private=1))

        # Test response...

        # and privates

        response.mustcontain('gist: %s' % gist.gist_access_id)

    def test_create_missing_description(self):

        self.log_user()

        response = self.app.post(url('gists'),

                                 status=200)

        response.mustcontain('Missing value')

    def test_create(self):

        self.log_user()

        response = self.app.post(url('gists'),

                                 params={'lifetime': -1,

                                         'content': 'gist test',

                                         'filename': 'foo',

                                         'public': 'public',

                                 status=302)

        response = response.follow()

        response.mustcontain('added file: foo')

        response.mustcontain('gist test')

        response.mustcontain('<div class="label label-success">Public Gist</div>')

    def test_create_with_path_with_dirs(self):

        self.log_user()

        response = self.app.post(url('gists'),

                                 params={'lifetime': -1,

                                         'content': 'gist test',

                                         'filename': '/home/foo',

                                         'public': 'public',

                                 status=200)

        response.mustcontain('Filename cannot be inside a directory')

    def test_access_expired_gist(self):

        self.log_user()

        gist = _create_gist('never-see-me')

        gist.gist_expires = 0  # 1970

        Session().commit()

        response = self.app.get(url('gist', gist_id=gist.gist_access_id), status=404)

    def test_create_private(self):

        self.log_user()

        response = self.app.post(url('gists'),

                                 params={'lifetime': -1,

                                         'content': 'private gist test',

                                         'filename': 'private-foo',

                                         'private': 'private',

                                 status=302)

        response = response.follow()

        response.mustcontain('added file: private-foo<')

        response.mustcontain('private gist test')

        response.mustcontain('<div class="label label-warning">Private Gist</div>')

    def test_create_with_description(self):

        self.log_user()

        response = self.app.post(url('gists'),

                                 params={'lifetime': -1,

                                         'content': 'gist test',

                                         'filename': 'foo-desc',

                                         'description': 'gist-desc',

                                         'public': 'public',

                                 status=302)

        response = response.follow()

        response.mustcontain('added file: foo-desc')

        response.mustcontain('gist test')

        response.mustcontain('gist-desc')

        response.mustcontain('<div class="label label-success">Public Gist</div>')

    def test_new(self):

        self.log_user()

        response = self.app.get(url('new_gist'))

    def test_delete(self):

        self.log_user()

        gist = _create_gist('delete-me')

        response = self.app.post(url('gist_delete', gist_id=gist.gist_id),

    def test_delete_normal_user_his_gist(self):

        self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)

        gist = _create_gist('delete-me', owner=TEST_USER_REGULAR_LOGIN)

        response = self.app.post(url('gist_delete', gist_id=gist.gist_id),

    def test_delete_normal_user_not_his_own_gist(self):

        self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)

        gist = _create_gist('delete-me')

        response = self.app.post(url('gist_delete', gist_id=gist.gist_id), status=403,

    def test_show(self):

        gist = _create_gist('gist-show-me')

        response = self.app.get(url('gist', gist_id=gist.gist_access_id))

        response.mustcontain('added file: gist-show-me<')

        response.mustcontain('%s - created' % TEST_USER_ADMIN_LOGIN)

        response.mustcontain('gist-desc')

        response.mustcontain('<div class="label label-success">Public Gist</div>')

    def test_show_as_raw(self):

        gist = _create_gist('gist-show-me', content='GIST CONTENT')

        response = self.app.get(url('formatted_gist',

                                    gist_id=gist.gist_access_id, format='raw'))

        assert response.body == 'GIST CONTENT'

    def test_show_as_raw_individual_file(self):

        gist = _create_gist('gist-show-me-raw', content='GIST BODY')

        response = self.app.get(url('formatted_gist_file',

                                    gist_id=gist.gist_access_id, format='raw',

                                    revision='tip', f_path='gist-show-me-raw'))

        assert response.body == 'GIST BODY'

    def test_edit(self):

        response = self.app.get(url('edit_gist', gist_id=1))

from tg.util.webtest import test_context

class TestAdminPermissionsController(TestController):

    def test_index(self):

        self.log_user()

        response = self.app.get(url('admin_permissions'))

        # Test response...

    def test_index_ips(self):

        self.log_user()

        response = self.app.get(url('admin_permissions_ips'))

        # Test response...

        response.mustcontain('All IP addresses are allowed')

    def test_add_delete_ips(self, auto_clear_ip_permissions):

        self.log_user()

        default_user_id = User.get_default_user().user_id

        # Add IP and verify it is shown in UI and both gives access and rejects

        response = self.app.post(url('edit_user_ips_update', id=default_user_id),

                                 params=dict(new_ip='0.0.0.0/24',

        invalidate_all_caches()

        response = self.app.get(url('admin_permissions_ips'),

                                extra_environ={'REMOTE_ADDR': '0.0.0.1'})

        response.mustcontain('0.0.0.0/24')

        response.mustcontain('0.0.0.0 - 0.0.0.255')

        response = self.app.get(url('admin_permissions_ips'),

                                extra_environ={'REMOTE_ADDR': '0.0.1.1'}, status=403)

        # Add another IP and verify previously rejected now works

        response = self.app.post(url('edit_user_ips_update', id=default_user_id),

                                 params=dict(new_ip='0.0.1.0/24',

        invalidate_all_caches()

        response = self.app.get(url('admin_permissions_ips'),

                                extra_environ={'REMOTE_ADDR': '0.0.1.1'})

        # Delete latest IP and verify same IP is rejected again

        x = UserIpMap.query().filter_by(ip_addr='0.0.1.0/24').first()

        response = self.app.post(url('edit_user_ips_delete', id=default_user_id),

                                 params=dict(del_ip_id=x.ip_id,

        invalidate_all_caches()

        response = self.app.get(url('admin_permissions_ips'),

                                extra_environ={'REMOTE_ADDR': '0.0.1.1'}, status=403)

        # Delete first IP and verify unlimited access again

        x = UserIpMap.query().filter_by(ip_addr='0.0.0.0/24').first()

        response = self.app.post(url('edit_user_ips_delete', id=default_user_id),

                                 params=dict(del_ip_id=x.ip_id,

        invalidate_all_caches()

        response = self.app.get(url('admin_permissions_ips'),

                                extra_environ={'REMOTE_ADDR': '0.0.1.1'})

    def test_index_overview(self):

        self.log_user()

        response = self.app.get(url('admin_permissions_perms'))

        # Test response...

    def test_edit_permissions_permissions(self):

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        # Test unauthenticated access - it will redirect to login page

        response = self.app.post(

            url('edit_repo_perms_update', repo_name=HG_REPO),

            params=dict(

                perm_new_member_1='repository.read',

                perm_new_member_name_1=user.username,

                perm_new_member_type_1='user',

            status=302)

        assert not response.location.endswith(url('edit_repo_perms_update', repo_name=HG_REPO))

        assert response.location.endswith(url('login_home', came_from=url('edit_repo_perms_update', repo_name=HG_REPO)))

        response = self.app.post(

            url('edit_repo_perms_revoke', repo_name=HG_REPO),

            params=dict(

                obj_type='user',

                user_id=user.user_id,

            status=302)

        assert response.location.endswith(url('login_home', came_from=url('edit_repo_perms_revoke', repo_name=HG_REPO)))

        # Test authenticated access

        self.log_user()

        response = self.app.post(

            url('edit_repo_perms_update', repo_name=HG_REPO),

            params=dict(

                perm_new_member_1='repository.read',

                perm_new_member_name_1=user.username,

                perm_new_member_type_1='user',

            status=302)

        assert response.location.endswith(url('edit_repo_perms_update', repo_name=HG_REPO))

        response = self.app.post(

            url('edit_repo_perms_revoke', repo_name=HG_REPO),

            params=dict(

                obj_type='user',

                user_id=user.user_id,

            status=200)

        assert not response.body

from kallithea.model.db import Repository

from kallithea.model.meta import Session

from kallithea.model.repo_group import RepoGroupModel

from kallithea.tests.base import TestController, url

from kallithea.tests.fixture import Fixture

fixture = Fixture()

class TestRepoGroupsController(TestController):

    def test_case_insensitivity(self):

        self.log_user()

        group_name = u'newgroup'

        response = self.app.post(url('repos_groups'),

                                 fixture._get_repo_group_create_params(group_name=group_name,

        # try to create repo group with swapped case

        swapped_group_name = group_name.swapcase()

        response = self.app.post(url('repos_groups'),

                                 fixture._get_repo_group_create_params(group_name=swapped_group_name,

        response.mustcontain('already exists')

        RepoGroupModel().delete(group_name)

        Session().commit()

class _BaseTestCase(TestController):

    """

    Write all tests here

    """

    REPO = None

    REPO_TYPE = None

    NEW_REPO = None

    OTHER_TYPE_REPO = None

    OTHER_TYPE = None

    def test_index(self):

        self.log_user()

        response = self.app.get(url('repos'))

    def test_create(self):

        self.log_user()

        repo_name = self.NEW_REPO

        description = u'description for newly created repo'

        response = self.app.post(url('repos'),

                        fixture._get_repo_create_params(repo_private=False,

                                                repo_name=repo_name,

                                                repo_type=self.REPO_TYPE,

                                                repo_description=description,

        ## run the check page that triggers the flash message

        response = self.app.get(url('repo_check_home', repo_name=repo_name))

        assert response.json == {u'result': True}

        self.checkSessionFlash(response,

                               'Created repository <a href="/%s">%s</a>'

                               % (repo_name, repo_name))

        # test if the repo was created in the database

        new_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name).one()

        assert new_repo.repo_name == repo_name

        assert new_repo.description == description

        # test if the repository is visible in the list ?

        response = self.app.get(url('summary_home', repo_name=repo_name))

        response.mustcontain(repo_name)

        response.mustcontain(self.REPO_TYPE)

        # test if the repository was created on filesystem

        try:

            vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)))

        except vcs.exceptions.VCSError:

            pytest.fail('no repo %s in filesystem' % repo_name)

        RepoModel().delete(repo_name)

        Session().commit()

    def test_case_insensitivity(self):

        self.log_user()

        repo_name = self.NEW_REPO

        description = u'description for newly created repo'

        response = self.app.post(url('repos'),

                                 fixture._get_repo_create_params(repo_private=False,

                                                                 repo_name=repo_name,

                                                                 repo_type=self.REPO_TYPE,

                                                                 repo_description=description,

        # try to create repo with swapped case

        swapped_repo_name = repo_name.swapcase()

        response = self.app.post(url('repos'),

                                 fixture._get_repo_create_params(repo_private=False,

                                                                 repo_name=swapped_repo_name,

                                                                 repo_type=self.REPO_TYPE,

                                                                 repo_description=description,

        response.mustcontain('already exists')

        RepoModel().delete(repo_name)