self.checkSessionFlash(response, 'Updated application settings')

        assert Setting.get_app_settings()['captcha_private_key'] == ''

        response = self.app.get(base.url('register'))

        response.mustcontain(no=['captcha'])

    def test_title_change(self):

        self.log_user()

        old_title = 'Kallithea'

        new_title = old_title + '_changed'

        old_realm = 'Kallithea authentication'

        for new_title in ['Changed', 'Żółwik', old_title]:

            response = self.app.post(base.url('admin_settings_global'),

                        params=dict(title=new_title,

                                 realm=old_realm,

                                 ga_code='',

                                 captcha_private_key='',

                                 captcha_public_key='',

                                 _session_csrf_secret_token=self.session_csrf_secret_token(),

                                ))

            self.checkSessionFlash(response, 'Updated application settings')

            response = response.follow()

            response.mustcontain("""<span class="branding">%s</span>""" % new_title)

        response = self.app.post(base.url('files_add_home',

                                      repo_name=repo.repo_name,

                                      revision='tip', f_path='/'),

                                 params={

                                    'content': "foo",

                                    'filename': filename,

                                    'location': location,

                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),

                                 },

                                 status=302)

        try:

            self.checkSessionFlash(response, 'Successfully committed to %s'

                                   % posixpath.join(location, filename))

        finally:

            fixture.destroy_repo(repo.repo_name)

    # Hg - EDIT

    def test_edit_file_view_hg(self):

        self.log_user()

        response = self.app.get(base.url('files_edit_home',

                                      repo_name=base.HG_REPO,

                                      revision='tip', f_path='vcs/nodes.py'))

        # Odd error when on tip ...

        self.checkSessionFlash(response, "You can only edit files with revision being a valid branch")

        # Specify branch head revision to avoid "valid branch" error and get coverage of edit form

        response = self.app.get(base.url('files_edit_home',

                                      repo_name=base.HG_REPO,

                                      revision='96507bd11ecc815ebc6270fdf6db110928c09c1e', f_path='vcs/nodes.py'))

    def test_edit_file_view_not_on_branch_hg(self):

        self.log_user()

        repo = fixture.create_repo(u'test-edit-repo', repo_type='hg')

        ## add file

        location = 'vcs'

        filename = 'nodes.py'

        response = self.app.post(base.url('files_add_home',

                                      repo_name=repo.repo_name,

                                      revision='tip', f_path='/'),

                                 params={

                                    'content': "def py():\n print 'hello'\n",

                                    'filename': filename,

                                    'location': location,

                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),

                                 },

                                 status=302)

        response.follow()

        try:

            self.checkSessionFlash(response, 'Successfully committed to %s'

                                   % posixpath.join(location, filename))

            response = self.app.get(base.url('files_edit_home',

                                          repo_name=repo.repo_name,

        response.mustcontain('Invalid username or password')

    def test_login_non_ascii(self):

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

                                 {'username': base.TEST_USER_REGULAR_LOGIN,

                                  'password': 'blåbærgrød',

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        response.mustcontain('>Invalid username or password<')

    # verify that get arguments are correctly passed along login redirection

    @base.parametrize('args', [

        {'foo':'one', 'bar':'two'},

        {'blue': u'blå', 'green': u'grøn'},

    ])

    def test_redirection_to_login_form_preserves_get_args(self, args):

        with fixture.anon_access(False):

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

                                        repo_name=base.HG_REPO,

                                        **args))

            assert response.status == '302 Found'

            came_from = urllib.parse.parse_qs(urllib.parse.urlparse(response.location).query)['came_from'][0]

            came_from_qs = urllib.parse.parse_qsl(urllib.parse.urlparse(came_from).query)

    @base.parametrize('args,args_encoded', [

        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),

        ({'blue': u'blå', 'green':u'grøn'},

             ('blue=bl%C3%A5', 'green=gr%C3%B8n')),

    ])

    def test_login_form_preserves_get_args(self, args, args_encoded):

        response = self.app.get(base.url(controller='login', action='index',

                                    came_from=base.url('/_admin/users', **args)))

        came_from = urllib.parse.parse_qs(urllib.parse.urlparse(response.form.action).query)['came_from'][0]

        for encoded in args_encoded:

            assert encoded in came_from

    @base.parametrize('args,args_encoded', [

        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),

        ({'blue': u'blå', 'green':u'grøn'},

             ('blue=bl%C3%A5', 'green=gr%C3%B8n')),

    ])

    def test_redirection_after_successful_login_preserves_get_args(self, args, args_encoded):

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

                                     came_from=base.url('/_admin/users', **args)),

                                 {'username': base.TEST_USER_ADMIN_LOGIN,

                                  'password': base.TEST_USER_ADMIN_PASS,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

}

class TestDiffLib(base.TestController):

    @base.parametrize('diff_fixture', DIFF_FIXTURES)

    def test_diff(self, diff_fixture):

        raw_diff = fixture.load_resource(diff_fixture, strip=False)

        vcs = 'hg'

        if diff_fixture.startswith('git_'):

            vcs = 'git'

        diff_processor = DiffProcessor(raw_diff, vcs=vcs)

        data = [(x['filename'], x['operation'], x['stats']) for x in diff_processor.parsed]

        expected_data = DIFF_FIXTURES[diff_fixture]

        assert expected_data == data

    def test_diff_markup(self):

        raw_diff = fixture.load_resource('markuptest.diff', strip=False)

        diff_processor = DiffProcessor(raw_diff)

        chunks = diff_processor.parsed[0]['chunks']

        assert not chunks[0]

        #from pprint import pprint; pprint(chunks[1])

        l = ['\n']

        for d in chunks[1]:

            l.append('%(action)-7s %(new_lineno)3s %(old_lineno)3s %(line)r\n' % d)

        s = ''.join(l)

        assert s == r'''

context ... ... '@@ -51,6 +51,13 @@\n'

unmod    51  51 '<u>\t</u>begin();\n'

unmod    52  52 '<u>\t</u>\n'

add      53     '<u>\t</u>int foo;<u class="cr"></u>\n'

add      54     '<u>\t</u>int bar; <u class="cr"></u>\n'

add      55     '<u>\t</u>int baz;<u>\t</u><u class="cr"></u>\n'

add      56     '<u>\t</u>int space; <i></i>'

add      57     '<u>\t</u>int tab;<u>\t</u>\n'

add      58     '<u>\t</u>\n'

unmod    59  53 ' <i></i>'

del          54 '<u>\t</u>#define MAX_STEPS (48)\n'

add      60     '<u>\t</u><u class="cr"></u>\n'

add      61     '<u>\t</u>#define MAX_STEPS (64)<u class="cr"></u>\n'

unmod    62  55 '\n'

del          56 '<u>\t</u>#define MIN_STEPS (<del>48</del>)\n'

add      63     '<u>\t</u>#define MIN_STEPS (<ins>42</ins>)\n'

'''

This file was forked by the Kallithea project in July 2014.

Original author and date, and relevant copyright and licensing information is below:

:created_on: Dec 30, 2010

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

from __future__ import print_function

import json

import os

import re

import tempfile

import time

import urllib.request

from subprocess import PIPE, Popen

from tempfile import _RandomNameSequence

import pytest

from kallithea import CONFIG

from kallithea.model.db import CacheInvalidation, Repository, Ui, User, UserIpMap, UserLog

from kallithea.model.meta import Session

from kallithea.model.ssh_key import SshKeyModel

from kallithea.model.user import UserModel

from kallithea.tests import base

from kallithea.tests.fixture import Fixture

DEBUG = True

HOST = '127.0.0.1:4999'  # test host

fixture = Fixture()

# Parameterize different kinds of VCS testing - both the kind of VCS and the

# access method (HTTP/SSH)

# Mixin for using HTTP and SSH URLs

class HttpVcsTest(object):

    @staticmethod

    def repo_url_param(webserver, repo_name, **kwargs):

        return webserver.repo_url(repo_name, **kwargs)

class SshVcsTest(object):

    def execute(self, *args, **environ):

        """

        Runs command on the system with given ``args`` using simple space

        join without safe quoting.

        """

        command = ' '.join(args)

        ignoreReturnCode = environ.pop('ignoreReturnCode', False)

        if DEBUG:

            print('*** CMD %s ***' % command)

        testenv = dict(os.environ)

        testenv['LANG'] = 'en_US.UTF-8'

        testenv['LANGUAGE'] = 'en_US:en'

        testenv['HGPLAIN'] = ''

        testenv['HGRCPATH'] = ''

        testenv.update(environ)

        p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd, env=testenv)

        stdout, stderr = p.communicate()

        if DEBUG:

            if stdout:

                print('stdout:', stdout)

            if stderr:

                print('stderr:', stderr)

        if not ignoreReturnCode:

            assert p.returncode == 0

def _get_tmp_dir(prefix='vcs_operations-', suffix=''):

    return tempfile.mkdtemp(dir=base.TESTS_TMP_PATH, prefix=prefix, suffix=suffix)

def _add_files(vcs, dest_dir, files_no=3):

    """

    Generate some files, add it to dest_dir repo and push back

    vcs is git or hg and defines what VCS we want to make those files for

    :param vcs:

    :param dest_dir:

    """

    added_file = '%ssetup.py' % next(_RandomNameSequence())

    open(os.path.join(dest_dir, added_file), 'a').close()

    Command(dest_dir).execute(vcs, 'add', added_file)

    email = 'me@example.com'

    if os.name == 'nt':

        author_str = 'User <%s>' % email

    else:

        author_str = 'User ǝɯɐᴎ <%s>' % email

    for i in range(files_no):