Changeset - 313930ae35ad
Parent rev.
Child rev.
[Not reviewed]
default
0 4 0
Thomas De Schampheleire - 10 years ago 2016-05-06 21:21:50
thomas.de_schampheleire@nokia.com
pytest migration: other: convert to TestControllerPytest
4 files changed with 18 insertions and 20 deletions:
kallithea/tests/other/manual_test_vcs_operations.py
2
2
kallithea/tests/other/test_libs.py
12
11
kallithea/tests/other/test_mail.py
1
1
kallithea/tests/other/test_validators.py
3
6
0 comments (0 inline, 0 general)
kallithea/tests/other/manual_test_vcs_operations.py
➞
Show inline comments
 
# -*- 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/>.
 
"""
 
kallithea.tests.other.manual_test_vcs_operations
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

			




	
	
	
		
 
Test suite for making push/pull operations.

			




	
	
	
		
 

			




	
	
	
		
 
Run it in two terminals::

			




	
	
	
		
 
 paster serve kallithea/tests/test.ini

			




	
	
	
		
 
 KALLITHEA_WHOOSH_TEST_DISABLE=1 KALLITHEA_NO_TMP_PATH=1 nosetests kallithea/tests/other/manual_test_vcs_operations.py

			




	
	
	
		
 

			




	
	
	
		
 
You must have git > 1.8.1 for tests to work fine

			




	
	
	
		
 

			




	
	
	
		
 
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.

			




	
	
	
		
 

			




	
	
	
		
 
"""

			




	
	
	
		
 

			




	
	
	
		
 
import re

			




	
	
	
		
 
import tempfile

			




	
	
	
		
 
import time

			




	
	
	
		
 
from os.path import join as jn

			




	
	
	
		
 

			




	
	
	
		
 
from tempfile import _RandomNameSequence

			




	
	
	
		
 
from subprocess import Popen, PIPE

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.tests import *

			




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

			




	
	
	
		
 
from kallithea.model.meta import Session

			




	
	
	
		
 
from kallithea.model.repo import RepoModel

			




	
	
	
		
 
from kallithea.model.user import UserModel

			




	
	
	
		
 

			




	
	
	
		
 
DEBUG = True

			




	
	
	
		
 
HOST = '127.0.0.1:4999'  # test host

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class Command(object):

			




	
	
	
		
 

			




	
	
	
		
 
    def __init__(self, cwd):

			




	
	
	
		
 
        self.cwd = cwd

			




	
	
	
		
 

			




	
	
	
		
 
    def execute(self, cmd, *args):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Runs command on the system with given ``args``.

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




	
	
	
		
 
        command = cmd + ' ' + ' '.join(args)

			




	
	
	
		
 
        if DEBUG:

			




	
	
	
		
 
            print '*** CMD %s ***' % command

			




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

			




	
	
	
		
 
        stdout, stderr = p.communicate()

			




	
	
	
		
 
        if DEBUG:

			




	
	
	
		
 
            print 'stdout:', repr(stdout)

			




	
	
	
		
 
            print 'stderr:', repr(stderr)

			




	
	
	
		
 
        return stdout, stderr

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def _get_tmp_dir():

			




	
	
	
		
 
    return tempfile.mkdtemp(prefix='rc_integration_test')

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def _construct_url(repo, dest=None, **kwargs):

			




	
	
	
		
 
    if dest is None:

			




	
	
	
		
 
        #make temp clone

			




	
	
	
		
 
        dest = _get_tmp_dir()

			




	
	
	
		
 
    params = {

			




	
	
	
		
 
        'user': TEST_USER_ADMIN_LOGIN,

			




	
	
	
		
 
        'passwd': TEST_USER_ADMIN_PASS,

			




	
	
	
		
 
        'host': HOST,

			




	
	
	
		
 
        'cloned_repo': repo,

			




	
	
	
		
 
        'dest': dest

			




	
	
	
		
 
    }

			




	
	
	
		
 
    params.update(**kwargs)

			




	
	
	
		
 
    if params['user'] and params['passwd']:

			




	
	
	
		
 
        _url = 'http://%(user)s:%(passwd)s@%(host)s/%(cloned_repo)s %(dest)s' % params

			




	
	
	
		
 
    else:

			




	
	
	
		
 
        _url = 'http://(host)s/%(cloned_repo)s %(dest)s' % params

			




	
	
	
		
 
    return _url

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def _add_files_and_push(vcs, DEST, **kwargs):

			




	
	
	
		
 
    """

			




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

			




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

			




	
	
	
		
 

			




	
	
	
		
 
    :param vcs:

			




	
	
	
		
 
    :param DEST:

			




	
	
	
		
 
    """

			




	
	
	
		
 
    # commit some stuff into this repo

			




	
	
	
		
 
    cwd = path = jn(DEST)

			




	
	
	
		
 
    #added_file = jn(path, '%ssetupążźć.py' % _RandomNameSequence().next())

			




	
	
	
		
 
    added_file = jn(path, '%ssetup.py' % _RandomNameSequence().next())

			




	
	
	
		
 
    Command(cwd).execute('touch %s' % added_file)

			




	
	
	
		
 
    Command(cwd).execute('%s add %s' % (vcs, added_file))

			




	
	
	
		
 

			




	
	
	
		
 
    for i in xrange(kwargs.get('files_no', 3)):

			




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

			




	
	
	
		
 
        Command(cwd).execute(cmd)

			




	
	
	
		
 
        author_str = 'User ǝɯɐᴎ <me@example.com>'

			




	
	
	
		
 
        if vcs == 'hg':

			




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

			




	
	
	
		
 
                i, author_str, added_file

			




	
	
	
		
 
            )

			




	
	
	
		
 
        elif vcs == 'git':

			




	
	
	
		
 
            cmd = """EMAIL="me@example.com" git commit -m 'committed new %s' --author '%s' %s """ % (

			




	
	
	
		
 
                i, author_str, added_file

			




	
	
	
		
 
            )

			




	
	
	
		
 
        Command(cwd).execute(cmd)

			




	
	
	
		
 

			




	
	
	
		
 
    # PUSH it back

			




	
	
	
		
 
    _REPO = None

			




	
	
	
		
 
    if vcs == 'hg':

			




	
	
	
		
 
        _REPO = HG_REPO

			




	
	
	
		
 
    elif vcs == 'git':

			




	
	
	
		
 
        _REPO = GIT_REPO

			




	
	
	
		
 

			




	
	
	
		
 
    kwargs['dest'] = ''

			




	
	
	
		
 
    clone_url = _construct_url(_REPO, **kwargs)

			




	
	
	
		
 
    if 'clone_url' in kwargs:

			




	
	
	
		
 
        clone_url = kwargs['clone_url']

			




	
	
	
		
 
    stdout = stderr = None

			




	
	
	
		
 
    if vcs == 'hg':

			




	
	
	
		
 
        stdout, stderr = Command(cwd).execute('hg push --verbose', clone_url)

			




	
	
	
		
 
    elif vcs == 'git':

			




	
	
	
		
 
        stdout, stderr = Command(cwd).execute('git push --verbose', clone_url + " master")

			




	
	
	
		
 

			




	
	
	
		
 
    return stdout, stderr

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def set_anonymous_access(enable=True):

			




	
	
	
		
 
    user = User.get_by_username(User.DEFAULT_USER)

			




	
	
	
		
 
    user.active = enable

			




	
	
	
		
 
    Session().add(user)

			




	
	
	
		
 
    Session().commit()

			




	
	
	
		
 
    print '\tanonymous access is now:', enable

			




	
	
	
		
 
    if enable != User.get_by_username(User.DEFAULT_USER).active:

			




	
	
	
		
 
        raise Exception('Cannot set anonymous access')

			




	
	
	
		
 

			




	
	
	
		
 

			




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

			




	
	
	
		
 
# TESTS

			




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

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def _check_proper_git_push(stdout, stderr):

			




	
	
	
		
 
    #WTF Git stderr is output ?!

			




	
	
	
		
 
    assert 'fatal' not in stderr

			




	
	
	
		
 
    assert 'rejected' not in stderr

			




	
	
	
		
 
    assert 'Pushing to' in stderr

			




	
	
	
		
 
    assert 'master -> master' in stderr

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TestVCSOperations(BaseTestCase):

			




	
	
	
		
 
class TestVCSOperations(TestControllerPytest):

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def setup_class(cls):

			




	
	
	
		
 
        #DISABLE ANONYMOUS ACCESS

			




	
	
	
		
 
        set_anonymous_access(False)

			




	
	
	
		
 

			




	
	
	
		
 
    def setUp(self):

			




	
	
	
		
 
    def setup_method(self, method):

			




	
	
	
		
 
        r = Repository.get_by_repo_name(GIT_REPO)

			




	
	
	
		
 
        Repository.unlock(r)

			




	
	
	
		
 
        r.enable_locking = False

			




	
	
	
		
 
        Session().add(r)

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
        r = Repository.get_by_repo_name(HG_REPO)

			




	
	
	
		
 
        Repository.unlock(r)

			




	
	
	
		
 
        r.enable_locking = False

			




	
	
	
		
 
        Session().add(r)

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_hg_repo_by_admin(self):

			




	
	
	
		
 
        clone_url = _construct_url(HG_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'requesting all changes' in stdout

			




	
	
	
		
 
        assert 'adding changesets' in stdout

			




	
	
	
		
 
        assert 'adding manifests' in stdout

			




	
	
	
		
 
        assert 'adding file changes' in stdout

			




	
	
	
		
 

			




	
	
	
		
 
        assert stderr == ''

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_git_repo_by_admin(self):

			




	
	
	
		
 
        clone_url = _construct_url(GIT_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'Cloning into' in stdout + stderr

			




	
	
	
		
 
        assert stderr == '' or stdout == ''

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_wrong_credentials_hg(self):

			




	
	
	
		
 
        clone_url = _construct_url(HG_REPO, passwd='bad!')

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)

			




	
	
	
		
 
        assert 'abort: authorization failed' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_wrong_credentials_git(self):

			




	
	
	
		
 
        clone_url = _construct_url(GIT_REPO, passwd='bad!')

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)

			




	
	
	
		
 
        assert 'fatal: Authentication failed' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_git_dir_as_hg(self):

			




	
	
	
		
 
        clone_url = _construct_url(GIT_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)

			




	
	
	
		
 
        assert 'HTTP Error 404: Not Found' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_hg_repo_as_git(self):

			




	
	
	
		
 
        clone_url = _construct_url(HG_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)

			




	
	
	
		
 
        assert 'not found' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_non_existing_path_hg(self):

			




	
	
	
		
 
        clone_url = _construct_url('trololo')

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)

			




	
	
	
		
 
        assert 'HTTP Error 404: Not Found' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_non_existing_path_git(self):

			




	
	
	
		
 
        clone_url = _construct_url('trololo')

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)

			




	
	
	
		
 
        assert 'not found' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_new_file_hg(self):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = _construct_url(HG_REPO, dest=DEST)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push('hg', DEST)

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'pushing to' in stdout

			




	
	
	
		
 
        assert 'Repository size' in stdout

			




	
	
	
		
 
        assert 'Last revision is now' in stdout

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_new_file_git(self):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = _construct_url(GIT_REPO, dest=DEST)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        # commit some stuff into this repo

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push('git', DEST)

			




	
	
	
		
 

			




	
	
	
		
 
        print [(x.repo_full_path,x.repo_path) for x in Repository.get_all()]

			




	
	
	
		
 
        _check_proper_git_push(stdout, stderr)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_invalidates_cache_hg(self):

			




	
	
	
		
 
        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

			




	
	
	
		
 
                                               ==HG_REPO).scalar()

			




	
	
	
		
 
        if not key:

			




	
	
	
		
 
            key = CacheInvalidation(HG_REPO, HG_REPO)

			




	
	
	
		
 

			




	
	
	
		
 
        key.cache_active = True

			




	
	
	
		
 
        Session().add(key)

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = _construct_url(HG_REPO, dest=DEST)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push('hg', DEST, files_no=1)

			




	
	
	
		
 

			




	
	
	
		
 
        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

			




	
	
	
		
 
                                               ==HG_REPO).all()

			




	
	
	
		
 
        self.assertEqual(key, [])

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_invalidates_cache_git(self):

			




	
	
	
		
 
        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

			




	
	
	
		
 
                                               ==GIT_REPO).scalar()

			




	
	
	
		
 
        if not key:

			




	
	
	
		
 
            key = CacheInvalidation(GIT_REPO, GIT_REPO)

			




	
	
	
		
 

			




	
	
	
		
 
        key.cache_active = True

			




	
	
	
		
 
        Session().add(key)

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = _construct_url(GIT_REPO, dest=DEST)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        # commit some stuff into this repo

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push('git', DEST, files_no=1)

			




	
	
	
		
 
        _check_proper_git_push(stdout, stderr)

			




	
	
	
		
 

			




	
	
	
		
 
        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

			




	
	
	
		
 
                                               ==GIT_REPO).all()

			




	
	
	
		
 
        self.assertEqual(key, [])

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_wrong_credentials_hg(self):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = _construct_url(HG_REPO, dest=DEST)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push('hg', DEST, user='bad',

			




	
	
	
		
 
                                             passwd='name')

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'abort: authorization failed' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_wrong_credentials_git(self):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = _construct_url(GIT_REPO, dest=DEST)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push('git', DEST, user='bad',

			




	
	
	
		
 
                                             passwd='name')

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'fatal: Authentication failed' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_back_to_wrong_url_hg(self):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = _construct_url(HG_REPO, dest=DEST)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push('hg', DEST,

			




	
	
	
		
 
                                    clone_url='http://%s/tmp' % HOST)

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'HTTP Error 404: Not Found' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_push_back_to_wrong_url_git(self):

			




	
	
	
		
 
        DEST = _get_tmp_dir()

			




	
	
	
		
 
        clone_url = _construct_url(GIT_REPO, dest=DEST)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        stdout, stderr = _add_files_and_push('git', DEST,

			




	
	
	
		
 
                                    clone_url='http://%s/tmp' % HOST)

			




	
	
	
		
 

			




	
	
	
		
 
        assert 'not found' in stderr

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_and_create_lock_hg(self):

			




	
	
	
		
 
        # enable locking

			




	
	
	
		
 
        r = Repository.get_by_repo_name(HG_REPO)

			




	
	
	
		
 
        r.enable_locking = True

			




	
	
	
		
 
        Session().add(r)

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 
        # clone

			




	
	
	
		
 
        clone_url = _construct_url(HG_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        #check if lock was made

			




	
	
	
		
 
        r = Repository.get_by_repo_name(HG_REPO)

			




	
	
	
		
 
        assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id

			




	
	
	
		
 

			




	
	
	
		
 
    def test_clone_and_create_lock_git(self):

			




	
	
	
		
 
        # enable locking

			




	
	
	
		
 
        r = Repository.get_by_repo_name(GIT_REPO)

			




	
	
	
		
 
        r.enable_locking = True

			




	
	
	
		
 
        Session().add(r)

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 
        # clone

			




	
	
	
		
 
        clone_url = _construct_url(GIT_REPO)

			




	
	
	
		
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)

			




	
	
	
		
 

			




	
	
	
		
 
        #check if lock was made

			




	
	
	
		
 
        r = Repository.get_by_repo_name(GIT_REPO)

			




	
	
	
		
 
        assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id

			




	
	
	
		
 

			




        

    


    
    

    

        

                

                    kallithea/tests/other/test_libs.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 
# -*- 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/>.

			




	
	
	
		
 
"""

			




	
	
	
		
 
kallithea.tests.other.test_libs

			




	
	
	
		
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

			




	
	
	
		
 

			




	
	
	
		
 
Package for testing various lib/helper functions in kallithea

			




	
	
	
		
 

			




	
	
	
		
 
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: Jun 9, 2011

			




	
	
	
		
 
:author: marcink

			




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

			




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

			




	
	
	
		
 
"""

			




	
	
	
		
 

			




	
	
	
		
 
import datetime

			




	
	
	
		
 
import hashlib

			




	
	
	
		
 
import mock

			




	
	
	
		
 
from kallithea.tests import *

			




	
	
	
		
 
from kallithea.lib.utils2 import AttributeDict

			




	
	
	
		
 
from kallithea.model.db import Repository

			




	
	
	
		
 

			




	
	
	
		
 
proto = 'http'

			




	
	
	
		
 
TEST_URLS = [

			




	
	
	
		
 
    ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

			




	
	
	
		
 
     '%s://127.0.0.1' % proto),

			




	
	
	
		
 
    ('%s://username@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

			




	
	
	
		
 
     '%s://127.0.0.1' % proto),

			




	
	
	
		
 
    ('%s://username:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

			




	
	
	
		
 
     '%s://127.0.0.1' % proto),

			




	
	
	
		
 
    ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],

			




	
	
	
		
 
     '%s://127.0.0.1:8080' % proto),

			




	
	
	
		
 
    ('%s://example.com' % proto, ['%s://' % proto, 'example.com'],

			




	
	
	
		
 
     '%s://example.com' % proto),

			




	
	
	
		
 
    ('%s://user:pass@example.com:8080' % proto, ['%s://' % proto, 'example.com',

			




	
	
	
		
 
                                                '8080'],

			




	
	
	
		
 
     '%s://example.com:8080' % proto),

			




	
	
	
		
 
]

			




	
	
	
		
 

			




	
	
	
		
 
proto = 'https'

			




	
	
	
		
 
TEST_URLS += [

			




	
	
	
		
 
    ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

			




	
	
	
		
 
     '%s://127.0.0.1' % proto),

			




	
	
	
		
 
    ('%s://username@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

			




	
	
	
		
 
     '%s://127.0.0.1' % proto),

			




	
	
	
		
 
    ('%s://username:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

			




	
	
	
		
 
     '%s://127.0.0.1' % proto),

			




	
	
	
		
 
    ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],

			




	
	
	
		
 
     '%s://127.0.0.1:8080' % proto),

			




	
	
	
		
 
    ('%s://example.com' % proto, ['%s://' % proto, 'example.com'],

			




	
	
	
		
 
     '%s://example.com' % proto),

			




	
	
	
		
 
    ('%s://user:pass@example.com:8080' % proto, ['%s://' % proto, 'example.com',

			




	
	
	
		
 
                                                '8080'],

			




	
	
	
		
 
     '%s://example.com:8080' % proto),

			




	
	
	
		
 
]

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TestLibs(BaseTestCase):

			




	
	
	
		
 
class TestLibs(TestControllerPytest):

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand(TEST_URLS)

			




	
	
	
		
 
    @parametrize('test_url,expected,expected_creds', TEST_URLS)

			




	
	
	
		
 
    def test_uri_filter(self, test_url, expected, expected_creds):

			




	
	
	
		
 
        from kallithea.lib.utils2 import uri_filter

			




	
	
	
		
 
        self.assertEqual(uri_filter(test_url), expected)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand(TEST_URLS)

			




	
	
	
		
 
    @parametrize('test_url,expected,expected_creds', TEST_URLS)

			




	
	
	
		
 
    def test_credentials_filter(self, test_url, expected, expected_creds):

			




	
	
	
		
 
        from kallithea.lib.utils2 import credentials_filter

			




	
	
	
		
 
        self.assertEqual(credentials_filter(test_url), expected_creds)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([('t', True),

			




	
	
	
		
 
    @parametrize('str_bool,expected', [

			




	
	
	
		
 
                           ('t', True),

			




	
	
	
		
 
                           ('true', True),

			




	
	
	
		
 
                           ('y', True),

			




	
	
	
		
 
                           ('yes', True),

			




	
	
	
		
 
                           ('on', True),

			




	
	
	
		
 
                           ('1', True),

			




	
	
	
		
 
                           ('Y', True),

			




	
	
	
		
 
                           ('yeS', True),

			




	
	
	
		
 
                           ('Y', True),

			




	
	
	
		
 
                           ('TRUE', True),

			




	
	
	
		
 
                           ('T', True),

			




	
	
	
		
 
                           ('False', False),

			




	
	
	
		
 
                           ('F', False),

			




	
	
	
		
 
                           ('FALSE', False),

			




	
	
	
		
 
                           ('0', False),

			




	
	
	
		
 
                           ('-1', False),

			




	
	
	
		
 
                           ('', False)

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_str2bool(self, str_bool, expected):

			




	
	
	
		
 
        from kallithea.lib.utils2 import str2bool

			




	
	
	
		
 
        self.assertEqual(str2bool(str_bool), expected)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_mention_extractor(self):

			




	
	
	
		
 
        from kallithea.lib.utils2 import extract_mentioned_usernames

			




	
	
	
		
 
        sample = (

			




	
	
	
		
 
            "@first hi there @world here's my email username@example.com "

			




	
	
	
		
 
            "@lukaszb check @one_more22 it pls @ ttwelve @D[] @one@two@three "

			




	
	
	
		
 
            "@UPPER    @cAmEL @2one_more22 @john please see this http://org.pl "

			




	
	
	
		
 
            "@marian.user just do it @marco-polo and next extract @marco_polo "

			




	
	
	
		
 
            "user.dot  hej ! not-needed maril@example.com"

			




	
	
	
		
 
        )

			




	
	
	
		
 

			




	
	
	
		
 
        expected = set([

			




	
	
	
		
 
            '2one_more22', 'first', 'lukaszb', 'one', 'one_more22', 'UPPER', 'cAmEL', 'john',

			




	
	
	
		
 
            'marian.user', 'marco-polo', 'marco_polo', 'world'])

			




	
	
	
		
 
        self.assertEqual(expected, set(extract_mentioned_usernames(sample)))

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
    @parametrize('age_args,expected', [

			




	
	
	
		
 
        (dict(), u'just now'),

			




	
	
	
		
 
        (dict(seconds= -1), u'1 second ago'),

			




	
	
	
		
 
        (dict(seconds= -60 * 2), u'2 minutes ago'),

			




	
	
	
		
 
        (dict(hours= -1), u'1 hour ago'),

			




	
	
	
		
 
        (dict(hours= -24), u'1 day ago'),

			




	
	
	
		
 
        (dict(hours= -24 * 5), u'5 days ago'),

			




	
	
	
		
 
        (dict(months= -1), u'1 month ago'),

			




	
	
	
		
 
        (dict(months= -1, days= -2), u'1 month and 2 days ago'),

			




	
	
	
		
 
        (dict(months= -1, days= -20), u'1 month and 19 days ago'),

			




	
	
	
		
 
        (dict(years= -1, months= -1), u'1 year and 1 month ago'),

			




	
	
	
		
 
        (dict(years= -1, months= -10), u'1 year and 10 months ago'),

			




	
	
	
		
 
        (dict(years= -2, months= -4), u'2 years and 4 months ago'),

			




	
	
	
		
 
        (dict(years= -2, months= -11), u'2 years and 11 months ago'),

			




	
	
	
		
 
        (dict(years= -3, months= -2), u'3 years and 2 months ago'),

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_age(self, age_args, expected):

			




	
	
	
		
 
        from kallithea.lib.utils2 import age

			




	
	
	
		
 
        from dateutil import relativedelta

			




	
	
	
		
 
        n = datetime.datetime(year=2012, month=5, day=17)

			




	
	
	
		
 
        delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)

			




	
	
	
		
 
        self.assertEqual(age(n + delt(**age_args), now=n), expected)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
    @parametrize('age_args,expected', [

			




	
	
	
		
 
        (dict(), u'just now'),

			




	
	
	
		
 
        (dict(seconds= -1), u'1 second ago'),

			




	
	
	
		
 
        (dict(seconds= -60 * 2), u'2 minutes ago'),

			




	
	
	
		
 
        (dict(hours= -1), u'1 hour ago'),

			




	
	
	
		
 
        (dict(hours= -24), u'1 day ago'),

			




	
	
	
		
 
        (dict(hours= -24 * 5), u'5 days ago'),

			




	
	
	
		
 
        (dict(months= -1), u'1 month ago'),

			




	
	
	
		
 
        (dict(months= -1, days= -2), u'1 month ago'),

			




	
	
	
		
 
        (dict(months= -1, days= -20), u'1 month ago'),

			




	
	
	
		
 
        (dict(years= -1, months= -1), u'13 months ago'),

			




	
	
	
		
 
        (dict(years= -1, months= -10), u'22 months ago'),

			




	
	
	
		
 
        (dict(years= -2, months= -4), u'2 years ago'),

			




	
	
	
		
 
        (dict(years= -2, months= -11), u'3 years ago'),

			




	
	
	
		
 
        (dict(years= -3, months= -2), u'3 years ago'),

			




	
	
	
		
 
        (dict(years= -4, months= -8), u'5 years ago'),

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_age_short(self, age_args, expected):

			




	
	
	
		
 
        from kallithea.lib.utils2 import age

			




	
	
	
		
 
        from dateutil import relativedelta

			




	
	
	
		
 
        n = datetime.datetime(year=2012, month=5, day=17)

			




	
	
	
		
 
        delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)

			




	
	
	
		
 
        self.assertEqual(age(n + delt(**age_args), show_short_version=True, now=n), expected)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
    @parametrize('age_args,expected', [

			




	
	
	
		
 
        (dict(), u'just now'),

			




	
	
	
		
 
        (dict(seconds=1), u'in 1 second'),

			




	
	
	
		
 
        (dict(seconds=60 * 2), u'in 2 minutes'),

			




	
	
	
		
 
        (dict(hours=1), u'in 1 hour'),

			




	
	
	
		
 
        (dict(hours=24), u'in 1 day'),

			




	
	
	
		
 
        (dict(hours=24 * 5), u'in 5 days'),

			




	
	
	
		
 
        (dict(months=1), u'in 1 month'),

			




	
	
	
		
 
        (dict(months=1, days=1), u'in 1 month and 1 day'),

			




	
	
	
		
 
        (dict(years=1, months=1), u'in 1 year and 1 month')

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_age_in_future(self, age_args, expected):

			




	
	
	
		
 
        from kallithea.lib.utils2 import age

			




	
	
	
		
 
        from dateutil import relativedelta

			




	
	
	
		
 
        n = datetime.datetime(year=2012, month=5, day=17)

			




	
	
	
		
 
        delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)

			




	
	
	
		
 
        self.assertEqual(age(n + delt(**age_args), now=n), expected)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_tag_extractor(self):

			




	
	
	
		
 
        sample = (

			




	
	
	
		
 
            "hello pta[tag] gog [[]] [[] sda ero[or]d [me =>>< sa]"

			




	
	
	
		
 
            "[requires] [stale] [see<>=>] [see => http://example.com]"

			




	
	
	
		
 
            "[requires => url] [lang => python] [just a tag]"

			




	
	
	
		
 
            "[,d] [ => ULR ] [obsolete] [desc]]"

			




	
	
	
		
 
        )

			




	
	
	
		
 
        from kallithea.lib.helpers import urlify_text

			




	
	
	
		
 
        res = urlify_text(sample, stylize=True)

			




	
	
	
		
 
        self.assertIn('<div class="metatag" tag="tag">tag</div>', res)

			




	
	
	
		
 
        self.assertIn('<div class="metatag" tag="obsolete">obsolete</div>', res)

			




	
	
	
		
 
        self.assertIn('<div class="metatag" tag="stale">stale</div>', res)

			




	
	
	
		
 
        self.assertIn('<div class="metatag" tag="lang">python</div>', res)

			




	
	
	
		
 
        self.assertIn('<div class="metatag" tag="requires">requires =&gt; <a href="/url">url</a></div>', res)

			




	
	
	
		
 
        self.assertIn('<div class="metatag" tag="tag">tag</div>', res)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_alternative_gravatar(self):

			




	
	
	
		
 
        from kallithea.lib.helpers import gravatar_url

			




	
	
	
		
 
        _md5 = lambda s: hashlib.md5(s).hexdigest()

			




	
	
	
		
 

			




	
	
	
		
 
        #mock pylons.url

			




	
	
	
		
 
        class fake_url(object):

			




	
	
	
		
 
            @classmethod

			




	
	
	
		
 
            def current(cls, *args, **kwargs):

			




	
	
	
		
 
                return 'https://example.com'

			




	
	
	
		
 

			




	
	
	
		
 
        #mock pylons.tmpl_context

			




	
	
	
		
 
        def fake_tmpl_context(_url):

			




	
	
	
		
 
            _c = AttributeDict()

			




	
	
	
		
 
            _c.visual = AttributeDict()

			




	
	
	
		
 
            _c.visual.use_gravatar = True

			




	
	
	
		
 
            _c.visual.gravatar_url = _url

			




	
	
	
		
 

			




	
	
	
		
 
            return _c

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
        with mock.patch('pylons.url', fake_url):

			




	
	
	
		
 
            fake = fake_tmpl_context(_url='http://example.com/{email}')

			




	
	
	
		
 
            with mock.patch('pylons.tmpl_context', fake):

			




	
	
	
		
 
                    from pylons import url

			




	
	
	
		
 
                    assert url.current() == 'https://example.com'

			




	
	
	
		
 
                    grav = gravatar_url(email_address='test@example.com', size=24)

			




	
	
	
		
 
                    assert grav == 'http://example.com/test@example.com'

			




	
	
	
		
 

			




	
	
	
		
 
            fake = fake_tmpl_context(_url='http://example.com/{email}')

			




	
	
	
		
 
            with mock.patch('pylons.tmpl_context', fake):

			




	
	
	
		
 
                grav = gravatar_url(email_address='test@example.com', size=24)

			




	
	
	
		
 
                assert grav == 'http://example.com/test@example.com'

			




	
	
	
		
 

			




	
	
	
		
 
            fake = fake_tmpl_context(_url='http://example.com/{md5email}')

			




	
	
	
		
 
            with mock.patch('pylons.tmpl_context', fake):

			




	
	
	
		
 
                em = 'test@example.com'

			




	
	
	
		
 
                grav = gravatar_url(email_address=em, size=24)

			




	
	
	
		
 
                assert grav == 'http://example.com/%s' % (_md5(em))

			




	
	
	
		
 

			




	
	
	
		
 
            fake = fake_tmpl_context(_url='http://example.com/{md5email}/{size}')

			




	
	
	
		
 
            with mock.patch('pylons.tmpl_context', fake):

			




	
	
	
		
 
                em = 'test@example.com'

			




	
	
	
		
 
                grav = gravatar_url(email_address=em, size=24)

			




	
	
	
		
 
                assert grav == 'http://example.com/%s/%s' % (_md5(em), 24)

			




	
	
	
		
 

			




	
	
	
		
 
            fake = fake_tmpl_context(_url='{scheme}://{netloc}/{md5email}/{size}')

			




	
	
	
		
 
            with mock.patch('pylons.tmpl_context', fake):

			




	
	
	
		
 
                em = 'test@example.com'

			




	
	
	
		
 
                grav = gravatar_url(email_address=em, size=24)

			




	
	
	
		
 
                assert grav == 'https://example.com/%s/%s' % (_md5(em), 24)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
    @parametrize('tmpl,repo_name,overrides,prefix,expected', [

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '', 'http://vps1:8000/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '/prefix', 'http://vps1:8000/prefix/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/prefix', 'http://user@vps1:8000/prefix/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '/prefix', 'http://username@vps1:8000/prefix/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/prefix/', 'http://user@vps1:8000/prefix/group/repo1'),

			




	
	
	
		
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '/prefix/', 'http://username@vps1:8000/prefix/group/repo1'),

			




	
	
	
		
 
        ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {}, '', 'http://vps1:8000/_23'),

			




	
	
	
		
 
        ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/_23'),

			




	
	
	
		
 
        ('http://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/_23'),

			




	
	
	
		
 
        ('http://{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://vps1:8000/_23'),

			




	
	
	
		
 
        ('https://{user}@proxy1.example.com/{repo}', 'group/repo1', {'user': 'username'}, '', 'https://username@proxy1.example.com/group/repo1'),

			




	
	
	
		
 
        ('https://{user}@proxy1.example.com/{repo}', 'group/repo1', {}, '', 'https://proxy1.example.com/group/repo1'),

			




	
	
	
		
 
        ('https://proxy1.example.com/{user}/{repo}', 'group/repo1', {'user': 'username'}, '', 'https://proxy1.example.com/username/group/repo1'),

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_clone_url_generator(self, tmpl, repo_name, overrides, prefix, expected):

			




	
	
	
		
 
        from kallithea.lib.utils2 import get_clone_url

			




	
	
	
		
 
        clone_url = get_clone_url(uri_tmpl=tmpl, qualified_home_url='http://vps1:8000'+prefix,

			




	
	
	
		
 
                                  repo_name=repo_name, repo_id=23, **overrides)

			




	
	
	
		
 
        self.assertEqual(clone_url, expected)

			




	
	
	
		
 

			




	
	
	
		
 
    def _quick_url(self, text, tmpl="""<a class="revision-link" href="%s">%s</a>""", url_=None):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Changes `some text url[foo]` => `some text <a href="/">foo</a>

			




	
	
	
		
 

			




	
	
	
		
 
        :param text:

			




	
	
	
		
 
        """

			




	
	
	
		
 
        import re

			




	
	
	
		
 
        # quickly change expected url[] into a link

			




	
	
	
		
 
        URL_PAT = re.compile(r'(?:url\[)(.+?)(?:\])')

			




	
	
	
		
 

			




	
	
	
		
 
        def url_func(match_obj):

			




	
	
	
		
 
            _url = match_obj.groups()[0]

			




	
	
	
		
 
            return tmpl % (url_ or '/some-url', _url)

			




	
	
	
		
 
        return URL_PAT.sub(url_func, text)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
    @parametrize('sample,expected', [

			




	
	
	
		
 
      ("",

			




	
	
	
		
 
       ""),

			




	
	
	
		
 
      ("git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68",

			




	
	
	
		
 
       "git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68"),

			




	
	
	
		
 
      ("from rev 000000000000",

			




	
	
	
		
 
       "from rev url[000000000000]"),

			




	
	
	
		
 
      ("from rev 000000000000123123 also rev 000000000000",

			




	
	
	
		
 
       "from rev url[000000000000123123] also rev url[000000000000]"),

			




	
	
	
		
 
      ("this should-000 00",

			




	
	
	
		
 
       "this should-000 00"),

			




	
	
	
		
 
      ("longtextffffffffff rev 123123123123",

			




	
	
	
		
 
       "longtextffffffffff rev url[123123123123]"),

			




	
	
	
		
 
      ("rev ffffffffffffffffffffffffffffffffffffffffffffffffff",

			




	
	
	
		
 
       "rev ffffffffffffffffffffffffffffffffffffffffffffffffff"),

			




	
	
	
		
 
      ("ffffffffffff some text traalaa",

			




	
	
	
		
 
       "url[ffffffffffff] some text traalaa"),

			




	
	
	
		
 
       ("""Multi line

			




	
	
	
		
 
       123123123123

			




	
	
	
		
 
       some text 123123123123

			




	
	
	
		
 
       sometimes !

			




	
	
	
		
 
       """,

			




	
	
	
		
 
       """Multi line

			




	
	
	
		
 
       url[123123123123]

			




	
	
	
		
 
       some text url[123123123123]

			




	
	
	
		
 
       sometimes !

			




	
	
	
		
 
       """)

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_urlify_changesets(self, sample, expected):

			




	
	
	
		
 
        def fake_url(self, *args, **kwargs):

			




	
	
	
		
 
            return '/some-url'

			




	
	
	
		
 

			




	
	
	
		
 
        expected = self._quick_url(expected)

			




	
	
	
		
 

			




	
	
	
		
 
        with mock.patch('pylons.url', fake_url):

			




	
	
	
		
 
            from kallithea.lib.helpers import urlify_changesets

			




	
	
	
		
 
            self.assertEqual(urlify_changesets(sample, 'repo_name'), expected)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
    @parametrize('sample,expected,url_', [

			




	
	
	
		
 
      ("",

			




	
	
	
		
 
       "",

			




	
	
	
		
 
       ""),

			




	
	
	
		
 
      ("https://svn.apache.org/repos",

			




	
	
	
		
 
       "url[https://svn.apache.org/repos]",

			




	
	
	
		
 
       "https://svn.apache.org/repos"),

			




	
	
	
		
 
      ("http://svn.apache.org/repos",

			




	
	
	
		
 
       "url[http://svn.apache.org/repos]",

			




	
	
	
		
 
       "http://svn.apache.org/repos"),

			




	
	
	
		
 
      ("from rev a also rev http://google.com",

			




	
	
	
		
 
       "from rev a also rev url[http://google.com]",

			




	
	
	
		
 
       "http://google.com"),

			




	
	
	
		
 
       ("""Multi line

			




	
	
	
		
 
       https://foo.bar.example.com

			




	
	
	
		
 
       some text lalala""",

			




	
	
	
		
 
       """Multi line

			




	
	
	
		
 
       url[https://foo.bar.example.com]

			




	
	
	
		
 
       some text lalala""",

			




	
	
	
		
 
       "https://foo.bar.example.com")

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_urlify_test(self, sample, expected, url_):

			




	
	
	
		
 
        from kallithea.lib.helpers import urlify_text

			




	
	
	
		
 
        expected = self._quick_url(expected,

			




	
	
	
		
 
                                   tmpl="""<a href="%s">%s</a>""", url_=url_)

			




	
	
	
		
 
        self.assertEqual(urlify_text(sample), expected)

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
    @parametrize('test,expected', [

			




	
	
	
		
 
      ("", None),

			




	
	
	
		
 
      ("/_2", '2'),

			




	
	
	
		
 
      ("_2", '2'),

			




	
	
	
		
 
      ("/_2/", '2'),

			




	
	
	
		
 
      ("_2/", '2'),

			




	
	
	
		
 

			




	
	
	
		
 
      ("/_21", '21'),

			




	
	
	
		
 
      ("_21", '21'),

			




	
	
	
		
 
      ("/_21/", '21'),

			




	
	
	
		
 
      ("_21/", '21'),

			




	
	
	
		
 

			




	
	
	
		
 
      ("/_21/foobar", '21'),

			




	
	
	
		
 
      ("_21/121", '21'),

			




	
	
	
		
 
      ("/_21/_12", '21'),

			




	
	
	
		
 
      ("_21/prefix/foo", '21'),

			




	
	
	
		
 
    ])

			




	
	
	
		
 
    def test_get_repo_by_id(self, test, expected):

			




	
	
	
		
 
        from kallithea.lib.utils import _extract_id_from_repo_name

			




	
	
	
		
 
        _test = _extract_id_from_repo_name(test)

			




	
	
	
		
 
        self.assertEqual(_test, expected, msg='url:%s, got:`%s` expected: `%s`'

			




	
	
	
		
 
                                              % (test, _test, expected))

			




        

    


    
    

    

        

                

                    kallithea/tests/other/test_mail.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 
import mock

			




	
	
	
		
 

			




	
	
	
		
 
import kallithea

			




	
	
	
		
 
from kallithea.tests import *

			




	
	
	
		
 
from kallithea.model.db import User

			




	
	
	
		
 

			




	
	
	
		
 
class smtplib_mock(object):

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def SMTP(cls, server, port, local_hostname):

			




	
	
	
		
 
        return smtplib_mock()

			




	
	
	
		
 

			




	
	
	
		
 
    def ehlo(self):

			




	
	
	
		
 
        pass

			




	
	
	
		
 
    def quit(self):

			




	
	
	
		
 
        pass

			




	
	
	
		
 
    def sendmail(self, sender, dest, msg):

			




	
	
	
		
 
        smtplib_mock.lastsender = sender

			




	
	
	
		
 
        smtplib_mock.lastdest = dest

			




	
	
	
		
 
        smtplib_mock.lastmsg = msg

			




	
	
	
		
 
        pass

			




	
	
	
		
 

			




	
	
	
		
 
@mock.patch('kallithea.lib.rcmail.smtp_mailer.smtplib', smtplib_mock)

			




	
	
	
		
 
class TestMail(BaseTestCase):

			




	
	
	
		
 
class TestMail(TestControllerPytest):

			




	
	
	
		
 

			




	
	
	
		
 
    def test_send_mail_trivial(self):

			




	
	
	
		
 
        mailserver = 'smtp.mailserver.org'

			




	
	
	
		
 
        recipients = ['rcpt1', 'rcpt2']

			




	
	
	
		
 
        envelope_from = 'noreply@mailserver.org'

			




	
	
	
		
 
        subject = 'subject'

			




	
	
	
		
 
        body = 'body'

			




	
	
	
		
 
        html_body = 'html_body'

			




	
	
	
		
 

			




	
	
	
		
 
        config_mock = {

			




	
	
	
		
 
            'smtp_server': mailserver,

			




	
	
	
		
 
            'app_email_from': envelope_from,

			




	
	
	
		
 
        }

			




	
	
	
		
 
        with mock.patch('kallithea.lib.celerylib.tasks.config', config_mock):

			




	
	
	
		
 
            kallithea.lib.celerylib.tasks.send_email(recipients, subject, body, html_body)

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertSetEqual(smtplib_mock.lastdest, set(recipients))

			




	
	
	
		
 
        self.assertEqual(smtplib_mock.lastsender, envelope_from)

			




	
	
	
		
 
        self.assertIn('From: %s' % envelope_from, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn('Subject: %s' % subject, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn(body, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn(html_body, smtplib_mock.lastmsg)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_send_mail_no_recipients(self):

			




	
	
	
		
 
        mailserver = 'smtp.mailserver.org'

			




	
	
	
		
 
        recipients = []

			




	
	
	
		
 
        envelope_from = 'noreply@mailserver.org'

			




	
	
	
		
 
        email_to = 'admin@mailserver.org'

			




	
	
	
		
 
        subject = 'subject'

			




	
	
	
		
 
        body = 'body'

			




	
	
	
		
 
        html_body = 'html_body'

			




	
	
	
		
 

			




	
	
	
		
 
        config_mock = {

			




	
	
	
		
 
            'smtp_server': mailserver,

			




	
	
	
		
 
            'app_email_from': envelope_from,

			




	
	
	
		
 
            'email_to': email_to,

			




	
	
	
		
 
        }

			




	
	
	
		
 
        with mock.patch('kallithea.lib.celerylib.tasks.config', config_mock):

			




	
	
	
		
 
            kallithea.lib.celerylib.tasks.send_email(recipients, subject, body, html_body)

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertSetEqual(smtplib_mock.lastdest, set([TEST_USER_ADMIN_EMAIL, email_to]))

			




	
	
	
		
 
        self.assertEqual(smtplib_mock.lastsender, envelope_from)

			




	
	
	
		
 
        self.assertIn('From: %s' % envelope_from, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn('Subject: %s' % subject, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn(body, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn(html_body, smtplib_mock.lastmsg)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_send_mail_no_recipients_no_email_to(self):

			




	
	
	
		
 
        mailserver = 'smtp.mailserver.org'

			




	
	
	
		
 
        recipients = []

			




	
	
	
		
 
        envelope_from = 'noreply@mailserver.org'

			




	
	
	
		
 
        subject = 'subject'

			




	
	
	
		
 
        body = 'body'

			




	
	
	
		
 
        html_body = 'html_body'

			




	
	
	
		
 

			




	
	
	
		
 
        config_mock = {

			




	
	
	
		
 
            'smtp_server': mailserver,

			




	
	
	
		
 
            'app_email_from': envelope_from,

			




	
	
	
		
 
        }

			




	
	
	
		
 
        with mock.patch('kallithea.lib.celerylib.tasks.config', config_mock):

			




	
	
	
		
 
            kallithea.lib.celerylib.tasks.send_email(recipients, subject, body, html_body)

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertSetEqual(smtplib_mock.lastdest, set([TEST_USER_ADMIN_EMAIL]))

			




	
	
	
		
 
        self.assertEqual(smtplib_mock.lastsender, envelope_from)

			




	
	
	
		
 
        self.assertIn('From: %s' % envelope_from, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn('Subject: %s' % subject, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn(body, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn(html_body, smtplib_mock.lastmsg)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_send_mail_with_author(self):

			




	
	
	
		
 
        mailserver = 'smtp.mailserver.org'

			




	
	
	
		
 
        recipients = ['rcpt1', 'rcpt2']

			




	
	
	
		
 
        envelope_from = 'noreply@mailserver.org'

			




	
	
	
		
 
        subject = 'subject'

			




	
	
	
		
 
        body = 'body'

			




	
	
	
		
 
        html_body = 'html_body'

			




	
	
	
		
 
        author = User.get_by_username(TEST_USER_REGULAR_LOGIN)

			




	
	
	
		
 

			




	
	
	
		
 
        config_mock = {

			




	
	
	
		
 
            'smtp_server': mailserver,

			




	
	
	
		
 
            'app_email_from': envelope_from,

			




	
	
	
		
 
        }

			




	
	
	
		
 
        with mock.patch('kallithea.lib.celerylib.tasks.config', config_mock):

			




	
	
	
		
 
            kallithea.lib.celerylib.tasks.send_email(recipients, subject, body, html_body, author=author)

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertSetEqual(smtplib_mock.lastdest, set(recipients))

			




	
	
	
		
 
        self.assertEqual(smtplib_mock.lastsender, envelope_from)

			




	
	
	
		
 
        self.assertIn('From: "Kallithea Admin (no-reply)" <%s>' % envelope_from, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn('Subject: %s' % subject, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn(body, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn(html_body, smtplib_mock.lastmsg)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_send_mail_with_author_full_mail_from(self):

			




	
	
	
		
 
        mailserver = 'smtp.mailserver.org'

			




	
	
	
		
 
        recipients = ['rcpt1', 'rcpt2']

			




	
	
	
		
 
        envelope_addr = 'noreply@mailserver.org'

			




	
	
	
		
 
        envelope_from = 'Some Name <%s>' % envelope_addr

			




	
	
	
		
 
        subject = 'subject'

			




	
	
	
		
 
        body = 'body'

			




	
	
	
		
 
        html_body = 'html_body'

			




	
	
	
		
 
        author = User.get_by_username(TEST_USER_REGULAR_LOGIN)

			




	
	
	
		
 

			




	
	
	
		
 
        config_mock = {

			




	
	
	
		
 
            'smtp_server': mailserver,

			




	
	
	
		
 
            'app_email_from': envelope_from,

			




	
	
	
		
 
        }

			




	
	
	
		
 
        with mock.patch('kallithea.lib.celerylib.tasks.config', config_mock):

			




	
	
	
		
 
            kallithea.lib.celerylib.tasks.send_email(recipients, subject, body, html_body, author=author)

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertSetEqual(smtplib_mock.lastdest, set(recipients))

			




	
	
	
		
 
        self.assertEqual(smtplib_mock.lastsender, envelope_from)

			




	
	
	
		
 
        self.assertIn('From: "Kallithea Admin (no-reply)" <%s>' % envelope_addr, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn('Subject: %s' % subject, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn(body, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn(html_body, smtplib_mock.lastmsg)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_send_mail_extra_headers(self):

			




	
	
	
		
 
        mailserver = 'smtp.mailserver.org'

			




	
	
	
		
 
        recipients = ['rcpt1', 'rcpt2']

			




	
	
	
		
 
        envelope_from = 'noreply@mailserver.org'

			




	
	
	
		
 
        subject = 'subject'

			




	
	
	
		
 
        body = 'body'

			




	
	
	
		
 
        html_body = 'html_body'

			




	
	
	
		
 
        author = User(name='foo', lastname=u'(fubar) "baz"')

			




	
	
	
		
 
        headers = {'extra': 'yes'}

			




	
	
	
		
 

			




	
	
	
		
 
        config_mock = {

			




	
	
	
		
 
            'smtp_server': mailserver,

			




	
	
	
		
 
            'app_email_from': envelope_from,

			




	
	
	
		
 
        }

			




	
	
	
		
 
        with mock.patch('kallithea.lib.celerylib.tasks.config', config_mock):

			




	
	
	
		
 
            kallithea.lib.celerylib.tasks.send_email(recipients, subject, body, html_body,

			




	
	
	
		
 
                                                     author=author, headers=headers)

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertSetEqual(smtplib_mock.lastdest, set(recipients))

			




	
	
	
		
 
        self.assertEqual(smtplib_mock.lastsender, envelope_from)

			




	
	
	
		
 
        self.assertIn(r'From: "foo (fubar) \"baz\" (no-reply)" <%s>' % envelope_from, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn('Subject: %s' % subject, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn(body, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn(html_body, smtplib_mock.lastmsg)

			




	
	
	
		
 
        self.assertIn('Extra: yes', smtplib_mock.lastmsg)

			




	
	
	
		
 
        # verify that headers dict hasn't mutated by send_email

			




	
	
	
		
 
        self.assertDictEqual(headers, {'extra': 'yes'})

			




        

    


    
    

    

        

                

                    kallithea/tests/other/test_validators.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



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

			




	
	
	
		
 
import formencode

			




	
	
	
		
 
import tempfile

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.tests import *

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.model import validators as v

			




	
	
	
		
 
from kallithea.model.user_group import UserGroupModel

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.model.meta import Session

			




	
	
	
		
 
from kallithea.model.repo_group import RepoGroupModel

			




	
	
	
		
 
from kallithea.tests.fixture import Fixture

			




	
	
	
		
 

			




	
	
	
		
 
fixture = Fixture()

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TestRepoGroups(BaseTestCase):

			




	
	
	
		
 
class TestRepoGroups(TestControllerPytest):

			




	
	
	
		
 

			




	
	
	
		
 
    def setUp(self):

			




	
	
	
		
 
        pass

			




	
	
	
		
 

			




	
	
	
		
 
    def tearDown(self):

			




	
	
	
		
 
    def teardown_method(self, method):

			




	
	
	
		
 
        Session.remove()

			




	
	
	
		
 

			




	
	
	
		
 
    def test_Message_extractor(self):

			




	
	
	
		
 
        validator = v.ValidUsername()

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid, validator.to_python, 'default')

			




	
	
	
		
 

			




	
	
	
		
 
        class StateObj(object):

			




	
	
	
		
 
            pass

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid,

			




	
	
	
		
 
                          validator.to_python, 'default', StateObj)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidUsername(self):

			




	
	
	
		
 
        validator = v.ValidUsername()

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid, validator.to_python, 'default')

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid, validator.to_python, 'new_user')

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid, validator.to_python, '.,')

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid, validator.to_python,

			




	
	
	
		
 
                          TEST_USER_ADMIN_LOGIN)

			




	
	
	
		
 
        self.assertEqual('test', validator.to_python('test'))

			




	
	
	
		
 

			




	
	
	
		
 
        validator = v.ValidUsername(edit=True, old_data={'user_id': 1})

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidRepoUser(self):

			




	
	
	
		
 
        validator = v.ValidRepoUser()

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid, validator.to_python, 'nouser')

			




	
	
	
		
 
        self.assertEqual(TEST_USER_ADMIN_LOGIN,

			




	
	
	
		
 
                         validator.to_python(TEST_USER_ADMIN_LOGIN))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidUserGroup(self):

			




	
	
	
		
 
        validator = v.ValidUserGroup()

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid, validator.to_python, u'default')

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid, validator.to_python, u'.,')

			




	
	
	
		
 

			




	
	
	
		
 
        gr = fixture.create_user_group(u'test')

			




	
	
	
		
 
        gr2 = fixture.create_user_group(u'tes2')

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid, validator.to_python, u'test')

			




	
	
	
		
 
        assert gr.users_group_id is not None

			




	
	
	
		
 
        validator = v.ValidUserGroup(edit=True,

			




	
	
	
		
 
                                    old_data={'users_group_id':

			




	
	
	
		
 
                                              gr2.users_group_id})

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid, validator.to_python, u'test')

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid, validator.to_python, u'TesT')

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid, validator.to_python, u'TEST')

			




	
	
	
		
 
        UserGroupModel().delete(gr)

			




	
	
	
		
 
        UserGroupModel().delete(gr2)

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidRepoGroup(self):

			




	
	
	
		
 
        validator = v.ValidRepoGroup()

			




	
	
	
		
 
        model = RepoGroupModel()

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid, validator.to_python,

			




	
	
	
		
 
                          {'group_name': HG_REPO, })

			




	
	
	
		
 
        gr = model.create(group_name=u'test_gr', group_description=u'desc',

			




	
	
	
		
 
                          parent=None,

			




	
	
	
		
 
                          just_db=True,

			




	
	
	
		
 
                          owner=TEST_USER_ADMIN_LOGIN)

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid,

			




	
	
	
		
 
                          validator.to_python, {'group_name': gr.group_name, })

			




	
	
	
		
 

			




	
	
	
		
 
        validator = v.ValidRepoGroup(edit=True,

			




	
	
	
		
 
                                      old_data={'group_id':  gr.group_id})

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid,

			




	
	
	
		
 
                          validator.to_python, {

			




	
	
	
		
 
                                        'group_name': gr.group_name + 'n',

			




	
	
	
		
 
                                        'group_parent_id': gr.group_id

			




	
	
	
		
 
                                        })

			




	
	
	
		
 
        model.delete(gr)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidPassword(self):

			




	
	
	
		
 
        validator = v.ValidPassword()

			




	
	
	
		
 
        self.assertEqual('lol', validator.to_python('lol'))

			




	
	
	
		
 
        self.assertEqual(None, validator.to_python(None))

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid, validator.to_python, 'ąćżź')

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidPasswordsMatch(self):

			




	
	
	
		
 
        validator = v.ValidPasswordsMatch('new_password', 'password_confirmation')

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid,

			




	
	
	
		
 
                    validator.to_python, {'new_password': 'pass',

			




	
	
	
		
 
                                          'password_confirmation': 'pass2'})

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid,

			




	
	
	
		
 
                    validator.to_python, {'new_password': 'pass',

			




	
	
	
		
 
                                          'password_confirmation': 'pass2'})

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertEqual({'new_password': 'pass',

			




	
	
	
		
 
                          'password_confirmation': 'pass'},

			




	
	
	
		
 
                    validator.to_python({'new_password': 'pass',

			




	
	
	
		
 
                                         'password_confirmation': 'pass'}))

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertEqual({'new_password': 'pass',

			




	
	
	
		
 
                          'password_confirmation': 'pass'},

			




	
	
	
		
 
                    validator.to_python({'new_password': 'pass',

			




	
	
	
		
 
                                         'password_confirmation': 'pass'}))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidAuth(self):

			




	
	
	
		
 
        validator = v.ValidAuth()

			




	
	
	
		
 
        valid_creds = {

			




	
	
	
		
 
            'username': TEST_USER_REGULAR2_LOGIN,

			




	
	
	
		
 
            'password': TEST_USER_REGULAR2_PASS,

			




	
	
	
		
 
        }

			




	
	
	
		
 
        invalid_creds = {

			




	
	
	
		
 
            'username': 'err',

			




	
	
	
		
 
            'password': 'err',

			




	
	
	
		
 
        }

			




	
	
	
		
 
        self.assertEqual(valid_creds, validator.to_python(valid_creds))

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid,

			




	
	
	
		
 
                          validator.to_python, invalid_creds)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidAuthToken(self):

			




	
	
	
		
 
        validator = v.ValidAuthToken()

			




	
	
	
		
 
        # this is untestable without a threadlocal

			




	
	
	
		
 
#        self.assertRaises(formencode.Invalid,

			




	
	
	
		
 
#                          validator.to_python, 'BadToken')

			




	
	
	
		
 
        validator

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidRepoName(self):

			




	
	
	
		
 
        validator = v.ValidRepoName()

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid,

			




	
	
	
		
 
                          validator.to_python, {'repo_name': ''})

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid,

			




	
	
	
		
 
                          validator.to_python, {'repo_name': HG_REPO})

			




	
	
	
		
 

			




	
	
	
		
 
        gr = RepoGroupModel().create(group_name=u'group_test',

			




	
	
	
		
 
                                      group_description=u'desc',

			




	
	
	
		
 
                                      parent=None,

			




	
	
	
		
 
                                      owner=TEST_USER_ADMIN_LOGIN)

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid,

			




	
	
	
		
 
                          validator.to_python, {'repo_name': gr.group_name})

			




	
	
	
		
 

			




	
	
	
		
 
        #TODO: write an error case for that ie. create a repo withinh a group

			




	
	
	
		
 
#        self.assertRaises(formencode.Invalid,

			




	
	
	
		
 
#                          validator.to_python, {'repo_name': 'some',

			




	
	
	
		
 
#                                                'repo_group': gr.group_id})

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidForkName(self):

			




	
	
	
		
 
        # this uses ValidRepoName validator

			




	
	
	
		
 
        assert True

			




	
	
	
		
 

			




	
	
	
		
 
    @parameterized.expand([

			




	
	
	
		
 
    @parametrize('name,expected', [

			




	
	
	
		
 
        ('test', 'test'), ('lolz!', 'lolz'), ('  aavv', 'aavv'),

			




	
	
	
		
 
        ('ala ma kota', 'ala-ma-kota'), ('@nooo', 'nooo'),

			




	
	
	
		
 
        ('$!haha lolz !', 'haha-lolz'), ('$$$$$', ''), ('{}OK!', 'OK'),

			




	
	
	
		
 
        ('/]re po', 're-po')])

			




	
	
	
		
 
    def test_SlugifyName(self, name, expected):

			




	
	
	
		
 
        validator = v.SlugifyName()

			




	
	
	
		
 
        self.assertEqual(expected, validator.to_python(name))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidCloneUri(self):

			




	
	
	
		
 
            #TODO: write this one

			




	
	
	
		
 
            pass

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidForkType(self):

			




	
	
	
		
 
            validator = v.ValidForkType(old_data={'repo_type': 'hg'})

			




	
	
	
		
 
            self.assertEqual('hg', validator.to_python('hg'))

			




	
	
	
		
 
            self.assertRaises(formencode.Invalid, validator.to_python, 'git')

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidPerms(self):

			




	
	
	
		
 
            #TODO: write this one

			




	
	
	
		
 
            pass

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidSettings(self):

			




	
	
	
		
 
        validator = v.ValidSettings()

			




	
	
	
		
 
        self.assertEqual({'pass': 'pass'},

			




	
	
	
		
 
                         validator.to_python(value={'user': 'test',

			




	
	
	
		
 
                                                    'pass': 'pass'}))

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertEqual({'user2': 'test', 'pass': 'pass'},

			




	
	
	
		
 
                         validator.to_python(value={'user2': 'test',

			




	
	
	
		
 
                                                    'pass': 'pass'}))

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidPath(self):

			




	
	
	
		
 
            validator = v.ValidPath()

			




	
	
	
		
 
            self.assertEqual(tempfile.gettempdir(),

			




	
	
	
		
 
                             validator.to_python(tempfile.gettempdir()))

			




	
	
	
		
 
            self.assertRaises(formencode.Invalid, validator.to_python,

			




	
	
	
		
 
                              '/no_such_dir')

			




	
	
	
		
 

			




	
	
	
		
 
    def test_UniqSystemEmail(self):

			




	
	
	
		
 
        validator = v.UniqSystemEmail(old_data={})

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertEqual('mail@python.org',

			




	
	
	
		
 
                         validator.to_python('MaiL@Python.org'))

			




	
	
	
		
 

			




	
	
	
		
 
        email = TEST_USER_REGULAR2_EMAIL

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid, validator.to_python, email)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_ValidSystemEmail(self):

			




	
	
	
		
 
        validator = v.ValidSystemEmail()

			




	
	
	
		
 
        email = TEST_USER_REGULAR2_EMAIL

			




	
	
	
		
 

			




	
	
	
		
 
        self.assertEqual(email, validator.to_python(email))

			




	
	
	
		
 
        self.assertRaises(formencode.Invalid, validator.to_python, 'err')

			




	
	
	
		
 

			




	
	
	
		
 
    def test_LdapLibValidator(self):

			




	
	
	
		
 
        if ldap_lib_installed:

			




	
	
	
		
 
            validator = v.LdapLibValidator()

			




	
	
	
		
 
            self.assertEqual("DN", validator.to_python('DN'))

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            validator = v.LdapLibValidator()

			




	
	
	
		
 
            self.assertRaises(v.LdapImportError, validator.to_python, 'err')

			




	
	
	
		
 

			




	
	
	
		
 
    def test_AttrLoginValidator(self):

			




	
	
	
		
 
        validator = v.AttrLoginValidator()

			




	
	
	
		
 
        self.assertEqual('DN_attr', validator.to_python('DN_attr'))

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








    
        
    
    
        This site is powered by
            Kallithea 0.7.0,
        which is
        © 2010–2025 by various authors & licensed under GPLv3.