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.
26 files changed:
0 comments (0 inline, 0 general)
kallithea/tests/__init__.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
# This program is free software: you can redistribute it and/or modify
 
# it under the terms of the GNU General Public License as published by
 
# the Free Software Foundation, either version 3 of the License, or
 
# (at your option) any later version.
 
#
 
# This program is distributed in the hope that it will be useful,
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
# GNU General Public License for more details.
 
#
 
# You should have received a copy of the GNU General Public License
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 

	
 
"""
 
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()
 
        return response.session['authuser']
 

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

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

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

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

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

	
kallithea/tests/api/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):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/admin', action='index',
 
                                    filter='repository:xxx'))
 
        response.mustcontain('3 Entries')
 

	
 
    def test_filter_journal_filter_exact_match_on_repository_CamelCase(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/admin', action='index',
 
                                    filter='repository:XxX'))
 
        response.mustcontain('3 Entries')
 

	
 
    def test_filter_journal_filter_wildcard_on_repository(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/admin', action='index',
 
                                    filter='repository:*test*'))
 
        response.mustcontain('862 Entries')
 

	
 
    def test_filter_journal_filter_prefix_on_repository(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/admin', action='index',
 
                                    filter='repository:test*'))
 
        response.mustcontain('257 Entries')
 

	
 
    def test_filter_journal_filter_prefix_on_repository_CamelCase(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/admin', action='index',
 
                                    filter='repository:Test*'))
 
        response.mustcontain('257 Entries')
 

	
 
    def test_filter_journal_filter_prefix_on_repository_and_user(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/admin', action='index',
 
                                    filter='repository:test* AND username:demo'))
 
        response.mustcontain('130 Entries')
 

	
 
    def test_filter_journal_filter_prefix_on_repository_or_other_repo(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/admin', action='index',
 
                                    filter='repository:test* OR repository:xxx'))
 
        response.mustcontain('260 Entries')  # 257 + 3
 

	
 
    def test_filter_journal_filter_exact_match_on_username(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/admin', action='index',
 
                                    filter='username:demo'))
 
        response.mustcontain('1087 Entries')
 

	
 
    def test_filter_journal_filter_exact_match_on_username_camelCase(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/admin', action='index',
 
                                    filter='username:DemO'))
 
        response.mustcontain('1087 Entries')
 

	
 
    def test_filter_journal_filter_wildcard_on_username(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/admin', action='index',
 
                                    filter='username:*test*'))
 
        response.mustcontain('100 Entries')
 

	
 
    def test_filter_journal_filter_prefix_on_username(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/admin', action='index',
 
                                    filter='username:demo*'))
 
        response.mustcontain('1101 Entries')
 

	
 
    def test_filter_journal_filter_prefix_on_user_or_other_user(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/admin', action='index',
 
                                    filter='username:demo OR username:volcan'))
 
        response.mustcontain('1095 Entries')  # 1087 + 8
 

	
 
    def test_filter_journal_filter_wildcard_on_action(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/admin', action='index',
 
                                    filter='action:*pull_request*'))
 
        response.mustcontain('187 Entries')
 

	
 
    def test_filter_journal_filter_on_date(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/admin', action='index',
 
                                    filter='date:20121010'))
 
        response.mustcontain('47 Entries')
 

	
 
    def test_filter_journal_filter_on_date_2(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='admin/admin', action='index',
 
                                    filter='date:20121020'))
 
        response.mustcontain('17 Entries')
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):
 
        self.log_user()
 

	
 
        params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_ldap')
 
        params.update({'auth_ldap_host': '',
 
                       'auth_ldap_port': 'i-should-be-number',  # bad port num
 
                       'auth_ldap_tls_kind': 'PLAIN',
 
                       'auth_ldap_tls_reqcert': 'NEVER',
 
                       'auth_ldap_dn_user': '',
 
                       'auth_ldap_dn_pass': '',
 
                       'auth_ldap_base_dn': '',
 
                       'auth_ldap_filter': '',
 
                       'auth_ldap_search_scope': 'BASE',
 
                       'auth_ldap_attr_login': '',
 
                       'auth_ldap_attr_firstname': '',
 
                       'auth_ldap_attr_lastname': '',
 
                       'auth_ldap_attr_email': ''})
 
        test_url = url(controller='admin/auth_settings',
 
                       action='auth_settings')
 

	
 
        response = self.app.post(url=test_url, params=params)
 

	
 
        response.mustcontain("""<span class="error-message">"""
 
                             """Please enter a number</span>""")
 

	
 
    @skipif(not ldap_lib_installed, reason='skipping due to missing ldap lib')
 
    def test_ldap_error_form(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': 'Host',
 
                       'auth_ldap_port': '123',
 
                       'auth_ldap_tls_kind': 'PLAIN',
 
                       'auth_ldap_tls_reqcert': 'NEVER',
 
                       'auth_ldap_dn_user': '',
 
                       'auth_ldap_dn_pass': '',
 
                       'auth_ldap_base_dn': '',
 
                       'auth_ldap_filter': '',
 
                       'auth_ldap_search_scope': 'BASE',
 
                       'auth_ldap_attr_login': '',  # <----- missing required input
 
                       'auth_ldap_attr_firstname': '',
 
                       'auth_ldap_attr_lastname': '',
 
                       'auth_ldap_attr_email': ''})
 

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

	
 
        response = self.app.post(url=test_url, params=params)
 

	
 
        response.mustcontain("""<span class="error-message">The LDAP Login"""
 
                             """ attribute of the CN must be specified""")
 

	
 
    def test_ldap_login(self):
 
        pass
 

	
 
    def test_ldap_login_incorrect(self):
 
        pass
 

	
 
    def _container_auth_setup(self, **settings):
 
        self.log_user()
 

	
 
        params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_container')
 
        params.update(settings)
 

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

	
 
        response = self.app.post(url=test_url, params=params)
 
        response = response.follow()
 
        response.click('Log Out') # end admin login session
 

	
 
    def _container_auth_verify_login(self, resulting_username, **get_kwargs):
 
        response = self.app.get(
 
            url=url(controller='admin/my_account', action='my_account'),
 
            **get_kwargs
 
        )
 
        response.mustcontain('My Account %s' % resulting_username)
 

	
 
    def test_container_auth_login_header(self):
 
        self._container_auth_setup(
 
            auth_container_header='THE_USER_NAME',
 
            auth_container_email_header='',
 
            auth_container_firstname_header='',
 
            auth_container_lastname_header='',
 
            auth_container_fallback_header='',
 
            auth_container_clean_username='False',
 
        )
 
        self._container_auth_verify_login(
 
            extra_environ={'THE_USER_NAME': 'john@example.org'},
 
            resulting_username='john@example.org',
 
        )
 

	
 
    def test_container_auth_login_header_attr(self):
 
        self._container_auth_setup(
 
            auth_container_header='THE_USER_NAME',
 
            auth_container_email_header='THE_USER_EMAIL',
 
            auth_container_firstname_header='THE_USER_FIRSTNAME',
 
            auth_container_lastname_header='THE_USER_LASTNAME',
 
            auth_container_fallback_header='',
 
            auth_container_clean_username='False',
 
        )
 
        response = self.app.get(
 
            url=url(controller='admin/my_account', action='my_account'),
 
            extra_environ={'THE_USER_NAME': 'johnd',
 
                           'THE_USER_EMAIL': 'john@example.org',
 
                           'THE_USER_FIRSTNAME': 'John',
 
                           'THE_USER_LASTNAME': 'Doe',
 
                           }
 
        )
 
        assert response.form['email'].value == 'john@example.org'
 
        assert response.form['firstname'].value == 'John'
 
        assert response.form['lastname'].value == 'Doe'
 

	
 

	
 
    def test_container_auth_login_fallback_header(self):
 
        self._container_auth_setup(
 
            auth_container_header='THE_USER_NAME',
 
            auth_container_email_header='',
 
            auth_container_firstname_header='',
 
            auth_container_lastname_header='',
 
            auth_container_fallback_header='HTTP_X_YZZY',
 
            auth_container_clean_username='False',
 
        )
 
        self._container_auth_verify_login(
 
            headers={'X-Yzzy': r'foo\bar'},
 
            resulting_username=r'foo\bar',
 
        )
 

	
 
    def test_container_auth_clean_username_at(self):
 
        self._container_auth_setup(
 
            auth_container_header='REMOTE_USER',
 
            auth_container_email_header='',
 
            auth_container_firstname_header='',
 
            auth_container_lastname_header='',
 
            auth_container_fallback_header='',
 
            auth_container_clean_username='True',
 
        )
 
        self._container_auth_verify_login(
 
            extra_environ={'REMOTE_USER': 'john@example.org'},
 
            resulting_username='john',
 
        )
 

	
 
    def test_container_auth_clean_username_backslash(self):
 
        self._container_auth_setup(
 
            auth_container_header='REMOTE_USER',
 
            auth_container_email_header='',
 
            auth_container_firstname_header='',
 
            auth_container_lastname_header='',
 
            auth_container_fallback_header='',
 
            auth_container_clean_username='True',
 
        )
 
        self._container_auth_verify_login(
 
            extra_environ={'REMOTE_USER': r'example\jane'},
 
            resulting_username=r'jane',
 
        )
 

	
 
    def test_container_auth_no_logout(self):
 
        self._container_auth_setup(
 
            auth_container_header='REMOTE_USER',
 
            auth_container_email_header='',
 
            auth_container_firstname_header='',
 
            auth_container_lastname_header='',
 
            auth_container_fallback_header='',
 
            auth_container_clean_username='True',
 
        )
 
        response = self.app.get(
 
            url=url(controller='admin/my_account', action='my_account'),
 
            extra_environ={'REMOTE_USER': 'john'},
 
        )
 
        assert 'Log Out' not in response.normal_body
 

	
 
    def test_crowd_save_settings(self):
 
        self.log_user()
 

	
 
        params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_crowd')
 
        params.update({'auth_crowd_host': ' hostname ',
 
                       'auth_crowd_app_password': 'secret',
 
                       'auth_crowd_admin_groups': 'mygroup',
 
                       'auth_crowd_port': '123',
 
                       'auth_crowd_app_name': 'xyzzy'})
 

	
 
        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_crowd_host'] == u'hostname', 'fail db write compare'
 

	
 
    @skipif(not pam_lib_installed, reason='skipping due to missing pam lib')
 
    def test_pam_save_settings(self):
 
        self.log_user()
 

	
 
        params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_pam')
 
        params.update({'auth_pam_service': 'kallithea',
 
                       'auth_pam_gecos': '^foo-.*'})
 

	
 
        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_pam_service'] == u'kallithea', 'fail db write compare'
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(),
 
        }
 
        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_browser_fakeout(self):
 
        response = self.app.post(url('default', id=1), params=dict(_method='put', _authentication_token=self.authentication_token()))
 

	
 
    def test_delete(self):
 
        # Not possible due to CSRF protection.
 
        response = self.app.delete(url('default', id=1), status=405)
 

	
 
    def test_delete_browser_fakeout(self):
 
        response = self.app.post(url('default', id=1), params=dict(_method='delete', _authentication_token=self.authentication_token()))
 

	
 
    def test_show(self):
 
        response = self.app.get(url('default', id=1))
 

	
 
    def test_show_as_xml(self):
 
        response = self.app.get(url('formatted_default', id=1, format='xml'))
 

	
 
    def test_edit(self):
 
        response = self.app.get(url('edit_default', id=1))
 

	
 
    def test_edit_as_xml(self):
 
        response = self.app.get(url('formatted_edit_default', id=1, format='xml'))
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',
 
                                         'public': 'public',
 
                                         '_authentication_token': self.authentication_token()},
 
                                 status=302)
 
        response = response.follow()
 
        response.mustcontain('added file: foo')
 
        response.mustcontain('gist test')
 
        response.mustcontain('<div class="btn btn-mini btn-success disabled">Public Gist</div>')
 

	
 
    def test_create_with_path_with_dirs(self):
 
        self.log_user()
 
        response = self.app.post(url('gists'),
 
                                 params={'lifetime': -1,
 
                                         'content': 'gist test',
 
                                         'filename': '/home/foo',
 
                                         'public': 'public',
 
                                         '_authentication_token': self.authentication_token()},
 
                                 status=200)
 
        response.mustcontain('Filename cannot be inside a directory')
 

	
 
    def test_access_expired_gist(self):
 
        self.log_user()
 
        gist = _create_gist('never-see-me')
 
        gist.gist_expires = 0  # 1970
 
        Session().add(gist)
 
        Session().commit()
 

	
 
        response = self.app.get(url('gist', gist_id=gist.gist_access_id), status=404)
 

	
 
    def test_create_private(self):
 
        self.log_user()
 
        response = self.app.post(url('gists'),
 
                                 params={'lifetime': -1,
 
                                         'content': 'private gist test',
 
                                         'filename': 'private-foo',
 
                                         'private': 'private',
 
                                         '_authentication_token': self.authentication_token()},
 
                                 status=302)
 
        response = response.follow()
 
        response.mustcontain('added file: private-foo<')
 
        response.mustcontain('private gist test')
 
        response.mustcontain('<div class="btn btn-mini btn-warning disabled">Private Gist</div>')
 

	
 
    def test_create_with_description(self):
 
        self.log_user()
 
        response = self.app.post(url('gists'),
 
                                 params={'lifetime': -1,
 
                                         'content': 'gist test',
 
                                         'filename': 'foo-desc',
 
                                         'description': 'gist-desc',
 
                                         'public': 'public',
 
                                         '_authentication_token': self.authentication_token()},
 
                                 status=302)
 
        response = response.follow()
 
        response.mustcontain('added file: foo-desc')
 
        response.mustcontain('gist test')
 
        response.mustcontain('gist-desc')
 
        response.mustcontain('<div class="btn btn-mini btn-success disabled">Public Gist</div>')
 

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

	
 
    # Functionality currently not implemented
 
    #def test_update(self):
 
    #    response = self.app.put(url('gist', gist_id=1))
 

	
 
    def test_delete(self):
 
        self.log_user()
 
        gist = _create_gist('delete-me')
 
        response = self.app.post(url('gist', gist_id=gist.gist_id),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 

	
 
    def test_delete_normal_user_his_gist(self):
 
        self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
 
        gist = _create_gist('delete-me', owner=TEST_USER_REGULAR_LOGIN)
 
        response = self.app.post(url('gist', gist_id=gist.gist_id),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 

	
 
    def test_delete_normal_user_not_his_own_gist(self):
 
        self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
 
        gist = _create_gist('delete-me')
 
        response = self.app.post(url('gist', gist_id=gist.gist_id), status=403,
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 

	
 
    def test_show(self):
 
        gist = _create_gist('gist-show-me')
 
        response = self.app.get(url('gist', gist_id=gist.gist_access_id))
 
        response.mustcontain('added file: gist-show-me<')
 
        response.mustcontain('%s - created' % TEST_USER_ADMIN_LOGIN)
 
        response.mustcontain('gist-desc')
 
        response.mustcontain('<div class="btn btn-mini btn-success disabled">Public Gist</div>')
 

	
 
    def test_show_as_raw(self):
 
        gist = _create_gist('gist-show-me', content='GIST CONTENT')
 
        response = self.app.get(url('formatted_gist',
 
                                    gist_id=gist.gist_access_id, format='raw'))
 
        assert response.body == 'GIST CONTENT'
 

	
 
    def test_show_as_raw_individual_file(self):
 
        gist = _create_gist('gist-show-me-raw', content='GIST BODY')
 
        response = self.app.get(url('formatted_gist_file',
 
                                    gist_id=gist.gist_access_id, format='raw',
 
                                    revision='tip', f_path='gist-show-me-raw'))
 
        assert response.body == 'GIST BODY'
 

	
 
    def test_edit(self):
 
        response = self.app.get(url('edit_gist', gist_id=1))
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
 

	
 
        response = self.app.post(
 
            url('notification', notification_id=notification.notification_id),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 
        assert response.body == 'ok'
 

	
 
        cur_user = User.get(cur_usr_id)
 
        assert cur_user.notifications == []
 

	
 
    def test_show(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')
 

	
 
        subject = u'test'
 
        notif_body = u'hi there'
 
        notification = NotificationModel().create(created_by=cur_user,
 
                                                  subject=subject,
 
                                                  body=notif_body,
 
                                                  recipients=[cur_user, u1, u2])
 

	
 
        response = self.app.get(url('notification',
 
                                    notification_id=notification.notification_id))
 

	
 
        response.mustcontain(subject)
 
        response.mustcontain(notif_body)
 

	
 
    def test_description_with_age(self):
 
        self.log_user()
 
        cur_user = self._get_logged_user()
 
        subject = u'test'
 
        notify_body = u'hi there'
 
        notification = NotificationModel().create(created_by = cur_user,
 
                                                  subject    = subject,
 
                                                  body       = notify_body)
 

	
 
        description = NotificationModel().make_description(notification)
 
        assert description == "{0} sent message {1}".format(
 
                cur_user.username,
 
                h.age(notification.created_on)
 
                )
 

	
 
    def test_description_with_datetime(self):
 
        self.log_user()
 
        cur_user = self._get_logged_user()
 
        subject = u'test'
 
        notify_body = u'hi there'
 
        notification = NotificationModel().create(created_by = cur_user,
 
                                                  subject    = subject,
 
                                                  body       = notify_body)
 

	
 
        description = NotificationModel().make_description(notification, False)
 
        assert description == "{0} sent message at {1}".format(
 
                cur_user.username,
 
                h.fmt_date(notification.created_on)
 
                )
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'])
 

	
 

	
 
    def test_index_overview(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_permissions_perms'))
 
        # Test response...
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)
 

	
 
        RepoModel().delete(repo_name)
 
        Session().commit()
 

	
 
    def test_create_in_group(self):
 
        self.log_user()
 

	
 
        ## create GROUP
 
        group_name = u'sometest_%s' % self.REPO_TYPE
 
        gr = RepoGroupModel().create(group_name=group_name,
 
                                     group_description=u'test',
 
                                     owner=TEST_USER_ADMIN_LOGIN)
 
        Session().commit()
 

	
 
        repo_name = u'ingroup'
 
        repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
 
        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,
 
                                                repo_group=gr.group_id,
 
                                                _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_full))
 
        assert response.json == {u'result': True}
 
        self.checkSessionFlash(response,
 
                               'Created repository <a href="/%s">%s</a>'
 
                               % (repo_name_full, repo_name_full))
 
        # test if the repo was created in the database
 
        new_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == repo_name_full).one()
 
        new_repo_id = new_repo.repo_id
 

	
 
        assert new_repo.repo_name == repo_name_full
 
        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_full))
 
        response.mustcontain(repo_name_full)
 
        response.mustcontain(self.REPO_TYPE)
 

	
 
        inherited_perms = UserRepoToPerm.query() \
 
            .filter(UserRepoToPerm.repository_id == new_repo_id).all()
 
        assert len(inherited_perms) == 1
 

	
 
        # 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_full)))
 
        except vcs.exceptions.VCSError:
 
            RepoGroupModel().delete(group_name)
 
            Session().commit()
 
            pytest.fail('no repo %s in filesystem' % repo_name)
 

	
 
        RepoModel().delete(repo_name_full)
 
        RepoGroupModel().delete(group_name)
 
        Session().commit()
 

	
 
    def test_create_in_group_without_needed_permissions(self):
 
        usr = self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
 
        # avoid spurious RepoGroup DetachedInstanceError ...
 
        authentication_token = self.authentication_token()
 
        # revoke
 
        user_model = UserModel()
 
        # disable fork and create on default user
 
        user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
 
        user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
 
        user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
 
        user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
 

	
 
        # disable on regular user
 
        user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
 
        user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
 
        user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
 
        user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
 
        Session().commit()
 

	
 
        ## create GROUP
 
        group_name = u'reg_sometest_%s' % self.REPO_TYPE
 
        gr = RepoGroupModel().create(group_name=group_name,
 
                                     group_description=u'test',
 
                                     owner=TEST_USER_ADMIN_LOGIN)
 
        Session().commit()
 

	
 
        group_name_allowed = u'reg_sometest_allowed_%s' % self.REPO_TYPE
 
        gr_allowed = RepoGroupModel().create(group_name=group_name_allowed,
 
                                     group_description=u'test',
 
                                     owner=TEST_USER_REGULAR_LOGIN)
 
        Session().commit()
 

	
 
        repo_name = u'ingroup'
 
        repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
 
        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,
 
                                                repo_group=gr.group_id,
 
                                                _authentication_token=authentication_token))
 

	
 
        response.mustcontain('Invalid value')
 

	
 
        # user is allowed to create in this group
 
        repo_name = u'ingroup'
 
        repo_name_full = RepoGroup.url_sep().join([group_name_allowed, repo_name])
 
        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,
 
                                                repo_group=gr_allowed.group_id,
 
                                                _authentication_token=authentication_token))
 

	
 
        ## run the check page that triggers the flash message
 
        response = self.app.get(url('repo_check_home', repo_name=repo_name_full))
 
        assert response.json == {u'result': True}
 
        self.checkSessionFlash(response,
 
                               'Created repository <a href="/%s">%s</a>'
 
                               % (repo_name_full, repo_name_full))
 
        # test if the repo was created in the database
 
        new_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == repo_name_full).one()
 
        new_repo_id = new_repo.repo_id
 

	
 
        assert new_repo.repo_name == repo_name_full
 
        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_full))
 
        response.mustcontain(repo_name_full)
 
        response.mustcontain(self.REPO_TYPE)
 

	
 
        inherited_perms = UserRepoToPerm.query() \
 
            .filter(UserRepoToPerm.repository_id == new_repo_id).all()
 
        assert len(inherited_perms) == 1
 

	
 
        # 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_full)))
 
        except vcs.exceptions.VCSError:
 
            RepoGroupModel().delete(group_name)
 
            Session().commit()
 
            pytest.fail('no repo %s in filesystem' % repo_name)
 

	
 
        RepoModel().delete(repo_name_full)
 
        RepoGroupModel().delete(group_name)
 
        RepoGroupModel().delete(group_name_allowed)
 
        Session().commit()
 

	
 
    def test_create_in_group_inherit_permissions(self):
 
        self.log_user()
 

	
 
        ## create GROUP
 
        group_name = u'sometest_%s' % self.REPO_TYPE
 
        gr = RepoGroupModel().create(group_name=group_name,
 
                                     group_description=u'test',
 
                                     owner=TEST_USER_ADMIN_LOGIN)
 
        perm = Permission.get_by_key('repository.write')
 
        RepoGroupModel().grant_user_permission(gr, TEST_USER_REGULAR_LOGIN, perm)
 

	
 
        ## add repo permissions
 
        Session().commit()
 

	
 
        repo_name = u'ingroup_inherited_%s' % self.REPO_TYPE
 
        repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
 
        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,
 
                                                repo_group=gr.group_id,
 
                                                repo_copy_permissions=True,
 
                                                _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_full))
 
        self.checkSessionFlash(response,
 
                               'Created repository <a href="/%s">%s</a>'
 
                               % (repo_name_full, repo_name_full))
 
        # test if the repo was created in the database
 
        new_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == repo_name_full).one()
 
        new_repo_id = new_repo.repo_id
 

	
 
        assert new_repo.repo_name == repo_name_full
 
        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_full))
 
        response.mustcontain(repo_name_full)
 
        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_full)))
 
        except vcs.exceptions.VCSError:
 
            RepoGroupModel().delete(group_name)
 
            Session().commit()
 
            pytest.fail('no repo %s in filesystem' % repo_name)
 

	
 
        #check if inherited permissiona are applied
 
        inherited_perms = UserRepoToPerm.query() \
 
            .filter(UserRepoToPerm.repository_id == new_repo_id).all()
 
        assert len(inherited_perms) == 2
 

	
 
        assert TEST_USER_REGULAR_LOGIN in [x.user.username
 
                                                    for x in inherited_perms]
 
        assert 'repository.write' in [x.permission.permission_name
 
                                               for x in inherited_perms]
 

	
 
        RepoModel().delete(repo_name_full)
 
        RepoGroupModel().delete(group_name)
 
        Session().commit()
 

	
 
    def test_create_remote_repo_wrong_clone_uri(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,
 
                                                clone_uri='http://127.0.0.1/repo',
 
                                                _authentication_token=self.authentication_token()))
 
        response.mustcontain('Invalid repository URL')
 

	
 

	
 
    def test_create_remote_repo_wrong_clone_uri_hg_svn(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,
 
                                                clone_uri='svn+http://127.0.0.1/repo',
 
                                                _authentication_token=self.authentication_token()))
 
        response.mustcontain('Invalid repository URL')
 

	
 

	
 
    def test_delete(self):
 
        self.log_user()
 
        repo_name = u'vcs_test_new_to_delete_%s' % self.REPO_TYPE
 
        description = u'description for newly created repo'
 
        response = self.app.post(url('repos'),
 
                        fixture._get_repo_create_params(repo_private=False,
 
                                                repo_type=self.REPO_TYPE,
 
                                                repo_name=repo_name,
 
                                                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))
 
        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)
 

	
 
        response = self.app.post(url('delete_repo', repo_name=repo_name),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 

	
 
        self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name))
 

	
 
        response.follow()
 

	
 
        #check if repo was deleted from db
 
        deleted_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == repo_name).scalar()
 

	
 
        assert deleted_repo == None
 

	
 
        assert os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)) == False
 

	
 
    def test_delete_non_ascii(self):
 
        self.log_user()
 
        non_ascii = "ąęł"
 
        repo_name = "%s%s" % (safe_str(self.NEW_REPO), non_ascii)
 
        repo_name_unicode = safe_unicode(repo_name)
 
        description = 'description for newly created repo' + non_ascii
 
        description_unicode = safe_unicode(description)
 
        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,
 
                               u'Created repository <a href="/%s">%s</a>'
 
                               % (urllib.quote(repo_name), repo_name_unicode))
 
        # test if the repo was created in the database
 
        new_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == repo_name_unicode).one()
 

	
 
        assert new_repo.repo_name == repo_name_unicode
 
        assert new_repo.description == description_unicode
 

	
 
        # 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_unicode)))
 
        except vcs.exceptions.VCSError:
 
            pytest.fail('no repo %s in filesystem' % repo_name)
 

	
 
        response = self.app.post(url('delete_repo', repo_name=repo_name),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name_unicode))
 
        response.follow()
 

	
 
        #check if repo was deleted from db
 
        deleted_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == repo_name_unicode).scalar()
 

	
 
        assert deleted_repo == None
 

	
 
        assert os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_unicode)) == False
 

	
 
    def test_delete_repo_with_group(self):
 
        #TODO:
 
        pass
 

	
 
    def test_delete_browser_fakeout(self):
 
        response = self.app.post(url('delete_repo', repo_name=self.REPO),
 
                                 params=dict(_method='delete', _authentication_token=self.authentication_token()))
 

	
 
    def test_show(self):
 
        self.log_user()
 
        response = self.app.get(url('summary_home', repo_name=self.REPO))
 

	
 
    def test_edit(self):
 
        response = self.app.get(url('edit_repo', repo_name=self.REPO))
 

	
 
    def test_set_private_flag_sets_default_to_none(self):
 
        self.log_user()
 
        #initially repository perm should be read
 
        perm = _get_permission_for_user(user='default', repo=self.REPO)
 
        assert len(perm), 1
 
        assert perm[0].permission.permission_name == 'repository.read'
 
        assert Repository.get_by_repo_name(self.REPO).private == False
 

	
 
        response = self.app.put(url('put_repo', repo_name=self.REPO),
 
                        fixture._get_repo_create_params(repo_private=1,
 
                                                repo_name=self.REPO,
 
                                                repo_type=self.REPO_TYPE,
 
                                                user=TEST_USER_ADMIN_LOGIN,
 
                                                _authentication_token=self.authentication_token()))
 
        self.checkSessionFlash(response,
 
                               msg='Repository %s updated successfully' % (self.REPO))
 
        assert Repository.get_by_repo_name(self.REPO).private == True
 

	
 
        #now the repo default permission should be None
 
        perm = _get_permission_for_user(user='default', repo=self.REPO)
 
        assert len(perm), 1
 
        assert perm[0].permission.permission_name == 'repository.none'
 

	
 
        response = self.app.put(url('put_repo', repo_name=self.REPO),
 
                        fixture._get_repo_create_params(repo_private=False,
 
                                                repo_name=self.REPO,
 
                                                repo_type=self.REPO_TYPE,
 
                                                user=TEST_USER_ADMIN_LOGIN,
 
                                                _authentication_token=self.authentication_token()))
 
        self.checkSessionFlash(response,
 
                               msg='Repository %s updated successfully' % (self.REPO))
 
        assert Repository.get_by_repo_name(self.REPO).private == False
 

	
 
        #we turn off private now the repo default permission should stay None
 
        perm = _get_permission_for_user(user='default', repo=self.REPO)
 
        assert len(perm), 1
 
        assert perm[0].permission.permission_name == 'repository.none'
 

	
 
        #update this permission back
 
        perm[0].permission = Permission.get_by_key('repository.read')
 
        Session().add(perm[0])
 
        Session().commit()
 

	
 
    def test_set_repo_fork_has_no_self_id(self):
 
        self.log_user()
 
        repo = Repository.get_by_repo_name(self.REPO)
 
        response = self.app.get(url('edit_repo_advanced', repo_name=self.REPO))
 
        opt = """<option value="%s">%s</option>""" % (repo.repo_id, self.REPO)
 
        response.mustcontain(no=[opt])
 

	
 
    def test_set_fork_of_other_repo(self):
 
        self.log_user()
 
        other_repo = u'other_%s' % self.REPO_TYPE
 
        fixture.create_repo(other_repo, repo_type=self.REPO_TYPE)
 
        repo = Repository.get_by_repo_name(self.REPO)
 
        repo2 = Repository.get_by_repo_name(other_repo)
 
        response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
 
                                params=dict(id_fork_of=repo2.repo_id, _authentication_token=self.authentication_token()))
 
        repo = Repository.get_by_repo_name(self.REPO)
 
        repo2 = Repository.get_by_repo_name(other_repo)
 
        self.checkSessionFlash(response,
 
            'Marked repository %s as fork of %s' % (repo.repo_name, repo2.repo_name))
 

	
 
        assert repo.fork == repo2
 
        response = response.follow()
 
        # check if given repo is selected
 

	
 
        opt = """<option value="%s" selected="selected">%s</option>""" % (
 
                    repo2.repo_id, repo2.repo_name)
 
        response.mustcontain(opt)
 

	
 
        fixture.destroy_repo(other_repo, forks='detach')
 

	
 
    def test_set_fork_of_other_type_repo(self):
 
        self.log_user()
 
        repo = Repository.get_by_repo_name(self.REPO)
 
        repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
 
        response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
 
                                params=dict(id_fork_of=repo2.repo_id, _authentication_token=self.authentication_token()))
 
        repo = Repository.get_by_repo_name(self.REPO)
 
        repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
 
        self.checkSessionFlash(response,
 
            'Cannot set repository as fork of repository with other type')
 

	
 
    def test_set_fork_of_none(self):
 
        self.log_user()
 
        ## mark it as None
 
        response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
 
                                params=dict(id_fork_of=None, _authentication_token=self.authentication_token()))
 
        repo = Repository.get_by_repo_name(self.REPO)
 
        repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
 
        self.checkSessionFlash(response,
 
                               'Marked repository %s as fork of %s'
 
                               % (repo.repo_name, "Nothing"))
 
        assert repo.fork is None
 

	
 
    def test_set_fork_of_same_repo(self):
 
        self.log_user()
 
        repo = Repository.get_by_repo_name(self.REPO)
 
        response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
 
                                params=dict(id_fork_of=repo.repo_id, _authentication_token=self.authentication_token()))
 
        self.checkSessionFlash(response,
 
                               'An error occurred during this operation')
 

	
 
    def test_create_on_top_level_without_permissions(self):
 
        usr = self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
 
        # revoke
 
        user_model = UserModel()
 
        # disable fork and create on default user
 
        user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
 
        user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
 
        user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
 
        user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
 

	
 
        # disable on regular user
 
        user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
 
        user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
 
        user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
 
        user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
 
        Session().commit()
 

	
 

	
 
        user = User.get(usr['user_id'])
 

	
 
        repo_name = self.NEW_REPO + u'no_perms'
 
        description = '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()))
 

	
 
        response.mustcontain('<span class="error-message">Invalid value</span>')
 

	
 
        RepoModel().delete(repo_name)
 
        Session().commit()
 

	
 
    @mock.patch.object(RepoModel, '_create_filesystem_repo', error_function)
 
    def test_create_repo_when_filesystem_op_fails(self):
 
        self.log_user()
 
        repo_name = self.NEW_REPO
 
        description = '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()))
 

	
 
        self.checkSessionFlash(response,
 
                               'Error creating repository %s' % repo_name)
 
        # repo must not be in db
 
        repo = Repository.get_by_repo_name(repo_name)
 
        assert repo == None
 

	
 
        # repo must not be in filesystem !
 
        assert not os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name))
 

	
 
class TestAdminReposControllerGIT(_BaseTestCase):
 
    REPO = GIT_REPO
 
    REPO_TYPE = 'git'
 
    NEW_REPO = NEW_GIT_REPO
 
    OTHER_TYPE_REPO = HG_REPO
 
    OTHER_TYPE = 'hg'
 

	
 

	
 
class TestAdminReposControllerHG(_BaseTestCase):
 
    REPO = HG_REPO
 
    REPO_TYPE = 'hg'
 
    NEW_REPO = NEW_HG_REPO
 
    OTHER_TYPE_REPO = GIT_REPO
 
    OTHER_TYPE = 'git'
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
 
        ## delete
 
        self.app.post(url('admin_settings_hooks'),
 
                        params=dict(hook_id=hook_id, _authentication_token=self.authentication_token()))
 
        response = self.app.get(url('admin_settings_hooks'))
 
        response.mustcontain(no=['test_hooks_2'])
 
        response.mustcontain(no=['cd %s2' % tempfile.gettempdir()])
 

	
 
    def test_index_search(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_settings_search'))
 

	
 
    def test_index_system(self):
 
        self.log_user()
 
        response = self.app.get(url('admin_settings_system'))
 

	
 
    def test_ga_code_active(self):
 
        self.log_user()
 
        old_title = 'Kallithea'
 
        old_realm = 'Kallithea authentication'
 
        new_ga_code = 'ga-test-123456789'
 
        response = self.app.post(url('admin_settings_global'),
 
                        params=dict(title=old_title,
 
                                 realm=old_realm,
 
                                 ga_code=new_ga_code,
 
                                 captcha_private_key='',
 
                                 captcha_public_key='',
 
                                 _authentication_token=self.authentication_token(),
 
                                 ))
 

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

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

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

	
 
    def test_ga_code_inactive(self):
 
        self.log_user()
 
        old_title = 'Kallithea'
 
        old_realm = 'Kallithea authentication'
 
        new_ga_code = ''
 
        response = self.app.post(url('admin_settings_global'),
 
                        params=dict(title=old_title,
 
                                 realm=old_realm,
 
                                 ga_code=new_ga_code,
 
                                 captcha_private_key='',
 
                                 captcha_public_key='',
 
                                 _authentication_token=self.authentication_token(),
 
                                 ))
 

	
 
        self.checkSessionFlash(response, 'Updated application settings')
 
        assert Setting.get_app_settings()['ga_code'] == new_ga_code
 

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

	
 
    def test_captcha_activate(self):
 
        self.log_user()
 
        old_title = 'Kallithea'
 
        old_realm = 'Kallithea authentication'
 
        new_ga_code = ''
 
        response = self.app.post(url('admin_settings_global'),
 
                        params=dict(title=old_title,
 
                                 realm=old_realm,
 
                                 ga_code=new_ga_code,
 
                                 captcha_private_key='1234567890',
 
                                 captcha_public_key='1234567890',
 
                                 _authentication_token=self.authentication_token(),
 
                                 ))
 

	
 
        self.checkSessionFlash(response, 'Updated application settings')
 
        assert Setting.get_app_settings()['captcha_private_key'] == '1234567890'
 

	
 
        response = self.app.get(url('register'))
 
        response.mustcontain('captcha')
 

	
 
    def test_captcha_deactivate(self):
 
        self.log_user()
 
        old_title = 'Kallithea'
 
        old_realm = 'Kallithea authentication'
 
        new_ga_code = ''
 
        response = self.app.post(url('admin_settings_global'),
 
                        params=dict(title=old_title,
 
                                 realm=old_realm,
 
                                 ga_code=new_ga_code,
 
                                 captcha_private_key='',
 
                                 captcha_public_key='1234567890',
 
                                 _authentication_token=self.authentication_token(),
 
                                 ))
 

	
 
        self.checkSessionFlash(response, 'Updated application settings')
 
        assert Setting.get_app_settings()['captcha_private_key'] == ''
 

	
 
        response = self.app.get(url('register'))
 
        response.mustcontain(no=['captcha'])
 

	
 
    def test_title_change(self):
 
        self.log_user()
 
        old_title = 'Kallithea'
 
        new_title = old_title + '_changed'
 
        old_realm = 'Kallithea authentication'
 

	
 
        for new_title in ['Changed', 'Żółwik', old_title]:
 
            response = self.app.post(url('admin_settings_global'),
 
                        params=dict(title=new_title,
 
                                 realm=old_realm,
 
                                 ga_code='',
 
                                 captcha_private_key='',
 
                                 captcha_public_key='',
 
                                 _authentication_token=self.authentication_token(),
 
                                ))
 

	
 
            self.checkSessionFlash(response, 'Updated application settings')
 
            assert Setting.get_app_settings()['title'] == new_title.decode('utf-8')
 

	
 
            response = response.follow()
 
            response.mustcontain("""<div class="branding">%s</div>""" % new_title)
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),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 

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

	
 
        assert gr == None
 

	
 
    def test_default_perms_enable_repository_read_on_group(self):
 
        self.log_user()
 
        users_group_name = TEST_USER_GROUP + 'another2'
 
        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()
 

	
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        self.checkSessionFlash(response,
 
                               'Created user group ')
 
        ## ENABLE REPO CREATE ON A GROUP
 
        response = self.app.put(url('edit_user_group_default_perms',
 
                                    id=ug.users_group_id),
 
                                 {'create_repo_perm': True,
 
                                  '_authentication_token': self.authentication_token()})
 
        response.follow()
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        p = Permission.get_by_key('hg.create.repository')
 
        p2 = Permission.get_by_key('hg.usergroup.create.false')
 
        p3 = Permission.get_by_key('hg.fork.none')
 
        # check if user has this perms, they should be here since
 
        # defaults are on
 
        perms = UserGroupToPerm.query() \
 
            .filter(UserGroupToPerm.users_group == ug).all()
 

	
 
        assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
 
                    [ug.users_group_id, p2.permission_id],
 
                    [ug.users_group_id, p3.permission_id]])
 

	
 
        ## DISABLE REPO CREATE ON A GROUP
 
        response = self.app.put(
 
            url('edit_user_group_default_perms', id=ug.users_group_id),
 
            params={'_authentication_token': self.authentication_token()})
 

	
 
        response.follow()
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        p = Permission.get_by_key('hg.create.none')
 
        p2 = Permission.get_by_key('hg.usergroup.create.false')
 
        p3 = Permission.get_by_key('hg.fork.none')
 

	
 
        # check if user has this perms, they should be here since
 
        # defaults are on
 
        perms = UserGroupToPerm.query() \
 
            .filter(UserGroupToPerm.users_group == ug).all()
 

	
 
        assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
 
                    [ug.users_group_id, p2.permission_id],
 
                    [ug.users_group_id, p3.permission_id]])
 

	
 
        # DELETE !
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        ugid = ug.users_group_id
 
        response = self.app.post(url('users_group', id=ug.users_group_id),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 
        response = response.follow()
 
        gr = Session().query(UserGroup) \
 
            .filter(UserGroup.users_group_name == users_group_name).scalar()
 

	
 
        assert gr == None
 
        p = Permission.get_by_key('hg.create.repository')
 
        perms = UserGroupToPerm.query() \
 
            .filter(UserGroupToPerm.users_group_id == ugid).all()
 
        perms = [[x.users_group_id,
 
                  x.permission_id, ] for x in perms]
 
        assert perms == []
 

	
 
    def test_default_perms_enable_repository_fork_on_group(self):
 
        self.log_user()
 
        users_group_name = TEST_USER_GROUP + 'another2'
 
        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()
 

	
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        self.checkSessionFlash(response,
 
                               'Created user group ')
 
        ## ENABLE REPO CREATE ON A GROUP
 
        response = self.app.put(url('edit_user_group_default_perms',
 
                                    id=ug.users_group_id),
 
                                {'fork_repo_perm': True, '_authentication_token': self.authentication_token()})
 

	
 
        response.follow()
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        p = Permission.get_by_key('hg.create.none')
 
        p2 = Permission.get_by_key('hg.usergroup.create.false')
 
        p3 = Permission.get_by_key('hg.fork.repository')
 
        # check if user has this perms, they should be here since
 
        # defaults are on
 
        perms = UserGroupToPerm.query() \
 
            .filter(UserGroupToPerm.users_group == ug).all()
 

	
 
        assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
 
                    [ug.users_group_id, p2.permission_id],
 
                    [ug.users_group_id, p3.permission_id]])
 

	
 
        ## DISABLE REPO CREATE ON A GROUP
 
        response = self.app.put(url('edit_user_group_default_perms', id=ug.users_group_id),
 
            params={'_authentication_token': self.authentication_token()})
 

	
 
        response.follow()
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        p = Permission.get_by_key('hg.create.none')
 
        p2 = Permission.get_by_key('hg.usergroup.create.false')
 
        p3 = Permission.get_by_key('hg.fork.none')
 
        # check if user has this perms, they should be here since
 
        # defaults are on
 
        perms = UserGroupToPerm.query() \
 
            .filter(UserGroupToPerm.users_group == ug).all()
 

	
 
        assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
 
                    [ug.users_group_id, p2.permission_id],
 
                    [ug.users_group_id, p3.permission_id]])
 

	
 
        # DELETE !
 
        ug = UserGroup.get_by_group_name(users_group_name)
 
        ugid = ug.users_group_id
 
        response = self.app.post(url('users_group', id=ug.users_group_id),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 
        response = response.follow()
 
        gr = Session().query(UserGroup) \
 
                           .filter(UserGroup.users_group_name ==
 
                                   users_group_name).scalar()
 

	
 
        assert gr == None
 
        p = Permission.get_by_key('hg.fork.repository')
 
        perms = UserGroupToPerm.query() \
 
            .filter(UserGroupToPerm.users_group_id == ugid).all()
 
        perms = [[x.users_group_id,
 
                  x.permission_id, ] for x in perms]
 
        assert perms == []
 

	
 
    def test_delete_browser_fakeout(self):
 
        response = self.app.post(url('users_group', id=1),
 
                                 params=dict(_method='delete', _authentication_token=self.authentication_token()))
 

	
 
    def test_show(self):
 
        response = self.app.get(url('users_group', id=1))
 

	
 
    def test_edit(self):
 
        response = self.app.get(url('edit_users_group', id=1))
 

	
 
    def test_assign_members(self):
 
        pass
 

	
 
    def test_add_create_permission(self):
 
        pass
 

	
 
    def test_revoke_members(self):
 
        pass
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()
 
        response.mustcontain("""newtestuser""")
 

	
 
    def test_create_err(self):
 
        self.log_user()
 
        username = 'new_user'
 
        password = ''
 
        name = u'name'
 
        lastname = u'lastname'
 
        email = 'errmail.example.com'
 

	
 
        response = self.app.post(url('users'), {'username': username,
 
                                               'password': password,
 
                                               'name': name,
 
                                               'active': False,
 
                                               'lastname': lastname,
 
                                               'email': email,
 
                                               '_authentication_token': self.authentication_token()})
 

	
 
        msg = validators.ValidUsername(False, {})._messages['system_invalid_username']
 
        msg = h.html_escape(msg % {'username': 'new_user'})
 
        response.mustcontain("""<span class="error-message">%s</span>""" % msg)
 
        response.mustcontain("""<span class="error-message">Please enter a value</span>""")
 
        response.mustcontain("""<span class="error-message">An email address must contain a single @</span>""")
 

	
 
        def get_user():
 
            Session().query(User).filter(User.username == username).one()
 

	
 
        with pytest.raises(NoResultFound):
 
            get_user(), 'found user in database'
 

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

	
 
    @parametrize('name,attrs',
 
        [('firstname', {'firstname': 'new_username'}),
 
         ('lastname', {'lastname': 'new_username'}),
 
         ('admin', {'admin': True}),
 
         ('admin', {'admin': False}),
 
         ('extern_type', {'extern_type': 'ldap'}),
 
         ('extern_type', {'extern_type': None}),
 
         ('extern_name', {'extern_name': 'test'}),
 
         ('extern_name', {'extern_name': None}),
 
         ('active', {'active': False}),
 
         ('active', {'active': True}),
 
         ('email', {'email': 'someemail@example.com'}),
 
        # ('new_password', {'new_password': 'foobar123',
 
        #                   'password_confirmation': 'foobar123'})
 
        ])
 
    def test_update(self, name, attrs):
 
        self.log_user()
 
        usr = fixture.create_user(self.test_user_1, password='qweqwe',
 
                                  email='testme@example.com',
 
                                  extern_type='internal',
 
                                  extern_name=self.test_user_1,
 
                                  skip_if_exists=True)
 
        Session().commit()
 
        params = usr.get_api_data(True)
 
        params.update({'password_confirmation': ''})
 
        params.update({'new_password': ''})
 
        params.update(attrs)
 
        if name == 'email':
 
            params['emails'] = [attrs['email']]
 
        if name == 'extern_type':
 
            #cannot update this via form, expected value is original one
 
            params['extern_type'] = "internal"
 
        if name == 'extern_name':
 
            #cannot update this via form, expected value is original one
 
            params['extern_name'] = self.test_user_1
 
            # special case since this user is not
 
                                          # logged in yet his data is not filled
 
                                          # so we use creation data
 

	
 
        params.update({'_authentication_token': self.authentication_token()})
 
        response = self.app.put(url('user', id=usr.user_id), params)
 
        self.checkSessionFlash(response, 'User updated successfully')
 
        params.pop('_authentication_token')
 

	
 
        updated_user = User.get_by_username(self.test_user_1)
 
        updated_params = updated_user.get_api_data(True)
 
        updated_params.update({'password_confirmation': ''})
 
        updated_params.update({'new_password': ''})
 

	
 
        assert params == updated_params
 

	
 
    def test_delete(self):
 
        self.log_user()
 
        username = 'newtestuserdeleteme'
 

	
 
        fixture.create_user(name=username)
 

	
 
        new_user = Session().query(User) \
 
            .filter(User.username == username).one()
 
        response = self.app.post(url('user', id=new_user.user_id),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 

	
 
        self.checkSessionFlash(response, 'Successfully deleted user')
 

	
 
    def test_delete_repo_err(self):
 
        self.log_user()
 
        username = 'repoerr'
 
        reponame = u'repoerr_fail'
 

	
 
        fixture.create_user(name=username)
 
        fixture.create_repo(name=reponame, cur_user=username)
 

	
 
        new_user = Session().query(User) \
 
            .filter(User.username == username).one()
 
        response = self.app.post(url('user', id=new_user.user_id),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'User "%s" still '
 
                               'owns 1 repositories and cannot be removed. '
 
                               'Switch owners or remove those repositories: '
 
                               '%s' % (username, reponame))
 

	
 
        response = self.app.post(url('delete_repo', repo_name=reponame),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'Deleted repository %s' % reponame)
 

	
 
        response = self.app.post(url('user', id=new_user.user_id),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'Successfully deleted user')
 

	
 
    def test_delete_repo_group_err(self):
 
        self.log_user()
 
        username = 'repogrouperr'
 
        groupname = u'repogroup_fail'
 

	
 
        fixture.create_user(name=username)
 
        fixture.create_repo_group(name=groupname, cur_user=username)
 

	
 
        new_user = Session().query(User) \
 
            .filter(User.username == username).one()
 
        response = self.app.post(url('user', id=new_user.user_id),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'User "%s" still '
 
                               'owns 1 repository groups and cannot be removed. '
 
                               'Switch owners or remove those repository groups: '
 
                               '%s' % (username, groupname))
 

	
 
        # Relevant _if_ the user deletion succeeded to make sure we can render groups without owner
 
        # rg = RepoGroup.get_by_group_name(group_name=groupname)
 
        # response = self.app.get(url('repos_groups', id=rg.group_id))
 

	
 
        response = self.app.post(url('delete_repo_group', group_name=groupname),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'Removed repository group %s' % groupname)
 

	
 
        response = self.app.post(url('user', id=new_user.user_id),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'Successfully deleted user')
 

	
 
    def test_delete_user_group_err(self):
 
        self.log_user()
 
        username = 'usergrouperr'
 
        groupname = u'usergroup_fail'
 

	
 
        fixture.create_user(name=username)
 
        ug = fixture.create_user_group(name=groupname, cur_user=username)
 

	
 
        new_user = Session().query(User) \
 
            .filter(User.username == username).one()
 
        response = self.app.post(url('user', id=new_user.user_id),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'User "%s" still '
 
                               'owns 1 user groups and cannot be removed. '
 
                               'Switch owners or remove those user groups: '
 
                               '%s' % (username, groupname))
 

	
 
        # TODO: why do this fail?
 
        #response = self.app.delete(url('delete_users_group', id=groupname))
 
        #self.checkSessionFlash(response, 'Removed user group %s' % groupname)
 

	
 
        fixture.destroy_user_group(ug.users_group_id)
 

	
 
        response = self.app.post(url('user', id=new_user.user_id),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'Successfully deleted user')
 

	
 
    def test_show(self):
 
        response = self.app.get(url('user', id=1))
 

	
 
    def test_edit(self):
 
        self.log_user()
 
        user = User.get_by_username(TEST_USER_ADMIN_LOGIN)
 
        response = self.app.get(url('edit_user', id=user.user_id))
 

	
 
    def test_add_perm_create_repo(self):
 
        self.log_user()
 
        perm_none = Permission.get_by_key('hg.create.none')
 
        perm_create = Permission.get_by_key('hg.create.repository')
 

	
 
        user = UserModel().create_or_update(username='dummy', password='qwe',
 
                                            email='dummy', firstname=u'a',
 
                                            lastname=u'b')
 
        Session().commit()
 
        uid = user.user_id
 

	
 
        try:
 
            #User should have None permission on creation repository
 
            assert UserModel().has_perm(user, perm_none) == False
 
            assert UserModel().has_perm(user, perm_create) == False
 

	
 
            response = self.app.post(url('edit_user_perms', id=uid),
 
                                     params=dict(_method='put',
 
                                                 create_repo_perm=True,
 
                                                 _authentication_token=self.authentication_token()))
 

	
 
            perm_none = Permission.get_by_key('hg.create.none')
 
            perm_create = Permission.get_by_key('hg.create.repository')
 

	
 
            #User should have None permission on creation repository
 
            assert UserModel().has_perm(uid, perm_none) == False
 
            assert UserModel().has_perm(uid, perm_create) == True
 
        finally:
 
            UserModel().delete(uid)
 
            Session().commit()
 

	
 
    def test_revoke_perm_create_repo(self):
 
        self.log_user()
 
        perm_none = Permission.get_by_key('hg.create.none')
 
        perm_create = Permission.get_by_key('hg.create.repository')
 

	
 
        user = UserModel().create_or_update(username='dummy', password='qwe',
 
                                            email='dummy', firstname=u'a',
 
                                            lastname=u'b')
 
        Session().commit()
 
        uid = user.user_id
 

	
 
        try:
 
            #User should have None permission on creation repository
 
            assert UserModel().has_perm(user, perm_none) == False
 
            assert UserModel().has_perm(user, perm_create) == False
 

	
 
            response = self.app.post(url('edit_user_perms', id=uid),
 
                                     params=dict(_method='put', _authentication_token=self.authentication_token()))
 

	
 
            perm_none = Permission.get_by_key('hg.create.none')
 
            perm_create = Permission.get_by_key('hg.create.repository')
 

	
 
            #User should have None permission on creation repository
 
            assert UserModel().has_perm(uid, perm_none) == True
 
            assert UserModel().has_perm(uid, perm_create) == False
 
        finally:
 
            UserModel().delete(uid)
 
            Session().commit()
 

	
 
    def test_add_perm_fork_repo(self):
 
        self.log_user()
 
        perm_none = Permission.get_by_key('hg.fork.none')
 
        perm_fork = Permission.get_by_key('hg.fork.repository')
 

	
 
        user = UserModel().create_or_update(username='dummy', password='qwe',
 
                                            email='dummy', firstname=u'a',
 
                                            lastname=u'b')
 
        Session().commit()
 
        uid = user.user_id
 

	
 
        try:
 
            #User should have None permission on creation repository
 
            assert UserModel().has_perm(user, perm_none) == False
 
            assert UserModel().has_perm(user, perm_fork) == False
 

	
 
            response = self.app.post(url('edit_user_perms', id=uid),
 
                                     params=dict(_method='put',
 
                                                 create_repo_perm=True,
 
                                                 _authentication_token=self.authentication_token()))
 

	
 
            perm_none = Permission.get_by_key('hg.create.none')
 
            perm_create = Permission.get_by_key('hg.create.repository')
 

	
 
            #User should have None permission on creation repository
 
            assert UserModel().has_perm(uid, perm_none) == False
 
            assert UserModel().has_perm(uid, perm_create) == True
 
        finally:
 
            UserModel().delete(uid)
 
            Session().commit()
 

	
 
    def test_revoke_perm_fork_repo(self):
 
        self.log_user()
 
        perm_none = Permission.get_by_key('hg.fork.none')
 
        perm_fork = Permission.get_by_key('hg.fork.repository')
 

	
 
        user = UserModel().create_or_update(username='dummy', password='qwe',
 
                                            email='dummy', firstname=u'a',
 
                                            lastname=u'b')
 
        Session().commit()
 
        uid = user.user_id
 

	
 
        try:
 
            #User should have None permission on creation repository
 
            assert UserModel().has_perm(user, perm_none) == False
 
            assert UserModel().has_perm(user, perm_fork) == False
 

	
 
            response = self.app.post(url('edit_user_perms', id=uid),
 
                                     params=dict(_method='put', _authentication_token=self.authentication_token()))
 

	
 
            perm_none = Permission.get_by_key('hg.create.none')
 
            perm_create = Permission.get_by_key('hg.create.repository')
 

	
 
            #User should have None permission on creation repository
 
            assert UserModel().has_perm(uid, perm_none) == True
 
            assert UserModel().has_perm(uid, perm_create) == False
 
        finally:
 
            UserModel().delete(uid)
 
            Session().commit()
 

	
 
    def test_ips(self):
 
        self.log_user()
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
 
        response = self.app.get(url('edit_user_ips', id=user.user_id))
 
        response.mustcontain('All IP addresses are allowed')
 

	
 
    @parametrize('test_name,ip,ip_range,failure', [
 
        ('127/24', '127.0.0.1/24', '127.0.0.0 - 127.0.0.255', False),
 
        ('10/32', '10.0.0.10/32', '10.0.0.10 - 10.0.0.10', False),
 
        ('0/16', '0.0.0.0/16', '0.0.0.0 - 0.0.255.255', False),
 
        ('0/8', '0.0.0.0/8', '0.0.0.0 - 0.255.255.255', False),
 
        ('127_bad_mask', '127.0.0.1/99', '127.0.0.1 - 127.0.0.1', True),
 
        ('127_bad_ip', 'foobar', 'foobar', True),
 
    ])
 
    def test_add_ip(self, test_name, ip, ip_range, failure):
 
        self.log_user()
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
 
        user_id = user.user_id
 

	
 
        response = self.app.put(url('edit_user_ips', id=user_id),
 
                                params=dict(new_ip=ip, _authentication_token=self.authentication_token()))
 

	
 
        if failure:
 
            self.checkSessionFlash(response, 'Please enter a valid IPv4 or IPv6 address')
 
            response = self.app.get(url('edit_user_ips', id=user_id))
 
            response.mustcontain(no=[ip])
 
            response.mustcontain(no=[ip_range])
 

	
 
        else:
 
            response = self.app.get(url('edit_user_ips', id=user_id))
 
            response.mustcontain(ip)
 
            response.mustcontain(ip_range)
 

	
 
        ## cleanup
 
        for del_ip in UserIpMap.query().filter(UserIpMap.user_id == user_id).all():
 
            Session().delete(del_ip)
 
            Session().commit()
 

	
 
    def test_delete_ip(self):
 
        self.log_user()
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
 
        user_id = user.user_id
 
        ip = '127.0.0.1/32'
 
        ip_range = '127.0.0.1 - 127.0.0.1'
 
        new_ip = UserModel().add_extra_ip(user_id, ip)
 
        Session().commit()
 
        new_ip_id = new_ip.ip_id
 

	
 
        response = self.app.get(url('edit_user_ips', id=user_id))
 
        response.mustcontain(ip)
 
        response.mustcontain(ip_range)
 

	
 
        self.app.post(url('edit_user_ips', id=user_id),
 
                      params=dict(_method='delete', del_ip_id=new_ip_id, _authentication_token=self.authentication_token()))
 

	
 
        response = self.app.get(url('edit_user_ips', id=user_id))
 
        response.mustcontain('All IP addresses are allowed')
 
        response.mustcontain(no=[ip])
 
        response.mustcontain(no=[ip_range])
 

	
 
    def test_api_keys(self):
 
        self.log_user()
 

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

	
 
    @parametrize('desc,lifetime', [
 
        ('forever', -1),
 
        ('5mins', 60*5),
 
        ('30days', 60*60*24*30),
 
    ])
 
    def test_add_api_keys(self, desc, lifetime):
 
        self.log_user()
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
 
        user_id = user.user_id
 

	
 
        response = self.app.post(url('edit_user_api_keys', id=user_id),
 
                 {'description': desc, 'lifetime': lifetime, '_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'API key successfully created')
 
        try:
 
            response = response.follow()
 
            user = User.get(user_id)
 
            for api_key in user.api_keys:
 
                response.mustcontain(api_key)
 
        finally:
 
            for api_key in UserApiKeys.query().filter(UserApiKeys.user_id == user_id).all():
 
                Session().delete(api_key)
 
                Session().commit()
 

	
 
    def test_remove_api_key(self):
 
        self.log_user()
 
        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
 
        user_id = user.user_id
 

	
 
        response = self.app.post(url('edit_user_api_keys', id=user_id),
 
                {'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()
 
        response = self.app.get(url('edit_user_emails', id=user.user_id), status=404)
 

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

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

	
 
    # IP addresses
 
    # Add/delete of IP addresses for the default user is used to maintain
 
    # the global IP whitelist and thus allowed. Only 'edit' is forbidden.
 
    def test_edit_ip_default_user(self):
 
        self.log_user()
 
        user = User.get_default_user()
 
        response = self.app.get(url('edit_user_ips', id=user.user_id), status=404)
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>"""
 
        )
 

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

	
 
        response.mustcontain('''id="chg_20" class="container "''') # why no mergerow for git?
 
        response.mustcontain(
 
            """<input class="changeset_range" """
 
            """id="95f9a91d775b0084b2368ae7779e44931c849c0e" """
 
            """name="95f9a91d775b0084b2368ae7779e44931c849c0e" """
 
            """type="checkbox" value="1" />"""
 
        )
 

	
 
        response.mustcontain(
 
            """<span class="changeset_hash">r613:95f9a91d775b</span>"""
 
        )
 

	
 
        response.mustcontain("""fixing stupid typo in context for mercurial""")
 

	
 
        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, 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], [[0, 2], [[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, 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], [[0, 2], [[0, 0, 4, 0], [1, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 1, 5, 0], [0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 5, 0]], 0, 0], [[1, 5], [[0, 0, 4, 0], [1, 1, 5, 0]], 0, 0], [[1, 5], [[0, 0, 4, 0], [1, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 1, 6, 0], [0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 6, 0]], 0, 0], [[1, 6], [[0, 0, 4, 0], [1, 1, 6, 0]], 0, 0], [[1, 6], [[0, 0, 4, 0], [1, 1, 6, 0], [1, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 6, 0]], 0, 0], [[1, 6], [[0, 0, 4, 0], [1, 0, 4, 0]], 0, 0], [[0, 4], [[0, 1, 7, 0], [0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 7, 0]], 0, 0], [[1, 7], [[0, 0, 4, 0], [1, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 1, 8, 0], [0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 8, 0]], 0, 0], [[1, 8], [[0, 0, 4, 0], [1, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 1, 9, 0], [0, 0, 4, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[1, 9], [[0, 0, 4, 0], [1, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[1, 9], [[0, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[0, 4], [[0, 0, 4, 0], [1, 1, 9, 0]], 0, 0], [[0, 4], [[0, 0, 9, 0], [1, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 1, 10, 0], [0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 1, 11, 0], [0, 0, 9, 0], [1, 2, 10, 0]], 0, 0], [[2, 10], [[0, 0, 9, 0], [1, 1, 11, 0], [2, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[1, 11], [[0, 0, 9, 0], [1, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[1, 11], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[0, 9], [[0, 1, 11, 0], [0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0], [1, 1, 11, 0]], 0, 0], [[1, 11], [[0, 0, 9, 0], [1, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0], [[0, 9], [[0, 0, 9, 0]], 0, 0]];""")
 

	
 
#        response.mustcontain(
 
#            """<div id="changed_total_5e204e7583b9c8e7b93a020bd036564b1e731dae" """
 
#            """style="float:right;" class="changed_total tooltip" """
 
#            """title="Affected number of files, click to show """
 
#            """more details">3</div>"""
 
#        )
 

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

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

	
 
        response.mustcontain(
 
            """<span class="changeset_hash">r515:636ed213f2f1</span>"""
 
        )
 

	
 
    def test_index_hg_with_filenode(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='changelog', action='index',
 
                                    revision='tip', f_path='/vcs/exceptions.py',
 
                                    repo_name=HG_REPO))
 
        #history commits messages
 
        response.mustcontain('Added exceptions module, this time for real')
 
        response.mustcontain('Added not implemented hg backend test case')
 
        response.mustcontain('Added BaseChangeset class')
 
        # Test response...
 

	
 
    def test_index_git_with_filenode(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='changelog', action='index',
 
                                    revision='tip', f_path='/vcs/exceptions.py',
 
                                    repo_name=GIT_REPO))
 
        #history commits messages
 
        response.mustcontain('Added exceptions module, this time for real')
 
        response.mustcontain('Added not implemented hg backend test case')
 
        response.mustcontain('Added BaseChangeset class')
 

	
 
    def test_index_hg_with_filenode_that_is_dirnode(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='changelog', action='index',
 
                                    revision='tip', f_path='/tests',
 
                                    repo_name=HG_REPO))
 
        assert response.status == '302 Found'
 

	
 
    def test_index_git_with_filenode_that_is_dirnode(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='changelog', action='index',
 
                                    revision='tip', f_path='/tests',
 
                                    repo_name=GIT_REPO))
 
        assert response.status == '302 Found'
 

	
 
    def test_index_hg_with_filenode_not_existing(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='changelog', action='index',
 
                                    revision='tip', f_path='/wrong_path',
 
                                    repo_name=HG_REPO))
 
        assert response.status == '302 Found'
 

	
 
    def test_index_git_with_filenode_not_existing(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='changelog', action='index',
 
                                    revision='tip', f_path='/wrong_path',
 
                                    repo_name=GIT_REPO))
 
        assert response.status == '302 Found'
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'
 

	
 
        params = {'text': text, 'f_path': f_path, 'line': line, '_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 (1 inline, 0 general)'''
 
        )
 
        response.mustcontain(
 
            '''<div class="comments-list-chunk" '''
 
            '''data-f_path="vcs/web/simplevcs/views/repository.py" '''
 
            '''data-line_no="n1" data-target-id="vcswebsimplevcsviewsrepositorypy_n1">'''
 
        )
 

	
 
        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_with_mention(self):
 
        self.log_user()
 

	
 
        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
 
        text = u'@%s check CommentOnRevision' % TEST_USER_REGULAR_LOGIN
 

	
 
        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() == 2
 
        users = [x.user.username for x in UserNotification.query().all()]
 

	
 
        # test_regular gets notification by @mention
 
        assert sorted(users) == [TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN]
 

	
 
    def test_delete(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'})
 

	
 
        comments = ChangesetComment.query().all()
 
        assert len(comments) == 1
 
        comment_id = comments[0].comment_id
 

	
 
        self.app.post(url(controller='changeset',
 
                                    action='delete_comment',
 
                                    repo_name=HG_REPO,
 
                                    comment_id=comment_id),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 

	
 
        comments = ChangesetComment.query().all()
 
        assert len(comments) == 0
 

	
 
        response = self.app.get(url(controller='changeset', action='index',
 
                                repo_name=HG_REPO, revision=rev))
 
        response.mustcontain(
 
            '''<div class="comments-number">'''
 
            ''' 0 comments (0 inline, 0 general)'''
 
        )
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',))
 

	
 
        response.mustcontain('%s@%s' % (repo1.repo_name, rev2))
 
        response.mustcontain('%s@%s' % (repo2.repo_name, rev1))
 
        response.mustcontain("""Showing 2 commits""")
 
        response.mustcontain("""1 file changed with 2 insertions and 0 deletions""")
 

	
 
        response.mustcontain(_commit_div(cs1.raw_id, 'commit2'))
 
        response.mustcontain(_commit_div(cs2.raw_id, 'commit3'))
 

	
 
        response.mustcontain("""<a href="/%s/changeset/%s">r1:%s</a>""" % (repo2.repo_name, cs1.raw_id, cs1.short_id))
 
        response.mustcontain("""<a href="/%s/changeset/%s">r2:%s</a>""" % (repo2.repo_name, cs2.raw_id, cs2.short_id))
 
        ## files
 
        response.mustcontain("""<a href="#C--826e8142e6ba">file1</a>""")
 
        #swap
 
        response.mustcontain("""<a class="btn btn-small" href="/%s/compare/branch@%s...branch@%s?other_repo=%s&amp;merge=True"><i class="icon-arrows-cw"></i> Swap</a>""" % (repo2.repo_name, rev1, rev2, repo1.repo_name))
 

	
 
    def test_compare_forks_on_branch_extra_commits_git(self):
 
        self.log_user()
 
        repo1 = fixture.create_repo(u'one-git', repo_type='git',
 
                                    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='git',
 
                parent=None, newfile=True)
 

	
 
        #fork this repo
 
        repo2 = fixture.create_fork(u'one-git', u'one-git-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='git',
 
                parent=cs0)
 

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

	
 
        rev1 = 'master'
 
        rev2 = 'master'
 

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

	
 
        response.mustcontain('%s@%s' % (repo1.repo_name, rev2))
 
        response.mustcontain('%s@%s' % (repo2.repo_name, rev1))
 
        response.mustcontain("""Showing 2 commits""")
 
        response.mustcontain("""1 file changed with 2 insertions and 0 deletions""")
 

	
 
        response.mustcontain(_commit_div(cs1.raw_id, 'commit2'))
 
        response.mustcontain(_commit_div(cs2.raw_id, 'commit3'))
 

	
 
        response.mustcontain("""<a href="/%s/changeset/%s">r1:%s</a>""" % (repo2.repo_name, cs1.raw_id, cs1.short_id))
 
        response.mustcontain("""<a href="/%s/changeset/%s">r2:%s</a>""" % (repo2.repo_name, cs2.raw_id, cs2.short_id))
 
        ## files
 
        response.mustcontain("""<a href="#C--826e8142e6ba">file1</a>""")
 
        #swap
 
        response.mustcontain("""<a class="btn btn-small" href="/%s/compare/branch@%s...branch@%s?other_repo=%s&amp;merge=True"><i class="icon-arrows-cw"></i> Swap</a>""" % (repo2.repo_name, rev1, rev2, repo1.repo_name))
 

	
 
    def test_compare_forks_on_branch_extra_commits_origin_has_incoming_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
 

	
 
        #now commit something to origin repo
 
        cs1_prim = fixture.commit_change(repo1.repo_name, filename='file2',
 
                content='line1file2\n', message='commit2', vcs_type='hg',
 
                parent=cs0, newfile=True)
 

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

	
 
        response.mustcontain('%s@%s' % (repo1.repo_name, rev2))
 
        response.mustcontain('%s@%s' % (repo2.repo_name, rev1))
 
        response.mustcontain("""Showing 2 commits""")
 
        response.mustcontain("""1 file changed with 2 insertions and 0 deletions""")
 

	
 
        response.mustcontain(_commit_div(cs1.raw_id, 'commit2'))
 
        response.mustcontain(_commit_div(cs2.raw_id, 'commit3'))
 

	
 
        response.mustcontain("""<a href="/%s/changeset/%s">r1:%s</a>""" % (repo2.repo_name, cs1.raw_id, cs1.short_id))
 
        response.mustcontain("""<a href="/%s/changeset/%s">r2:%s</a>""" % (repo2.repo_name, cs2.raw_id, cs2.short_id))
 
        ## files
 
        response.mustcontain("""<a href="#C--826e8142e6ba">file1</a>""")
 
        #swap
 
        response.mustcontain("""<a class="btn btn-small" href="/%s/compare/branch@%s...branch@%s?other_repo=%s&amp;merge=True"><i class="icon-arrows-cw"></i> Swap</a>""" % (repo2.repo_name, rev1, rev2, repo1.repo_name))
 

	
 
    def test_compare_forks_on_branch_extra_commits_origin_has_incoming_git(self):
 
        self.log_user()
 

	
 
        repo1 = fixture.create_repo(u'one-git', repo_type='git',
 
                                    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='git',
 
                parent=None, newfile=True)
 

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

	
 
        #now commit something to origin repo
 
        cs1_prim = fixture.commit_change(repo1.repo_name, filename='file2',
 
                content='line1file2\n', message='commit2', vcs_type='git',
 
                parent=cs0, newfile=True)
 

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

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

	
 
        rev1 = 'master'
 
        rev2 = 'master'
 

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

	
 
        response.mustcontain('%s@%s' % (repo1.repo_name, rev2))
 
        response.mustcontain('%s@%s' % (repo2.repo_name, rev1))
 
        response.mustcontain("""Showing 2 commits""")
 
        response.mustcontain("""1 file changed with 2 insertions and 0 deletions""")
 

	
 
        response.mustcontain(_commit_div(cs1.raw_id, 'commit2'))
 
        response.mustcontain(_commit_div(cs2.raw_id, 'commit3'))
 

	
 
        response.mustcontain("""<a href="/%s/changeset/%s">r1:%s</a>""" % (repo2.repo_name, cs1.raw_id, cs1.short_id))
 
        response.mustcontain("""<a href="/%s/changeset/%s">r2:%s</a>""" % (repo2.repo_name, cs2.raw_id, cs2.short_id))
 
        ## files
 
        response.mustcontain("""<a href="#C--826e8142e6ba">file1</a>""")
 
        #swap
 
        response.mustcontain("""<a class="btn btn-small" href="/%s/compare/branch@%s...branch@%s?other_repo=%s&amp;merge=True"><i class="icon-arrows-cw"></i> Swap</a>""" % (repo2.repo_name, rev1, rev2, repo1.repo_name))
 

	
 
    def test_compare_cherry_pick_changesets_from_bottom(self):
 

	
 
#        repo1:
 
#            cs0:
 
#            cs1:
 
#        repo1-fork- in which we will cherry pick bottom changesets
 
#            cs0:
 
#            cs1:
 
#            cs2: x
 
#            cs3: x
 
#            cs4: x
 
#            cs5:
 
        #make repo1, and cs1+cs2
 
        self.log_user()
 

	
 
        repo1 = fixture.create_repo(u'repo1', 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)
 
        cs1 = fixture.commit_change(repo1.repo_name, filename='file1',
 
                content='line1\nline2\n', message='commit2', vcs_type='hg',
 
                parent=cs0)
 
        #fork this repo
 
        repo2 = fixture.create_fork(u'repo1', u'repo1-fork')
 
        self.r2_id = repo2.repo_id
 
        #now make cs3-6
 
        cs2 = fixture.commit_change(repo1.repo_name, filename='file1',
 
                content='line1\nline2\nline3\n', message='commit3',
 
                vcs_type='hg', parent=cs1)
 
        cs3 = fixture.commit_change(repo1.repo_name, filename='file1',
 
                content='line1\nline2\nline3\nline4\n', message='commit4',
 
                vcs_type='hg', parent=cs2)
 
        cs4 = fixture.commit_change(repo1.repo_name, filename='file1',
 
                content='line1\nline2\nline3\nline4\nline5\n',
 
                message='commit5', vcs_type='hg', parent=cs3)
 
        cs5 = fixture.commit_change(repo1.repo_name, filename='file1',
 
                content='line1\nline2\nline3\nline4\nline5\nline6\n',
 
                message='commit6', vcs_type='hg', parent=cs4)
 

	
 
        response = self.app.get(url('compare_url',
 
                                    repo_name=repo2.repo_name,
 
                                    org_ref_type="rev",
 
                                    org_ref_name=cs1.short_id,  # parent of cs2, in repo2
 
                                    other_repo=repo1.repo_name,
 
                                    other_ref_type="rev",
 
                                    other_ref_name=cs4.short_id,
 
                                    merge='True',
 
                                    ))
 
        response.mustcontain('%s@%s' % (repo2.repo_name, cs1.short_id))
 
        response.mustcontain('%s@%s' % (repo1.repo_name, cs4.short_id))
 
        response.mustcontain("""Showing 3 commits""")
 
        response.mustcontain("""1 file changed with 3 insertions and 0 deletions""")
 

	
 
        response.mustcontain(_commit_div(cs2.raw_id, 'commit3'))
 
        response.mustcontain(_commit_div(cs3.raw_id, 'commit4'))
 
        response.mustcontain(_commit_div(cs4.raw_id, 'commit5'))
 

	
 
        response.mustcontain("""<a href="/%s/changeset/%s">r2:%s</a>""" % (repo1.repo_name, cs2.raw_id, cs2.short_id))
 
        response.mustcontain("""<a href="/%s/changeset/%s">r3:%s</a>""" % (repo1.repo_name, cs3.raw_id, cs3.short_id))
 
        response.mustcontain("""<a href="/%s/changeset/%s">r4:%s</a>""" % (repo1.repo_name, cs4.raw_id, cs4.short_id))
 
        ## files
 
        response.mustcontain("""#C--826e8142e6ba">file1</a>""")
 

	
 
    def test_compare_cherry_pick_changesets_from_top(self):
 
#        repo1:
 
#            cs0:
 
#            cs1:
 
#        repo1-fork- in which we will cherry pick bottom changesets
 
#            cs0:
 
#            cs1:
 
#            cs2:
 
#            cs3: x
 
#            cs4: x
 
#            cs5: x
 
#
 
        #make repo1, and cs1+cs2
 
        self.log_user()
 
        repo1 = fixture.create_repo(u'repo1', 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)
 
        cs1 = fixture.commit_change(repo1.repo_name, filename='file1',
 
                content='line1\nline2\n', message='commit2', vcs_type='hg',
 
                parent=cs0)
 
        #fork this repo
 
        repo2 = fixture.create_fork(u'repo1', u'repo1-fork')
 
        self.r2_id = repo2.repo_id
 
        #now make cs3-6
 
        cs2 = fixture.commit_change(repo1.repo_name, filename='file1',
 
                content='line1\nline2\nline3\n', message='commit3',
 
                vcs_type='hg', parent=cs1)
 
        cs3 = fixture.commit_change(repo1.repo_name, filename='file1',
 
                content='line1\nline2\nline3\nline4\n', message='commit4',
 
                vcs_type='hg', parent=cs2)
 
        cs4 = fixture.commit_change(repo1.repo_name, filename='file1',
 
                content='line1\nline2\nline3\nline4\nline5\n',
 
                message='commit5', vcs_type='hg', parent=cs3)
 
        cs5 = fixture.commit_change(repo1.repo_name, filename='file1',
 
                content='line1\nline2\nline3\nline4\nline5\nline6\n',
 
                message='commit6', vcs_type='hg', parent=cs4)
 

	
 
        response = self.app.get(url('compare_url',
 
                                    repo_name=repo1.repo_name,
 
                                    org_ref_type="rev",
 
                                    org_ref_name=cs2.short_id, # parent of cs3, not in repo2
 
                                    other_ref_type="rev",
 
                                    other_ref_name=cs5.short_id,
 
                                    merge='1',))
 

	
 
        response.mustcontain('%s@%s' % (repo1.repo_name, cs2.short_id))
 
        response.mustcontain('%s@%s' % (repo1.repo_name, cs5.short_id))
 
        response.mustcontain("""Showing 3 commits""")
 
        response.mustcontain("""1 file changed with 3 insertions and 0 deletions""")
 

	
 
        response.mustcontain(_commit_div(cs3.raw_id, 'commit4'))
 
        response.mustcontain(_commit_div(cs4.raw_id, 'commit5'))
 
        response.mustcontain(_commit_div(cs5.raw_id, 'commit6'))
 

	
 
        response.mustcontain("""<a href="/%s/changeset/%s">r3:%s</a>""" % (repo1.repo_name, cs3.raw_id, cs3.short_id))
 
        response.mustcontain("""<a href="/%s/changeset/%s">r4:%s</a>""" % (repo1.repo_name, cs4.raw_id, cs4.short_id))
 
        response.mustcontain("""<a href="/%s/changeset/%s">r5:%s</a>""" % (repo1.repo_name, cs5.raw_id, cs5.short_id))
 
        ## files
 
        response.mustcontain("""#C--826e8142e6ba">file1</a>""")
 

	
 
    def test_compare_cherry_pick_changeset_mixed_branches(self):
 
        #TODO: write this
 
        assert 1
 

	
 
    def test_compare_remote_branches_hg(self):
 
        self.log_user()
 

	
 
        repo2 = fixture.create_fork(HG_REPO, HG_FORK)
 
        self.r2_id = repo2.repo_id
 
        rev1 = '56349e29c2af'
 
        rev2 = '7d4bc8ec6be5'
 

	
 
        response = self.app.get(url('compare_url',
 
                                    repo_name=HG_REPO,
 
                                    org_ref_type="rev",
 
                                    org_ref_name=rev1,
 
                                    other_ref_type="rev",
 
                                    other_ref_name=rev2,
 
                                    other_repo=HG_FORK,
 
                                    merge='1',))
 

	
 
        response.mustcontain('%s@%s' % (HG_REPO, rev1))
 
        response.mustcontain('%s@%s' % (HG_FORK, rev2))
 
        ## outgoing changesets between those revisions
 

	
 
        response.mustcontain("""<a href="/%s/changeset/2dda4e345facb0ccff1a191052dd1606dba6781d">r4:2dda4e345fac</a>""" % (HG_FORK))
 
        response.mustcontain("""<a href="/%s/changeset/6fff84722075f1607a30f436523403845f84cd9e">r5:6fff84722075</a>""" % (HG_FORK))
 
        response.mustcontain("""<a href="/%s/changeset/7d4bc8ec6be56c0f10425afb40b6fc315a4c25e7">r6:%s</a>""" % (HG_FORK, rev2))
 

	
 
        ## files
 
        response.mustcontain("""<a href="#C--9c390eb52cd6">vcs/backends/hg.py</a>""")
 
        response.mustcontain("""<a href="#C--41b41c1f2796">vcs/backends/__init__.py</a>""")
 
        response.mustcontain("""<a href="#C--2f574d260608">vcs/backends/base.py</a>""")
 

	
 
    def test_compare_remote_branches_git(self):
 
        self.log_user()
 

	
 
        repo2 = fixture.create_fork(GIT_REPO, GIT_FORK)
 
        self.r2_id = repo2.repo_id
 
        rev1 = '102607b09cdd60e2793929c4f90478be29f85a17'
 
        rev2 = 'd7e0d30fbcae12c90680eb095a4f5f02505ce501'
 

	
 
        response = self.app.get(url('compare_url',
 
                                    repo_name=GIT_REPO,
 
                                    org_ref_type="rev",
 
                                    org_ref_name=rev1,
 
                                    other_ref_type="rev",
 
                                    other_ref_name=rev2,
 
                                    other_repo=GIT_FORK,
 
                                    merge='1',))
 

	
 
        response.mustcontain('%s@%s' % (GIT_REPO, rev1))
 
        response.mustcontain('%s@%s' % (GIT_FORK, rev2))
 
        ## outgoing changesets between those revisions
 

	
 
        response.mustcontain("""<a href="/%s/changeset/49d3fd156b6f7db46313fac355dca1a0b94a0017">r4:49d3fd156b6f</a>""" % (GIT_FORK))
 
        response.mustcontain("""<a href="/%s/changeset/2d1028c054665b962fa3d307adfc923ddd528038">r5:2d1028c05466</a>""" % (GIT_FORK))
 
        response.mustcontain("""<a href="/%s/changeset/d7e0d30fbcae12c90680eb095a4f5f02505ce501">r6:%s</a>""" % (GIT_FORK, rev2[:12]))
 

	
 
        ## files
 
        response.mustcontain("""<a href="#C--9c390eb52cd6">vcs/backends/hg.py</a>""")
 
        response.mustcontain("""<a href="#C--41b41c1f2796">vcs/backends/__init__.py</a>""")
 
        response.mustcontain("""<a href="#C--2f574d260608">vcs/backends/base.py</a>""")
 

	
 
    def test_org_repo_new_commits_after_forking_simple_diff_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
 
        r1_name = repo1.repo_name
 

	
 
        cs0 = fixture.commit_change(repo=r1_name, filename='file1',
 
                content='line1', message='commit1', vcs_type='hg', newfile=True)
 
        Session().commit()
 
        assert repo1.scm_instance.revisions == [cs0.raw_id]
 
        #fork the repo1
 
        repo2 = fixture.create_repo(u'one-fork', repo_type='hg',
 
                                    repo_description='diff-test',
 
                                    cur_user=TEST_USER_ADMIN_LOGIN,
 
                                    clone_uri=repo1.repo_full_path,
 
                                    fork_of='one')
 
        Session().commit()
 
        assert repo2.scm_instance.revisions == [cs0.raw_id]
 
        self.r2_id = repo2.repo_id
 
        r2_name = repo2.repo_name
 

	
 

	
 
        cs1 = fixture.commit_change(repo=r2_name, filename='file1-fork',
 
                content='file1-line1-from-fork', message='commit1-fork',
 
                vcs_type='hg', parent=repo2.scm_instance[-1], newfile=True)
 

	
 
        cs2 = fixture.commit_change(repo=r2_name, filename='file2-fork',
 
                content='file2-line1-from-fork', message='commit2-fork',
 
                vcs_type='hg', parent=cs1, newfile=True)
 

	
 
        cs3 = fixture.commit_change(repo=r2_name, filename='file3-fork',
 
                content='file3-line1-from-fork', message='commit3-fork',
 
                vcs_type='hg', parent=cs2, newfile=True)
 
        #compare !
 
        rev1 = 'default'
 
        rev2 = 'default'
 

	
 
        response = self.app.get(url('compare_url',
 
                                    repo_name=r2_name,
 
                                    org_ref_type="branch",
 
                                    org_ref_name=rev1,
 
                                    other_ref_type="branch",
 
                                    other_ref_name=rev2,
 
                                    other_repo=r1_name,
 
                                    merge='1',))
 

	
 
        response.mustcontain('%s@%s' % (r2_name, rev1))
 
        response.mustcontain('%s@%s' % (r1_name, rev2))
 
        response.mustcontain('No files')
 
        response.mustcontain('No changesets')
 

	
 
        cs0 = fixture.commit_change(repo=r1_name, filename='file2',
 
                content='line1-added-after-fork', message='commit2-parent',
 
                vcs_type='hg', parent=None, newfile=True)
 

	
 
        #compare !
 
        rev1 = 'default'
 
        rev2 = 'default'
 
        response = self.app.get(url('compare_url',
 
                                    repo_name=r2_name,
 
                                    org_ref_type="branch",
 
                                    org_ref_name=rev1,
 
                                    other_ref_type="branch",
 
                                    other_ref_name=rev2,
 
                                    other_repo=r1_name,
 
                                    merge='1',
 
                                    ))
 

	
 
        response.mustcontain('%s@%s' % (r2_name, rev1))
 
        response.mustcontain('%s@%s' % (r1_name, rev2))
 

	
 
        response.mustcontain("""commit2-parent""")
 
        response.mustcontain("""1 file changed with 1 insertions and 0 deletions""")
 
        response.mustcontain("""line1-added-after-fork""")
 

	
 
    def test_org_repo_new_commits_after_forking_simple_diff_git(self):
 
        self.log_user()
 

	
 
        repo1 = fixture.create_repo(u'one-git', repo_type='git',
 
                                    repo_description='diff-test',
 
                                    cur_user=TEST_USER_ADMIN_LOGIN)
 

	
 
        self.r1_id = repo1.repo_id
 
        r1_name = repo1.repo_name
 

	
 
        cs0 = fixture.commit_change(repo=r1_name, filename='file1',
 
                content='line1', message='commit1', vcs_type='git',
 
                newfile=True)
 
        Session().commit()
 
        assert repo1.scm_instance.revisions == [cs0.raw_id]
 
        #fork the repo1
 
        repo2 = fixture.create_repo(u'one-git-fork', repo_type='git',
 
                                    repo_description='diff-test',
 
                                    cur_user=TEST_USER_ADMIN_LOGIN,
 
                                    clone_uri=repo1.repo_full_path,
 
                                    fork_of='one-git')
 
        Session().commit()
 
        assert repo2.scm_instance.revisions == [cs0.raw_id]
 
        self.r2_id = repo2.repo_id
 
        r2_name = repo2.repo_name
 

	
 

	
 
        cs1 = fixture.commit_change(repo=r2_name, filename='file1-fork',
 
                content='file1-line1-from-fork', message='commit1-fork',
 
                vcs_type='git', parent=repo2.scm_instance[-1], newfile=True)
 

	
 
        cs2 = fixture.commit_change(repo=r2_name, filename='file2-fork',
 
                content='file2-line1-from-fork', message='commit2-fork',
 
                vcs_type='git', parent=cs1, newfile=True)
 

	
 
        cs3 = fixture.commit_change(repo=r2_name, filename='file3-fork',
 
                content='file3-line1-from-fork', message='commit3-fork',
 
                vcs_type='git', parent=cs2, newfile=True)
 
        #compare !
 
        rev1 = 'master'
 
        rev2 = 'master'
 

	
 
        response = self.app.get(url('compare_url',
 
                                    repo_name=r2_name,
 
                                    org_ref_type="branch",
 
                                    org_ref_name=rev1,
 
                                    other_ref_type="branch",
 
                                    other_ref_name=rev2,
 
                                    other_repo=r1_name,
 
                                    merge='1',))
 

	
 
        response.mustcontain('%s@%s' % (r2_name, rev1))
 
        response.mustcontain('%s@%s' % (r1_name, rev2))
 
        response.mustcontain('No files')
 
        response.mustcontain('No changesets')
 

	
 
        cs0 = fixture.commit_change(repo=r1_name, filename='file2',
 
                content='line1-added-after-fork', message='commit2-parent',
 
                vcs_type='git', parent=None, newfile=True)
 

	
 
        #compare !
 
        rev1 = 'master'
 
        rev2 = 'master'
 
        response = self.app.get(url('compare_url',
 
                                    repo_name=r2_name,
 
                                    org_ref_type="branch",
 
                                    org_ref_name=rev1,
 
                                    other_ref_type="branch",
 
                                    other_ref_name=rev2,
 
                                    other_repo=r1_name,
 
                                    merge='1',
 
                                    ))
 

	
 
        response.mustcontain('%s@%s' % (r2_name, rev1))
 
        response.mustcontain('%s@%s' % (r1_name, rev2))
 

	
 
        response.mustcontain("""commit2-parent""")
 
        response.mustcontain("""1 file changed with 1 insertions and 0 deletions""")
 
        response.mustcontain("""line1-added-after-fork""")
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>''')
 
        response.mustcontain('''<div class="node">
 
                             <i class="icon-diff-M"></i>
 
                             <a href="#C--ebb592c595c0">vcs/utils/__init__.py</a>''')
 
        response.mustcontain('''<div class="node">
 
                             <i class="icon-diff-M"></i>
 
                             <a href="#C--7abc741b5052">vcs/utils/annotate.py</a>''')
 
        response.mustcontain('''<div class="node">
 
                             <i class="icon-diff-M"></i>
 
                             <a href="#C--2ef0ef106c56">vcs/utils/diffs.py</a>''')
 
        response.mustcontain('''<div class="node">
 
                             <i class="icon-diff-M"></i>
 
                             <a href="#C--3150cb87d4b7">vcs/utils/lazy.py</a>''')
 

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

	
 
        ## outgoing changesets between tags
 
        response.mustcontain('''<a href="/%s/changeset/794bbdd31545c199f74912709ea350dedcd189a2">r113:794bbdd31545</a>''' % GIT_REPO)
 
        response.mustcontain('''<a href="/%s/changeset/e36d8c5025329bdd4212bd53d4ed8a70ff44985f">r115:e36d8c502532</a>''' % GIT_REPO)
 
        response.mustcontain('''<a href="/%s/changeset/5c9ff4f6d7508db0e72b1d2991c357d0d8e07af2">r116:5c9ff4f6d750</a>''' % GIT_REPO)
 
        response.mustcontain('''<a href="/%s/changeset/b7187fa2b8c1d773ec35e9dee12f01f74808c879">r117:b7187fa2b8c1</a>''' % GIT_REPO)
 
        response.mustcontain('''<a href="/%s/changeset/5f3b74262014a8de2dc7dade1152de9fd0c8efef">r118:5f3b74262014</a>''' % GIT_REPO)
 
        response.mustcontain('''<a href="/%s/changeset/17438a11f72b93f56d0e08e7d1fa79a378578a82">r119:17438a11f72b</a>''' % GIT_REPO)
 
        response.mustcontain('''<a href="/%s/changeset/5a3a8fb005554692b16e21dee62bf02667d8dc3e">r120:5a3a8fb00555</a>''' % GIT_REPO)
 

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

	
 
        #files
 
        response.mustcontain('''<a href="#C--1c5cf9e91c12">docs/api/utils/index.rst</a>''')
 
        response.mustcontain('''<a href="#C--e3305437df55">test_and_report.sh</a>''')
 
        response.mustcontain('''<a href="#C--c8e92ef85cd1">.hgignore</a>''')
 
        response.mustcontain('''<a href="#C--6e08b694d687">.hgtags</a>''')
 
        response.mustcontain('''<a href="#C--2c14b00f3393">docs/api/index.rst</a>''')
 
        response.mustcontain('''<a href="#C--430ccbc82bdf">vcs/__init__.py</a>''')
 
        response.mustcontain('''<a href="#C--9c390eb52cd6">vcs/backends/hg.py</a>''')
 
        response.mustcontain('''<a href="#C--ebb592c595c0">vcs/utils/__init__.py</a>''')
 
        response.mustcontain('''<a href="#C--7abc741b5052">vcs/utils/annotate.py</a>''')
 
        response.mustcontain('''<a href="#C--2ef0ef106c56">vcs/utils/diffs.py</a>''')
 
        response.mustcontain('''<a href="#C--3150cb87d4b7">vcs/utils/lazy.py</a>''')
 

	
 
    def test_index_branch_hg(self):
 
        self.log_user()
 
        response = self.app.get(url('compare_url',
 
                                    repo_name=HG_REPO,
 
                                    org_ref_type="branch",
 
                                    org_ref_name='default',
 
                                    other_ref_type="branch",
 
                                    other_ref_name='default',
 
                                    ))
 

	
 
        response.mustcontain('%s@default' % (HG_REPO))
 
        response.mustcontain('%s@default' % (HG_REPO))
 
        # branch are equal
 
        response.mustcontain('<span class="empty_data">No files</span>')
 
        response.mustcontain('<span class="empty_data">No changesets</span>')
 

	
 
    def test_index_branch_git(self):
 
        self.log_user()
 
        response = self.app.get(url('compare_url',
 
                                    repo_name=GIT_REPO,
 
                                    org_ref_type="branch",
 
                                    org_ref_name='master',
 
                                    other_ref_type="branch",
 
                                    other_ref_name='master',
 
                                    ))
 

	
 
        response.mustcontain('%s@master' % (GIT_REPO))
 
        response.mustcontain('%s@master' % (GIT_REPO))
 
        # branch are equal
 
        response.mustcontain('<span class="empty_data">No files</span>')
 
        response.mustcontain('<span class="empty_data">No changesets</span>')
 

	
 
    def test_compare_revisions_hg(self):
 
        self.log_user()
 
        rev1 = 'b986218ba1c9'
 
        rev2 = '3d8f361e72ab'
 

	
 
        response = self.app.get(url('compare_url',
 
                                    repo_name=HG_REPO,
 
                                    org_ref_type="rev",
 
                                    org_ref_name=rev1,
 
                                    other_ref_type="rev",
 
                                    other_ref_name=rev2,
 
                                    ))
 
        response.mustcontain('%s@%s' % (HG_REPO, rev1))
 
        response.mustcontain('%s@%s' % (HG_REPO, rev2))
 

	
 
        ## outgoing changesets between those revisions
 
        response.mustcontain("""<a href="/%s/changeset/3d8f361e72ab303da48d799ff1ac40d5ac37c67e">r1:%s</a>""" % (HG_REPO, rev2))
 

	
 
        response.mustcontain('1 file changed with 7 insertions and 0 deletions')
 
        ## files
 
        response.mustcontain("""<a href="#C--c8e92ef85cd1">.hgignore</a>""")
 

	
 
    def test_compare_revisions_git(self):
 
        self.log_user()
 
        rev1 = 'c1214f7e79e02fc37156ff215cd71275450cffc3'
 
        rev2 = '38b5fe81f109cb111f549bfe9bb6b267e10bc557'
 

	
 
        response = self.app.get(url('compare_url',
 
                                    repo_name=GIT_REPO,
 
                                    org_ref_type="rev",
 
                                    org_ref_name=rev1,
 
                                    other_ref_type="rev",
 
                                    other_ref_name=rev2,
 
                                    ))
 
        response.mustcontain('%s@%s' % (GIT_REPO, rev1))
 
        response.mustcontain('%s@%s' % (GIT_REPO, rev2))
 

	
 
        ## outgoing changesets between those revisions
 
        response.mustcontain("""<a href="/%s/changeset/38b5fe81f109cb111f549bfe9bb6b267e10bc557">r1:%s</a>""" % (GIT_REPO, rev2[:12]))
 
        response.mustcontain('1 file changed with 7 insertions and 0 deletions')
 

	
 
        ## files
 
        response.mustcontain("""<a href="#C--c8e92ef85cd1">.hgignore</a>""")
 

	
 
    def test_compare_revisions_hg_as_form(self):
 
        self.log_user()
 
        rev1 = 'b986218ba1c9'
 
        rev2 = '3d8f361e72ab'
 

	
 
        response = self.app.get(url('compare_url',
 
                                    repo_name=HG_REPO,
 
                                    org_ref_type="rev",
 
                                    org_ref_name=rev1,
 
                                    other_ref_type="rev",
 
                                    other_ref_name=rev2,
 
                                    as_form=True,
 
                                    ),
 
                                extra_environ={'HTTP_X_PARTIAL_XHR': '1'},)
 

	
 
        ## outgoing changesets between those revisions
 
        response.mustcontain("""<a href="/%s/changeset/3d8f361e72ab303da48d799ff1ac40d5ac37c67e">r1:%s</a>""" % (HG_REPO, rev2))
 

	
 
        response.mustcontain('Common ancestor')
 
        response.mustcontain("""<a href="/%s/changeset/b986218ba1c9b0d6a259fac9b050b1724ed8e545">%s</a>""" % (HG_REPO, rev1))
 

	
 
    def test_compare_revisions_git_as_form(self):
 
        self.log_user()
 
        rev1 = 'c1214f7e79e02fc37156ff215cd71275450cffc3'
 
        rev2 = '38b5fe81f109cb111f549bfe9bb6b267e10bc557'
 

	
 
        response = self.app.get(url('compare_url',
 
                                    repo_name=GIT_REPO,
 
                                    org_ref_type="rev",
 
                                    org_ref_name=rev1,
 
                                    other_ref_type="rev",
 
                                    other_ref_name=rev2,
 
                                    as_form=True,
 
                                    ),
 
                                extra_environ={'HTTP_X_PARTIAL_XHR': '1'},)
 
        ## outgoing changesets between those revisions
 
        response.mustcontain("""<a href="/%s/changeset/38b5fe81f109cb111f549bfe9bb6b267e10bc557">r1:%s</a>""" % (GIT_REPO, rev2[:12]))
 

	
 
        response.mustcontain('Common ancestor')
 
        response.mustcontain("""<a href="/%s/changeset/c1214f7e79e02fc37156ff215cd71275450cffc3">%s</a>""" % (GIT_REPO, rev1[:12]))
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>""")
 

	
 
    def test_index_paging(self):
 
        self.log_user()
 

	
 
        for r in [(73, 'a066b25d5df7016b45a41b7e2a78c33b57adc235'),
 
                  (92, 'cc66b61b8455b264a7a8a2d8ddc80fcfc58c221e'),
 
                  (109, '75feb4c33e81186c87eac740cee2447330288412'),
 
                  (1, '3d8f361e72ab303da48d799ff1ac40d5ac37c67e'),
 
                  (0, 'b986218ba1c9b0d6a259fac9b050b1724ed8e545')]:
 

	
 
            response = self.app.get(url(controller='files', action='index',
 
                                    repo_name=HG_REPO,
 
                                    revision=r[1],
 
                                    f_path='/'))
 

	
 
            response.mustcontain("""@ r%s:%s""" % (r[0], r[1][:12]))
 

	
 
    def test_file_source(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='index',
 
                                    repo_name=HG_REPO,
 
                                    revision='8911406ad776fdd3d0b9932a2e89677e57405a48',
 
                                    f_path='vcs/nodes.py'))
 

	
 
        response.mustcontain("""<div class="commit">Partially implemented <a class="issue-tracker-link" href="https://issues.example.com/vcs_test_hg/issue/16">#16</a>. filecontent/commit message/author/node name are safe_unicode now.
 
In addition some other __str__ are unicode as well
 
Added test for unicode
 
Improved test to clone into uniq repository.
 
removed extra unicode conversion in diff.</div>
 
""")
 

	
 
        response.mustcontain("""<option selected="selected" value="8911406ad776fdd3d0b9932a2e89677e57405a48">default at 8911406ad776</option>""")
 

	
 
    def test_file_source_history(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='history',
 
                                    repo_name=HG_REPO,
 
                                    revision='tip',
 
                                    f_path='vcs/nodes.py'),
 
                                extra_environ={'HTTP_X_PARTIAL_XHR': '1'},)
 
        assert response.body == HG_NODE_HISTORY
 

	
 
    def test_file_source_history_git(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='history',
 
                                    repo_name=GIT_REPO,
 
                                    revision='master',
 
                                    f_path='vcs/nodes.py'),
 
                                extra_environ={'HTTP_X_PARTIAL_XHR': '1'},)
 
        assert response.body == GIT_NODE_HISTORY
 

	
 
    def test_file_annotation(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='index',
 
                                    repo_name=HG_REPO,
 
                                    revision='tip',
 
                                    f_path='vcs/nodes.py',
 
                                    annotate=True))
 

	
 
        response.mustcontain("""r356:25213a5fbb04""")
 

	
 
    def test_file_annotation_git(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='index',
 
                                    repo_name=GIT_REPO,
 
                                    revision='master',
 
                                    f_path='vcs/nodes.py',
 
                                    annotate=True))
 
        response.mustcontain("""r345:c994f0de03b2""")
 

	
 
    def test_file_annotation_history(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='history',
 
                                    repo_name=HG_REPO,
 
                                    revision='tip',
 
                                    f_path='vcs/nodes.py',
 
                                    annotate=True),
 
                                extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
 

	
 
        assert response.body == HG_NODE_HISTORY
 

	
 
    def test_file_annotation_history_git(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='history',
 
                                    repo_name=GIT_REPO,
 
                                    revision='master',
 
                                    f_path='vcs/nodes.py',
 
                                    annotate=True),
 
                                extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
 

	
 
        assert response.body == GIT_NODE_HISTORY
 

	
 
    def test_file_authors(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='authors',
 
                                    repo_name=HG_REPO,
 
                                    revision='tip',
 
                                    f_path='vcs/nodes.py',
 
                                    annotate=True))
 
        response.mustcontain('Marcin Kuzminski')
 
        response.mustcontain('Lukasz Balcerzak')
 

	
 
    def test_file_authors_git(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='authors',
 
                                    repo_name=GIT_REPO,
 
                                    revision='master',
 
                                    f_path='vcs/nodes.py',
 
                                    annotate=True))
 
        response.mustcontain('Marcin Kuzminski')
 
        response.mustcontain('Lukasz Balcerzak')
 

	
 
    def test_archival(self):
 
        self.log_user()
 
        _set_downloads(HG_REPO, set_to=True)
 
        for arch_ext, info in ARCHIVE_SPECS.items():
 
            short = '27cd5cce30c9%s' % arch_ext
 
            fname = '27cd5cce30c96924232dffcd24178a07ffeb5dfc%s' % arch_ext
 
            filename = '%s-%s' % (HG_REPO, short)
 
            response = self.app.get(url(controller='files',
 
                                        action='archivefile',
 
                                        repo_name=HG_REPO,
 
                                        fname=fname))
 

	
 
            assert response.status == '200 OK'
 
            heads = [
 
                ('Pragma', 'no-cache'),
 
                ('Cache-Control', 'no-cache'),
 
                ('Content-Disposition', 'attachment; filename=%s' % filename),
 
                ('Content-Type', '%s; charset=utf-8' % info[0]),
 
            ]
 
            assert response.response._headers.items() == heads
 

	
 
    def test_archival_wrong_ext(self):
 
        self.log_user()
 
        _set_downloads(HG_REPO, set_to=True)
 
        for arch_ext in ['tar', 'rar', 'x', '..ax', '.zipz']:
 
            fname = '27cd5cce30c96924232dffcd24178a07ffeb5dfc%s' % arch_ext
 

	
 
            response = self.app.get(url(controller='files',
 
                                        action='archivefile',
 
                                        repo_name=HG_REPO,
 
                                        fname=fname))
 
            response.mustcontain('Unknown archive type')
 

	
 
    def test_archival_wrong_revision(self):
 
        self.log_user()
 
        _set_downloads(HG_REPO, set_to=True)
 
        for rev in ['00x000000', 'tar', 'wrong', '@##$@$42413232', '232dffcd']:
 
            fname = '%s.zip' % rev
 

	
 
            response = self.app.get(url(controller='files',
 
                                        action='archivefile',
 
                                        repo_name=HG_REPO,
 
                                        fname=fname))
 
            response.mustcontain('Unknown revision')
 

	
 
    #==========================================================================
 
    # RAW FILE
 
    #==========================================================================
 
    def test_raw_file_ok(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='rawfile',
 
                                    repo_name=HG_REPO,
 
                                    revision='27cd5cce30c96924232dffcd24178a07ffeb5dfc',
 
                                    f_path='vcs/nodes.py'))
 

	
 
        assert response.content_disposition == "attachment; filename=nodes.py"
 
        assert response.content_type == mimetypes.guess_type("nodes.py")[0]
 

	
 
    def test_raw_file_wrong_cs(self):
 
        self.log_user()
 
        rev = u'ERRORce30c96924232dffcd24178a07ffeb5dfc'
 
        f_path = 'vcs/nodes.py'
 

	
 
        response = self.app.get(url(controller='files', action='rawfile',
 
                                    repo_name=HG_REPO,
 
                                    revision=rev,
 
                                    f_path=f_path), status=404)
 

	
 
        msg = """Such revision does not exist for this repository"""
 
        response.mustcontain(msg)
 

	
 
    def test_raw_file_wrong_f_path(self):
 
        self.log_user()
 
        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
 
        f_path = 'vcs/ERRORnodes.py'
 
        response = self.app.get(url(controller='files', action='rawfile',
 
                                    repo_name=HG_REPO,
 
                                    revision=rev,
 
                                    f_path=f_path), status=404)
 

	
 
        msg = "There is no file nor directory at the given path: &#39;%s&#39; at revision %s" % (f_path, rev[:12])
 
        response.mustcontain(msg)
 

	
 
    #==========================================================================
 
    # RAW RESPONSE - PLAIN
 
    #==========================================================================
 
    def test_raw_ok(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='raw',
 
                                    repo_name=HG_REPO,
 
                                    revision='27cd5cce30c96924232dffcd24178a07ffeb5dfc',
 
                                    f_path='vcs/nodes.py'))
 

	
 
        assert response.content_type == "text/plain"
 

	
 
    def test_raw_wrong_cs(self):
 
        self.log_user()
 
        rev = u'ERRORcce30c96924232dffcd24178a07ffeb5dfc'
 
        f_path = 'vcs/nodes.py'
 

	
 
        response = self.app.get(url(controller='files', action='raw',
 
                                    repo_name=HG_REPO,
 
                                    revision=rev,
 
                                    f_path=f_path), status=404)
 

	
 
        msg = """Such revision does not exist for this repository"""
 
        response.mustcontain(msg)
 

	
 
    def test_raw_wrong_f_path(self):
 
        self.log_user()
 
        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
 
        f_path = 'vcs/ERRORnodes.py'
 
        response = self.app.get(url(controller='files', action='raw',
 
                                    repo_name=HG_REPO,
 
                                    revision=rev,
 
                                    f_path=f_path), status=404)
 
        msg = "There is no file nor directory at the given path: &#39;%s&#39; at revision %s" % (f_path, rev[:12])
 
        response.mustcontain(msg)
 

	
 
    def test_ajaxed_files_list(self):
 
        self.log_user()
 
        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
 
        response = self.app.get(
 
            url('files_nodelist_home', repo_name=HG_REPO, f_path='/',
 
                revision=rev),
 
            extra_environ={'HTTP_X_PARTIAL_XHR': '1'},
 
        )
 
        response.mustcontain("vcs/web/simplevcs/views/repository.py")
 

	
 
    # Hg - ADD FILE
 
    def test_add_file_view_hg(self):
 
        self.log_user()
 
        response = self.app.get(url('files_add_home',
 
                                      repo_name=HG_REPO,
 
                                      revision='tip', f_path='/'))
 

	
 
    def test_add_file_into_hg_missing_content(self):
 
        self.log_user()
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=HG_REPO,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': '',
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 

	
 
        self.checkSessionFlash(response, 'No content')
 

	
 
    def test_add_file_into_hg_missing_filename(self):
 
        self.log_user()
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=HG_REPO,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "foo",
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 

	
 
        self.checkSessionFlash(response, 'No filename')
 

	
 
    @parametrize('location,filename', [
 
        ('/abs', 'foo'),
 
        ('../rel', 'foo'),
 
        ('file/../foo', 'foo'),
 
    ])
 
    def test_add_file_into_hg_bad_filenames(self, location, filename):
 
        self.log_user()
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=HG_REPO,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "foo",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 

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

	
 
    @parametrize('cnt,location,filename', [
 
        (1, '', 'foo.txt'),
 
        (2, 'dir', 'foo.rst'),
 
        (3, 'rel/dir', 'foo.bar'),
 
    ])
 
    def test_add_file_into_hg(self, cnt, location, filename):
 
        self.log_user()
 
        repo = fixture.create_repo(u'commit-test-%s' % cnt, repo_type='hg')
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "foo",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    # Git - add file
 
    def test_add_file_view_git(self):
 
        self.log_user()
 
        response = self.app.get(url('files_add_home',
 
                                      repo_name=GIT_REPO,
 
                                      revision='tip', f_path='/'))
 

	
 
    def test_add_file_into_git_missing_content(self):
 
        self.log_user()
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=GIT_REPO,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                     'content': '',
 
                                     '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        self.checkSessionFlash(response, 'No content')
 

	
 
    def test_add_file_into_git_missing_filename(self):
 
        self.log_user()
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=GIT_REPO,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "foo",
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 

	
 
        self.checkSessionFlash(response, 'No filename')
 

	
 
    @parametrize('location,filename', [
 
        ('/abs', 'foo'),
 
        ('../rel', 'foo'),
 
        ('file/../foo', 'foo'),
 
    ])
 
    def test_add_file_into_git_bad_filenames(self, location, filename):
 
        self.log_user()
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=GIT_REPO,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "foo",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 

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

	
 
    @parametrize('cnt,location,filename', [
 
        (1, '', 'foo.txt'),
 
        (2, 'dir', 'foo.rst'),
 
        (3, 'rel/dir', 'foo.bar'),
 
    ])
 
    def test_add_file_into_git(self, cnt, location, filename):
 
        self.log_user()
 
        repo = fixture.create_repo(u'commit-test-%s' % cnt, repo_type='git')
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "foo",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    # Hg - EDIT
 
    def test_edit_file_view_hg(self):
 
        self.log_user()
 
        response = self.app.get(url('files_edit_home',
 
                                      repo_name=HG_REPO,
 
                                      revision='tip', f_path='vcs/nodes.py'))
 

	
 
    def test_edit_file_view_not_on_branch_hg(self):
 
        self.log_user()
 
        repo = fixture.create_repo(u'test-edit-repo', repo_type='hg')
 

	
 
        ## add file
 
        location = 'vcs'
 
        filename = 'nodes.py'
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "def py():\n print 'hello'\n",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        response.follow()
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
            response = self.app.get(url('files_edit_home',
 
                                          repo_name=repo.repo_name,
 
                                          revision='tip', f_path=posixpath.join(location, filename)),
 
                                    status=302)
 
            self.checkSessionFlash(response,
 
                'You can only edit files with revision being a valid branch')
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    def test_edit_file_view_commit_changes_hg(self):
 
        self.log_user()
 
        repo = fixture.create_repo(u'test-edit-repo', repo_type='hg')
 

	
 
        ## add file
 
        location = 'vcs'
 
        filename = 'nodes.py'
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip',
 
                                      f_path='/'),
 
                                 params={
 
                                    'content': "def py():\n print 'hello'\n",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        response.follow()
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
            response = self.app.post(url('files_edit_home',
 
                                          repo_name=repo.repo_name,
 
                                          revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
 
                                          f_path=posixpath.join(location, filename)),
 
                                     params={
 
                                        'content': "def py():\n print 'hello world'\n",
 
                                        'message': 'i committed',
 
                                        '_authentication_token': self.authentication_token(),
 
                                     },
 
                                    status=302)
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    # Git - edit
 
    def test_edit_file_view_git(self):
 
        self.log_user()
 
        response = self.app.get(url('files_edit_home',
 
                                      repo_name=GIT_REPO,
 
                                      revision='tip', f_path='vcs/nodes.py'))
 

	
 
    def test_edit_file_view_not_on_branch_git(self):
 
        self.log_user()
 
        repo = fixture.create_repo(u'test-edit-repo', repo_type='git')
 

	
 
        ## add file
 
        location = 'vcs'
 
        filename = 'nodes.py'
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "def py():\n print 'hello'\n",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        response.follow()
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
            response = self.app.get(url('files_edit_home',
 
                                          repo_name=repo.repo_name,
 
                                          revision='tip', f_path=posixpath.join(location, filename)),
 
                                    status=302)
 
            self.checkSessionFlash(response,
 
                'You can only edit files with revision being a valid branch')
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    def test_edit_file_view_commit_changes_git(self):
 
        self.log_user()
 
        repo = fixture.create_repo(u'test-edit-repo', repo_type='git')
 

	
 
        ## add file
 
        location = 'vcs'
 
        filename = 'nodes.py'
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip',
 
                                      f_path='/'),
 
                                 params={
 
                                    'content': "def py():\n print 'hello'\n",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        response.follow()
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
            response = self.app.post(url('files_edit_home',
 
                                          repo_name=repo.repo_name,
 
                                          revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
 
                                          f_path=posixpath.join(location, filename)),
 
                                     params={
 
                                        'content': "def py():\n print 'hello world'\n",
 
                                        'message': 'i committed',
 
                                        '_authentication_token': self.authentication_token(),
 
                                     },
 
                                    status=302)
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    # Hg - delete
 
    def test_delete_file_view_hg(self):
 
        self.log_user()
 
        response = self.app.get(url('files_delete_home',
 
                                     repo_name=HG_REPO,
 
                                     revision='tip', f_path='vcs/nodes.py'))
 

	
 
    def test_delete_file_view_not_on_branch_hg(self):
 
        self.log_user()
 
        repo = fixture.create_repo(u'test-delete-repo', repo_type='hg')
 

	
 
        ## add file
 
        location = 'vcs'
 
        filename = 'nodes.py'
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "def py():\n print 'hello'\n",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        response.follow()
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
            response = self.app.get(url('files_delete_home',
 
                                          repo_name=repo.repo_name,
 
                                          revision='tip', f_path=posixpath.join(location, filename)),
 
                                    status=302)
 
            self.checkSessionFlash(response,
 
                'You can only delete files with revision being a valid branch')
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    def test_delete_file_view_commit_changes_hg(self):
 
        self.log_user()
 
        repo = fixture.create_repo(u'test-delete-repo', repo_type='hg')
 

	
 
        ## add file
 
        location = 'vcs'
 
        filename = 'nodes.py'
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip',
 
                                      f_path='/'),
 
                                 params={
 
                                    'content': "def py():\n print 'hello'\n",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        response.follow()
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
            response = self.app.post(url('files_delete_home',
 
                                          repo_name=repo.repo_name,
 
                                          revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
 
                                          f_path=posixpath.join(location, filename)),
 
                                     params={
 
                                        'message': 'i committed',
 
                                        '_authentication_token': self.authentication_token(),
 
                                     },
 
                                    status=302)
 
            self.checkSessionFlash(response,
 
                                   'Successfully deleted file %s' % posixpath.join(location, filename))
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    # Git - delete
 
    def test_delete_file_view_git(self):
 
        self.log_user()
 
        response = self.app.get(url('files_delete_home',
 
                                     repo_name=HG_REPO,
 
                                     revision='tip', f_path='vcs/nodes.py'))
 

	
 
    def test_delete_file_view_not_on_branch_git(self):
 
        self.log_user()
 
        repo = fixture.create_repo(u'test-delete-repo', repo_type='git')
 

	
 
        ## add file
 
        location = 'vcs'
 
        filename = 'nodes.py'
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "def py():\n print 'hello'\n",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        response.follow()
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
            response = self.app.get(url('files_delete_home',
 
                                          repo_name=repo.repo_name,
 
                                          revision='tip', f_path=posixpath.join(location, filename)),
 
                                    status=302)
 
            self.checkSessionFlash(response,
 
                'You can only delete files with revision being a valid branch')
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    def test_delete_file_view_commit_changes_git(self):
 
        self.log_user()
 
        repo = fixture.create_repo(u'test-delete-repo', repo_type='git')
 

	
 
        ## add file
 
        location = 'vcs'
 
        filename = 'nodes.py'
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip',
 
                                      f_path='/'),
 
                                 params={
 
                                    'content': "def py():\n print 'hello'\n",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        response.follow()
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
            response = self.app.post(url('files_delete_home',
 
                                          repo_name=repo.repo_name,
 
                                          revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
 
                                          f_path=posixpath.join(location, filename)),
 
                                     params={
 
                                        'message': 'i committed',
 
                                        '_authentication_token': self.authentication_token(),
 
                                     },
 
                                    status=302)
 
            self.checkSessionFlash(response,
 
                                   'Successfully deleted file %s' % posixpath.join(location, filename))
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
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'
 
        repo_name = self.REPO
 
        org_repo = Repository.get_by_repo_name(repo_name)
 
        creation_args = {
 
            'repo_name': fork_name,
 
            'repo_group': u'-1',
 
            'fork_parent_id': org_repo.repo_id,
 
            'repo_type': self.REPO_TYPE,
 
            'description': description,
 
            'private': 'False',
 
            'landing_rev': 'rev:tip',
 
            '_authentication_token': self.authentication_token()}
 

	
 
        self.app.post(url(controller='forks', action='fork_create',
 
                          repo_name=repo_name), creation_args)
 

	
 
        response = self.app.get(url(controller='forks', action='forks',
 
                                    repo_name=repo_name))
 

	
 
        response.mustcontain(
 
            """<a href="/%s">%s</a>""" % (fork_name, fork_name)
 
        )
 

	
 
        # remove this fork
 
        response = self.app.post(url('delete_repo', repo_name=fork_name),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 

	
 
    def test_fork_create_into_group(self):
 
        self.log_user()
 
        group = fixture.create_repo_group(u'vc')
 
        group_id = group.group_id
 
        fork_name = self.REPO_FORK
 
        fork_name_full = 'vc/%s' % fork_name
 
        description = 'fork of vcs test'
 
        repo_name = self.REPO
 
        org_repo = Repository.get_by_repo_name(repo_name)
 
        creation_args = {
 
            'repo_name': fork_name,
 
            'repo_group': group_id,
 
            'fork_parent_id': org_repo.repo_id,
 
            'repo_type': self.REPO_TYPE,
 
            'description': description,
 
            'private': 'False',
 
            'landing_rev': 'rev:tip',
 
            '_authentication_token': self.authentication_token()}
 
        self.app.post(url(controller='forks', action='fork_create',
 
                          repo_name=repo_name), creation_args)
 
        repo = Repository.get_by_repo_name(fork_name_full)
 
        assert repo.fork.repo_name == self.REPO
 

	
 
        ## run the check page that triggers the flash message
 
        response = self.app.get(url('repo_check_home', repo_name=fork_name_full))
 
        #test if we have a message that fork is ok
 
        self.checkSessionFlash(response,
 
                'Forked repository %s as <a href="/%s">%s</a>'
 
                % (repo_name, fork_name_full, fork_name_full))
 

	
 
        #test if the fork was created in the database
 
        fork_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == fork_name_full).one()
 

	
 
        assert fork_repo.repo_name == fork_name_full
 
        assert fork_repo.fork.repo_name == repo_name
 

	
 
        # test if the repository is visible in the list ?
 
        response = self.app.get(url('summary_home', repo_name=fork_name_full))
 
        response.mustcontain(fork_name_full)
 
        response.mustcontain(self.REPO_TYPE)
 
        response.mustcontain('Fork of "<a href="/%s">%s</a>"' % (repo_name, repo_name))
 

	
 
        fixture.destroy_repo(fork_name_full)
 
        fixture.destroy_repo_group(group_id)
 

	
 
    def test_z_fork_create(self):
 
        self.log_user()
 
        fork_name = self.REPO_FORK
 
        description = 'fork of vcs test'
 
        repo_name = self.REPO
 
        org_repo = Repository.get_by_repo_name(repo_name)
 
        creation_args = {
 
            'repo_name': fork_name,
 
            'repo_group': u'-1',
 
            'fork_parent_id': org_repo.repo_id,
 
            'repo_type': self.REPO_TYPE,
 
            'description': description,
 
            'private': 'False',
 
            'landing_rev': 'rev:tip',
 
            '_authentication_token': self.authentication_token()}
 
        self.app.post(url(controller='forks', action='fork_create',
 
                          repo_name=repo_name), creation_args)
 
        repo = Repository.get_by_repo_name(self.REPO_FORK)
 
        assert repo.fork.repo_name == self.REPO
 

	
 
        ## run the check page that triggers the flash message
 
        response = self.app.get(url('repo_check_home', repo_name=fork_name))
 
        #test if we have a message that fork is ok
 
        self.checkSessionFlash(response,
 
                'Forked repository %s as <a href="/%s">%s</a>'
 
                % (repo_name, fork_name, fork_name))
 

	
 
        #test if the fork was created in the database
 
        fork_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == fork_name).one()
 

	
 
        assert fork_repo.repo_name == fork_name
 
        assert fork_repo.fork.repo_name == repo_name
 

	
 
        # test if the repository is visible in the list ?
 
        response = self.app.get(url('summary_home', repo_name=fork_name))
 
        response.mustcontain(fork_name)
 
        response.mustcontain(self.REPO_TYPE)
 
        response.mustcontain('Fork of "<a href="/%s">%s</a>"' % (repo_name, repo_name))
 

	
 
    def test_zz_fork_permission_page(self):
 
        usr = self.log_user(self.username, self.password)['user_id']
 
        repo_name = self.REPO
 

	
 
        forks = Repository.query() \
 
            .filter(Repository.repo_type == self.REPO_TYPE) \
 
            .filter(Repository.fork_id != None).all()
 
        assert 1 == len(forks)
 

	
 
        # set read permissions for this
 
        RepoModel().grant_user_permission(repo=forks[0],
 
                                          user=usr,
 
                                          perm='repository.read')
 
        Session().commit()
 

	
 
        response = self.app.get(url(controller='forks', action='forks',
 
                                    repo_name=repo_name))
 

	
 
        response.mustcontain('<div style="padding:5px 3px 3px 42px;">fork of vcs test</div>')
 

	
 
    def test_zzz_fork_permission_page(self):
 
        usr = self.log_user(self.username, self.password)['user_id']
 
        repo_name = self.REPO
 

	
 
        forks = Repository.query() \
 
            .filter(Repository.repo_type == self.REPO_TYPE) \
 
            .filter(Repository.fork_id != None).all()
 
        assert 1 == len(forks)
 

	
 
        # set none
 
        RepoModel().grant_user_permission(repo=forks[0],
 
                                          user=usr, perm='repository.none')
 
        Session().commit()
 
        # fork shouldn't be there
 
        response = self.app.get(url(controller='forks', action='forks',
 
                                    repo_name=repo_name))
 
        response.mustcontain('There are no forks yet')
 

	
 

	
 
class TestGIT(_BaseTestCase):
 
    REPO = GIT_REPO
 
    NEW_REPO = NEW_GIT_REPO
 
    REPO_TYPE = 'git'
 
    REPO_FORK = GIT_FORK
 

	
 

	
 
class TestHG(_BaseTestCase):
 
    REPO = HG_REPO
 
    NEW_REPO = NEW_HG_REPO
 
    REPO_TYPE = 'hg'
 
    REPO_FORK = HG_FORK
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'),)

Changeset was too big and was cut off... Show full diff anyway

0 comments (0 inline, 0 general)