old_title = 'Kallithea'

        old_realm = 'Kallithea authentication'

        new_ga_code = 'ga-test-123456789'

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

                        params=dict(title=old_title,

                                 realm=old_realm,

                                 ga_code=new_ga_code,

                                 captcha_private_key='',

                                 captcha_public_key='',

                                 _session_csrf_secret_token=self.session_csrf_secret_token(),

                                 ))

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

        assert Setting.get_app_settings()['ga_code'] == new_ga_code

        response = response.follow()

        response.mustcontain("""_gaq.push(['_setAccount', '%s']);""" % new_ga_code)

    def test_ga_code_inactive(self):

        self.log_user()

        old_title = 'Kallithea'

        old_realm = 'Kallithea authentication'

        new_ga_code = ''

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

                        params=dict(title=old_title,

                                 realm=old_realm,

                                 ga_code=new_ga_code,

                                 captcha_private_key='',

                                 captcha_public_key='',

                                 _session_csrf_secret_token=self.session_csrf_secret_token(),

                                 ))

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

        assert Setting.get_app_settings()['ga_code'] == new_ga_code

        response = response.follow()

        response.mustcontain(no=["_gaq.push(['_setAccount', '%s']);" % new_ga_code])

    def test_captcha_activate(self):

        self.log_user()

        old_title = 'Kallithea'

        old_realm = 'Kallithea authentication'

        new_ga_code = ''

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

                        params=dict(title=old_title,

                                 realm=old_realm,

                                 ga_code=new_ga_code,

                                 captcha_private_key='1234567890',

                                 captcha_public_key='1234567890',

                                 _session_csrf_secret_token=self.session_csrf_secret_token(),

                                 ))

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

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

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

        response.mustcontain('captcha')

    def test_captcha_deactivate(self):

        self.log_user()

        old_title = 'Kallithea'

        old_realm = 'Kallithea authentication'

        new_ga_code = ''

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

                        params=dict(title=old_title,

                                 realm=old_realm,

                                 ga_code=new_ga_code,

                                 captcha_private_key='',

                                 captcha_public_key='1234567890',

                                 _session_csrf_secret_token=self.session_csrf_secret_token(),

                                 ))

        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)

                                    '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)

    # Git - add file

    def test_add_file_view_git(self):

        self.log_user()

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

                                      repo_name=base.GIT_REPO,

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

    def test_add_file_into_git_missing_content(self):

        self.log_user()

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

                                      repo_name=base.GIT_REPO,

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

                                 params={

                                     'content': '',

                                     '_session_csrf_secret_token': self.session_csrf_secret_token(),

                                 },

                                 status=302)

        self.checkSessionFlash(response, 'No content')

    def test_add_file_into_git_missing_filename(self):

        self.log_user()

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

                                      repo_name=base.GIT_REPO,

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

                                 params={

                                    'content': "foo",

                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),

                                 },

                                 status=302)

        self.checkSessionFlash(response, 'No filename')

    @base.parametrize('location,filename', [

        ('/abs', 'foo'),

        ('../rel', 'foo'),

        ('file/../foo', 'foo'),

    ])

    def test_add_file_into_git_bad_filenames(self, location, filename):

        self.log_user()

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

                                      repo_name=base.GIT_REPO,

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

                                 params={

                                    'content': "foo",

                                    'filename': filename,

                                    'location': location,

                                    '_session_csrf_secret_token': self.session_csrf_secret_token(),

                                 },

                                 status=302)

        self.checkSessionFlash(response, 'Location must be relative path and must not contain .. in path')

    @base.parametrize('cnt,location,filename', [

        (1, '', 'foo.txt'),

        (2, 'dir', 'foo.rst'),

        (3, 'rel/dir', 'foo.bar'),

    ])

    def test_add_file_into_git(self, cnt, location, filename):

        self.log_user()

        repo = fixture.create_repo(u'commit-test-%s' % cnt, repo_type='git')

        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,

                                          revision='tip', f_path=posixpath.join(location, filename)),

                                    status=302)

            self.checkSessionFlash(response,

                'You can only edit files with revision being a valid branch')

        finally:

            fixture.destroy_repo(repo.repo_name)

    def test_edit_file_view_commit_changes_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.post(base.url('files_edit_home',

                                          repo_name=repo.repo_name,

                                          revision=repo.scm_instance.DEFAULT_BRANCH_NAME,

                                          f_path=posixpath.join(location, filename)),

                                     params={

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

                                        'message': 'i committed',

                                        '_session_csrf_secret_token': self.session_csrf_secret_token(),

                                     },

                                    status=302)

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

                                   % posixpath.join(location, filename))

        finally:

            fixture.destroy_repo(repo.repo_name)

    # Git - edit

    def test_edit_file_view_git(self):

        self.log_user()

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

                                      repo_name=base.GIT_REPO,

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

    def test_edit_file_view_not_on_branch_git(self):

        self.log_user()

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

        ## 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.post(base.url(controller='login', action='index'),

                                 {'username': base.TEST_USER_REGULAR_LOGIN,

                                  'password': base.TEST_USER_REGULAR_PASS,

                                  'remember': False,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        assert 'Set-Cookie' in response.headers

        for cookie in response.headers.getall('Set-Cookie'):

            assert not re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE), 'Cookie %r has expiration date, but should be a session cookie' % cookie

    def test_login_remember(self):

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

                                 {'username': base.TEST_USER_REGULAR_LOGIN,

                                  'password': base.TEST_USER_REGULAR_PASS,

                                  'remember': True,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        assert 'Set-Cookie' in response.headers

        for cookie in response.headers.getall('Set-Cookie'):

            assert re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE), 'Cookie %r should have expiration date, but is a session cookie' % cookie

    def test_logout(self):

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

                                 {'username': base.TEST_USER_REGULAR_LOGIN,

                                  'password': base.TEST_USER_REGULAR_PASS,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        # Verify that a login session has been established.

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

        response = response.follow()

        assert 'authuser' in response.session

        response.click('Log Out')

        # Verify that the login session has been terminated.

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

        assert 'authuser' not in response.session

    @base.parametrize('url_came_from', [

          ('data:text/html,<script>window.alert("xss")</script>',),

          ('mailto:test@example.com',),

          ('file:///etc/passwd',),

          ('ftp://ftp.example.com',),

          ('http://other.example.com/bl%C3%A5b%C3%A6rgr%C3%B8d',),

          ('//evil.example.com/',),

          ('/\r\nX-Header-Injection: boo',),

          ('/invälid_url_bytes',),

          ('non-absolute-path',),

    ])

    def test_login_bad_came_froms(self, url_came_from):

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

                                     came_from=url_came_from),

                                 {'username': base.TEST_USER_ADMIN_LOGIN,

                                  'password': base.TEST_USER_ADMIN_PASS,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()},

                                 status=400)

    def test_login_short_password(self):

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

                                 {'username': base.TEST_USER_ADMIN_LOGIN,

                                  'password': 'as',

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        assert response.status == '200 OK'

        response.mustcontain('Enter 3 characters or more')

    def test_login_wrong_username_password(self):

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

                                 {'username': 'error',

                                  'password': 'test12',

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        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()})

        assert response.status == '302 Found'

        for encoded in args_encoded:

            assert encoded in response.location

    @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_after_incorrect_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': 'error',

                                  'password': 'test12',

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        response.mustcontain('Invalid username or password')

        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

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

    # REGISTRATIONS

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

    def test_register(self):

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

        response.mustcontain('Sign Up')

    def test_register_err_same_username(self):

        uname = base.TEST_USER_ADMIN_LOGIN

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

                                            {'username': uname,

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

                                             'email': 'goodmail@example.com',

                                             'firstname': 'test',

                                             'lastname': 'test',

                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})

        with test_context(self.app):

            msg = validators.ValidUsername()._messages['username_exists']

        msg = h.html_escape(msg % {'username': uname})

        response.mustcontain(msg)

    def test_register_err_same_email(self):

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

                                            {'username': 'test_admin_0',

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

                                             'email': base.TEST_USER_ADMIN_EMAIL,

                                             'firstname': 'test',

                                             'lastname': 'test',

                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})

        with test_context(self.app):

            msg = validators.UniqSystemEmail()()._messages['email_taken']

        response.mustcontain(msg)

    def test_register_err_same_email_case_sensitive(self):

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

                                            {'username': 'test_admin_1',

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

                                             'email': base.TEST_USER_ADMIN_EMAIL.title(),

                                             'firstname': 'test',

                                             'lastname': 'test',

                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})

        with test_context(self.app):

            msg = validators.UniqSystemEmail()()._messages['email_taken']

        response.mustcontain(msg)

    def test_register_err_wrong_data(self):

         {'added': 18,

          'deleted': 2,

          'binary': False,

          'ops': {MOD_FILENODE: 'modified file'}}),

        ('vcs/backends/git/repository.py', 'modified',

         {'added': 46,

          'deleted': 15,

          'binary': False,

          'ops': {MOD_FILENODE: 'modified file'}}),

        ('vcs/backends/hg.py', 'modified',

         {'added': 22,

          'deleted': 3,

          'binary': False,

          'ops': {MOD_FILENODE: 'modified file'}}),

        ('vcs/tests/test_git.py', 'modified',

         {'added': 5,

          'deleted': 5,

          'binary': False,

          'ops': {MOD_FILENODE: 'modified file'}}),

        ('vcs/tests/test_repository.py', 'modified',

         {'added': 174,

          'deleted': 2,

          'binary': False,

          'ops': {MOD_FILENODE: 'modified file'}}),

    ],

    'git_diff_modify_binary_file.diff': [

        ('file.name', 'modified',

         {'added': 0,

          'deleted': 0,

          'binary': True,

          'ops': {MOD_FILENODE: 'modified file',

                  BIN_FILENODE: 'binary diff not shown'}})

    ],

    'hg_diff_copy_file.diff': [

        ('file2', 'modified',

         {'added': 0,

          'deleted': 0,

          'binary': True,

          'ops': {COPIED_FILENODE: 'file copied from file1 to file2'}}),

    ],

    'hg_diff_copy_and_modify_file.diff': [

        ('file3', 'modified',

         {'added': 1,

          'deleted': 0,

          'binary': False,

          'ops': {COPIED_FILENODE: 'file copied from file2 to file3',

                  MOD_FILENODE: 'modified file'}}),

    ],

    'hg_diff_copy_and_chmod_file.diff': [

        ('file4', 'modified',

         {'added': 0,

          'deleted': 0,

          'binary': True,

          'ops': {COPIED_FILENODE: 'file copied from file3 to file4',

                  CHMOD_FILENODE: 'modified file chmod 100644 => 100755'}}),

    ],

    'hg_diff_copy_chmod_and_edit_file.diff': [

        ('file5', 'modified',

         {'added': 2,

          'deleted': 1,

          'binary': False,

          'ops': {COPIED_FILENODE: 'file copied from file4 to file5',

                  CHMOD_FILENODE: 'modified file chmod 100755 => 100644',

                  MOD_FILENODE: 'modified file'}}),

    ],

    'hg_diff_rename_space_cr.diff': [

        ('oh yes', 'renamed',

         {'added': 3,

          'deleted': 2,

          'binary': False,

          'ops': {RENAMED_FILENODE: 'file renamed from oh no to oh yes'}}),

    ],

}

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'

'''

# -*- coding: utf-8 -*-

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# 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 General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""

Test suite for vcs push/pull operations.

The tests need Git > 1.8.1.

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):

    public_keys = {

        base.TEST_USER_REGULAR_LOGIN: u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUQ== kallithea@localhost',

        base.TEST_USER_ADMIN_LOGIN: u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6Ycnc2oUZHQnQwuqgZqTTdMDZD7ataf3JM7oG2Fw8JR6cdmz4QZLe5mfDwaFwG2pWHLRpVqzfrD/Pn3rIO++bgCJH5ydczrl1WScfryV1hYMJ/4EzLGM657J1/q5EI+b9SntKjf4ax+KP322L0TNQGbZUHLbfG2MwHMrYBQpHUq== kallithea@localhost',

    }

    @classmethod

    def repo_url_param(cls, webserver, repo_name, username=base.TEST_USER_ADMIN_LOGIN, password=base.TEST_USER_ADMIN_PASS, client_ip=base.IP_ADDR):

        user = User.get_by_username(username)

        if user.ssh_keys:

            ssh_key = user.ssh_keys[0]

        else:

            sshkeymodel = SshKeyModel()

            ssh_key = sshkeymodel.create(user, u'test key', cls.public_keys[user.username])

            Session().commit()

        return cls._ssh_param(repo_name, user, ssh_key, client_ip)

# Mixins for using Mercurial and Git

class HgVcsTest(object):

    repo_type = 'hg'

    repo_name = base.HG_REPO

class GitVcsTest(object):

    repo_type = 'git'

    repo_name = base.GIT_REPO

# Combine mixins to give the combinations we want to parameterize tests with

class HgHttpVcsTest(HgVcsTest, HttpVcsTest):

    pass

class GitHttpVcsTest(GitVcsTest, HttpVcsTest):

    pass

class HgSshVcsTest(HgVcsTest, SshVcsTest):

    @staticmethod

    def _ssh_param(repo_name, user, ssh_key, client_ip):

        # Specify a custom ssh command on the command line

        return r"""--config ui.ssh="bash -c 'SSH_ORIGINAL_COMMAND=\"\$2\" SSH_CONNECTION=\"%s 1024 127.0.0.1 22\" kallithea-cli ssh-serve -c %s %s %s' --" ssh://someuser@somehost/%s""" % (

            client_ip,

            CONFIG['__file__'],

            user.user_id,

            ssh_key.user_ssh_key_id,

            repo_name)

class GitSshVcsTest(GitVcsTest, SshVcsTest):

    @staticmethod

    def _ssh_param(repo_name, user, ssh_key, client_ip):

        # Set a custom ssh command in the global environment

        os.environ['GIT_SSH_COMMAND'] = r"""bash -c 'SSH_ORIGINAL_COMMAND="$2" SSH_CONNECTION="%s 1024 127.0.0.1 22" kallithea-cli ssh-serve -c %s %s %s' --""" % (

            client_ip,

            CONFIG['__file__'],

            user.user_id,

            ssh_key.user_ssh_key_id)

        return "ssh://someuser@somehost/%s""" % repo_name

parametrize_vcs_test = base.parametrize('vt', [

    HgHttpVcsTest,

    GitHttpVcsTest,

    HgSshVcsTest,

    GitSshVcsTest,

])

parametrize_vcs_test_hg = base.parametrize('vt', [

    HgHttpVcsTest,

    HgSshVcsTest,

])

parametrize_vcs_test_http = base.parametrize('vt', [

    HgHttpVcsTest,

    GitHttpVcsTest,

])

class Command(object):

    def __init__(self, cwd):

        self.cwd = cwd

    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):

        cmd = """echo "added_line%s" >> %s""" % (i, added_file)

        Command(dest_dir).execute(cmd)

        if vcs == 'hg':

            cmd = """hg commit -m "committed new %s" -u "%s" "%s" """ % (

                i, author_str, added_file

            )

        elif vcs == 'git':

            cmd = """git commit -m "committed new %s" --author "%s" "%s" """ % (

                i, author_str, added_file

            )

        # git commit needs EMAIL on some machines

        Command(dest_dir).execute(cmd, EMAIL=email)

def _add_files_and_push(webserver, vt, dest_dir, clone_url, ignoreReturnCode=False, files_no=3):

    _add_files(vt.repo_type, dest_dir, files_no=files_no)

    # PUSH it back

    stdout = stderr = None

    if vt.repo_type == 'hg':

        stdout, stderr = Command(dest_dir).execute('hg push -f --verbose', clone_url, ignoreReturnCode=ignoreReturnCode)

    elif vt.repo_type == 'git':

        stdout, stderr = Command(dest_dir).execute('git push -f --verbose', clone_url, "master", ignoreReturnCode=ignoreReturnCode)

    return stdout, stderr

def _check_outgoing(vcs, cwd, clone_url):

    if vcs == 'hg':

        # hg removes the password from default URLs, so we have to provide it here via the clone_url

        return Command(cwd).execute('hg -q outgoing', clone_url, ignoreReturnCode=True)

    elif vcs == 'git':

        Command(cwd).execute('git remote update')

        return Command(cwd).execute('git log origin/master..master')

def set_anonymous_access(enable=True):

    user = User.get_default_user()

    user.active = enable

    Session().commit()

    if enable != User.get_default_user().active:

        raise Exception('Cannot set anonymous access')

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

# TESTS

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

def _check_proper_git_push(stdout, stderr):

    assert 'fatal' not in stderr

    assert 'rejected' not in stderr

    assert 'Pushing to' in stderr

    assert 'master -> master' in stderr