Changeset - 7f2aa3ec2931
[Not reviewed]
default
0 45 0
Thomas De Schampheleire - 10 years ago 2016-05-15 22:37:37
thomas.de.schampheleire@gmail.com
pytest migration: rename TestControllerPytest back to TestController

The name TestControllerPytest was introduced to allow a temporary situation
where nose/unittest and pytest-based tests could coexist.
This situation is now over, so the base test class can be renamed again.
45 files changed with 51 insertions and 51 deletions:
0 comments (0 inline, 0 general)
kallithea/tests/__init__.py
Show inline comments
 
@@ -16,184 +16,184 @@
 
Pylons application test package
 

	
 
This package assumes the Pylons environment is already loaded.
 

	
 
This module initializes the application via ``websetup`` (`paster
 
setup-app`) and provides the base testing objects.
 

	
 
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 os.path import join as jn
 

	
 
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.tests.parameterized import parameterized
 
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', 'TestControllerPytest',
 
    'skipif', 'parametrize', 'environ', 'url', 'TestController',
 
    'ldap_lib_installed', 'pam_lib_installed',
 
    '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 = jn(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 = jn(TESTS_TMP_PATH, GIT_REPO)
 
TEST_GIT_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)
 
TEST_GIT_REPO_PULL = jn(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)
 

	
 

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

	
 
TEST_HG_REPO = jn(TESTS_TMP_PATH, HG_REPO)
 
TEST_HG_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)
 
TEST_HG_REPO_PULL = jn(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 = jn(TESTS_TMP_PATH, GIT_REPO)
 
HG_REMOTE_REPO = jn(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
 

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

	
 
class TestControllerPytest(object):
 
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):
 
        Notification.query().delete()
 

	
 
        # Because query().delete() does not (by default) trigger cascades.
 
        # http://docs.sqlalchemy.org/en/rel_0_7/orm/collections.html#passive-deletes
 
        UserNotification.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()
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 TestControllerPytest, GIT_REPO
 
from kallithea.tests import TestController, GIT_REPO
 
from kallithea.tests.api.api_base import _BaseTestApi
 

	
 

	
 
class TestGitApi(_BaseTestApi, TestControllerPytest):
 
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 TestControllerPytest, HG_REPO
 
from kallithea.tests import TestController, HG_REPO
 
from kallithea.tests.api.api_base import _BaseTestApi
 

	
 

	
 
class TestHgApi(_BaseTestApi, TestControllerPytest):
 
class TestHgApi(_BaseTestApi, TestController):
 
    REPO = HG_REPO
 
    REPO_TYPE = 'hg'
kallithea/tests/functional/test_admin.py
Show inline comments
 
import os
 
import csv
 
import datetime
 
from kallithea.tests import *
 
from kallithea.model.db import UserLog
 
from kallithea.model.meta import Session
 
from kallithea.lib.utils2 import safe_unicode
 

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

	
 

	
 
class TestAdminController(TestControllerPytest):
 
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()
 
        response = self.app.get(url(controller='admin/admin', action='index'))
 
        response.mustcontain('Admin Journal')
 

	
 
    def test_filter_all_entries(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/admin', action='index',))
 
        response.mustcontain('2034 Entries')
 

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

	
 

	
 
class TestAuthSettingsController(TestControllerPytest):
 
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()
 
        assert new_settings['auth_ldap_host'] == u'dc.example.com', 'fail db write compare'
 

	
 
    @skipif(not ldap_lib_installed, reason='skipping due to missing ldap lib')
 
    def test_ldap_error_form_wrong_port_number(self):
kallithea/tests/functional/test_admin_defaults.py
Show inline comments
 
from kallithea.tests import *
 
from kallithea.model.db import Setting
 

	
 

	
 
class TestDefaultsController(TestControllerPytest):
 
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_index_as_xml(self):
 
        response = self.app.get(url('formatted_defaults', format='xml'))
 

	
 
    def test_create(self):
 
        response = self.app.post(url('defaults'),
 
            {'_authentication_token': self.authentication_token()})
 

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

	
 
    def test_new_as_xml(self):
 
        response = self.app.get(url('formatted_new_default', format='xml'))
 

	
 
    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.put(url('default', 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(),
kallithea/tests/functional/test_admin_gists.py
Show inline comments
 
from kallithea.tests 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(TestControllerPytest):
 
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')
 
        response = self.app.get(url('gists', private=1))
 
        # Test response...
 

	
 
        #and privates
 
        response.mustcontain('gist: %s' % gist.gist_access_id)
 

	
 
    def test_create_missing_description(self):
 
        self.log_user()
 
        response = self.app.post(url('gists'),
 
                                 params={'lifetime': -1, '_authentication_token': self.authentication_token()},
 
                                 status=200)
 

	
 
        response.mustcontain('Missing value')
 

	
 
    def test_create(self):
 
        self.log_user()
 
        response = self.app.post(url('gists'),
 
                                 params={'lifetime': -1,
 
                                         'content': 'gist test',
 
                                         'filename': 'foo',
kallithea/tests/functional/test_admin_notifications.py
Show inline comments
 
from kallithea.tests 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(TestControllerPytest):
 
class TestNotificationsController(TestController):
 
    def setup_method(self, method):
 
        self.remove_all_notifications()
 

	
 
    def test_index(self):
 
        self.log_user()
 

	
 
        u1 = UserModel().create_or_update(username='u1', password='qweqwe',
 
                                          email='u1@example.com',
 
                                          firstname=u'u1', lastname=u'u1')
 
        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):
 
        self.log_user()
 
        cur_user = self._get_logged_user()
 

	
 
        u1 = UserModel().create_or_update(username='u1', password='qweqwe',
 
                                               email='u1@example.com',
 
                                               firstname=u'u1', lastname=u'u1')
 
        u2 = UserModel().create_or_update(username='u2', password='qweqwe',
 
                                               email='u2@example.com',
 
                                               firstname=u'u2', lastname=u'u2')
 

	
 
        # make notifications
 
        notification = NotificationModel().create(created_by=cur_user,
 
                                                  subject=u'test',
 
                                                  body=u'hi there',
 
                                                  recipients=[cur_user, u1, u2])
 
        Session().commit()
 
        u1 = User.get(u1.user_id)
 
        u2 = User.get(u2.user_id)
 

	
 
        # check DB
 
        get_notif = lambda un: [x.notification for x in un]
 
        assert get_notif(cur_user.notifications) == [notification]
 
        assert get_notif(u1.notifications) == [notification]
 
        assert get_notif(u2.notifications) == [notification]
 
        cur_usr_id = cur_user.user_id
kallithea/tests/functional/test_admin_permissions.py
Show inline comments
 
import time
 

	
 
from kallithea.model.db import User, UserIpMap
 
from kallithea.tests import *
 

	
 
class TestAdminPermissionsController(TestControllerPytest):
 
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):
 
        self.log_user()
 
        default_user_id = User.get_default_user().user_id
 
        response = self.app.put(url('edit_user_ips', id=default_user_id),
 
                                 params=dict(new_ip='127.0.0.0/24',
 
                                 _authentication_token=self.authentication_token()))
 

	
 
        # sleep more than beaker.cache.sql_cache_short.expire to expire user cache
 
        time.sleep(1.5)
 
        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')
 

	
 
        ## delete
 
        default_user_id = User.get_default_user().user_id
 
        del_ip_id = UserIpMap.query().filter(UserIpMap.user_id ==
 
                                             default_user_id).first().ip_id
 

	
 
        response = self.app.post(url('edit_user_ips', id=default_user_id),
 
                                 params=dict(_method='delete',
 
                                             del_ip_id=del_ip_id,
 
                                             _authentication_token=self.authentication_token()),
 
                                 extra_environ={'REMOTE_ADDR': '127.0.0.1'})
 

	
 
        # sleep more than beaker.cache.sql_cache_short.expire to expire user cache
 
        time.sleep(1.5)
 

	
 
        response = self.app.get(url('admin_permissions_ips'))
 
        response.mustcontain('All IP addresses are allowed')
 
        response.mustcontain(no=['127.0.0.0/24'])
 
        response.mustcontain(no=['127.0.0.0 - 127.0.0.255'])
 

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

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

	
 
import os
 
import mock
 
import urllib
 

	
 
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.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(TestControllerPytest):
 
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))
 

	
 
        # test if the repo was created in the database
 
        new_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == repo_name).one()
 

	
 
        assert new_repo.repo_name == repo_name
 
        assert new_repo.description == description
 

	
 
        # test if the repository is visible in the list ?
 
        response = self.app.get(url('summary_home', repo_name=repo_name))
 
        response.mustcontain(repo_name)
 
        response.mustcontain(self.REPO_TYPE)
 

	
 
        # test if the repository was created on filesystem
 
        try:
 
            vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)))
 
        except vcs.exceptions.VCSError:
 
            pytest.fail('no repo %s in filesystem' % repo_name)
 

	
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.fixture import Fixture
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestAdminSettingsController(TestControllerPytest):
 
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()))
 

	
 
        response = response.follow()
 
        response.mustcontain('test_hooks_2')
 
        response.mustcontain('cd %s2' % tempfile.gettempdir())
 

	
 
        hook_id = Ui.get_by_key('hooks', 'test_hooks_2').ui_id
kallithea/tests/functional/test_admin_user_groups.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
from kallithea.tests import *
 
from kallithea.model.db import UserGroup, UserGroupToPerm, Permission
 
from kallithea.model.meta import Session
 

	
 
TEST_USER_GROUP = u'admins_test'
 

	
 

	
 
class TestAdminUsersGroupsController(TestControllerPytest):
 
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.put(url('users_group', id=1), status=403)
 

	
 
    def test_update_browser_fakeout(self):
 
        response = self.app.post(url('users_group', id=1),
 
                                 params=dict(_method='put', _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()
 

	
 
        self.checkSessionFlash(response,
 
                               'Created user group ')
 

	
 
        gr = Session().query(UserGroup) \
 
            .filter(UserGroup.users_group_name == users_group_name).one()
 

	
 
        response = self.app.post(url('users_group', id=gr.users_group_id),
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
 

	
 
import pytest
 
from kallithea.tests 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()
 

	
 

	
 
class TestAdminUsersController(TestControllerPytest):
 
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'
 

	
 
        response = self.app.post(url('users'),
 
            {'username': username,
 
             'password': password,
 
             'password_confirmation': password_confirmation,
 
             'firstname': name,
 
             'active': True,
 
             'lastname': lastname,
 
             'extern_name': 'internal',
 
             'extern_type': 'internal',
 
             'email': email,
 
             '_authentication_token': self.authentication_token()})
 

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

	
 
        new_user = Session().query(User). \
 
            filter(User.username == username).one()
 

	
 
        assert new_user.username == username
 
        assert check_password(password, new_user.password) == True
 
        assert new_user.name == name
 
        assert new_user.lastname == lastname
 
        assert new_user.email == email
 

	
 
        response.follow()
 
        response = response.follow()
 
@@ -485,97 +485,97 @@ class TestAdminUsersController(TestContr
 
                {'description': 'desc', 'lifetime': -1, '_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'API key successfully created')
 
        response = response.follow()
 

	
 
        #now delete our key
 
        keys = UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all()
 
        assert 1 == len(keys)
 

	
 
        response = self.app.post(url('edit_user_api_keys', id=user_id),
 
                 {'_method': 'delete', 'del_api_key': keys[0].api_key, '_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'API key successfully deleted')
 
        keys = UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all()
 
        assert 0 == len(keys)
 

	
 
    def test_reset_main_api_key(self):
 
        self.log_user()
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
 
        user_id = user.user_id
 
        api_key = user.api_key
 
        response = self.app.get(url('edit_user_api_keys', id=user_id))
 
        response.mustcontain(api_key)
 
        response.mustcontain('Expires: Never')
 

	
 
        response = self.app.post(url('edit_user_api_keys', id=user_id),
 
                 {'_method': 'delete', 'del_api_key_builtin': api_key, '_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'API key successfully reset')
 
        response = response.follow()
 
        response.mustcontain(no=[api_key])
 

	
 

	
 
class TestAdminUsersController_unittest(object):
 
    """ Unit tests for the users controller """
 

	
 
    def test_get_user_or_raise_if_default(self, monkeypatch):
 
        # flash complains about an non-existing session
 
        def flash_mock(*args, **kwargs):
 
            pass
 
        monkeypatch.setattr(h, 'flash', flash_mock)
 

	
 
        u = UsersController()
 
        # a regular user should work correctly
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
 
        assert u._get_user_or_raise_if_default(user.user_id) == user
 
        # the default user should raise
 
        with pytest.raises(HTTPNotFound):
 
            u._get_user_or_raise_if_default(User.get_default_user().user_id)
 

	
 

	
 
class TestAdminUsersControllerForDefaultUser(TestControllerPytest):
 
class TestAdminUsersControllerForDefaultUser(TestController):
 
    """
 
    Edit actions on the default user are not allowed.
 
    Validate that they throw a 404 exception.
 
    """
 
    def test_edit_default_user(self):
 
        self.log_user()
 
        user = User.get_default_user()
 
        response = self.app.get(url('edit_user', id=user.user_id), status=404)
 

	
 
    def test_edit_advanced_default_user(self):
 
        self.log_user()
 
        user = User.get_default_user()
 
        response = self.app.get(url('edit_user_advanced', id=user.user_id), status=404)
 

	
 
    # API keys
 
    def test_edit_api_keys_default_user(self):
 
        self.log_user()
 
        user = User.get_default_user()
 
        response = self.app.get(url('edit_user_api_keys', id=user.user_id), status=404)
 

	
 
    def test_add_api_keys_default_user(self):
 
        self.log_user()
 
        user = User.get_default_user()
 
        response = self.app.post(url('edit_user_api_keys', id=user.user_id),
 
                 {'_method': 'put', '_authentication_token': self.authentication_token()}, status=404)
 

	
 
    def test_delete_api_keys_default_user(self):
 
        self.log_user()
 
        user = User.get_default_user()
 
        response = self.app.post(url('edit_user_api_keys', id=user.user_id),
 
                 {'_method': 'delete', '_authentication_token': self.authentication_token()}, status=404)
 

	
 
    # Permissions
 
    def test_edit_perms_default_user(self):
 
        self.log_user()
 
        user = User.get_default_user()
 
        response = self.app.get(url('edit_user_perms', id=user.user_id), status=404)
 

	
 
    def test_update_perms_default_user(self):
 
        self.log_user()
 
        user = User.get_default_user()
 
        response = self.app.post(url('edit_user_perms', id=user.user_id),
 
                 {'_method': 'put', '_authentication_token': self.authentication_token()}, status=404)
 

	
 
    # Emails
 
    def test_edit_emails_default_user(self):
 
        self.log_user()
 
        user = User.get_default_user()
kallithea/tests/functional/test_branches.py
Show inline comments
 
from kallithea.tests import *
 

	
 

	
 
class TestBranchesController(TestControllerPytest):
 
class TestBranchesController(TestController):
 

	
 
    def test_index_hg(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='branches',
 
                                    action='index', repo_name=HG_REPO))
 
        response.mustcontain("""<a href="/%s/changelog?branch=default">default</a>""" % HG_REPO)
 

	
 
        # closed branches
 
        response.mustcontain("""<a href="/%s/changelog?branch=git">git [closed]</a>""" % HG_REPO)
 
        response.mustcontain("""<a href="/%s/changelog?branch=web">web [closed]</a>""" % HG_REPO)
 

	
 
    def test_index_git(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='branches',
 
                                    action='index', repo_name=GIT_REPO))
 
        response.mustcontain("""<a href="/%s/changelog?branch=master">master</a>""" % GIT_REPO)
kallithea/tests/functional/test_changelog.py
Show inline comments
 
from kallithea.tests import *
 

	
 

	
 
class TestChangelogController(TestControllerPytest):
 
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(
 
            """<span class="changeset_hash">r640:0a4e54a44604</span>"""
 
        )
 
        response.mustcontain("""code garden""")
 

	
 
        response.mustcontain("""var jsdata = [[[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 1, 2, 0], [0, 0, 1, 0]], 0, 0], [[0, 1], [[0, 0, 1, 0], [1, 1, 2, 0]], 0, 0], [[0, 1], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 3, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 3, 0]], 0, 0], [[1, 3], [[0, 0, 2, 0], [1, 1, 3, 0]], 0, 0], [[1, 3], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 4, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 1, 4, 0]], 0, 0], [[1, 4], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 5, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 5, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 5, 0]], 0, 0], [[1, 5], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 6, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 6, 0]], 0, 0], [[1, 6], [[0, 0, 2, 0], [1, 1, 6, 0]], 0, 0], [[1, 6], [[0, 0, 2, 0], [1, 1, 6, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 6, 0]], 0, 0], [[1, 6], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 7, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 7, 0]], 0, 0], [[1, 7], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 8, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 8, 0]], 0, 0], [[1, 8], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 9, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 10, 0], [0, 0, 2, 0], [1, 2, 9, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 10, 0], [2, 2, 9, 0]], 0, 0], [[2, 9], [[0, 0, 2, 0], [1, 1, 10, 0], [2, 1, 10, 0]], 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 1, 10, 0]], 0, 0], [[1, 10], [[0, 0, 2, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[1, 11], [[0, 0, 2, 0], [1, 1, 11, 0]], 1, 0], [[2, 12], [[0, 0, 2, 0], [2, 1, 12, 0]], 1, 0], [[0, 2], [[0, 1, 13, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 13, 0]], 0, 0], [[1, 13], [[0, 0, 2, 0], [1, 1, 13, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 13, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 13, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0], [1, 1, 13, 0]], 0, 0], [[1, 13], [[0, 0, 2, 0], [1, 1, 13, 0], [1, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 1, 14, 0], [0, 0, 2, 0]], 0, 0], [[0, 2], [[0, 0, 2, 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" />"""
 
        )
 

	
 
        response.mustcontain(
 
            """<span class="changeset_hash">r539:22baf968d547</span>"""
 
        )
kallithea/tests/functional/test_changeset.py
Show inline comments
 
from kallithea.tests import *
 

	
 
class TestChangesetController(TestControllerPytest):
 
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.model.db import ChangesetComment, Notification, \
 
    UserNotification
 
from kallithea.model.meta import Session
 

	
 

	
 
class TestChangeSetCommentsController(TestControllerPytest):
 
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):
 
        self.log_user()
 
        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
 
        text = u'CommentOnRevision'
 
        f_path = 'vcs/web/simplevcs/views/repository.py'
 
        line = 'n1'
 

	
kallithea/tests/functional/test_compare.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
from kallithea.tests 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(TestControllerPytest):
 
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)
 

	
 
        rev1 = 'default'
 
        rev2 = 'default'
 

	
 
        response = self.app.get(url('compare_url',
 
                                    repo_name=repo1.repo_name,
 
                                    org_ref_type="branch",
 
                                    org_ref_name=rev2,
 
                                    other_repo=repo2.repo_name,
 
                                    other_ref_type="branch",
 
                                    other_ref_name=rev1,
 
                                    merge='1',))
kallithea/tests/functional/test_compare_local.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
from kallithea.tests import *
 

	
 
class TestCompareController(TestControllerPytest):
 
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 href="/%s/changeset/c5ddebc06eaaba3010c2d66ea6ec9d074eb0f678">r112:c5ddebc06eaa</a>''' % HG_REPO)
 
        response.mustcontain('''<a href="/%s/changeset/70d4cef8a37657ee4cf5aabb3bd9f68879769816">r115:70d4cef8a376</a>''' % HG_REPO)
 
        response.mustcontain('''<a href="/%s/changeset/9749bfbfc0d2eba208d7947de266303b67c87cda">r116:9749bfbfc0d2</a>''' % HG_REPO)
 
        response.mustcontain('''<a href="/%s/changeset/41fda979f02fda216374bf8edac4e83f69e7581c">r117:41fda979f02f</a>''' % HG_REPO)
 
        response.mustcontain('''<a href="/%s/changeset/bb1a3ab98cc45cb934a77dcabf87a5a598b59e97">r118:bb1a3ab98cc4</a>''' % HG_REPO)
 
        response.mustcontain('''<a href="/%s/changeset/36e0fc9d2808c5022a24f49d6658330383ed8666">r119:36e0fc9d2808</a>''' % HG_REPO)
 
        response.mustcontain('''<a 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>''')
 
        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">
 
                             <i class="icon-diff-M"></i>
 
                             <a href="#C--2c14b00f3393">docs/api/index.rst</a>''')
 
        response.mustcontain('''<div class="node">
 
                             <i class="icon-diff-M"></i>
 
                             <a href="#C--430ccbc82bdf">vcs/__init__.py</a>''')
 
        response.mustcontain('''<div class="node">
 
                             <i class="icon-diff-M"></i>
 
                             <a href="#C--9c390eb52cd6">vcs/backends/hg.py</a>''')
kallithea/tests/functional/test_feed.py
Show inline comments
 
from kallithea.tests import *
 

	
 
class TestFeedController(TestControllerPytest):
 
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.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(TestControllerPytest):
 
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()
 

	
 
        response = self.app.get(
 
            url(controller='files', action='index',
 
                repo_name=HG_REPO,
 
                revision='7ba66bec8d6dbba14a2155be32408c435c5f4492',
 
                f_path='/')
 
        )
 

	
 
        #Test response...
 

	
 
        response.mustcontain('<a class="browser-dir ypjax-link" href="/%s/files/7ba66bec8d6dbba14a2155be32408c435c5f4492/docs"><i class="icon-folder-open"></i><span>docs</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-dir ypjax-link" href="/%s/files/7ba66bec8d6dbba14a2155be32408c435c5f4492/tests"><i class="icon-folder-open"></i><span>tests</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/7ba66bec8d6dbba14a2155be32408c435c5f4492/README.rst"><i class="icon-doc"></i><span>README.rst</span></a>' % HG_REPO)
 
        response.mustcontain('1.1 KiB')
 

	
 
    def test_index_different_branch(self):
 
        self.log_user()
 

	
 
        response = self.app.get(url(controller='files', action='index',
 
                                    repo_name=HG_REPO,
 
                                    revision='97e8b885c04894463c51898e14387d80c30ed1ee',
 
                                    f_path='/'))
 

	
 
        response.mustcontain("""<option selected="selected" value="97e8b885c04894463c51898e14387d80c30ed1ee">git at 97e8b885c048</option>""")
kallithea/tests/functional/test_followers.py
Show inline comments
 
from kallithea.tests import *
 

	
 

	
 
class TestFollowersController(TestControllerPytest):
 
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.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(TestControllerPytest):
 
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
 
        repo_name = self.REPO
 
        self.app.post(url(controller='forks', action='fork_create',
 
                          repo_name=repo_name), {'_authentication_token': self.authentication_token()}, status=403)
 

	
 
    def test_index_with_fork(self):
 
        self.log_user()
 

	
 
        # create a fork
 
        fork_name = self.REPO_FORK
 
        description = 'fork of vcs test'
kallithea/tests/functional/test_home.py
Show inline comments
 
from kallithea.tests 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(TestControllerPytest):
 
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.getAll()))
 
        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'))
 

	
 
        try:
 
            response.mustcontain(u"gr1/repo_in_group")
 
        finally:
 
            RepoModel().delete(u'gr1/repo_in_group')
 
            RepoGroupModel().delete(repo_group=u'gr1', force_delete=True)
 
            Session().commit()
kallithea/tests/functional/test_journal.py
Show inline comments
 
from kallithea.tests import *
 
import datetime
 

	
 

	
 
class TestJournalController(TestControllerPytest):
 
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.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(TestControllerPytest):
 
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})
 

	
 
        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_ok_came_from(self):
 
        test_came_from = '/_admin/users'
 
        response = self.app.post(url(controller='login', action='index',
 
                                     came_from=test_came_from),
 
                                 {'username': TEST_USER_ADMIN_LOGIN,
 
                                  'password': TEST_USER_ADMIN_PASS})
 
        assert response.status == '302 Found'
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.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(TestControllerPytest):
 
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'),
 
                                 {'new_email': TEST_USER_REGULAR_EMAIL, '_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'This email address is already in use')
 

	
 
    def test_my_account_my_emails_add_missing_email_in_form(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'),
 
            {'_authentication_token': self.authentication_token()})
kallithea/tests/functional/test_pullrequests.py
Show inline comments
 
import re
 

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

	
 
from kallithea.controllers.pullrequests import PullrequestsController
 

	
 
fixture = Fixture()
 

	
 
class TestPullrequestsController(TestControllerPytest):
 
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()
 
        assert response.status == '200 OK'
 
        response.mustcontain('This pull request has already been merged to default.')
 

	
 
    def test_create_with_invalid_reviewer(self):
 
        invalid_user_name = 'invalid_user'
 
        self.log_user()
 
        response = self.app.post(url(controller='pullrequests', action='create',
 
                                     repo_name=HG_REPO),
 
@@ -98,97 +98,97 @@ class TestPullrequestsController(TestCon
 
                                     repo_name=HG_REPO, pull_request_id=pull_request_id),
 
                                 {
 
                                  'updaterev': 'default',
 
                                  'pullrequest_title': 'title',
 
                                  'pullrequest_desc': 'description',
 
                                  'owner': TEST_USER_ADMIN_LOGIN,
 
                                  '_authentication_token': self.authentication_token(),
 
                                  'review_members': invalid_user_id,
 
                                 },
 
                                 status=400)
 
        response.mustcontain('Invalid reviewer &#34;%s&#34; specified' % invalid_user_id)
 

	
 
    def test_edit_with_invalid_reviewer(self):
 
        invalid_user_id = 99999
 
        self.log_user()
 
        # create a valid pull request
 
        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'
 
        # location is of the form:
 
        # http://localhost/vcs_test_hg/pull-request/54/_/title
 
        m = re.search('/pull-request/(\d+)/', response.location)
 
        assert m != None
 
        pull_request_id = m.group(1)
 

	
 
        # edit it
 
        response = self.app.post(url(controller='pullrequests', action='post',
 
                                     repo_name=HG_REPO, pull_request_id=pull_request_id),
 
                                 {
 
                                  'pullrequest_title': 'title',
 
                                  'pullrequest_desc': 'description',
 
                                  'owner': TEST_USER_ADMIN_LOGIN,
 
                                  '_authentication_token': self.authentication_token(),
 
                                  'review_members': invalid_user_id,
 
                                 },
 
                                 status=400)
 
        response.mustcontain('Invalid reviewer &#34;%s&#34; specified' % invalid_user_id)
 

	
 
class TestPullrequestsGetRepoRefs(TestControllerPytest):
 
class TestPullrequestsGetRepoRefs(TestController):
 

	
 
    def setup_method(self, method):
 
        self.main = fixture.create_repo(u'main', repo_type='hg')
 
        Session.add(self.main)
 
        Session.commit()
 
        self.c = PullrequestsController()
 

	
 
    def teardown_method(self, method):
 
        fixture.destroy_repo(u'main')
 
        Session.commit()
 
        Session.remove()
 

	
 
    def test_repo_refs_empty_repo(self):
 
        # empty repo with no commits, no branches, no bookmarks, just one tag
 
        refs, default = self.c._get_repo_refs(self.main.scm_instance)
 
        assert default == 'tag:null:0000000000000000000000000000000000000000'
 

	
 
    def test_repo_refs_one_commit_no_hints(self):
 
        cs0 = fixture.commit_change(self.main.repo_name, filename='file1',
 
                content='line1\n', message='commit1', vcs_type='hg',
 
                parent=None, newfile=True)
 

	
 
        refs, default = self.c._get_repo_refs(self.main.scm_instance)
 
        assert default == 'branch:default:%s' % cs0.raw_id
 
        assert ([('branch:default:%s' % cs0.raw_id, 'default (current tip)')],
 
                'Branches') in refs
 

	
 
    def test_repo_refs_one_commit_rev_hint(self):
 
        cs0 = fixture.commit_change(self.main.repo_name, filename='file1',
 
                content='line1\n', message='commit1', vcs_type='hg',
 
                parent=None, newfile=True)
 

	
 
        refs, default = self.c._get_repo_refs(self.main.scm_instance, rev=cs0.raw_id)
 
        expected = 'branch:default:%s' % cs0.raw_id
 
        assert default == expected
 
        assert ([(expected, 'default (current tip)')], 'Branches') in refs
 

	
 
    def test_repo_refs_two_commits_no_hints(self):
 
        cs0 = fixture.commit_change(self.main.repo_name, filename='file1',
 
                content='line1\n', message='commit1', vcs_type='hg',
 
                parent=None, newfile=True)
 
        cs1 = fixture.commit_change(self.main.repo_name, filename='file2',
 
                content='line2\n', message='commit2', vcs_type='hg',
 
                parent=None, newfile=True)
 

	
 
        refs, default = self.c._get_repo_refs(self.main.scm_instance)
 
        expected = 'branch:default:%s' % cs1.raw_id
 
        assert default == expected
kallithea/tests/functional/test_repo_groups.py
Show inline comments
 
from kallithea.tests import *
 

	
 

	
 
class TestRepoGroupsController(TestControllerPytest):
 
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_create(self):
 
#        response = self.app.post(url('repos_groups'))
 

	
 
    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)
 
#
 
#    def test_update(self):
 
#        response = self.app.put(url('repos_group', group_name=1))
 
#
 
#    def test_delete(self):
 
#        self.log_user()
 
#        response = self.app.delete(url('repos_group', group_name=1))
 
#
 
#    def test_show(self):
 
#        response = self.app.get(url('repos_group', group_name=1))
 
#
 
#    def test_edit(self):
 
#        response = self.app.get(url('edit_repo_group', group_name=1))
kallithea/tests/functional/test_search.py
Show inline comments
 
import mock
 
import os
 
from kallithea.tests import *
 

	
 

	
 
class TestSearchController(TestControllerPytest):
 
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()
 
        response = self.app.get(url(controller='search', action='index'),
 
                    {'q': 'bother to ask where to fetch repo during tests',
 
                     'type': 'commit'})
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.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(TestControllerPytest):
 
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...
 
        response.mustcontain('''id="clone_url" readonly="readonly" value="http://%s@localhost:80/%s"''' % (TEST_USER_ADMIN_LOGIN, GIT_REPO))
 
        response.mustcontain('''id="clone_url_id" readonly="readonly" value="http://%s@localhost:80/_%s"''' % (TEST_USER_ADMIN_LOGIN, ID))
 

	
 
    def test_index_by_id_hg(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='_%s' % ID))
 

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

	
 

	
 
class TestTagsController(TestControllerPytest):
 
class TestTagsController(TestController):
 

	
 
    def test_index_hg(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='tags', action='index', repo_name=HG_REPO))
 
        response.mustcontain("""<a href="/%s/changeset/96507bd11ecc815ebc6270fdf6db110928c09c1e">tip</a>""" % HG_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/2c96c02def9a7c997f33047761a53943e6254396">v0.2.0</a>""" % HG_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/fef5bfe1dc17611d5fb59a7f6f95c55c3606f933">v0.1.11</a>""" % HG_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/92831aebf2f8dd4879e897024b89d09af214df1c">v0.1.10</a>""" % HG_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/8680b1d1cee3aa3c1ab3734b76ee164bbedbc5c9">v0.1.9</a>""" % HG_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/ecb25ba9c96faf1e65a0bc3fd914918420a2f116">v0.1.8</a>""" % HG_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/f67633a2894edaf28513706d558205fa93df9209">v0.1.7</a>""" % HG_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/02b38c0eb6f982174750c0e309ff9faddc0c7e12">v0.1.6</a>""" % HG_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/a6664e18181c6fc81b751a8d01474e7e1a3fe7fc">v0.1.5</a>""" % HG_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/fd4bdb5e9b2a29b4393a4ac6caef48c17ee1a200">v0.1.4</a>""" % HG_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/17544fbfcd33ffb439e2b728b5d526b1ef30bfcf">v0.1.3</a>""" % HG_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/a7e60bff65d57ac3a1a1ce3b12a70f8a9e8a7720">v0.1.2</a>""" % HG_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/eb3a60fc964309c1a318b8dfe26aa2d1586c85ae">v0.1.1</a>""" % HG_REPO)
 

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

	
 
        response.mustcontain("""<a href="/%s/changeset/137fea89f304a42321d40488091ee2ed419a3686">v0.2.2</a>""" % GIT_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/5051d0fa344d4408a2659d9a0348eb2d41868ecf">v0.2.1</a>""" % GIT_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/599ba911aa24d2981225f3966eb659dfae9e9f30">v0.2.0</a>""" % GIT_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/c60f01b77c42dce653d6b1d3b04689862c261929">v0.1.11</a>""" % GIT_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/10cddef6b794696066fb346434014f0a56810218">v0.1.10</a>""" % GIT_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/341d28f0eec5ddf0b6b77871e13c2bbd6bec685c">v0.1.9</a>""" % GIT_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/74ebce002c088b8a5ecf40073db09375515ecd68">v0.1.8</a>""" % GIT_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/4d78bf73b5c22c82b68f902f138f7881b4fffa2c">v0.1.7</a>""" % GIT_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/0205cb3f44223fb3099d12a77a69c81b798772d9">v0.1.6</a>""" % GIT_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/6c0ce52b229aa978889e91b38777f800e85f330b">v0.1.5</a>""" % GIT_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/7d735150934cd7645ac3051903add952390324a5">v0.1.4</a>""" % GIT_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/5a3a8fb005554692b16e21dee62bf02667d8dc3e">v0.1.3</a>""" % GIT_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/0ba5f8a4660034ff25c0cac2a5baabf5d2791d63">v0.1.2</a>""" % GIT_REPO)
 
        response.mustcontain("""<a href="/%s/changeset/e6ea6d16e2f26250124a1f4b4fe37a912f9d86a0">v0.1.1</a>""" % GIT_REPO)
kallithea/tests/models/test_changeset_status.py
Show inline comments
 
from kallithea.tests 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(TestControllerPytest):
 
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
 
@@ -226,90 +226,90 @@ DIFF_FIXTURES = {
 
         {'added': 0,
 
          'deleted': 0,
 
          'binary': True,
 
          'ops': {MOD_FILENODE: 'modified file',
 
                  BIN_FILENODE: 'binary diff not shown'}})
 
    ],
 
    'hg_diff_copy_file.diff': [
 
        ('file2', 'M',
 
         {'added': 0,
 
          'deleted': 0,
 
          'binary': True,
 
          'ops': {COPIED_FILENODE: 'file copied from file1 to file2'}}),
 
    ],
 
    'hg_diff_copy_and_modify_file.diff': [
 
        ('file3', 'M',
 
         {'added': 1,
 
          'deleted': 0,
 
          'binary': False,
 
          'ops': {COPIED_FILENODE: 'file copied from file2 to file3',
 
                  MOD_FILENODE: 'modified file'}}),
 
    ],
 
    'hg_diff_copy_and_chmod_file.diff': [
 
        ('file4', 'M',
 
         {'added': 0,
 
          'deleted': 0,
 
          'binary': True,
 
          'ops': {COPIED_FILENODE: 'file copied from file3 to file4',
 
                  CHMOD_FILENODE: 'modified file chmod 100644 => 100755'}}),
 
    ],
 
    'hg_diff_copy_chmod_and_edit_file.diff': [
 
        ('file5', 'M',
 
         {'added': 2,
 
          'deleted': 1,
 
          'binary': False,
 
          'ops': {COPIED_FILENODE: 'file copied from file4 to file5',
 
                  CHMOD_FILENODE: 'modified file chmod 100755 => 100644',
 
                  MOD_FILENODE: 'modified file'}}),
 
    ],
 
    'hg_diff_rename_space_cr.diff': [
 
        ('oh yes', 'R',
 
         {'added': 3,
 
          'deleted': 2,
 
          'binary': False,
 
          'ops': {RENAMED_FILENODE: 'file renamed from oh no to oh yes'}}),
 
    ],
 
}
 

	
 

	
 
class TestDiffLib(TestControllerPytest):
 
class TestDiffLib(TestController):
 

	
 
    @parametrize('diff_fixture', DIFF_FIXTURES)
 
    def test_diff(self, diff_fixture):
 
        diff = fixture.load_resource(diff_fixture, strip=False)
 
        vcs = 'hg'
 
        if diff_fixture.startswith('git_'):
 
            vcs = 'git'
 
        diff_proc = DiffProcessor(diff, vcs=vcs)
 
        diff_proc_d = diff_proc.prepare()
 
        data = [(x['filename'], x['operation'], x['stats']) for x in diff_proc_d]
 
        expected_data = DIFF_FIXTURES[diff_fixture]
 
        assert expected_data == data
 

	
 
    def test_diff_markup(self):
 
        diff = fixture.load_resource('markuptest.diff', strip=False)
 
        diff_proc = DiffProcessor(diff)
 
        diff_proc_d = diff_proc.prepare()
 
        chunks = diff_proc_d[0]['chunks']
 
        assert not chunks[0]
 
        #from pprint import pprint; pprint(chunks[1])
 
        l = ['\n']
 
        for d in chunks[1]:
 
            l.append('%(action)-7s %(new_lineno)3s %(old_lineno)3s %(line)r\n' % d)
 
        s = ''.join(l)
 
        print s
 
        assert s == r'''
 
context ... ... u'@@ -51,5 +51,12 @@\n'
 
unmod    51  51 u'<u>\t</u>begin();\n'
 
unmod    52  52 u'<u>\t</u>\n'
 
add      53     u'<u>\t</u>int foo;<u class="cr"></u>\n'
 
add      54     u'<u>\t</u>int bar; <u class="cr"></u>\n'
 
add      55     u'<u>\t</u>int baz;<u>\t</u><u class="cr"></u>\n'
 
add      56     u'<u>\t</u>int space; <i></i>'
 
add      57     u'<u>\t</u>int tab;<u>\t</u>\n'
 
add      58     u'<u>\t</u>\n'
 
unmod    59  53 u' <i></i>'
 
del          54 u'<u>\t</u><del>#define MAX_STEPS (48)</del>\n'
 
add      60     u'<u>\t</u><ins><u class="cr"></u></ins>\n'
 
add      61     u'<u>\t</u>#define MAX_STEPS (64)<u class="cr"></u>\n'
 
unmod    62  55 u'\n'
 
'''
kallithea/tests/models/test_notifications.py
Show inline comments
 
from kallithea.tests import *
 

	
 
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
 

	
 

	
 
class TestNotifications(TestControllerPytest):
 
class TestNotifications(TestController):
 

	
 
    def setup_method(self, method):
 
        Session.remove()
 
        self.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 = self.u1.user_id
 

	
 
        self.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 = self.u2.user_id
 

	
 
        self.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 = self.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]
 
        notification = NotificationModel().create(created_by=self.u1,
 
                                           subject=u'subj', body=u'hi there',
 
                                           recipients=usrs)
 
        Session().commit()
 
        u1 = User.get(self.u1)
 
        u2 = User.get(self.u2)
 
        u3 = User.get(self.u3)
 
        notifications = Notification.query().all()
 
        assert len(notifications) == 1
 

	
 
        assert notifications[0].recipients == [u1, u2]
 
        assert notification.notification_id == notifications[0].notification_id
 

	
 
        unotification = UserNotification.query() \
 
            .filter(UserNotification.notification == notification).all()
 

	
 
        assert len(unotification) == len(usrs)
 
        assert set([x.user.user_id for x in unotification]) == set(usrs)
kallithea/tests/models/test_permissions.py
Show inline comments
 
from kallithea.tests 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(TestControllerPytest):
 
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)
 

	
 
        UserModel().delete(self.u1)
 
        UserModel().delete(self.u2)
 
        UserModel().delete(self.u3)
 
        UserModel().delete(self.a1)
 
        if hasattr(self, 'g1'):
 
            RepoGroupModel().delete(self.g1.group_id)
 
        if hasattr(self, 'g2'):
 
            RepoGroupModel().delete(self.g2.group_id)
 

	
 
        if hasattr(self, 'ug1'):
 
            UserGroupModel().delete(self.ug1, force=True)
 
        if hasattr(self, 'ug2'):
 
            UserGroupModel().delete(self.ug2, force=True)
 

	
 
        Session().commit()
 

	
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.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(TestControllerPytest):
 
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)
 

	
 
    def _check_folders(self):
 
        print os.listdir(TESTS_TMP_PATH)
 

	
 
    def __delete_group(self, id_):
 
        RepoGroupModel().delete(id_)
 

	
 
    def test_create_group(self):
 
        g = fixture.create_repo_group(u'newGroup')
 
        Session().commit()
 
        assert g.full_path == 'newGroup'
 

	
 
        assert self.__check_path('newGroup')
 

	
 
    def test_create_same_name_group(self):
 
        with pytest.raises(IntegrityError):
 
            fixture.create_repo_group(u'newGroup')
 
        Session().rollback()
 

	
 
    def test_same_subgroup(self):
 
        sg1 = fixture.create_repo_group(u'sub1', group_parent_id=self.g1.group_id)
 
        assert sg1.parent_group == self.g1
 
        assert sg1.full_path == 'test1/sub1'
 
        assert self.__check_path('test1', 'sub1')
 

	
 
        ssg1 = fixture.create_repo_group(u'subsub1', group_parent_id=sg1.group_id)
 
        assert ssg1.parent_group == sg1
 
        assert ssg1.full_path == 'test1/sub1/subsub1'
 
        assert self.__check_path('test1', 'sub1', 'subsub1')
 

	
 
    def test_remove_group(self):
 
        sg1 = fixture.create_repo_group(u'deleteme')
kallithea/tests/models/test_repos.py
Show inline comments
 
import pytest
 
from kallithea.tests 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(TestControllerPytest):
 
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()
 

	
 
        RepoModel().delete(repo=repo, forks='delete')
 
        Session().commit()
 

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

	
 
    def test_remove_repo_detach_forks(self):
 
        repo = fixture.create_repo(name=u'test-repo-1')
 
        Session().commit()
kallithea/tests/models/test_user_groups.py
Show inline comments
 
from kallithea.model.db import User
 

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

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

	
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestUserGroups(TestControllerPytest):
 
class TestUserGroups(TestController):
 

	
 
    def teardown_method(self, method):
 
        # delete all groups
 
        for gr in UserGroupModel.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 UserGroupModel.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'})
 
            Session().commit()
 
            UserGroupModel().add_user_to_group(gr, user)
 
            Session().commit()
 

	
 
        UserGroupModel().enforce_groups(user, groups, 'container')
 
        Session().commit()
 

	
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
 
        in_groups = user.group_member
 
        assert expected == [x.users_group.users_group_name for x in in_groups]
kallithea/tests/models/test_users.py
Show inline comments
 
import pytest
 
from kallithea.tests 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(TestControllerPytest):
 
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',
 
                                     email=u'main_email@example.com',
 
                                     firstname=u'u1', lastname=u'u1')
 
        Session().commit()
 

	
 
        with pytest.raises(AttributeError):
 
            m = UserEmailMap()
 
            m.email = u'main_email@example.com'
 
            m.user = usr
 
            Session().add(m)
 
            Session().commit()
 

	
 
        UserModel().delete(usr.user_id)
 
        Session().commit()
 

	
 
    def test_extra_email_map(self):
 
        usr = UserModel().create_or_update(username=u'test_user',
 
                                           password=u'qweqwe',
 
                                     email=u'main_email@example.com',
 
                                     firstname=u'u1', lastname=u'u1')
 
        Session().commit()
 

	
 
        m = UserEmailMap()
 
        m.email = u'main_email2@example.com'
 
        m.user = usr
 
        Session().add(m)
 
        Session().commit()
 

	
 
        u = User.get_by_email(email='MAIN_email@example.com')
 
        assert usr.user_id == u.user_id
 
        assert usr.username == u.username
 

	
 
        u = User.get_by_email(email='main_email@example.com')
 
        assert usr.user_id == u.user_id
 
        assert usr.username == u.username
 

	
 
        u = User.get_by_email(email='main_email2@example.com')
 
        assert usr.user_id == u.user_id
 
        assert usr.username == u.username
 
        u = User.get_by_email(email='main_email3@example.com')
 
        assert None == u
 

	
 
        u = User.get_by_email(email='main_e%ail@example.com')
 
        assert None == u
 
        u = User.get_by_email(email='main_emai_@example.com')
 
        assert None == u
 

	
 

	
 
        UserModel().delete(usr.user_id)
 
        Session().commit()
 

	
 

	
 
class TestUsers(TestControllerPytest):
 
class TestUsers(TestController):
 

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

	
 
    def teardown_method(self, method):
 
        perm = Permission.query().all()
 
        for p in perm:
 
            UserModel().revoke_perm(self.u1, p)
 

	
 
        UserModel().delete(self.u1)
 
        Session().commit()
 
        Session.remove()
 

	
 
    def test_add_perm(self):
 
        perm = Permission.query().all()[0]
 
        UserModel().grant_perm(self.u1, perm)
 
        Session().commit()
 
        assert UserModel().has_perm(self.u1, perm) == True
 

	
 
    def test_has_perm(self):
 
        perm = Permission.query().all()
 
        for p in perm:
 
            has_p = UserModel().has_perm(self.u1, p)
 
            assert False == has_p
 

	
 
    def test_revoke_perm(self):
 
        perm = Permission.query().all()[0]
 
        UserModel().grant_perm(self.u1, perm)
 
        Session().commit()
 
        assert UserModel().has_perm(self.u1, perm) == True
 

	
 
        #revoke
 
        UserModel().revoke_perm(self.u1, perm)
 
        Session().commit()
 
        assert UserModel().has_perm(self.u1, perm) == False
kallithea/tests/other/manual_test_vcs_operations.py
Show inline comments
 
@@ -121,97 +121,97 @@ def _add_files_and_push(vcs, DEST, **kwa
 
            cmd = """EMAIL="me@example.com" git commit -m 'committed new %s' --author '%s' %s """ % (
 
                i, author_str, added_file
 
            )
 
        Command(cwd).execute(cmd)
 

	
 
    # PUSH it back
 
    _REPO = None
 
    if vcs == 'hg':
 
        _REPO = HG_REPO
 
    elif vcs == 'git':
 
        _REPO = GIT_REPO
 

	
 
    kwargs['dest'] = ''
 
    clone_url = _construct_url(_REPO, **kwargs)
 
    if 'clone_url' in kwargs:
 
        clone_url = kwargs['clone_url']
 
    stdout = stderr = None
 
    if vcs == 'hg':
 
        stdout, stderr = Command(cwd).execute('hg push --verbose', clone_url)
 
    elif vcs == 'git':
 
        stdout, stderr = Command(cwd).execute('git push --verbose', clone_url + " master")
 

	
 
    return stdout, stderr
 

	
 

	
 
def set_anonymous_access(enable=True):
 
    user = User.get_by_username(User.DEFAULT_USER)
 
    user.active = enable
 
    Session().add(user)
 
    Session().commit()
 
    print '\tanonymous access is now:', enable
 
    if enable != User.get_by_username(User.DEFAULT_USER).active:
 
        raise Exception('Cannot set anonymous access')
 

	
 

	
 
#==============================================================================
 
# TESTS
 
#==============================================================================
 

	
 

	
 
def _check_proper_git_push(stdout, stderr):
 
    #WTF Git stderr is output ?!
 
    assert 'fatal' not in stderr
 
    assert 'rejected' not in stderr
 
    assert 'Pushing to' in stderr
 
    assert 'master -> master' in stderr
 

	
 

	
 
class TestVCSOperations(TestControllerPytest):
 
class TestVCSOperations(TestController):
 

	
 
    @classmethod
 
    def setup_class(cls):
 
        #DISABLE ANONYMOUS ACCESS
 
        set_anonymous_access(False)
 

	
 
    def setup_method(self, method):
 
        r = Repository.get_by_repo_name(GIT_REPO)
 
        Repository.unlock(r)
 
        r.enable_locking = False
 
        Session().add(r)
 
        Session().commit()
 

	
 
        r = Repository.get_by_repo_name(HG_REPO)
 
        Repository.unlock(r)
 
        r.enable_locking = False
 
        Session().add(r)
 
        Session().commit()
 

	
 
    def test_clone_hg_repo_by_admin(self):
 
        clone_url = _construct_url(HG_REPO)
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
 

	
 
        assert 'requesting all changes' in stdout
 
        assert 'adding changesets' in stdout
 
        assert 'adding manifests' in stdout
 
        assert 'adding file changes' in stdout
 

	
 
        assert stderr == ''
 

	
 
    def test_clone_git_repo_by_admin(self):
 
        clone_url = _construct_url(GIT_REPO)
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
 

	
 
        assert 'Cloning into' in stdout + stderr
 
        assert stderr == '' or stdout == ''
 

	
 
    def test_clone_wrong_credentials_hg(self):
 
        clone_url = _construct_url(HG_REPO, passwd='bad!')
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url)
 
        assert 'abort: authorization failed' in stderr
 

	
 
    def test_clone_wrong_credentials_git(self):
 
        clone_url = _construct_url(GIT_REPO, passwd='bad!')
 
        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)
 
        assert 'fatal: Authentication failed' in stderr
 

	
 
    def test_clone_git_dir_as_hg(self):
kallithea/tests/other/test_libs.py
Show inline comments
 
@@ -22,97 +22,97 @@ Original author and date, and relevant c
 
:created_on: Jun 9, 2011
 
:author: marcink
 
:copyright: (c) 2013 RhodeCode GmbH, and others.
 
:license: GPLv3, see LICENSE.md for more details.
 
"""
 

	
 
import datetime
 
import hashlib
 
import mock
 
from kallithea.tests import *
 
from kallithea.lib.utils2 import AttributeDict
 
from kallithea.model.db import Repository
 

	
 
proto = 'http'
 
TEST_URLS = [
 
    ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
 
     '%s://127.0.0.1' % proto),
 
    ('%s://username@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
 
     '%s://127.0.0.1' % proto),
 
    ('%s://username:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
 
     '%s://127.0.0.1' % proto),
 
    ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],
 
     '%s://127.0.0.1:8080' % proto),
 
    ('%s://example.com' % proto, ['%s://' % proto, 'example.com'],
 
     '%s://example.com' % proto),
 
    ('%s://user:pass@example.com:8080' % proto, ['%s://' % proto, 'example.com',
 
                                                '8080'],
 
     '%s://example.com:8080' % proto),
 
]
 

	
 
proto = 'https'
 
TEST_URLS += [
 
    ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
 
     '%s://127.0.0.1' % proto),
 
    ('%s://username@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
 
     '%s://127.0.0.1' % proto),
 
    ('%s://username:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
 
     '%s://127.0.0.1' % proto),
 
    ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],
 
     '%s://127.0.0.1:8080' % proto),
 
    ('%s://example.com' % proto, ['%s://' % proto, 'example.com'],
 
     '%s://example.com' % proto),
 
    ('%s://user:pass@example.com:8080' % proto, ['%s://' % proto, 'example.com',
 
                                                '8080'],
 
     '%s://example.com:8080' % proto),
 
]
 

	
 

	
 
class TestLibs(TestControllerPytest):
 
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
 
        assert credentials_filter(test_url) == expected_creds
 

	
 
    @parametrize('str_bool,expected', [
 
                           ('t', True),
 
                           ('true', True),
 
                           ('y', True),
 
                           ('yes', True),
 
                           ('on', True),
 
                           ('1', True),
 
                           ('Y', True),
 
                           ('yeS', True),
 
                           ('Y', True),
 
                           ('TRUE', True),
 
                           ('T', True),
 
                           ('False', False),
 
                           ('F', False),
 
                           ('FALSE', False),
 
                           ('0', False),
 
                           ('-1', False),
 
                           ('', False)
 
    ])
 
    def test_str2bool(self, str_bool, expected):
 
        from kallithea.lib.utils2 import str2bool
 
        assert str2bool(str_bool) == expected
 

	
 
    def test_mention_extractor(self):
 
        from kallithea.lib.utils2 import extract_mentioned_usernames
 
        sample = (
 
            "@first hi there @world here's my email username@example.com "
 
            "@lukaszb check @one_more22 it pls @ ttwelve @D[] @one@two@three "
 
            "@UPPER    @cAmEL @2one_more22 @john please see this http://org.pl "
 
            "@marian.user just do it @marco-polo and next extract @marco_polo "
 
            "user.dot  hej ! not-needed maril@example.com"
 
        )
 

	
 
        expected = set([
 
            '2one_more22', 'first', 'lukaszb', 'one', 'one_more22', 'UPPER', 'cAmEL', 'john',
 
            'marian.user', 'marco-polo', 'marco_polo', 'world'])
 
        assert expected == set(extract_mentioned_usernames(sample))
kallithea/tests/other/test_mail.py
Show inline comments
 
import mock
 

	
 
import kallithea
 
from kallithea.tests import *
 
from kallithea.model.db import User
 

	
 
class smtplib_mock(object):
 

	
 
    @classmethod
 
    def SMTP(cls, server, port, local_hostname):
 
        return smtplib_mock()
 

	
 
    def ehlo(self):
 
        pass
 
    def quit(self):
 
        pass
 
    def sendmail(self, sender, dest, msg):
 
        smtplib_mock.lastsender = sender
 
        smtplib_mock.lastdest = dest
 
        smtplib_mock.lastmsg = msg
 
        pass
 

	
 
@mock.patch('kallithea.lib.rcmail.smtp_mailer.smtplib', smtplib_mock)
 
class TestMail(TestControllerPytest):
 
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'
 
        subject = 'subject'
 
        body = 'body'
 
        html_body = 'html_body'
 

	
 
        config_mock = {
 
            'smtp_server': mailserver,
 
            'app_email_from': envelope_from,
 
            'email_to': email_to,
 
        }
 
        with mock.patch('kallithea.lib.celerylib.tasks.config', config_mock):
 
            kallithea.lib.celerylib.tasks.send_email(recipients, subject, body, html_body)
 

	
 
        assert smtplib_mock.lastdest == set([TEST_USER_ADMIN_EMAIL, email_to])
 
        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_no_email_to(self):
kallithea/tests/other/test_validators.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
import formencode
 
import pytest
 
import tempfile
 

	
 
from kallithea.tests import *
 

	
 
from kallithea.model import validators as v
 
from kallithea.model.user_group import UserGroupModel
 

	
 
from kallithea.model.meta import Session
 
from kallithea.model.repo_group import RepoGroupModel
 
from kallithea.tests.fixture import Fixture
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestRepoGroups(TestControllerPytest):
 
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)
 

	
 
    def test_ValidUserGroup(self):
 
        validator = v.ValidUserGroup()
 
        with pytest.raises(formencode.Invalid):
 
            validator.to_python(u'default')
 
        with pytest.raises(formencode.Invalid):
 
            validator.to_python(u'.,')
 

	
 
        gr = fixture.create_user_group(u'test')
 
        gr2 = fixture.create_user_group(u'tes2')
 
        Session().commit()
 
        with pytest.raises(formencode.Invalid):
 
            validator.to_python(u'test')
0 comments (0 inline, 0 general)