Changeset - 8d98924c58b1
[Not reviewed]
default
0 53 1
Thomas De Schampheleire - 9 years ago 2016-09-18 21:44:21
thomas.de.schampheleire@gmail.com
tests: add as little code as possible in __init__.py

kallithea/tests/__init__.py contained quite a lot of code, including the test
base class TestController. This in itself may be considered bad practice.

Specifically, this poses a problem when using pytest 3.0+, in which asserts
in some files are not automatically rewritten to give improved assert
output. That problem can be fixed by explicitly registering such files for
assertion rewriting, but that register call should be executed _before_ said
files are imported. I.e. if the register call is in
kallithea/tests/__init__.py, assert calls in __init__.py itself can not be
rewritten.

Since the TestController base class does effectively contain asserts, and we
do not want to execute the register call from somewhere outside the
kallithea/tests directory, we need to move the TestController class to
another file (kallithea/tests/base.py) so we can have a register call in
__init__.py before loading base.py.

While not strictly necessary to fix the mentioned pytest problem, we take
the opportunity to fully clean __init__.py and move everything to
the new kallithea/tests/base.py. While doing so, unnecessary imports are
removed, and imports are ordered alphabetically. Explicit imports of symbols
from modules that were already imported as a whole, are removed in favor of
fully qualifying the references (e.g. tempfile._RandomNameSequence).
54 files changed with 58 insertions and 280 deletions:
0 comments (0 inline, 0 general)
kallithea/config/environment.py
Show inline comments
 
@@ -51,89 +51,89 @@ def load_environment(global_conf, app_co
 

	
 
    # Pylons paths
 
    root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 
    paths = dict(
 
        root=root,
 
        controllers=os.path.join(root, 'controllers'),
 
        static_files=os.path.join(root, 'public'),
 
        templates=[os.path.join(root, 'templates')]
 
    )
 

	
 
    # Initialize config with the basic options
 
    config.init_app(global_conf, app_conf, package='kallithea', paths=paths)
 

	
 
    # store some globals into kallithea
 
    kallithea.CELERY_ON = str2bool(config['app_conf'].get('use_celery'))
 
    kallithea.CELERY_EAGER = str2bool(config['app_conf'].get('celery.always.eager'))
 

	
 
    config['routes.map'] = make_map(config)
 
    config['pylons.app_globals'] = app_globals.Globals(config)
 
    config['pylons.h'] = helpers
 
    kallithea.CONFIG = config
 

	
 
    load_rcextensions(root_path=config['here'])
 

	
 
    # Setup cache object as early as possible
 
    pylons.cache._push_object(config['pylons.app_globals'].cache)
 

	
 
    # Create the Mako TemplateLookup, with the default auto-escaping
 
    config['pylons.app_globals'].mako_lookup = mako.lookup.TemplateLookup(
 
        directories=paths['templates'],
 
        strict_undefined=True,
 
        module_directory=os.path.join(app_conf['cache_dir'], 'templates'),
 
        input_encoding='utf-8', default_filters=['escape'],
 
        imports=['from webhelpers.html import escape'])
 

	
 
    # sets the c attribute access when don't existing attribute are accessed
 
    config['pylons.strict_tmpl_context'] = True
 
    test = os.path.split(config['__file__'])[-1] == 'test.ini'
 
    if test:
 
        if test_env is None:
 
            test_env = not int(os.environ.get('KALLITHEA_NO_TMP_PATH', 0))
 
        if test_index is None:
 
            test_index = not int(os.environ.get('KALLITHEA_WHOOSH_TEST_DISABLE', 0))
 
        if os.environ.get('TEST_DB'):
 
            # swap config if we pass enviroment variable
 
            config['sqlalchemy.url'] = os.environ.get('TEST_DB')
 

	
 
        from kallithea.lib.utils import create_test_env, create_test_index
 
        from kallithea.tests import TESTS_TMP_PATH
 
        from kallithea.tests.base import TESTS_TMP_PATH
 
        #set KALLITHEA_NO_TMP_PATH=1 to disable re-creating the database and
 
        #test repos
 
        if test_env:
 
            create_test_env(TESTS_TMP_PATH, config)
 
        #set KALLITHEA_WHOOSH_TEST_DISABLE=1 to disable whoosh index during tests
 
        if test_index:
 
            create_test_index(TESTS_TMP_PATH, config, True)
 

	
 
    # MULTIPLE DB configs
 
    # Setup the SQLAlchemy database engine
 
    sa_engine = engine_from_config(config, 'sqlalchemy.')
 
    init_model(sa_engine)
 

	
 
    set_available_permissions(config)
 
    repos_path = make_ui('db').configitems('paths')[0][1]
 
    config['base_path'] = repos_path
 
    set_app_settings(config)
 

	
 
    instance_id = kallithea.CONFIG.get('instance_id', '*')
 
    if instance_id == '*':
 
        instance_id = '%s-%s' % (platform.uname()[1], os.getpid())
 
        kallithea.CONFIG['instance_id'] = instance_id
 

	
 
    # CONFIGURATION OPTIONS HERE (note: all config options will override
 
    # any Pylons config options)
 

	
 
    # store config reference into our module to skip import magic of
 
    # pylons
 
    kallithea.CONFIG.update(config)
 
    set_vcs_config(kallithea.CONFIG)
 
    set_indexer_config(kallithea.CONFIG)
 

	
 
    #check git version
 
    check_git_version()
 

	
 
    if str2bool(config.get('initial_repo_scan', True)):
 
        repo2db_mapper(ScmModel().repo_scan(repos_path),
 
                       remove_obsolete=False, install_git_hooks=False)
 
    formencode.api.set_stdtranslation(languages=[config.get('lang')])
 
    return config
kallithea/lib/db_manage.py
Show inline comments
 
@@ -151,97 +151,97 @@ class DbManage(object):
 
        self.sa.add(def_user)
 
        self.sa.commit()
 

	
 
    def fix_settings(self):
 
        """
 
        Fixes kallithea settings adds ga_code key for google analytics
 
        """
 

	
 
        hgsettings3 = Setting('ga_code', '')
 

	
 
        self.sa.add(hgsettings3)
 
        self.sa.commit()
 

	
 
    def admin_prompt(self, second=False):
 
        if not self.tests:
 
            import getpass
 

	
 
            username = self.cli_args.get('username')
 
            password = self.cli_args.get('password')
 
            email = self.cli_args.get('email')
 

	
 
            def get_password():
 
                password = getpass.getpass('Specify admin password '
 
                                           '(min 6 chars):')
 
                confirm = getpass.getpass('Confirm password:')
 

	
 
                if password != confirm:
 
                    log.error('passwords mismatch')
 
                    return False
 
                if len(password) < 6:
 
                    log.error('password is to short use at least 6 characters')
 
                    return False
 

	
 
                return password
 
            if username is None:
 
                username = raw_input('Specify admin username:')
 
            if password is None:
 
                password = get_password()
 
                if not password:
 
                    #second try
 
                    password = get_password()
 
                    if not password:
 
                        sys.exit()
 
            if email is None:
 
                email = raw_input('Specify admin email:')
 
            self.create_user(username, password, email, True)
 
        else:
 
            log.info('creating admin and regular test users')
 
            from kallithea.tests import TEST_USER_ADMIN_LOGIN, \
 
            from kallithea.tests.base import TEST_USER_ADMIN_LOGIN, \
 
            TEST_USER_ADMIN_PASS, TEST_USER_ADMIN_EMAIL, \
 
            TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS, \
 
            TEST_USER_REGULAR_EMAIL, TEST_USER_REGULAR2_LOGIN, \
 
            TEST_USER_REGULAR2_PASS, TEST_USER_REGULAR2_EMAIL
 

	
 
            self.create_user(TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS,
 
                             TEST_USER_ADMIN_EMAIL, True)
 

	
 
            self.create_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS,
 
                             TEST_USER_REGULAR_EMAIL, False)
 

	
 
            self.create_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS,
 
                             TEST_USER_REGULAR2_EMAIL, False)
 

	
 
    def create_ui_settings(self, repo_store_path):
 
        """
 
        Creates ui settings, fills out hooks
 
        """
 

	
 
        #HOOKS
 
        hooks1_key = Ui.HOOK_UPDATE
 
        hooks1_ = self.sa.query(Ui) \
 
            .filter(Ui.ui_key == hooks1_key).scalar()
 

	
 
        hooks1 = Ui() if hooks1_ is None else hooks1_
 
        hooks1.ui_section = 'hooks'
 
        hooks1.ui_key = hooks1_key
 
        hooks1.ui_value = 'hg update >&2'
 
        hooks1.ui_active = False
 
        self.sa.add(hooks1)
 

	
 
        hooks2_key = Ui.HOOK_REPO_SIZE
 
        hooks2_ = self.sa.query(Ui) \
 
            .filter(Ui.ui_key == hooks2_key).scalar()
 
        hooks2 = Ui() if hooks2_ is None else hooks2_
 
        hooks2.ui_section = 'hooks'
 
        hooks2.ui_key = hooks2_key
 
        hooks2.ui_value = 'python:kallithea.lib.hooks.repo_size'
 
        self.sa.add(hooks2)
 

	
 
        hooks3 = Ui()
 
        hooks3.ui_section = 'hooks'
 
        hooks3.ui_key = Ui.HOOK_PUSH
 
        hooks3.ui_value = 'python:kallithea.lib.hooks.log_push_action'
 
        self.sa.add(hooks3)
 

	
 
        hooks4 = Ui()
 
        hooks4.ui_section = 'hooks'
kallithea/lib/utils.py
Show inline comments
 
@@ -584,97 +584,97 @@ def get_custom_lexer(extension):
 
    """
 
    returns a custom lexer if it's defined in rcextensions module, or None
 
    if there's no custom lexer defined
 
    """
 
    import kallithea
 
    from pygments import lexers
 
    #check if we didn't define this extension as other lexer
 
    if kallithea.EXTENSIONS and extension in kallithea.EXTENSIONS.EXTRA_LEXERS:
 
        _lexer_name = kallithea.EXTENSIONS.EXTRA_LEXERS[extension]
 
        return lexers.get_lexer_by_name(_lexer_name)
 

	
 

	
 
#==============================================================================
 
# TEST FUNCTIONS AND CREATORS
 
#==============================================================================
 
def create_test_index(repo_location, config, full_index):
 
    """
 
    Makes default test index
 

	
 
    :param config: test config
 
    :param full_index:
 
    """
 

	
 
    from kallithea.lib.indexers.daemon import WhooshIndexingDaemon
 
    from kallithea.lib.pidlock import DaemonLock, LockHeld
 

	
 
    repo_location = repo_location
 

	
 
    index_location = os.path.join(config['app_conf']['index_dir'])
 
    if not os.path.exists(index_location):
 
        os.makedirs(index_location)
 

	
 
    try:
 
        l = DaemonLock(file_=os.path.join(dirname(index_location), 'make_index.lock'))
 
        WhooshIndexingDaemon(index_location=index_location,
 
                             repo_location=repo_location) \
 
            .run(full_index=full_index)
 
        l.release()
 
    except LockHeld:
 
        pass
 

	
 

	
 
def create_test_env(repos_test_path, config):
 
    """
 
    Makes a fresh database and
 
    install test repository into tmp dir
 
    """
 
    from kallithea.lib.db_manage import DbManage
 
    from kallithea.tests import HG_REPO, GIT_REPO, TESTS_TMP_PATH
 
    from kallithea.tests.base import HG_REPO, GIT_REPO, TESTS_TMP_PATH
 

	
 
    # PART ONE create db
 
    dbconf = config['sqlalchemy.url']
 
    log.debug('making test db %s', dbconf)
 

	
 
    # create test dir if it doesn't exist
 
    if not os.path.isdir(repos_test_path):
 
        log.debug('Creating testdir %s', repos_test_path)
 
        os.makedirs(repos_test_path)
 

	
 
    dbmanage = DbManage(log_sql=True, dbconf=dbconf, root=config['here'],
 
                        tests=True)
 
    dbmanage.create_tables(override=True)
 
    # for tests dynamically set new root paths based on generated content
 
    dbmanage.create_settings(dbmanage.config_prompt(repos_test_path))
 
    dbmanage.create_default_user()
 
    dbmanage.admin_prompt()
 
    dbmanage.create_permissions()
 
    dbmanage.populate_default_permissions()
 
    Session().commit()
 
    # PART TWO make test repo
 
    log.debug('making test vcs repositories')
 

	
 
    idx_path = config['app_conf']['index_dir']
 
    data_path = config['app_conf']['cache_dir']
 

	
 
    #clean index and data
 
    if idx_path and os.path.exists(idx_path):
 
        log.debug('remove %s', idx_path)
 
        shutil.rmtree(idx_path)
 

	
 
    if data_path and os.path.exists(data_path):
 
        log.debug('remove %s', data_path)
 
        shutil.rmtree(data_path)
 

	
 
    #CREATE DEFAULT TEST REPOS
 
    cur_dir = dirname(dirname(abspath(__file__)))
 
    tar = tarfile.open(os.path.join(cur_dir, 'tests', 'fixtures', "vcs_test_hg.tar.gz"))
 
    tar.extractall(os.path.join(TESTS_TMP_PATH, HG_REPO))
 
    tar.close()
 

	
 
    cur_dir = dirname(dirname(abspath(__file__)))
 
    tar = tarfile.open(os.path.join(cur_dir, 'tests', 'fixtures', "vcs_test_git.tar.gz"))
 
    tar.extractall(os.path.join(TESTS_TMP_PATH, GIT_REPO))
 
    tar.close()
 

	
 
    #LOAD VCS test stuff
 
    from kallithea.tests.vcs import setup_package
kallithea/tests/__init__.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 test package
 

	
 
Refer to docs/contributing.rst for details on running the test suite.
 
"""
 
import os
 
import re
 
import time
 
import logging
 
import datetime
 
import hashlib
 
import tempfile
 

	
 
from tempfile import _RandomNameSequence
 

	
 
import pylons
 
import pylons.test
 
from pylons import config, url
 
from pylons.i18n.translation import _get_translator
 
from pylons.util import ContextObj
 

	
 
from routes.util import URLGenerator
 
from webtest import TestApp
 
import pytest
 

	
 
from kallithea.lib.compat import unittest
 
from kallithea import is_windows
 
from kallithea.model.db import Notification, User, UserNotification
 
from kallithea.model.meta import Session
 
from kallithea.lib.utils2 import safe_str
 

	
 

	
 
os.environ['TZ'] = 'UTC'
 
if not is_windows:
 
    time.tzset()
 

	
 
log = logging.getLogger(__name__)
 

	
 
skipif = pytest.mark.skipif
 
parametrize = pytest.mark.parametrize
 

	
 
__all__ = [
 
    'skipif', 'parametrize', 'environ', 'url', 'TestController',
 
    'ldap_lib_installed', 'pam_lib_installed', 'invalidate_all_caches',
 
    'TESTS_TMP_PATH', 'HG_REPO', 'GIT_REPO', 'NEW_HG_REPO', 'NEW_GIT_REPO',
 
    'HG_FORK', 'GIT_FORK', 'TEST_USER_ADMIN_LOGIN', 'TEST_USER_ADMIN_PASS',
 
    'TEST_USER_ADMIN_EMAIL', 'TEST_USER_REGULAR_LOGIN', 'TEST_USER_REGULAR_PASS',
 
    'TEST_USER_REGULAR_EMAIL', 'TEST_USER_REGULAR2_LOGIN',
 
    'TEST_USER_REGULAR2_PASS', 'TEST_USER_REGULAR2_EMAIL', 'TEST_HG_REPO',
 
    'TEST_HG_REPO_CLONE', 'TEST_HG_REPO_PULL', 'TEST_GIT_REPO',
 
    'TEST_GIT_REPO_CLONE', 'TEST_GIT_REPO_PULL', 'HG_REMOTE_REPO',
 
    'GIT_REMOTE_REPO', 'SCM_TESTS',
 
]
 

	
 
# Invoke websetup with the current config file
 
# SetupCommand('setup-app').run([config_file])
 

	
 
environ = {}
 

	
 
#SOME GLOBALS FOR TESTS
 

	
 
TESTS_TMP_PATH = os.path.join(tempfile.gettempdir(), 'rc_test_%s' % _RandomNameSequence().next())
 
TEST_USER_ADMIN_LOGIN = 'test_admin'
 
TEST_USER_ADMIN_PASS = 'test12'
 
TEST_USER_ADMIN_EMAIL = 'test_admin@example.com'
 

	
 
TEST_USER_REGULAR_LOGIN = 'test_regular'
 
TEST_USER_REGULAR_PASS = 'test12'
 
TEST_USER_REGULAR_EMAIL = 'test_regular@example.com'
 

	
 
TEST_USER_REGULAR2_LOGIN = 'test_regular2'
 
TEST_USER_REGULAR2_PASS = 'test12'
 
TEST_USER_REGULAR2_EMAIL = 'test_regular2@example.com'
 

	
 
HG_REPO = u'vcs_test_hg'
 
GIT_REPO = u'vcs_test_git'
 

	
 
NEW_HG_REPO = u'vcs_test_hg_new'
 
NEW_GIT_REPO = u'vcs_test_git_new'
 

	
 
HG_FORK = u'vcs_test_hg_fork'
 
GIT_FORK = u'vcs_test_git_fork'
 

	
 
## VCS
 
SCM_TESTS = ['hg', 'git']
 
uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))
 

	
 
GIT_REMOTE_REPO = 'git://github.com/codeinn/vcs.git'
 

	
 
TEST_GIT_REPO = os.path.join(TESTS_TMP_PATH, GIT_REPO)
 
TEST_GIT_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)
 
TEST_GIT_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)
 

	
 

	
 
HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'
 

	
 
TEST_HG_REPO = os.path.join(TESTS_TMP_PATH, HG_REPO)
 
TEST_HG_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)
 
TEST_HG_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcshgpull%s' % uniq_suffix)
 

	
 
TEST_DIR = tempfile.gettempdir()
 
TEST_REPO_PREFIX = 'vcs-test'
 

	
 
# cached repos if any !
 
# comment out to get some other repos from bb or github
 
GIT_REMOTE_REPO = os.path.join(TESTS_TMP_PATH, GIT_REPO)
 
HG_REMOTE_REPO = os.path.join(TESTS_TMP_PATH, HG_REPO)
 

	
 
#skip ldap tests if LDAP lib is not installed
 
ldap_lib_installed = False
 
try:
 
    import ldap
 
    ldap.API_VERSION
 
    ldap_lib_installed = True
 
except ImportError:
 
    # means that python-ldap is not installed
 
    pass
 

	
 
try:
 
    import pam
 
    pam.PAM_TEXT_INFO
 
    pam_lib_installed = True
 
except ImportError:
 
    pam_lib_installed = False
 

	
 
def invalidate_all_caches():
 
    """Invalidate all beaker caches currently configured.
 
    Useful when manipulating IP permissions in a test and changes need to take
 
    effect immediately.
 
    Note: Any use of this function is probably a workaround - it should be
 
    replaced with a more specific cache invalidation in code or test."""
 
    from beaker.cache import cache_managers
 
    for cache in cache_managers.values():
 
        cache.clear()
 

	
 
class NullHandler(logging.Handler):
 
    def emit(self, record):
 
        pass
 

	
 
class TestController(object):
 
    """Pytest-style test controller"""
 

	
 
    # Note: pytest base classes cannot have an __init__ method
 

	
 
    @pytest.fixture(autouse=True)
 
    def app_fixture(self):
 
        self.wsgiapp = pylons.test.pylonsapp
 
        self.init_stack(self.wsgiapp.config)
 
        self.app = TestApp(self.wsgiapp)
 
        self.maxDiff = None
 
        self.index_location = config['app_conf']['index_dir']
 
        return self.app
 

	
 
    def init_stack(self, config=None):
 
        if not config:
 
            config = pylons.test.pylonsapp.config
 
        url._push_object(URLGenerator(config['routes.map'], environ))
 
        pylons.app_globals._push_object(config['pylons.app_globals'])
 
        pylons.config._push_object(config)
 
        pylons.tmpl_context._push_object(ContextObj())
 
        # Initialize a translator for tests that utilize i18n
 
        translator = _get_translator(pylons.config.get('lang'))
 
        pylons.translator._push_object(translator)
 
        h = NullHandler()
 
        logging.getLogger("kallithea").addHandler(h)
 

	
 
    def remove_all_notifications(self):
 
        # query().delete() does not (by default) trigger cascades
 
        # ( http://docs.sqlalchemy.org/en/rel_0_7/orm/collections.html#passive-deletes )
 
        # so delete the UserNotification first to ensure referential integrity.
 
        UserNotification.query().delete()
 

	
 
        Notification.query().delete()
 
        Session().commit()
 

	
 
    def log_user(self, username=TEST_USER_ADMIN_LOGIN,
 
                 password=TEST_USER_ADMIN_PASS):
 
        self._logged_username = username
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': username,
 
                                  'password': password})
 

	
 
        if 'Invalid username or password' in response.body:
 
            pytest.fail('could not login using %s %s' % (username, password))
 

	
 
        assert response.status == '302 Found'
 
        self.assert_authenticated_user(response, username)
 

	
 
        response = response.follow()
 
        return response.session['authuser']
 

	
 
    def _get_logged_user(self):
 
        return User.get_by_username(self._logged_username)
 

	
 
    def assert_authenticated_user(self, response, expected_username):
 
        cookie = response.session.get('authuser')
 
        user = cookie and cookie.get('user_id')
 
        user = user and User.get(user)
 
        user = user and user.username
 
        assert user == expected_username
 

	
 
    def authentication_token(self):
 
        return self.app.get(url('authentication_token')).body
 

	
 
    def checkSessionFlash(self, response, msg=None, skip=0, _matcher=lambda msg, m: msg in m):
 
        if 'flash' not in response.session:
 
            pytest.fail(safe_str(u'msg `%s` not found - session has no flash:\n%s' % (msg, response)))
 
        try:
 
            level, m = response.session['flash'][-1 - skip]
 
            if _matcher(msg, m):
 
                return
 
        except IndexError:
 
            pass
 
        pytest.fail(safe_str(u'msg `%s` not found in session flash (skipping %s): %s' %
 
                           (msg, skip,
 
                            ', '.join('`%s`' % m for level, m in response.session['flash']))))
 

	
 
    def checkSessionFlashRegex(self, response, regex, skip=0):
 
        self.checkSessionFlash(response, regex, skip=skip, _matcher=re.search)
kallithea/tests/api/api_base.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/>.
 

	
 
"""
 
Tests for the JSON-RPC web api.
 
"""
 

	
 
import os
 
import random
 
import mock
 

	
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 
from kallithea.lib.compat import json
 
from kallithea.lib.auth import AuthUser
 
from kallithea.model.user import UserModel
 
from kallithea.model.user_group import UserGroupModel
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.repo_group import RepoGroupModel
 
from kallithea.model.meta import Session
 
from kallithea.model.scm import ScmModel
 
from kallithea.model.gist import GistModel
 
from kallithea.model.db import Repository, User, Setting, Ui
 
from kallithea.lib.utils2 import time_to_datetime
 

	
 

	
 
API_URL = '/_admin/api'
 
TEST_USER_GROUP = u'test_user_group'
 
TEST_REPO_GROUP = u'test_repo_group'
 

	
 
fixture = Fixture()
 

	
 

	
 
def _build_data(apikey, method, **kw):
 
    """
 
    Builds API data with given random ID
 

	
 
    :param random_id:
 
    """
 
    random_id = random.randrange(1, 9999)
 
    return random_id, json.dumps({
 
        "id": random_id,
 
        "api_key": apikey,
 
        "method": method,
 
        "args": kw
 
    })
 

	
 

	
 
jsonify = lambda obj: json.loads(json.dumps(obj))
 

	
 

	
 
def crash(*args, **kwargs):
 
    raise Exception('Total Crash !')
 

	
 

	
 
def api_call(test_obj, params):
 
    response = test_obj.app.post(API_URL, content_type='application/json',
 
                                 params=params)
 
    return response
 

	
kallithea/tests/api/test_api_git.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/>.
 

	
 
from kallithea.tests import TestController, GIT_REPO
 
from kallithea.tests.base import TestController, GIT_REPO
 
from kallithea.tests.api.api_base import _BaseTestApi
 

	
 

	
 
class TestGitApi(_BaseTestApi, TestController):
 
    REPO = GIT_REPO
 
    REPO_TYPE = 'git'
kallithea/tests/api/test_api_hg.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/>.
 

	
 
from kallithea.tests import TestController, HG_REPO
 
from kallithea.tests.base import TestController, HG_REPO
 
from kallithea.tests.api.api_base import _BaseTestApi
 

	
 

	
 
class TestHgApi(_BaseTestApi, TestController):
 
    REPO = HG_REPO
 
    REPO_TYPE = 'hg'
kallithea/tests/base.py
Show inline comments
 
file copied from kallithea/tests/__init__.py to kallithea/tests/base.py
 
# -*- 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 test package
 

	
 
Refer to docs/contributing.rst for details on running the test suite.
 
"""
 
import datetime
 
import logging
 
import os
 
import pytest
 
import re
 
import tempfile
 
import time
 
import logging
 
import datetime
 
import hashlib
 
import tempfile
 

	
 
from tempfile import _RandomNameSequence
 

	
 
import pylons
 
import pylons.test
 
from pylons import config, url
 
from pylons.i18n.translation import _get_translator
 
from pylons.util import ContextObj
 

	
 
from routes.util import URLGenerator
 
from webtest import TestApp
 
import pytest
 

	
 
from kallithea.lib.compat import unittest
 
from kallithea import is_windows
 
from kallithea.model.db import Notification, User, UserNotification
 
from kallithea.model.meta import Session
 
from kallithea.lib.utils2 import safe_str
 

	
 

	
 
os.environ['TZ'] = 'UTC'
 
if not is_windows:
 
    time.tzset()
 

	
 
log = logging.getLogger(__name__)
 

	
 
skipif = pytest.mark.skipif
 
parametrize = pytest.mark.parametrize
 

	
 
__all__ = [
 
    'skipif', 'parametrize', 'environ', 'url', 'TestController',
 
    'ldap_lib_installed', 'pam_lib_installed', 'invalidate_all_caches',
 
    'TESTS_TMP_PATH', 'HG_REPO', 'GIT_REPO', 'NEW_HG_REPO', 'NEW_GIT_REPO',
 
    'HG_FORK', 'GIT_FORK', 'TEST_USER_ADMIN_LOGIN', 'TEST_USER_ADMIN_PASS',
 
    'TEST_USER_ADMIN_EMAIL', 'TEST_USER_REGULAR_LOGIN', 'TEST_USER_REGULAR_PASS',
 
    'TEST_USER_REGULAR_EMAIL', 'TEST_USER_REGULAR2_LOGIN',
 
    'TEST_USER_REGULAR2_PASS', 'TEST_USER_REGULAR2_EMAIL', 'TEST_HG_REPO',
 
    'TEST_HG_REPO_CLONE', 'TEST_HG_REPO_PULL', 'TEST_GIT_REPO',
 
    'TEST_GIT_REPO_CLONE', 'TEST_GIT_REPO_PULL', 'HG_REMOTE_REPO',
 
    'GIT_REMOTE_REPO', 'SCM_TESTS',
 
]
 

	
 
# Invoke websetup with the current config file
 
# SetupCommand('setup-app').run([config_file])
 

	
 
environ = {}
 

	
 
#SOME GLOBALS FOR TESTS
 

	
 
TESTS_TMP_PATH = os.path.join(tempfile.gettempdir(), 'rc_test_%s' % _RandomNameSequence().next())
 
TESTS_TMP_PATH = os.path.join(tempfile.gettempdir(), 'rc_test_%s' % tempfile._RandomNameSequence().next())
 
TEST_USER_ADMIN_LOGIN = 'test_admin'
 
TEST_USER_ADMIN_PASS = 'test12'
 
TEST_USER_ADMIN_EMAIL = 'test_admin@example.com'
 

	
 
TEST_USER_REGULAR_LOGIN = 'test_regular'
 
TEST_USER_REGULAR_PASS = 'test12'
 
TEST_USER_REGULAR_EMAIL = 'test_regular@example.com'
 

	
 
TEST_USER_REGULAR2_LOGIN = 'test_regular2'
 
TEST_USER_REGULAR2_PASS = 'test12'
 
TEST_USER_REGULAR2_EMAIL = 'test_regular2@example.com'
 

	
 
HG_REPO = u'vcs_test_hg'
 
GIT_REPO = u'vcs_test_git'
 

	
 
NEW_HG_REPO = u'vcs_test_hg_new'
 
NEW_GIT_REPO = u'vcs_test_git_new'
 

	
 
HG_FORK = u'vcs_test_hg_fork'
 
GIT_FORK = u'vcs_test_git_fork'
 

	
 
## VCS
 
SCM_TESTS = ['hg', 'git']
 
uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))
 

	
 
GIT_REMOTE_REPO = 'git://github.com/codeinn/vcs.git'
 

	
 
TEST_GIT_REPO = os.path.join(TESTS_TMP_PATH, GIT_REPO)
 
TEST_GIT_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)
 
TEST_GIT_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)
 

	
 

	
 
HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'
 

	
 
TEST_HG_REPO = os.path.join(TESTS_TMP_PATH, HG_REPO)
 
TEST_HG_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)
 
TEST_HG_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcshgpull%s' % uniq_suffix)
 

	
 
TEST_DIR = tempfile.gettempdir()
 
TEST_REPO_PREFIX = 'vcs-test'
 

	
 
# cached repos if any !
 
# comment out to get some other repos from bb or github
 
GIT_REMOTE_REPO = os.path.join(TESTS_TMP_PATH, GIT_REPO)
 
HG_REMOTE_REPO = os.path.join(TESTS_TMP_PATH, HG_REPO)
 

	
 
#skip ldap tests if LDAP lib is not installed
 
ldap_lib_installed = False
 
@@ -186,48 +174,49 @@ class TestController(object):
 

	
 
        Notification.query().delete()
 
        Session().commit()
 

	
 
    def log_user(self, username=TEST_USER_ADMIN_LOGIN,
 
                 password=TEST_USER_ADMIN_PASS):
 
        self._logged_username = username
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': username,
 
                                  'password': password})
 

	
 
        if 'Invalid username or password' in response.body:
 
            pytest.fail('could not login using %s %s' % (username, password))
 

	
 
        assert response.status == '302 Found'
 
        self.assert_authenticated_user(response, username)
 

	
 
        response = response.follow()
 
        return response.session['authuser']
 

	
 
    def _get_logged_user(self):
 
        return User.get_by_username(self._logged_username)
 

	
 
    def assert_authenticated_user(self, response, expected_username):
 
        cookie = response.session.get('authuser')
 
        user = cookie and cookie.get('user_id')
 
        user = user and User.get(user)
 
        user = user and user.username
 
        assert user == expected_username
 

	
 
    def authentication_token(self):
 
        return self.app.get(url('authentication_token')).body
 

	
 
    def checkSessionFlash(self, response, msg=None, skip=0, _matcher=lambda msg, m: msg in m):
 
        if 'flash' not in response.session:
 
            pytest.fail(safe_str(u'msg `%s` not found - session has no flash:\n%s' % (msg, response)))
 
        try:
 
            level, m = response.session['flash'][-1 - skip]
 
            if _matcher(msg, m):
 
                return
 
        except IndexError:
 
            pass
 
        pytest.fail(safe_str(u'msg `%s` not found in session flash (skipping %s): %s' %
 
                           (msg, skip,
 
                            ', '.join('`%s`' % m for level, m in response.session['flash']))))
 

	
 
    def checkSessionFlashRegex(self, response, regex, skip=0):
 
        self.checkSessionFlash(response, regex, skip=skip, _matcher=re.search)
 

	
kallithea/tests/conftest.py
Show inline comments
 
import os
 
import sys
 
import logging
 

	
 
import pkg_resources
 
from paste.deploy import loadapp
 
import pylons.test
 
from pylons.i18n.translation import _get_translator
 
import pytest
 
from kallithea.model.user import UserModel
 
from kallithea.model.meta import Session
 
from kallithea.model.db import Setting, User, UserIpMap
 
from kallithea.tests import invalidate_all_caches, TEST_USER_REGULAR_LOGIN
 
from kallithea.tests.base import invalidate_all_caches, TEST_USER_REGULAR_LOGIN
 

	
 

	
 
def pytest_configure():
 
    path = os.getcwd()
 
    sys.path.insert(0, path)
 
    pkg_resources.working_set.add_entry(path)
 

	
 
    # Disable INFO logging of test database creation, restore with NOTSET
 
    logging.disable(logging.INFO)
 
    pylons.test.pylonsapp = loadapp('config:kallithea/tests/test.ini', relative_to=path)
 
    logging.disable(logging.NOTSET)
 

	
 
    # Setup the config and app_globals, only works if we can get
 
    # to the config object
 
    conf = getattr(pylons.test.pylonsapp, 'config')
 
    if conf:
 
        pylons.config._push_object(conf)
 

	
 
        if 'pylons.app_globals' in conf:
 
            pylons.app_globals._push_object(conf['pylons.app_globals'])
 

	
 
    # Initialize a translator for tests that utilize i18n
 
    translator = _get_translator(pylons.config.get('lang'))
 
    pylons.translator._push_object(translator)
 

	
 
    return pylons.test.pylonsapp
 

	
 

	
 
@pytest.yield_fixture
 
def create_test_user():
 
    """Provide users that automatically disappear after test is over."""
 
    test_user_ids = []
 
    def _create_test_user(user_form):
 
        user = UserModel().create(user_form)
 
        test_user_ids.append(user.user_id)
 
        return user
 
    yield _create_test_user
 
    for user_id in test_user_ids:
 
        UserModel().delete(user_id)
 
    Session().commit()
 

	
 

	
 
def _set_settings(*kvtseq):
 
    session = Session()
 
    for kvt in kvtseq:
 
        assert len(kvt) in (2, 3)
 
        k = kvt[0]
 
        v = kvt[1]
kallithea/tests/fixture.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/>.
 

	
 
"""
 
Helpers for fixture generation
 
"""
 
import os
 
import time
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.model.db import Repository, User, RepoGroup, UserGroup, Gist
 
from kallithea.model.meta import Session
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.user import UserModel
 
from kallithea.model.repo_group import RepoGroupModel
 
from kallithea.model.user_group import UserGroupModel
 
from kallithea.model.gist import GistModel
 
from kallithea.model.scm import ScmModel
 
from kallithea.lib.vcs.backends.base import EmptyChangeset
 
from os.path import dirname
 

	
 
FIXTURES = os.path.join(dirname(dirname(os.path.abspath(__file__))), 'tests', 'fixtures')
 

	
 

	
 
def error_function(*args, **kwargs):
 
    raise Exception('Total Crash !')
 

	
 

	
 
class Fixture(object):
 

	
 
    def __init__(self):
 
        pass
 

	
 
    def anon_access(self, status):
 
        """
 
        Context manager for controlling anonymous access.
 
        Anon access will be set and committed, but restored again when exiting the block.
 

	
 
        Usage:
 

	
 
        fixture = Fixture()
 
        with fixture.anon_access(False):
 
            stuff
 
        """
 

	
 
        class context(object):
 
            def __enter__(self):
 
                anon = User.get_default_user()
 
                self._before = anon.active
 
                anon.active = status
 
                Session().add(anon)
 
                Session().commit()
 
                invalidate_all_caches()
 

	
 
            def __exit__(self, exc_type, exc_val, exc_tb):
 
                anon = User.get_default_user()
 
                anon.active = self._before
 
                Session().add(anon)
kallithea/tests/functional/test_admin.py
Show inline comments
 
import os
 
import csv
 
import datetime
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.model.db import UserLog
 
from kallithea.model.meta import Session
 
from kallithea.lib.utils2 import safe_unicode
 
from os.path import dirname
 

	
 
FIXTURES = os.path.join(dirname(dirname(os.path.abspath(__file__))), 'fixtures')
 

	
 

	
 
class TestAdminController(TestController):
 

	
 
    @classmethod
 
    def setup_class(cls):
 
        UserLog.query().delete()
 
        Session().commit()
 

	
 
        def strptime(val):
 
            fmt = '%Y-%m-%d %H:%M:%S'
 
            if '.' not in val:
 
                return datetime.datetime.strptime(val, fmt)
 

	
 
            nofrag, frag = val.split(".")
 
            date = datetime.datetime.strptime(nofrag, fmt)
 

	
 
            frag = frag[:6]  # truncate to microseconds
 
            frag += (6 - len(frag)) * '0'  # add 0s
 
            return date.replace(microsecond=int(frag))
 

	
 
        with open(os.path.join(FIXTURES, 'journal_dump.csv')) as f:
 
            for row in csv.DictReader(f):
 
                ul = UserLog()
 
                for k, v in row.iteritems():
 
                    v = safe_unicode(v)
 
                    if k == 'action_date':
 
                        v = strptime(v)
 
                    if k in ['user_id', 'repository_id']:
 
                        # nullable due to FK problems
 
                        v = None
 
                    setattr(ul, k, v)
 
                Session().add(ul)
 
            Session().commit()
 

	
 
    @classmethod
 
    def teardown_class(cls):
 
        UserLog.query().delete()
 
        Session().commit()
 

	
 
    def test_index(self):
 
        self.log_user()
kallithea/tests/functional/test_admin_auth_settings.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.model.db import Setting
 

	
 

	
 
class TestAuthSettingsController(TestController):
 
    def _enable_plugins(self, plugins_list):
 
        test_url = url(controller='admin/auth_settings',
 
                       action='auth_settings')
 
        params={'auth_plugins': plugins_list, '_authentication_token': self.authentication_token()}
 

	
 
        for plugin in plugins_list.split(','):
 
            enable = plugin.partition('kallithea.lib.auth_modules.')[-1]
 
            params.update({'%s_enabled' % enable: True})
 
        response = self.app.post(url=test_url, params=params)
 
        return params
 
        #self.checkSessionFlash(response, 'Auth settings updated successfully')
 

	
 
    def test_index(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/auth_settings',
 
                                    action='index'))
 
        response.mustcontain('Authentication Plugins')
 

	
 
    @skipif(not ldap_lib_installed, reason='skipping due to missing ldap lib')
 
    def test_ldap_save_settings(self):
 
        self.log_user()
 

	
 
        params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_ldap')
 
        params.update({'auth_ldap_host': u'dc.example.com',
 
                       'auth_ldap_port': '999',
 
                       'auth_ldap_tls_kind': 'PLAIN',
 
                       'auth_ldap_tls_reqcert': 'NEVER',
 
                       'auth_ldap_dn_user': 'test_user',
 
                       'auth_ldap_dn_pass': 'test_pass',
 
                       'auth_ldap_base_dn': 'test_base_dn',
 
                       'auth_ldap_filter': 'test_filter',
 
                       'auth_ldap_search_scope': 'BASE',
 
                       'auth_ldap_attr_login': 'test_attr_login',
 
                       'auth_ldap_attr_firstname': 'ima',
 
                       'auth_ldap_attr_lastname': 'tester',
 
                       'auth_ldap_attr_email': 'test@example.com'})
 

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

	
 
        response = self.app.post(url=test_url, params=params)
 
        self.checkSessionFlash(response, 'Auth settings updated successfully')
 

	
 
        new_settings = Setting.get_auth_settings()
kallithea/tests/functional/test_admin_defaults.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.model.db import Setting
 

	
 

	
 
class TestDefaultsController(TestController):
 

	
 
    def test_index(self):
 
        self.log_user()
 
        response = self.app.get(url('defaults'))
 
        response.mustcontain('default_repo_private')
 
        response.mustcontain('default_repo_enable_statistics')
 
        response.mustcontain('default_repo_enable_downloads')
 
        response.mustcontain('default_repo_enable_locking')
 

	
 
    def test_update_params_true_hg(self):
 
        self.log_user()
 
        params = {
 
            'default_repo_enable_locking': True,
 
            'default_repo_enable_downloads': True,
 
            'default_repo_enable_statistics': True,
 
            'default_repo_private': True,
 
            'default_repo_type': 'hg',
 
            '_authentication_token': self.authentication_token(),
 
        }
 
        response = self.app.post(url('defaults_update', id='default'), params=params)
 
        self.checkSessionFlash(response, 'Default settings updated successfully')
 

	
 
        params.pop('_authentication_token')
 
        defs = Setting.get_default_repo_settings()
 
        assert params == defs
 

	
 
    def test_update_params_false_git(self):
 
        self.log_user()
 
        params = {
 
            'default_repo_enable_locking': False,
 
            'default_repo_enable_downloads': False,
 
            'default_repo_enable_statistics': False,
 
            'default_repo_private': False,
 
            'default_repo_type': 'git',
 
            '_authentication_token': self.authentication_token(),
 
        }
 
        response = self.app.post(url('defaults_update', id='default'), params=params)
 
        self.checkSessionFlash(response, 'Default settings updated successfully')
 

	
 
        params.pop('_authentication_token')
 
        defs = Setting.get_default_repo_settings()
 
        assert params == defs
kallithea/tests/functional/test_admin_gists.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.model.gist import GistModel
 
from kallithea.model.meta import Session
 
from kallithea.model.db import User, Gist
 

	
 

	
 
def _create_gist(f_name, content='some gist', lifetime=-1,
 
                 description=u'gist-desc', gist_type='public',
 
                 owner=TEST_USER_ADMIN_LOGIN):
 
    gist_mapping = {
 
        f_name: {'content': content}
 
    }
 
    user = User.get_by_username(owner)
 
    gist = GistModel().create(description, owner=user,
 
                       gist_mapping=gist_mapping, gist_type=gist_type,
 
                       lifetime=lifetime)
 
    Session().commit()
 
    return gist
 

	
 

	
 
class TestGistsController(TestController):
 

	
 
    def teardown_method(self, method):
 
        for g in Gist.get_all():
 
            GistModel().delete(g)
 
        Session().commit()
 

	
 
    def test_index(self):
 
        self.log_user()
 
        response = self.app.get(url('gists'))
 
        # Test response...
 
        response.mustcontain('There are no gists yet')
 

	
 
        g1 = _create_gist('gist1').gist_access_id
 
        g2 = _create_gist('gist2', lifetime=1400).gist_access_id
 
        g3 = _create_gist('gist3', description=u'gist3-desc').gist_access_id
 
        g4 = _create_gist('gist4', gist_type='private').gist_access_id
 
        response = self.app.get(url('gists'))
 
        # Test response...
 
        response.mustcontain('gist: %s' % g1)
 
        response.mustcontain('gist: %s' % g2)
 
        response.mustcontain('Expires: in 23 hours')  # we don't care about the end
 
        response.mustcontain('gist: %s' % g3)
 
        response.mustcontain('gist3-desc')
 
        response.mustcontain(no=['gist: %s' % g4])
 

	
 
    def test_index_private_gists(self):
 
        self.log_user()
 
        gist = _create_gist('gist5', gist_type='private')
kallithea/tests/functional/test_admin_notifications.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.model.db import User
 

	
 
from kallithea.model.user import UserModel
 
from kallithea.model.notification import NotificationModel
 
from kallithea.model.meta import Session
 
from kallithea.lib import helpers as h
 

	
 

	
 
class TestNotificationsController(TestController):
 
    def setup_method(self, method):
 
        self.remove_all_notifications()
 

	
 
    def test_index(self, create_test_user):
 
        self.log_user()
 

	
 
        u1 = create_test_user(dict(username='u1', password='qweqwe',
 
                                   email='u1@example.com',
 
                                   firstname=u'u1', lastname=u'u1',
 
                                   active=True))
 
        u1 = u1.user_id
 
        Session().commit()
 

	
 
        response = self.app.get(url('notifications'))
 
        response.mustcontain('<div class="table">No notifications here yet</div>')
 

	
 
        cur_user = self._get_logged_user()
 
        notif = NotificationModel().create(created_by=u1, subject=u'test_notification_1',
 
                                           body=u'notification_1', recipients=[cur_user])
 
        Session().commit()
 
        response = self.app.get(url('notifications'))
 
        response.mustcontain('id="notification_%s"' % notif.notification_id)
 

	
 
    def test_delete(self, create_test_user):
 
        self.log_user()
 
        cur_user = self._get_logged_user()
 

	
 
        u1 = create_test_user(dict(username='u1', password='qweqwe',
 
                                   email='u1@example.com',
 
                                   firstname=u'u1', lastname=u'u1',
 
                                   active=True))
 
        u2 = create_test_user(dict(username='u2', password='qweqwe',
 
                                   email='u2@example.com',
 
                                   firstname=u'u2', lastname=u'u2',
 
                                   active=True))
 

	
 
        # make notifications
 
        notification = NotificationModel().create(created_by=cur_user,
 
                                                  subject=u'test',
kallithea/tests/functional/test_admin_permissions.py
Show inline comments
 
import time
 

	
 
from kallithea.model.db import User, UserIpMap
 
from kallithea.model.user import UserModel
 
from kallithea.model.meta import Session
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 

	
 
class TestAdminPermissionsController(TestController):
 

	
 
    def test_index(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_permissions'))
 
        # Test response...
 

	
 
    def test_index_ips(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_permissions_ips'))
 
        # Test response...
 
        response.mustcontain('All IP addresses are allowed')
 

	
 
    def test_add_ips(self, auto_clear_ip_permissions):
 
        self.log_user()
 
        default_user_id = User.get_default_user().user_id
 
        response = self.app.post(url('edit_user_ips_update', id=default_user_id),
 
                                 params=dict(new_ip='127.0.0.0/24',
 
                                 _authentication_token=self.authentication_token()))
 

	
 
        # IP permissions are cached, need to invalidate this cache explicitly
 
        invalidate_all_caches()
 

	
 
        self.app.get(url('admin_permissions_ips'), status=302)
 

	
 
        # REMOTE_ADDR must match 127.0.0.0/24
 
        response = self.app.get(url('admin_permissions_ips'),
 
                                extra_environ={'REMOTE_ADDR': '127.0.0.1'})
 
        response.mustcontain('127.0.0.0/24')
 
        response.mustcontain('127.0.0.0 - 127.0.0.255')
 

	
 
    def test_delete_ips(self, auto_clear_ip_permissions):
 
        self.log_user()
 
        default_user_id = User.get_default_user().user_id
 

	
 
        ## first add
 
        new_ip = '127.0.0.0/24'
 
        user_model = UserModel()
 
        ip_obj = user_model.add_extra_ip(default_user_id, new_ip)
 
        Session().commit()
 

	
 
        ## double check that add worked
 
        # IP permissions are cached, need to invalidate this cache explicitly
 
        invalidate_all_caches()
 
        self.app.get(url('admin_permissions_ips'), status=302)
 
        # REMOTE_ADDR must match 127.0.0.0/24
 
        response = self.app.get(url('admin_permissions_ips'),
kallithea/tests/functional/test_admin_repo_groups.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 

	
 
class TestRepoGroupsController(TestController):
 
    pass
kallithea/tests/functional/test_admin_repos.py
Show inline comments
 
# -*- coding: utf-8 -*-
 

	
 
import os
 
import mock
 
import urllib
 

	
 
import pytest
 

	
 
from kallithea.lib import vcs
 
from kallithea.lib.utils2 import safe_str, safe_unicode
 
from kallithea.model.db import Repository, RepoGroup, UserRepoToPerm, User, \
 
    Permission, Ui
 
from kallithea.model.user import UserModel
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.model.repo_group import RepoGroupModel
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.meta import Session
 
from kallithea.tests.fixture import Fixture, error_function
 

	
 
fixture = Fixture()
 

	
 

	
 
def _get_permission_for_user(user, repo):
 
    perm = UserRepoToPerm.query() \
 
                .filter(UserRepoToPerm.repository ==
 
                        Repository.get_by_repo_name(repo)) \
 
                .filter(UserRepoToPerm.user == User.get_by_username(user)) \
 
                .all()
 
    return perm
 

	
 

	
 
class _BaseTestCase(TestController):
 
    """
 
    Write all tests here
 
    """
 
    REPO = None
 
    REPO_TYPE = None
 
    NEW_REPO = None
 
    OTHER_TYPE_REPO = None
 
    OTHER_TYPE = None
 

	
 
    def test_index(self):
 
        self.log_user()
 
        response = self.app.get(url('repos'))
 

	
 
    def test_create(self):
 
        self.log_user()
 
        repo_name = self.NEW_REPO
 
        description = u'description for newly created repo'
 
        response = self.app.post(url('repos'),
 
                        fixture._get_repo_create_params(repo_private=False,
 
                                                repo_name=repo_name,
 
                                                repo_type=self.REPO_TYPE,
 
                                                repo_description=description,
 
                                                _authentication_token=self.authentication_token()))
 
        ## run the check page that triggers the flash message
 
        response = self.app.get(url('repo_check_home', repo_name=repo_name))
 
        assert response.json == {u'result': True}
 
        self.checkSessionFlash(response,
 
                               'Created repository <a href="/%s">%s</a>'
 
                               % (repo_name, repo_name))
 

	
kallithea/tests/functional/test_admin_settings.py
Show inline comments
 
# -*- coding: utf-8 -*-
 

	
 
import tempfile
 

	
 
from kallithea.model.db import Setting, Ui
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestAdminSettingsController(TestController):
 

	
 
    def test_index_main(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_settings'))
 

	
 
    def test_index_mapping(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_settings_mapping'))
 

	
 
    def test_index_global(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_settings_global'))
 

	
 
    def test_index_visual(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_settings_visual'))
 

	
 
    def test_index_email(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_settings_email'))
 

	
 
    def test_index_hooks(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_settings_hooks'))
 

	
 
    def test_create_custom_hook(self):
 
        self.log_user()
 
        response = self.app.post(url('admin_settings_hooks'),
 
                                params=dict(new_hook_ui_key='test_hooks_1',
 
                                            new_hook_ui_value='cd %s' % tempfile.gettempdir(),
 
                                            _authentication_token=self.authentication_token()))
 

	
 
        response = response.follow()
 
        response.mustcontain('test_hooks_1')
 
        response.mustcontain('cd %s' % tempfile.gettempdir())
 

	
 
    def test_create_custom_hook_delete(self):
 
        self.log_user()
 
        response = self.app.post(url('admin_settings_hooks'),
 
                                params=dict(new_hook_ui_key='test_hooks_2',
 
                                            new_hook_ui_value='cd %s2' % tempfile.gettempdir(),
 
                                            _authentication_token=self.authentication_token()))
kallithea/tests/functional/test_admin_user_groups.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.model.db import UserGroup, UserGroupToPerm, Permission
 
from kallithea.model.meta import Session
 

	
 
TEST_USER_GROUP = u'admins_test'
 

	
 

	
 
class TestAdminUsersGroupsController(TestController):
 

	
 
    def test_index(self):
 
        self.log_user()
 
        response = self.app.get(url('users_groups'))
 
        # Test response...
 

	
 
    def test_create(self):
 
        self.log_user()
 
        users_group_name = TEST_USER_GROUP
 
        response = self.app.post(url('users_groups'),
 
                                 {'users_group_name': users_group_name,
 
                                  'user_group_description': u'DESC',
 
                                  'active': True,
 
                                  '_authentication_token': self.authentication_token()})
 
        response.follow()
 

	
 
        self.checkSessionFlash(response,
 
                               'Created user group <a href="/_admin/user_groups/')
 
        self.checkSessionFlash(response,
 
                               '/edit">%s</a>' % TEST_USER_GROUP)
 

	
 
    def test_new(self):
 
        response = self.app.get(url('new_users_group'))
 

	
 
    def test_update(self):
 
        response = self.app.post(url('update_users_group', id=1), status=403)
 

	
 
    def test_update_browser_fakeout(self):
 
        response = self.app.post(url('update_users_group', id=1),
 
                                 params=dict(_authentication_token=self.authentication_token()))
 

	
 
    def test_delete(self):
 
        self.log_user()
 
        users_group_name = TEST_USER_GROUP + 'another'
 
        response = self.app.post(url('users_groups'),
 
                                 {'users_group_name':users_group_name,
 
                                  'user_group_description': u'DESC',
 
                                  'active': True,
 
                                  '_authentication_token': self.authentication_token()})
 
        response.follow()
 

	
kallithea/tests/functional/test_admin_users.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/>.
 

	
 
from sqlalchemy.orm.exc import NoResultFound, ObjectDeletedError
 

	
 
import pytest
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 
from kallithea.controllers.admin.users import UsersController
 
from kallithea.model.db import User, Permission, UserIpMap, UserApiKeys
 
from kallithea.lib.auth import check_password
 
from kallithea.model.user import UserModel
 
from kallithea.model import validators
 
from kallithea.lib import helpers as h
 
from kallithea.model.meta import Session
 
from webob.exc import HTTPNotFound
 

	
 
fixture = Fixture()
 

	
 
@pytest.yield_fixture
 
def user_and_repo_group_fail():
 
    username = 'repogrouperr'
 
    groupname = u'repogroup_fail'
 
    user = fixture.create_user(name=username)
 
    repo_group = fixture.create_repo_group(name=groupname, cur_user=username)
 
    yield user, repo_group
 
    # cleanup
 
    try:
 
        fixture.destroy_repo_group(repo_group)
 
    except ObjectDeletedError:
 
        # delete already succeeded in test body
 
        pass
 

	
 
class TestAdminUsersController(TestController):
 
    test_user_1 = 'testme'
 

	
 
    @classmethod
 
    def teardown_class(cls):
 
        if User.get_by_username(cls.test_user_1):
 
            UserModel().delete(cls.test_user_1)
 
            Session().commit()
 

	
 
    def test_index(self):
 
        self.log_user()
 
        response = self.app.get(url('users'))
 
        # Test response...
 

	
 
    def test_create(self):
 
        self.log_user()
 
        username = 'newtestuser'
 
        password = 'test12'
 
        password_confirmation = password
 
        name = u'name'
 
        lastname = u'lastname'
 
        email = 'mail@example.com'
kallithea/tests/functional/test_changelog.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 

	
 

	
 
class TestChangelogController(TestController):
 

	
 
    def test_index_hg(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='changelog', action='index',
 
                                    repo_name=HG_REPO))
 

	
 
        response.mustcontain('''id="chg_20" class="container mergerow"''')
 
        response.mustcontain(
 
            """<input class="changeset_range" """
 
            """id="7b22a518347bb9bc19679f6af07cd0a61bfe16e7" """
 
            """name="7b22a518347bb9bc19679f6af07cd0a61bfe16e7" """
 
            """type="checkbox" value="1" />"""
 
        )
 
        #rev 640: code garden
 
        response.mustcontain(
 
            """<a class="changeset_hash" href="/%s/changeset/0a4e54a4460401d6dbbd6a3604b17cd2b3606b82">r640:0a4e54a44604</a>""" % HG_REPO
 
        )
 
        response.mustcontain("""code garden""")
 

	
 
        response.mustcontain("""var jsdata = [[[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 1, 2, 0], [0, 0, 1, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 1, 0], [1, 1, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 1], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 1, 3, 0], [0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 3, 0]], 0, 0, 0, 0, 0, 0], [[1, 3], [[0, 0, 2, 0], [1, 1, 3, 0]], 0, 0, 0, 0, 0, 0], [[1, 3], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 1, 4, 0], [0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0, 0, 0, 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0, 0, 0, 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0], [1, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0, 0, 0, 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0, 0, 0, 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0, 0, 0, 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0, 0, 0, 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0], [1, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0, 0, 0, 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0, 0, 0, 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 1, 5, 0], [0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 5, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 5, 0]], 0, 0, 0, 0, 0, 0], [[1, 5], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 1, 6, 0], [0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 6, 0]], 0, 0, 0, 0, 0, 0], [[1, 6], [[0, 0, 2, 0], [1, 1, 6, 0]], 0, 0, 0, 0, 0, 0], [[1, 6], [[0, 0, 2, 0], [1, 1, 6, 0], [1, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 6, 0]], 0, 0, 0, 0, 0, 0], [[1, 6], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 1, 7, 0], [0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 7, 0]], 0, 0, 0, 0, 0, 0], [[1, 7], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 1, 8, 0], [0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 8, 0]], 0, 0, 0, 0, 0, 0], [[1, 8], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 1, 9, 0], [0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 1, 10, 0], [0, 0, 2, 0], [1, 2, 9, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 10, 0], [2, 2, 9, 0]], 0, 0, 0, 0, 0, 0], [[2, 9], [[0, 0, 2, 0], [1, 1, 10, 0], [2, 1, 10, 0]], 0, 0, 0, 0, 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0, 0, 0, 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0, 0, 0, 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0, 0, 0, 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0, 0, 0, 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0, 0, 0, 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0, 0, 0, 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[1, 11], [[0, 0, 2, 0], [1, 1, 11, 0]], 1, 0, 0, 0, 0, 0], [[2, 12], [[0, 0, 2, 0], [2, 1, 12, 0]], 1, 0, 0, 0, 0, 0], [[0, 2], [[0, 1, 13, 0], [0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 13, 0]], 0, 0, 0, 0, 0, 0], [[1, 13], [[0, 0, 2, 0], [1, 1, 13, 0], [1, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 13, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 13, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 13, 0]], 0, 0, 0, 0, 0, 0], [[1, 13], [[0, 0, 2, 0], [1, 1, 13, 0], [1, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 1, 14, 0], [0, 0, 2, 0]], 0, 0, 0, 0, 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0, 0, 0, 0, 0]];""")
 

	
 
    def test_index_pagination_hg(self):
 
        self.log_user()
 
        #pagination
 
        self.app.get(url(controller='changelog', action='index',
 
                                    repo_name=HG_REPO), {'page': 1})
 
        self.app.get(url(controller='changelog', action='index',
 
                                    repo_name=HG_REPO), {'page': 2})
 
        self.app.get(url(controller='changelog', action='index',
 
                                    repo_name=HG_REPO), {'page': 3})
 
        self.app.get(url(controller='changelog', action='index',
 
                                    repo_name=HG_REPO), {'page': 4})
 
        self.app.get(url(controller='changelog', action='index',
 
                                    repo_name=HG_REPO), {'page': 5})
 
        response = self.app.get(url(controller='changelog', action='index',
 
                                    repo_name=HG_REPO), {'page': 6, 'size': 20})
 

	
 
        # Test response after pagination...
 
        response.mustcontain(
 
            """<input class="changeset_range" """
 
            """id="22baf968d547386b9516965ce89d189665003a31" """
 
            """name="22baf968d547386b9516965ce89d189665003a31" """
 
            """type="checkbox" value="1" />"""
 
        )
 

	
kallithea/tests/functional/test_changeset.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 

	
 
class TestChangesetController(TestController):
 

	
 
    def test_index(self):
 
        response = self.app.get(url(controller='changeset', action='index',
 
                                    repo_name=HG_REPO, revision='tip'))
 
        # Test response...
 

	
 
    def test_changeset_range(self):
 
        #print self.app.get(url(controller='changelog', action='index', repo_name=HG_REPO))
 

	
 
        response = self.app.get(url(controller='changeset', action='index',
 
                                    repo_name=HG_REPO, revision='a53d9201d4bc278910d416d94941b7ea007ecd52...96507bd11ecc815ebc6270fdf6db110928c09c1e'))
 

	
 
        response = self.app.get(url(controller='changeset', action='changeset_raw',
 
                                    repo_name=HG_REPO, revision='a53d9201d4bc278910d416d94941b7ea007ecd52'))
 

	
 
        response = self.app.get(url(controller='changeset', action='changeset_patch',
 
                                    repo_name=HG_REPO, revision='a53d9201d4bc278910d416d94941b7ea007ecd52'))
 

	
 
        response = self.app.get(url(controller='changeset', action='changeset_download',
 
                                    repo_name=HG_REPO, revision='a53d9201d4bc278910d416d94941b7ea007ecd52'))
kallithea/tests/functional/test_changeset_comments.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.model.db import ChangesetComment, Notification, \
 
    UserNotification
 
from kallithea.model.meta import Session
 

	
 

	
 
class TestChangeSetCommentsController(TestController):
 

	
 
    def setup_method(self, method):
 
        for x in ChangesetComment.query().all():
 
            Session().delete(x)
 
        Session().commit()
 

	
 
        self.remove_all_notifications()
 

	
 
    def test_create(self):
 
        self.log_user()
 
        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
 
        text = u'CommentOnRevision'
 

	
 
        params = {'text': text, '_authentication_token': self.authentication_token()}
 
        response = self.app.post(url(controller='changeset', action='comment',
 
                                     repo_name=HG_REPO, revision=rev),
 
                                     params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
 
        # Test response...
 
        assert response.status == '200 OK'
 

	
 
        response = self.app.get(url(controller='changeset', action='index',
 
                                repo_name=HG_REPO, revision=rev))
 
        # test DB
 
        assert ChangesetComment.query().count() == 1
 
        response.mustcontain(
 
            '''<div class="comments-number">'''
 
            ''' 1 comment (0 inline, 1 general)'''
 
        )
 

	
 
        assert Notification.query().count() == 1
 
        assert ChangesetComment.query().count() == 1
 

	
 
        notification = Notification.query().all()[0]
 

	
 
        ID = ChangesetComment.query().first().comment_id
 
        assert notification.type_ == Notification.TYPE_CHANGESET_COMMENT
 
        sbj = (u'/%s/changeset/'
 
               '27cd5cce30c96924232dffcd24178a07ffeb5dfc#comment-%s' % (HG_REPO, ID))
 
        print "%s vs %s" % (sbj, notification.subject)
 
        assert sbj in notification.subject
 

	
 
    def test_create_inline(self):
kallithea/tests/functional/test_compare.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.meta import Session
 
from kallithea.tests.fixture import Fixture
 

	
 
fixture = Fixture()
 

	
 
def _commit_div(sha, msg):
 
    return """<div id="C-%s" class="message">%s</div>""" % (sha, msg)
 

	
 

	
 
class TestCompareController(TestController):
 

	
 
    def setup_method(self, method):
 
        self.r1_id = None
 
        self.r2_id = None
 

	
 
    def teardown_method(self, method):
 
        if self.r2_id:
 
            RepoModel().delete(self.r2_id)
 
        if self.r1_id:
 
            RepoModel().delete(self.r1_id)
 
        Session().commit()
 
        Session.remove()
 

	
 
    def test_compare_forks_on_branch_extra_commits_hg(self):
 
        self.log_user()
 
        repo1 = fixture.create_repo(u'one', repo_type='hg',
 
                                    repo_description='diff-test',
 
                                    cur_user=TEST_USER_ADMIN_LOGIN)
 
        self.r1_id = repo1.repo_id
 
        #commit something !
 
        cs0 = fixture.commit_change(repo1.repo_name, filename='file1',
 
                content='line1\n', message='commit1', vcs_type='hg',
 
                parent=None, newfile=True)
 

	
 
        #fork this repo
 
        repo2 = fixture.create_fork(u'one', u'one-fork')
 
        self.r2_id = repo2.repo_id
 

	
 
        #add two extra commit into fork
 
        cs1 = fixture.commit_change(repo2.repo_name, filename='file1',
 
                content='line1\nline2\n', message='commit2', vcs_type='hg',
 
                parent=cs0)
 

	
 
        cs2 = fixture.commit_change(repo2.repo_name, filename='file1',
 
                content='line1\nline2\nline3\n', message='commit3',
 
                vcs_type='hg', parent=cs1)
 

	
kallithea/tests/functional/test_compare_local.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 

	
 
class TestCompareController(TestController):
 

	
 
    def test_compare_tag_hg(self):
 
        self.log_user()
 
        tag1 = 'v0.1.2'
 
        tag2 = 'v0.1.3'
 
        response = self.app.get(url('compare_url',
 
                                    repo_name=HG_REPO,
 
                                    org_ref_type="tag",
 
                                    org_ref_name=tag1,
 
                                    other_ref_type="tag",
 
                                    other_ref_name=tag2,
 
                                    ), status=200)
 
        response.mustcontain('%s@%s' % (HG_REPO, tag1))
 
        response.mustcontain('%s@%s' % (HG_REPO, tag2))
 

	
 
        ## outgoing changesets between tags
 
        response.mustcontain('''<a class="changeset_hash" href="/%s/changeset/c5ddebc06eaaba3010c2d66ea6ec9d074eb0f678">r112:c5ddebc06eaa</a>''' % HG_REPO)
 
        response.mustcontain('''<a class="changeset_hash" href="/%s/changeset/70d4cef8a37657ee4cf5aabb3bd9f68879769816">r115:70d4cef8a376</a>''' % HG_REPO)
 
        response.mustcontain('''<a class="changeset_hash" href="/%s/changeset/9749bfbfc0d2eba208d7947de266303b67c87cda">r116:9749bfbfc0d2</a>''' % HG_REPO)
 
        response.mustcontain('''<a class="changeset_hash" href="/%s/changeset/41fda979f02fda216374bf8edac4e83f69e7581c">r117:41fda979f02f</a>''' % HG_REPO)
 
        response.mustcontain('''<a class="changeset_hash" href="/%s/changeset/bb1a3ab98cc45cb934a77dcabf87a5a598b59e97">r118:bb1a3ab98cc4</a>''' % HG_REPO)
 
        response.mustcontain('''<a class="changeset_hash" href="/%s/changeset/36e0fc9d2808c5022a24f49d6658330383ed8666">r119:36e0fc9d2808</a>''' % HG_REPO)
 
        response.mustcontain('''<a class="changeset_hash" href="/%s/changeset/17544fbfcd33ffb439e2b728b5d526b1ef30bfcf">r120:17544fbfcd33</a>''' % HG_REPO)
 

	
 
        response.mustcontain('11 files changed with 94 insertions and 64 deletions')
 

	
 
        ## files diff
 
        response.mustcontain(
 
                   '''<div class="node">
 
                          <i class="icon-diff-A"></i>
 
                          <a href="#C--1c5cf9e91c12">docs/api/utils/index.rst</a>
 
                      </div>''')
 
        response.mustcontain(
 
                   '''<div class="node">
 
                          <i class="icon-diff-A"></i>
 
                          <a href="#C--e3305437df55">test_and_report.sh</a>''')
 
        response.mustcontain(
 
                   '''<div class="node">
 
                          <i class="icon-diff-M"></i>
 
                          <a href="#C--c8e92ef85cd1">.hgignore</a>''')
 
        response.mustcontain(
 
                   '''<div class="node">
 
                          <i class="icon-diff-M"></i>
 
                          <a href="#C--6e08b694d687">.hgtags</a>''')
 
        response.mustcontain(
 
                   '''<div class="node">
kallithea/tests/functional/test_feed.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 

	
 
class TestFeedController(TestController):
 

	
 
    def test_rss(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='feed', action='rss',
 
                                    repo_name=HG_REPO))
 

	
 

	
 

	
 
        assert response.content_type == "application/rss+xml"
 
        assert """<rss version="2.0">""" in response
 

	
 
    def test_atom(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='feed', action='atom',
 
                                    repo_name=HG_REPO))
 

	
 
        assert response.content_type == """application/atom+xml"""
 
        assert """<?xml version="1.0" encoding="utf-8"?>""" in response
 
        assert """<feed xmlns="http://www.w3.org/2005/Atom" xml:lang="en-us">""" in response
kallithea/tests/functional/test_files.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
import os
 
import posixpath
 
import mimetypes
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.model.db import Repository
 
from kallithea.model.meta import Session
 
from kallithea.tests.fixture import Fixture
 

	
 
fixture = Fixture()
 

	
 
ARCHIVE_SPECS = {
 
    '.tar.bz2': ('application/x-bzip2', 'tbz2', ''),
 
    '.tar.gz': ('application/x-gzip', 'tgz', ''),
 
    '.zip': ('application/zip', 'zip', ''),
 
}
 

	
 
HG_NODE_HISTORY = fixture.load_resource('hg_node_history_response.json')
 
GIT_NODE_HISTORY = fixture.load_resource('git_node_history_response.json')
 

	
 

	
 
def _set_downloads(repo_name, set_to):
 
    repo = Repository.get_by_repo_name(repo_name)
 
    repo.enable_downloads = set_to
 
    Session().add(repo)
 
    Session().commit()
 

	
 

	
 
class TestFilesController(TestController):
 

	
 
    def test_index(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='index',
 
                                    repo_name=HG_REPO,
 
                                    revision='tip',
 
                                    f_path='/'))
 
        # Test response...
 
        response.mustcontain('<a class="browser-dir ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/docs"><i class="icon-folder-open"></i><span>docs</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-dir ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/vcs"><i class="icon-folder-open"></i><span>vcs</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/.gitignore"><i class="icon-doc"></i><span>.gitignore</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/.hgignore"><i class="icon-doc"></i><span>.hgignore</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/.hgtags"><i class="icon-doc"></i><span>.hgtags</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/.travis.yml"><i class="icon-doc"></i><span>.travis.yml</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/MANIFEST.in"><i class="icon-doc"></i><span>MANIFEST.in</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/README.rst"><i class="icon-doc"></i><span>README.rst</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/run_test_and_report.sh"><i class="icon-doc"></i><span>run_test_and_report.sh</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/setup.cfg"><i class="icon-doc"></i><span>setup.cfg</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/setup.py"><i class="icon-doc"></i><span>setup.py</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/test_and_report.sh"><i class="icon-doc"></i><span>test_and_report.sh</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/tox.ini"><i class="icon-doc"></i><span>tox.ini</span></a>' % HG_REPO)
 

	
 
    def test_index_revision(self):
 
        self.log_user()
kallithea/tests/functional/test_followers.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 

	
 

	
 
class TestFollowersController(TestController):
 

	
 
    def test_index_hg(self):
 
        self.log_user()
 
        repo_name = HG_REPO
 
        response = self.app.get(url(controller='followers',
 
                                    action='followers',
 
                                    repo_name=repo_name))
 

	
 
        response.mustcontain(TEST_USER_ADMIN_LOGIN)
 
        response.mustcontain("""Started following""")
 

	
 
    def test_index_git(self):
 
        self.log_user()
 
        repo_name = GIT_REPO
 
        response = self.app.get(url(controller='followers',
 
                                    action='followers',
 
                                    repo_name=repo_name))
 

	
 
        response.mustcontain(TEST_USER_ADMIN_LOGIN)
 
        response.mustcontain("""Started following""")
kallithea/tests/functional/test_forks.py
Show inline comments
 
# -*- coding: utf-8 -*-
 

	
 
import unittest
 

	
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 

	
 
from kallithea.model.db import Repository
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.user import UserModel
 
from kallithea.model.meta import Session
 

	
 
fixture = Fixture()
 

	
 
class _BaseTestCase(TestController):
 
    """
 
    Write all tests here
 
    """
 
    REPO = None
 
    REPO_TYPE = None
 
    NEW_REPO = None
 
    REPO_FORK = None
 

	
 
    def setup_method(self, method):
 
        self.username = u'forkuser'
 
        self.password = u'qweqwe'
 
        self.u1 = fixture.create_user(self.username, password=self.password,
 
                                      email=u'fork_king@example.com')
 
        Session().commit()
 

	
 
    def teardown_method(self, method):
 
        Session().delete(self.u1)
 
        Session().commit()
 

	
 

	
 
    def test_index(self):
 
        self.log_user()
 
        repo_name = self.REPO
 
        response = self.app.get(url(controller='forks', action='forks',
 
                                    repo_name=repo_name))
 

	
 
        response.mustcontain("""There are no forks yet""")
 

	
 
    def test_no_permissions_to_fork(self):
 
        usr = self.log_user(TEST_USER_REGULAR_LOGIN,
 
                            TEST_USER_REGULAR_PASS)['user_id']
 
        user_model = UserModel()
 
        user_model.revoke_perm(usr, 'hg.fork.repository')
 
        user_model.grant_perm(usr, 'hg.fork.none')
 
        u = UserModel().get(usr)
 
        u.inherit_default_permissions = False
 
        Session().commit()
 
        # try create a fork
kallithea/tests/functional/test_home.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 
from kallithea.model.meta import Session
 
from kallithea.model.db import Repository
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.repo_group import RepoGroupModel
 

	
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestHomeController(TestController):
 

	
 
    def test_index(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='home', action='index'))
 
        #if global permission is set
 
        response.mustcontain('Add Repository')
 
        # html in javascript variable:
 
        response.mustcontain('var data = {"totalRecords": %s' % len(Repository.get_all()))
 
        response.mustcontain(r'href=\"/%s\"' % HG_REPO)
 

	
 
        response.mustcontain(r'<span class="repotag">git')
 
        response.mustcontain(r'<i class=\"icon-globe\"')
 

	
 
        response.mustcontain("""fixes issue with having custom format for git-log""")
 
        response.mustcontain("""/%s/changeset/5f2c6ee195929b0be80749243c18121c9864a3b3""" % GIT_REPO)
 

	
 
        response.mustcontain("""disable security checks on hg clone for travis""")
 
        response.mustcontain("""/%s/changeset/96507bd11ecc815ebc6270fdf6db110928c09c1e""" % HG_REPO)
 

	
 
    def test_repo_summary_with_anonymous_access_disabled(self):
 
        with fixture.anon_access(False):
 
            response = self.app.get(url(controller='summary',
 
                                        action='index', repo_name=HG_REPO),
 
                                        status=302)
 
            assert 'login' in response.location
 

	
 
    def test_index_with_anonymous_access_disabled(self):
 
        with fixture.anon_access(False):
 
            response = self.app.get(url(controller='home', action='index'),
 
                                    status=302)
 
            assert 'login' in response.location
 

	
 
    def test_index_page_on_groups(self):
 
        self.log_user()
 
        gr = fixture.create_repo_group(u'gr1')
 
        fixture.create_repo(name=u'gr1/repo_in_group', repo_group=gr)
 
        response = self.app.get(url('repos_group_home', group_name=u'gr1'))
kallithea/tests/functional/test_journal.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
import datetime
 

	
 

	
 
class TestJournalController(TestController):
 

	
 
    def test_index(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='journal', action='index'))
 

	
 
        response.mustcontain("""<div class="journal_day">%s</div>""" % datetime.date.today())
 

	
 
    def test_stop_following_repository(self):
 
        session = self.log_user()
 
#        usr = Session().query(User).filter(User.username == TEST_USER_ADMIN_LOGIN).one()
 
#        repo = Session().query(Repository).filter(Repository.repo_name == HG_REPO).one()
 
#
 
#        followings = Session().query(UserFollowing) \
 
#            .filter(UserFollowing.user == usr) \
 
#            .filter(UserFollowing.follows_repository == repo).all()
 
#
 
#        assert len(followings) == 1, 'Not following any repository'
 
#
 
#        response = self.app.post(url(controller='journal',
 
#                                     action='toggle_following'),
 
#                                     {'follows_repo_id':repo.repo_id})
 

	
 
    def test_start_following_repository(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='journal', action='index'),)
 

	
 
    def test_public_journal_atom(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='journal', action='public_journal_atom'),)
 

	
 
    def test_public_journal_rss(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='journal', action='public_journal_rss'),)
kallithea/tests/functional/test_login.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
import re
 
import time
 
import urlparse
 

	
 
import mock
 

	
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 
from kallithea.lib.utils2 import generate_api_key
 
from kallithea.lib.auth import check_password
 
from kallithea.lib import helpers as h
 
from kallithea.model.api_key import ApiKeyModel
 
from kallithea.model import validators
 
from kallithea.model.db import User, Notification
 
from kallithea.model.meta import Session
 
from kallithea.model.user import UserModel
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestLoginController(TestController):
 
    def setup_method(self, method):
 
        self.remove_all_notifications()
 
        assert Notification.query().all() == []
 

	
 
    def test_index(self):
 
        response = self.app.get(url(controller='login', action='index'))
 
        assert response.status == '200 OK'
 
        # Test response...
 

	
 
    def test_login_admin_ok(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': TEST_USER_ADMIN_LOGIN,
 
                                  'password': TEST_USER_ADMIN_PASS})
 
        assert response.status == '302 Found'
 
        self.assert_authenticated_user(response, TEST_USER_ADMIN_LOGIN)
 

	
 
        response = response.follow()
 
        response.mustcontain('/%s' % HG_REPO)
 

	
 
    def test_login_regular_ok(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': TEST_USER_REGULAR_LOGIN,
 
                                  'password': TEST_USER_REGULAR_PASS})
 

	
 
        assert response.status == '302 Found'
 
        self.assert_authenticated_user(response, TEST_USER_REGULAR_LOGIN)
 

	
 
        response = response.follow()
 
        response.mustcontain('/%s' % HG_REPO)
 

	
 
    def test_login_regular_email_ok(self):
 
        response = self.app.post(url(controller='login', action='index'),
 
                                 {'username': TEST_USER_REGULAR_EMAIL,
 
                                  'password': TEST_USER_REGULAR_PASS})
kallithea/tests/functional/test_my_account.py
Show inline comments
 
# -*- coding: utf-8 -*-
 

	
 
from kallithea.model.db import User, UserFollowing, Repository, UserApiKeys
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 
from kallithea.lib import helpers as h
 
from kallithea.model.user import UserModel
 
from kallithea.model.meta import Session
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestMyAccountController(TestController):
 
    test_user_1 = 'testme'
 

	
 
    @classmethod
 
    def teardown_class(cls):
 
        if User.get_by_username(cls.test_user_1):
 
            UserModel().delete(cls.test_user_1)
 
            Session().commit()
 

	
 
    def test_my_account(self):
 
        self.log_user()
 
        response = self.app.get(url('my_account'))
 

	
 
        response.mustcontain('value="%s' % TEST_USER_ADMIN_LOGIN)
 

	
 
    def test_my_account_my_repos(self):
 
        self.log_user()
 
        response = self.app.get(url('my_account_repos'))
 
        cnt = Repository.query().filter(Repository.user ==
 
                           User.get_by_username(TEST_USER_ADMIN_LOGIN)).count()
 
        response.mustcontain('"totalRecords": %s' % cnt)
 

	
 
    def test_my_account_my_watched(self):
 
        self.log_user()
 
        response = self.app.get(url('my_account_watched'))
 

	
 
        cnt = UserFollowing.query().filter(UserFollowing.user ==
 
                            User.get_by_username(TEST_USER_ADMIN_LOGIN)).count()
 
        response.mustcontain('"totalRecords": %s' % cnt)
 

	
 
    def test_my_account_my_emails(self):
 
        self.log_user()
 
        response = self.app.get(url('my_account_emails'))
 
        response.mustcontain('No additional emails specified')
 

	
 
    def test_my_account_my_emails_add_existing_email(self):
 
        self.log_user()
 
        response = self.app.get(url('my_account_emails'))
 
        response.mustcontain('No additional emails specified')
 
        response = self.app.post(url('my_account_emails'),
kallithea/tests/functional/test_pullrequests.py
Show inline comments
 
import re
 

	
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 
from kallithea.model.meta import Session
 

	
 
from kallithea.controllers.pullrequests import PullrequestsController
 

	
 
fixture = Fixture()
 

	
 
class TestPullrequestsController(TestController):
 

	
 
    def test_index(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='pullrequests', action='index',
 
                                    repo_name=HG_REPO))
 

	
 
    def test_create_trivial(self):
 
        self.log_user()
 
        response = self.app.post(url(controller='pullrequests', action='create',
 
                                     repo_name=HG_REPO),
 
                                 {'org_repo': HG_REPO,
 
                                  'org_ref': 'branch:default:default',
 
                                  'other_repo': HG_REPO,
 
                                  'other_ref': 'branch:default:default',
 
                                  'pullrequest_title': 'title',
 
                                  'pullrequest_desc': 'description',
 
                                  '_authentication_token': self.authentication_token(),
 
                                 }
 
                                )
 
        assert response.status == '302 Found'
 
        response = response.follow()
 
        assert response.status == '200 OK'
 
        response.mustcontain('This pull request has already been merged to default.')
 

	
 
    def test_create_with_existing_reviewer(self):
 
        self.log_user()
 
        response = self.app.post(url(controller='pullrequests', action='create',
 
                                     repo_name=HG_REPO),
 
                                 {'org_repo': HG_REPO,
 
                                  'org_ref': 'branch:default:default',
 
                                  'other_repo': HG_REPO,
 
                                  'other_ref': 'branch:default:default',
 
                                  'pullrequest_title': 'title',
 
                                  'pullrequest_desc': 'description',
 
                                  '_authentication_token': self.authentication_token(),
 
                                  'review_members': TEST_USER_ADMIN_LOGIN,
 
                                 }
 
                                )
 
        assert response.status == '302 Found'
 
        response = response.follow()
kallithea/tests/functional/test_repo_groups.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 

	
 

	
 
class TestRepoGroupsController(TestController):
 

	
 
    def test_index(self):
 
        self.log_user()
 
        response = self.app.get(url('repos_groups'))
 
        response.mustcontain('{"totalRecords": 0, "sort": null, "startIndex": 0, "dir": "asc", "records": []};')
 

	
 
    def test_new(self):
 
        self.log_user()
 
        response = self.app.get(url('new_repos_group'))
 

	
 
    def test_new_by_regular_user(self):
 
        self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
 
        response = self.app.get(url('new_repos_group'), status=403)
kallithea/tests/functional/test_search.py
Show inline comments
 
import mock
 
import os
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 

	
 

	
 
class TestSearchController(TestController):
 

	
 
    def test_index(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='search', action='index'))
 

	
 
        response.mustcontain('class="small" id="q" name="q" type="text"')
 
        # Test response...
 

	
 
    def test_empty_search(self, tmpdir):
 
        self.log_user()
 

	
 
        config_mock = {
 
            'app_conf': {
 
                # can be any existing dir that does not contain an actual index
 
                'index_dir': str(tmpdir),
 
            }
 
        }
 
        with mock.patch('kallithea.controllers.search.config', config_mock):
 
            response = self.app.get(url(controller='search', action='index'),
 
                                    {'q': HG_REPO})
 
            response.mustcontain('There is no index to search in. '
 
                                 'Please run whoosh indexer')
 

	
 
    def test_normal_search(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='search', action='index'),
 
                                {'q': 'def repo'})
 
        response.mustcontain('58 results')
 

	
 
    def test_repo_search(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='search', action='index'),
 
                                {'q': 'repository:%s def test' % HG_REPO})
 

	
 
        response.mustcontain('18 results')
 

	
 
    def test_search_last(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='search', action='index'),
 
                                {'q': 'last:t', 'type': 'commit'})
 

	
 
        response.mustcontain('2 results')
 

	
 
    def test_search_commit_message(self):
 
        self.log_user()
kallithea/tests/functional/test_summary.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/>.
 

	
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 
from kallithea.model.db import Repository
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.meta import Session
 
from kallithea.model.scm import ScmModel
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestSummaryController(TestController):
 

	
 
    def test_index(self):
 
        self.log_user()
 
        ID = Repository.get_by_repo_name(HG_REPO).repo_id
 
        response = self.app.get(url(controller='summary',
 
                                    action='index',
 
                                    repo_name=HG_REPO))
 

	
 
        #repo type
 
        response.mustcontain(
 
            """<span class="repotag">hg"""
 
        )
 
        #public/private
 
        response.mustcontain(
 
            """<i class="icon-globe">"""
 
        )
 

	
 
        # clone url...
 
        response.mustcontain('''id="clone_url" readonly="readonly" value="http://%s@localhost:80/%s"''' % (TEST_USER_ADMIN_LOGIN, HG_REPO))
 
        response.mustcontain('''id="clone_url_id" readonly="readonly" value="http://%s@localhost:80/_%s"''' % (TEST_USER_ADMIN_LOGIN, ID))
 

	
 
    def test_index_git(self):
 
        self.log_user()
 
        ID = Repository.get_by_repo_name(GIT_REPO).repo_id
 
        response = self.app.get(url(controller='summary',
 
                                    action='index',
 
                                    repo_name=GIT_REPO))
 

	
 
        #repo type
 
        response.mustcontain(
 
            """<span class="repotag">git"""
 
        )
 
        #public/private
 
        response.mustcontain(
 
            """<i class="icon-globe">"""
 
        )
 

	
 
        # clone url...
kallithea/tests/models/common.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 

	
 
from kallithea.model.repo_group import RepoGroupModel
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.db import RepoGroup, Repository, User
 
from kallithea.model.user import UserModel
 

	
 
from kallithea.lib.auth import AuthUser
 
from kallithea.model.meta import Session
 

	
 

	
 
fixture = Fixture()
 

	
 

	
 
def _destroy_project_tree(test_u1_id):
 
    Session.remove()
 
    repo_group = RepoGroup.get_by_group_name(group_name=u'g0')
 
    for el in reversed(repo_group.recursive_groups_and_repos()):
 
        if isinstance(el, Repository):
 
            RepoModel().delete(el)
 
        elif isinstance(el, RepoGroup):
 
            RepoGroupModel().delete(el, force_delete=True)
 

	
 
    u = User.get(test_u1_id)
 
    Session().delete(u)
 
    Session().commit()
 

	
 

	
 
def _create_project_tree():
 
    """
 
    Creates a tree of groups and repositories to test permissions
 

	
 
    structure
 
     [g0] - group `g0` with 3 subgroups
 
     |
 
     |__[g0_1] group g0_1 with 2 groups 0 repos
 
     |  |
 
     |  |__[g0_1_1] group g0_1_1 with 1 group 2 repos
 
     |  |   |__<g0/g0_1/g0_1_1/g0_1_1_r1>
 
     |  |   |__<g0/g0_1/g0_1_1/g0_1_1_r2>
 
     |  |__<g0/g0_1/g0_1_r1>
 
     |
 
     |__[g0_2] 2 repos
 
     |  |
 
     |  |__<g0/g0_2/g0_2_r1>
 
     |  |__<g0/g0_2/g0_2_r2>
 
     |
 
     |__[g0_3] 1 repo
kallithea/tests/models/test_changeset_status.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.model.changeset_status import ChangesetStatusModel
 
from kallithea.model.db import ChangesetStatus as CS
 

	
 
class CSM(object): # ChangesetStatusMock
 

	
 
    def __init__(self, status):
 
        self.status = status
 

	
 
class TestChangesetStatusCalculation(TestController):
 

	
 
    def setup_method(self, method):
 
        self.m = ChangesetStatusModel()
 

	
 
    @parametrize('name,expected_result,statuses', [
 
        ('empty list', CS.STATUS_UNDER_REVIEW, []),
 
        ('approve', CS.STATUS_APPROVED, [CSM(CS.STATUS_APPROVED)]),
 
        ('approve2', CS.STATUS_APPROVED, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_APPROVED)]),
 
        ('approve_reject', CS.STATUS_REJECTED, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_REJECTED)]),
 
        ('approve_underreview', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_UNDER_REVIEW)]),
 
        ('approve_notreviewed', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_NOT_REVIEWED)]),
 
        ('underreview', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_UNDER_REVIEW), CSM(CS.STATUS_UNDER_REVIEW)]),
 
        ('reject', CS.STATUS_REJECTED, [CSM(CS.STATUS_REJECTED)]),
 
        ('reject_underreview', CS.STATUS_REJECTED, [CSM(CS.STATUS_REJECTED), CSM(CS.STATUS_UNDER_REVIEW)]),
 
        ('reject_notreviewed', CS.STATUS_REJECTED, [CSM(CS.STATUS_REJECTED), CSM(CS.STATUS_NOT_REVIEWED)]),
 
        ('notreviewed', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_NOT_REVIEWED)]),
 
        ('approve_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_APPROVED), None]),
 
        ('approve2_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_APPROVED), None]),
 
        ('approve_reject_none', CS.STATUS_REJECTED, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_REJECTED), None]),
 
        ('approve_underreview_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_UNDER_REVIEW), None]),
 
        ('approve_notreviewed_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_APPROVED), CSM(CS.STATUS_NOT_REVIEWED), None]),
 
        ('underreview_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_UNDER_REVIEW), CSM(CS.STATUS_UNDER_REVIEW), None]),
 
        ('reject_none', CS.STATUS_REJECTED, [CSM(CS.STATUS_REJECTED), None]),
 
        ('reject_underreview_none', CS.STATUS_REJECTED, [CSM(CS.STATUS_REJECTED), CSM(CS.STATUS_UNDER_REVIEW), None]),
 
        ('reject_notreviewed_none', CS.STATUS_REJECTED, [CSM(CS.STATUS_REJECTED), CSM(CS.STATUS_NOT_REVIEWED), None]),
 
        ('notreviewed_none', CS.STATUS_UNDER_REVIEW, [CSM(CS.STATUS_NOT_REVIEWED), None]),
 
    ])
 
    def test_result(self, name, expected_result, statuses):
 
        result = self.m._calculate_status(statuses)
 
        assert result == expected_result
kallithea/tests/models/test_diff_parsers.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.lib.diffs import DiffProcessor, NEW_FILENODE, DEL_FILENODE, \
 
    MOD_FILENODE, RENAMED_FILENODE, CHMOD_FILENODE, BIN_FILENODE, COPIED_FILENODE
 
from kallithea.tests.fixture import Fixture
 

	
 
fixture = Fixture()
 

	
 

	
 
DIFF_FIXTURES = {
 
    'hg_diff_add_single_binary_file.diff': [
 
        ('US Warszawa.jpg', 'A',
 
         {'added': 0,
 
          'deleted': 0,
 
          'binary': True,
 
          'ops': {NEW_FILENODE: 'new file 100755',
 
                  BIN_FILENODE: 'binary diff not shown'}}),
 
    ],
 
    'hg_diff_mod_single_binary_file.diff': [
 
        ('US Warszawa.jpg', 'M',
 
         {'added': 0,
 
          'deleted': 0,
 
          'binary': True,
 
          'ops': {MOD_FILENODE: 'modified file',
 
                  BIN_FILENODE: 'binary diff not shown'}}),
 
    ],
 

	
 
    'hg_diff_mod_single_file_and_rename_and_chmod.diff': [
 
        ('README', 'R',
 
         {'added': 3,
 
          'deleted': 0,
 
          'binary': False,
 
          'ops': {RENAMED_FILENODE: 'file renamed from README.rst to README',
 
                  CHMOD_FILENODE: 'modified file chmod 100755 => 100644'}}),
 
    ],
 
    'hg_diff_mod_file_and_rename.diff': [
 
        ('README.rst', 'R',
 
         {'added': 3,
 
          'deleted': 0,
 
          'binary': False,
 
          'ops': {RENAMED_FILENODE: 'file renamed from README to README.rst'}}),
 
    ],
 
    'hg_diff_del_single_binary_file.diff': [
 
        ('US Warszawa.jpg', 'D',
 
         {'added': 0,
 
          'deleted': 0,
 
          'binary': True,
 
          'ops': {DEL_FILENODE: 'deleted file',
 
                  BIN_FILENODE: 'binary diff not shown'}}),
 
    ],
kallithea/tests/models/test_notifications.py
Show inline comments
 
import os
 
import re
 

	
 
import mock
 
import routes.util
 

	
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.lib import helpers as h
 
from kallithea.model.db import User, Notification, UserNotification
 
from kallithea.model.user import UserModel
 
from kallithea.model.meta import Session
 
from kallithea.model.notification import NotificationModel, EmailNotificationModel
 

	
 
import kallithea.lib.celerylib
 
import kallithea.lib.celerylib.tasks
 

	
 

	
 
class TestNotifications(TestController):
 

	
 
    def setup_method(self, method):
 
        Session.remove()
 
        u1 = UserModel().create_or_update(username=u'u1',
 
                                        password=u'qweqwe',
 
                                        email=u'u1@example.com',
 
                                        firstname=u'u1', lastname=u'u1')
 
        Session().commit()
 
        self.u1 = u1.user_id
 

	
 
        u2 = UserModel().create_or_update(username=u'u2',
 
                                        password=u'qweqwe',
 
                                        email=u'u2@example.com',
 
                                        firstname=u'u2', lastname=u'u3')
 
        Session().commit()
 
        self.u2 = u2.user_id
 

	
 
        u3 = UserModel().create_or_update(username=u'u3',
 
                                        password=u'qweqwe',
 
                                        email=u'u3@example.com',
 
                                        firstname=u'u3', lastname=u'u3')
 
        Session().commit()
 
        self.u3 = u3.user_id
 

	
 
        self.remove_all_notifications()
 
        assert [] == Notification.query().all()
 
        assert [] == UserNotification.query().all()
 

	
 
    def test_create_notification(self):
 
        usrs = [self.u1, self.u2]
 
        def send_email(recipients, subject, body='', html_body='', headers=None, author=None):
 
            assert recipients == ['u2@example.com']
 
            assert subject == 'Test Message'
 
            assert body == u"hi there"
 
            assert '>hi there<' in html_body
 
            assert author.username == 'u1'
 
        with mock.patch.object(kallithea.lib.celerylib.tasks, 'send_email', send_email):
kallithea/tests/models/test_permissions.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 
from kallithea.model.repo_group import RepoGroupModel
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.db import RepoGroup, User, UserGroupRepoGroupToPerm, \
 
    Permission, UserToPerm
 
from kallithea.model.user import UserModel
 

	
 
from kallithea.model.meta import Session
 
from kallithea.model.user_group import UserGroupModel
 
from kallithea.lib.auth import AuthUser
 
from kallithea.model.permission import PermissionModel
 

	
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestPermissions(TestController):
 

	
 
    @classmethod
 
    def setup_class(cls):
 
        #recreate default user to get a clean start
 
        PermissionModel().create_default_permissions(user=User.DEFAULT_USER,
 
                                                     force=True)
 
        Session().commit()
 

	
 
    def setup_method(self, method):
 
        self.u1 = UserModel().create_or_update(
 
            username=u'u1', password=u'qweqwe',
 
            email=u'u1@example.com', firstname=u'u1', lastname=u'u1'
 
        )
 
        self.u2 = UserModel().create_or_update(
 
            username=u'u2', password=u'qweqwe',
 
            email=u'u2@example.com', firstname=u'u2', lastname=u'u2'
 
        )
 
        self.u3 = UserModel().create_or_update(
 
            username=u'u3', password=u'qweqwe',
 
            email=u'u3@example.com', firstname=u'u3', lastname=u'u3'
 
        )
 
        self.anon = User.get_default_user()
 
        self.a1 = UserModel().create_or_update(
 
            username=u'a1', password=u'qweqwe',
 
            email=u'a1@example.com', firstname=u'a1', lastname=u'a1', admin=True
 
        )
 
        Session().commit()
 

	
 
    def teardown_method(self, method):
 
        if hasattr(self, 'test_repo'):
 
            RepoModel().delete(repo=self.test_repo)
kallithea/tests/models/test_repo_groups.py
Show inline comments
 
import os
 
import pytest
 
from sqlalchemy.exc import IntegrityError
 

	
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 

	
 
from kallithea.model.repo_group import RepoGroupModel
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.db import RepoGroup
 
from kallithea.model.meta import Session
 

	
 

	
 
fixture = Fixture()
 

	
 

	
 
def _update_group(id_, group_name, desc=u'desc', parent_id=None):
 
    form_data = fixture._get_group_create_params(group_name=group_name,
 
                                                 group_desc=desc,
 
                                                 group_parent_id=parent_id)
 
    gr = RepoGroupModel().update(id_, form_data)
 
    return gr
 

	
 

	
 
def _update_repo(name, **kwargs):
 
    form_data = fixture._get_repo_create_params(**kwargs)
 
    if not 'repo_name' in kwargs:
 
        form_data['repo_name'] = name
 
    if not 'perms_new' in kwargs:
 
        form_data['perms_new'] = []
 
    if not 'perms_updates' in kwargs:
 
        form_data['perms_updates'] = []
 
    r = RepoModel().update(name, **form_data)
 
    return r
 

	
 

	
 
class TestRepoGroups(TestController):
 

	
 
    def setup_method(self, method):
 
        self.g1 = fixture.create_repo_group(u'test1', skip_if_exists=True)
 
        self.g2 = fixture.create_repo_group(u'test2', skip_if_exists=True)
 
        self.g3 = fixture.create_repo_group(u'test3', skip_if_exists=True)
 

	
 
    def teardown_method(self, method):
 
        Session.remove()
 

	
 
    def __check_path(self, *path):
 
        """
 
        Checks the path for existence !
 
        """
 
        path = [TESTS_TMP_PATH] + list(path)
 
        path = os.path.join(*path)
 
        return os.path.isdir(path)
kallithea/tests/models/test_repos.py
Show inline comments
 
import pytest
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 

	
 
from kallithea.model.meta import Session
 
from kallithea.tests.fixture import Fixture
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.db import Repository
 
from kallithea.lib.exceptions import AttachedForksError
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestRepos(TestController):
 

	
 
    def teardown_method(self, method):
 
        Session.remove()
 

	
 
    def test_remove_repo(self):
 
        repo = fixture.create_repo(name=u'test-repo-1')
 
        Session().commit()
 

	
 
        RepoModel().delete(repo=repo)
 
        Session().commit()
 

	
 
        assert None == Repository.get_by_repo_name(repo_name=u'test-repo-1')
 

	
 
    def test_remove_repo_repo_raises_exc_when_attached_forks(self):
 
        repo = fixture.create_repo(name=u'test-repo-1')
 
        Session().commit()
 

	
 
        fixture.create_fork(repo.repo_name, u'test-repo-fork-1')
 
        Session().commit()
 

	
 
        with pytest.raises(AttachedForksError):
 
            RepoModel().delete(repo=repo)
 
        # cleanup
 
        RepoModel().delete(repo=u'test-repo-fork-1')
 
        RepoModel().delete(repo=u'test-repo-1')
 
        Session().commit()
 

	
 
    def test_remove_repo_delete_forks(self):
 
        repo = fixture.create_repo(name=u'test-repo-1')
 
        Session().commit()
 

	
 
        fork = fixture.create_fork(repo.repo_name, u'test-repo-fork-1')
 
        Session().commit()
 

	
 
        #fork of fork
 
        fixture.create_fork(fork.repo_name, u'test-repo-fork-fork-1')
 
        Session().commit()
kallithea/tests/models/test_user_groups.py
Show inline comments
 
from kallithea.model.db import User, UserGroup
 

	
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 

	
 
from kallithea.model.user_group import UserGroupModel
 
from kallithea.model.meta import Session
 

	
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestUserGroups(TestController):
 

	
 
    def teardown_method(self, method):
 
        # delete all groups
 
        for gr in UserGroup.get_all():
 
            fixture.destroy_user_group(gr)
 
        Session().commit()
 

	
 
    @parametrize('pre_existing,regular_should_be,external_should_be,groups,expected', [
 
        ([], [], [], [], []),
 
        ([], [u'regular'], [], [], [u'regular']),  # no changes of regular
 
        ([u'some_other'], [], [], [u'some_other'], []),   # not added to regular group
 
        ([], [u'regular'], [u'container'], [u'container'], [u'regular', u'container']),
 
        ([], [u'regular'], [], [u'container', u'container2'], [u'regular', u'container', u'container2']),
 
        ([], [u'regular'], [u'other'], [], [u'regular']),  # remove not used
 
        ([u'some_other'], [u'regular'], [u'other', u'container'], [u'container', u'container2'], [u'regular', u'container', u'container2']),
 
    ])
 
    def test_enforce_groups(self, pre_existing, regular_should_be,
 
                            external_should_be, groups, expected):
 
        # delete all groups
 
        for gr in UserGroup.get_all():
 
            fixture.destroy_user_group(gr)
 
        Session().commit()
 

	
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
 
        for gr in pre_existing:
 
            gr = fixture.create_user_group(gr)
 
        Session().commit()
 

	
 
        # make sure use is just in those groups
 
        for gr in regular_should_be:
 
            gr = fixture.create_user_group(gr)
 
            Session().commit()
 
            UserGroupModel().add_user_to_group(gr, user)
 
            Session().commit()
 

	
 
        # now special external groups created by auth plugins
 
        for gr in external_should_be:
 
            gr = fixture.create_user_group(gr, user_group_data={'extern_type': 'container'})
kallithea/tests/models/test_users.py
Show inline comments
 
import pytest
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 

	
 
from kallithea.model.db import User, UserGroup, UserGroupMember, UserEmailMap, \
 
    Permission
 
from kallithea.model.user import UserModel
 

	
 
from kallithea.model.meta import Session
 
from kallithea.model.user_group import UserGroupModel
 
from kallithea.tests.fixture import Fixture
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestUser(TestController):
 

	
 
    @classmethod
 
    def setup_class(cls):
 
        Session.remove()
 

	
 
    def teardown_method(self, method):
 
        Session.remove()
 

	
 
    def test_create_and_remove(self):
 
        usr = UserModel().create_or_update(username=u'test_user',
 
                                           password=u'qweqwe',
 
                                           email=u'u232@example.com',
 
                                           firstname=u'u1', lastname=u'u1')
 
        Session().commit()
 
        assert User.get_by_username(u'test_user') == usr
 
        assert User.get_by_username(u'test_USER', case_insensitive=True) == usr
 
        assert User.get_by_username(u'test_USER', case_insensitive=False) == None
 

	
 
        # make user group
 
        user_group = fixture.create_user_group(u'some_example_group')
 
        Session().commit()
 

	
 
        UserGroupModel().add_user_to_group(user_group, usr)
 
        Session().commit()
 

	
 
        assert UserGroup.get(user_group.users_group_id) == user_group
 
        assert UserGroupMember.query().count() == 1
 
        UserModel().delete(usr.user_id)
 
        Session().commit()
 

	
 
        assert UserGroupMember.query().all() == []
 

	
 
    def test_additional_email_as_main(self):
 
        usr = UserModel().create_or_update(username=u'test_user',
 
                                           password=u'qweqwe',
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 py.test 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 os
 
import re
 
import tempfile
 
import time
 

	
 
from tempfile import _RandomNameSequence
 
from subprocess import Popen, PIPE
 

	
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 
from kallithea.tests.fixture import Fixture
 
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
 

	
 
fixture = Fixture()
 

	
 

	
 
class Command(object):
 

	
 
    def __init__(self, cwd):
 
        self.cwd = cwd
 

	
 
    def execute(self, cmd, *args, **environ):
 
        """
 
        Runs command on the system with given ``args``.
 
        """
 

	
 
        command = cmd + ' ' + ' '.join(args)
 
        if DEBUG:
 
            print '*** CMD %s ***' % command
 
        testenv = dict(os.environ)
 
        testenv['LANG'] = 'en_US.UTF-8'
 
        testenv['LANGUAGE'] = 'en_US:en'
 
        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:', repr(stdout)
 
            if stderr:
 
                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 = {
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.tests.base 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(TestController):
 

	
 
    @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
 
        assert uri_filter(test_url) == expected
 

	
 
    @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
kallithea/tests/other/test_mail.py
Show inline comments
 
import mock
 

	
 
import kallithea
 
from kallithea.tests import *
 
from kallithea.tests.base 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(TestController):
 

	
 
    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)
 

	
 
        assert smtplib_mock.lastdest == set(recipients)
 
        assert smtplib_mock.lastsender == envelope_from
 
        assert 'From: %s' % envelope_from in smtplib_mock.lastmsg
 
        assert 'Subject: %s' % subject in smtplib_mock.lastmsg
 
        assert body in smtplib_mock.lastmsg
 
        assert html_body in 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'
kallithea/tests/other/test_validators.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
import formencode
 
import pytest
 
import tempfile
 

	
 
from kallithea.tests import *
 
from kallithea.tests.base 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(TestController):
 

	
 
    def teardown_method(self, method):
 
        Session.remove()
 

	
 
    def test_Message_extractor(self):
 
        validator = v.ValidUsername()
 
        with pytest.raises(formencode.Invalid):
 
            validator.to_python('default')
 

	
 
        class StateObj(object):
 
            pass
 

	
 
        with pytest.raises(formencode.Invalid):
 
            validator.to_python('default', StateObj)
 

	
 
    def test_ValidUsername(self):
 
        validator = v.ValidUsername()
 

	
 
        with pytest.raises(formencode.Invalid):
 
            validator.to_python('default')
 
        with pytest.raises(formencode.Invalid):
 
            validator.to_python('new_user')
 
        with pytest.raises(formencode.Invalid):
 
            validator.to_python('.,')
 
        with pytest.raises(formencode.Invalid):
 
            validator.to_python(TEST_USER_ADMIN_LOGIN)
 
        assert 'test' == validator.to_python('test')
 

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

	
 
    def test_ValidRepoUser(self):
 
        validator = v.ValidRepoUser()
 
        with pytest.raises(formencode.Invalid):
 
            validator.to_python('nouser')
 
        assert TEST_USER_ADMIN_LOGIN == validator.to_python(TEST_USER_ADMIN_LOGIN)
 

	
kallithea/tests/scripts/manual_test_concurrency.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.scripts.manual_test_concurrency
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

	
 
Test suite for making push/pull operations
 

	
 
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 os
 
import sys
 
import shutil
 
import logging
 
from os.path import dirname
 

	
 
from tempfile import _RandomNameSequence
 
from subprocess import Popen, PIPE
 

	
 
from paste.deploy import appconfig
 
from sqlalchemy import engine_from_config
 

	
 
from kallithea.lib.utils import setup_cache_regions
 
from kallithea.model import init_model
 
from kallithea.model import meta
 
from kallithea.model.db import User, Repository, Ui
 
from kallithea.lib.auth import get_crypt_password
 

	
 
from kallithea.tests import HG_REPO
 
from kallithea.tests.base import HG_REPO
 
from kallithea.config.environment import load_environment
 

	
 
rel_path = dirname(dirname(dirname(dirname(os.path.abspath(__file__)))))
 
conf = appconfig('config:development.ini', relative_to=rel_path)
 
load_environment(conf.global_conf, conf.local_conf)
 

	
 
setup_cache_regions(conf)
 

	
 
USER = TEST_USER_ADMIN_LOGIN
 
PASS = TEST_USER_ADMIN_PASS
 
HOST = 'server.local'
 
METHOD = 'pull'
 
DEBUG = True
 
log = logging.getLogger(__name__)
 

	
 

	
 
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)
 
        log.debug('Executing %s', command)
 
        if DEBUG:
 
            print command
 
        p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd)
 
        stdout, stderr = p.communicate()
 
        if DEBUG:
 
            print stdout, stderr
 
        return stdout, stderr
 

	
 

	
 
def get_session():
 
    engine = engine_from_config(conf, 'sqlalchemy.')
 
    init_model(engine)
 
    sa = meta.Session
 
    return sa
 

	
 

	
 
def create_test_user(force=True):
 
    print 'creating test user'
 
    sa = get_session()
 

	
 
    user = sa.query(User).filter(User.username == USER).scalar()
kallithea/tests/vcs/__init__.py
Show inline comments
 
"""
 
Unit tests for vcs_ library.
 

	
 
In order to run tests we need to prepare our environment first. Tests would be
 
run for each engine listed at ``conf.SCM_TESTS`` - keys are aliases from
 
``vcs.backends.BACKENDS``.
 

	
 
For each SCM we run tests for, we need some repository. We would use
 
repositories location from system environment variables or test suite defaults
 
- see ``conf`` module for more detail. We simply try to check if repository at
 
certain location exists, if not we would try to fetch them. At ``test_vcs`` or
 
``test_common`` we run unit tests common for each repository type and for
 
example specific mercurial tests are located at ``test_hg`` module.
 

	
 
Oh, and tests are run with ``unittest.collector`` wrapped by ``collector``
 
function at ``tests/__init__.py``.
 

	
 
.. _vcs: http://bitbucket.org/marcinkuzminski/vcs
 
.. _unittest: http://pypi.python.org/pypi/unittest
 

	
 
"""
 
from kallithea.lib.vcs.utils.compat import unittest
 
from kallithea.tests.vcs.conf import *
 
from kallithea.tests.vcs.utils import SCMFetcher
 

	
 
from kallithea.tests import *
 
from kallithea.tests.base import *
 

	
 

	
 
def setup_package():
 
    """
 
    Prepares whole package for tests which mainly means it would try to fetch
 
    test repositories or use already existing ones.
 
    """
 
    fetchers = {
 
        'hg': {
 
            'alias': 'hg',
 
            'test_repo_path': TEST_HG_REPO,
 
            'remote_repo': HG_REMOTE_REPO,
 
            'clone_cmd': 'hg clone --insecure',
 
        },
 
        'git': {
 
            'alias': 'git',
 
            'test_repo_path': TEST_GIT_REPO,
 
            'remote_repo': GIT_REMOTE_REPO,
 
            'clone_cmd': 'git clone --bare',
 
        },
 
    }
 

	
 
    for scm, fetcher_info in fetchers.items():
 
        fetcher = SCMFetcher(**fetcher_info)
 
        fetcher.setup()
 

	
 

	
 
def collector():
 
    setup_package()
 
    start_dir = os.path.abspath(os.path.dirname(__file__))
 
    return unittest.defaultTestLoader.discover(start_dir)
 

	
 

	
 
def main():
 
    collector()
 
    unittest.main()
 

	
 
#if __name__ == '__main__':
 
#    main()
kallithea/tests/vcs/conf.py
Show inline comments
 
@@ -39,49 +39,49 @@ TEST_HG_REPO_PULL = os.environ.get('VCS_
 

	
 
TEST_DIR = os.environ.get('VCS_TEST_ROOT', tempfile.gettempdir())
 
TEST_REPO_PREFIX = 'vcs-test'
 

	
 

	
 
def get_new_dir(title=None):
 
    """
 
    Calculates a path for a new, non-existant, unique sub-directory in TEST_DIR.
 

	
 
    Resulting directory name will have format:
 

	
 
    prefix-[title-]hexuuid
 

	
 
    Prefix is equal to value of variable TEST_REPO_PREFIX. The "hexuuid" is a
 
    hexadecimal value of a randomly generated UUID. Title will be added if
 
    specified.
 

	
 
    Args:
 
        title: Custom title to include as part of the resulting sub-directory
 
            name. Can be useful for debugging to identify destination. Defaults
 
            to None.
 

	
 
    Returns:
 
        Path to the new directory as a string.
 
    """
 

	
 
    if title:
 
        name = "%s-%s" % (TEST_REPO_PREFIX, title)
 
    else:
 
        name = TEST_REPO_PREFIX
 

	
 
    path = os.path.join(TEST_DIR, name)
 

	
 
    # Generate new hexes until we get a unique name (just in case).
 
    hex_uuid = uuid.uuid4().hex
 
    while os.path.exists("%s-%s" % (path, hex_uuid)):
 
        hex_uuid = uuid.uuid4().hex
 

	
 
    return "%s-%s" % (path, hex_uuid)
 

	
 

	
 
PACKAGE_DIR = os.path.abspath(os.path.join(
 
    os.path.dirname(__file__), '..'))
 
_dest = os.path.join(TEST_TMP_PATH, 'aconfig')
 
shutil.copy(os.path.join(THIS, 'aconfig'), _dest)
 
TEST_USER_CONFIG_FILE = _dest
 

	
 
#overide default configurations with kallithea ones
 
from kallithea.tests import *
 
from kallithea.tests.base import *
0 comments (0 inline, 0 general)