# slugify repo group just in case :)

            slug = repo_name_slug(group_name)

            # check for parent of self

            parent_of_self = lambda: (

                old_data['group_id'] == group_parent_id

                if group_parent_id else False

            )

            if edit and parent_of_self():

                msg = M(self, 'group_parent_id', state)

                raise formencode.Invalid(msg, value, state,

                    error_dict=dict(group_parent_id=msg)

                )

            old_gname = None

            if edit:

                old_gname = RepoGroup.get(old_data.get('group_id')).group_name

            if old_gname != group_name or not edit:

                # check group

                gr = RepoGroup.query()\

                      .filter(RepoGroup.group_name == slug)\

                      .filter(RepoGroup.group_parent_id == group_parent_id)\

                      .scalar()

                if gr is not None:

                    msg = M(self, 'group_exists', state, group_name=slug)

                    raise formencode.Invalid(msg, value, state,

                            error_dict=dict(group_name=msg)

                    )

                # check for same repo

                repo = Repository.query()\

                      .filter(Repository.repo_name == slug)\

                      .scalar()

                if repo is not None:

                    msg = M(self, 'repo_exists', state, group_name=slug)

                    raise formencode.Invalid(msg, value, state,

                            error_dict=dict(group_name=msg)

                    )

    return _validator

def ValidPassword():

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'invalid_password':

                _('Invalid characters (non-ascii) in password')

        }

        def validate_python(self, value, state):

            try:

                (value or '').decode('ascii')

            except UnicodeError:

                msg = M(self, 'invalid_password', state)

                raise formencode.Invalid(msg, value, state,)

    return _validator

def ValidOldPassword(username):

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'invalid_password': _('Invalid old password')

        }

        def validate_python(self, value, state):

            from kallithea.lib import auth_modules

            if auth_modules.authenticate(username, value, '') is None:

                msg = M(self, 'invalid_password', state)

                raise formencode.Invalid(msg, value, state,

                    error_dict=dict(current_password=msg)

                )

    return _validator

def ValidPasswordsMatch(password_field, password_confirmation_field):

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'password_mismatch': _('Passwords do not match'),

        }

        def validate_python(self, value, state):

            if value.get(password_field) != value[password_confirmation_field]:

                msg = M(self, 'password_mismatch', state)

                raise formencode.Invalid(msg, value, state,

                     error_dict={password_field:msg, password_confirmation_field: msg}

                )

    return _validator

def ValidAuth():

    class _validator(formencode.validators.FancyValidator):

        messages = {

        }

        def validate_python(self, value, state):

            from kallithea.lib import auth_modules

            password = value['password']

            username = value['username']

            # authenticate returns unused dict but has called

            # plugin._authenticate which has create_or_update'ed the username user in db

            if auth_modules.authenticate(username, password) is None:

                user = User.get_by_username(username)

                if user and not user.active:

                    log.warning('user %s is disabled' % username)

                    raise formencode.Invalid(msg, value, state,

                    )

                else:

                    log.warning('user %s failed to authenticate' % username)

                    raise formencode.Invalid(msg, value, state,

                    )

    return _validator

def ValidAuthToken():

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'invalid_token': _('Token mismatch')

        }

        def validate_python(self, value, state):

            if value != authentication_token():

                msg = M(self, 'invalid_token', state)

                raise formencode.Invalid(msg, value, state)

    return _validator

def ValidRepoName(edit=False, old_data={}):

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'invalid_repo_name':

                _('Repository name %(repo)s is not allowed'),

            'repository_exists':

                _('Repository named %(repo)s already exists'),

            'repository_in_group_exists': _('Repository "%(repo)s" already '

                                            'exists in group "%(group)s"'),

            'same_group_exists': _('Repository group with name "%(repo)s" '

                                   'already exists')

        }

        def _to_python(self, value, state):

            repo_name = repo_name_slug(value.get('repo_name', ''))

            repo_group = value.get('repo_group')

            if repo_group:

                gr = RepoGroup.get(repo_group)

                group_path = gr.full_path

                group_name = gr.group_name

                # value needs to be aware of group name in order to check

                # db key This is an actual just the name to store in the

                # database

                repo_name_full = group_path + RepoGroup.url_sep() + repo_name

            else:

                group_name = group_path = ''

                repo_name_full = repo_name

            value['repo_name'] = repo_name

            value['repo_name_full'] = repo_name_full

            value['group_path'] = group_path

            value['group_name'] = group_name

            return value

        def validate_python(self, value, state):

            repo_name = value.get('repo_name')

            repo_name_full = value.get('repo_name_full')

            group_path = value.get('group_path')

            group_name = value.get('group_name')

            if repo_name in [ADMIN_PREFIX, '']:

                msg = M(self, 'invalid_repo_name', state, repo=repo_name)

                raise formencode.Invalid(msg, value, state,

                    error_dict=dict(repo_name=msg)

                )

            rename = old_data.get('repo_name') != repo_name_full

            create = not edit

            if rename or create:

                if group_path != '':

                    if Repository.get_by_repo_name(repo_name_full):

                        msg = M(self, 'repository_in_group_exists', state,

                                repo=repo_name, group=group_name)

                        raise formencode.Invalid(msg, value, state,

                            error_dict=dict(repo_name=msg)

                        )

                elif RepoGroup.get_by_group_name(repo_name_full):

                        msg = M(self, 'same_group_exists', state,

                                repo=repo_name)

                        raise formencode.Invalid(msg, value, state,

                            error_dict=dict(repo_name=msg)

                        )

                elif Repository.get_by_repo_name(repo_name_full):

                        msg = M(self, 'repository_exists', state,

                                repo=repo_name)

                        raise formencode.Invalid(msg, value, state,

                            error_dict=dict(repo_name=msg)

                        )

            return value

    return _validator

def ValidForkName(*args, **kwargs):

    return ValidRepoName(*args, **kwargs)

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

def get_new_dir(title):

    """

    Returns always new directory path.

    """

    from kallithea.tests.vcs.utils import get_normalized_path

    name = TEST_REPO_PREFIX

    if title:

        name = '-'.join((name, title))

    hex = hashlib.sha1(str(time.time())).hexdigest()

    name = '-'.join((name, hex))

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

    return get_normalized_path(path)

import logging

class NullHandler(logging.Handler):

    def emit(self, record):

        pass

def init_stack(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)

class BaseTestCase(unittest.TestCase):

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

        self.wsgiapp = pylons.test.pylonsapp

        init_stack(self.wsgiapp.config)

        unittest.TestCase.__init__(self, *args, **kwargs)

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

class TestController(BaseTestCase):

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

        BaseTestCase.__init__(self, *args, **kwargs)

        self.app = TestApp(self.wsgiapp)

        self.maxDiff = None

        self.index_location = config['app_conf']['index_dir']

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

            self.fail('could not login using %s %s' % (username, password))

        self.assertEqual(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

        self.assertEqual(user, expected_username)

        self.assertEqual(cookie.get('is_authenticated'), True)

    def authentication_token(self):

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

    def checkSessionFlash(self, response, msg, skip=0):

        if 'flash' not in response.session:

            self.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 msg in m:

                return

        except IndexError:

            pass

        self.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']))))

        response = response.follow()

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

    def test_login_regular_ok(self):

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

                                 {'username': TEST_USER_REGULAR_LOGIN,

                                  'password': TEST_USER_REGULAR_PASS})

        self.assertEqual(response.status, '302 Found')

        self.assert_authenticated_user(response, TEST_USER_REGULAR_LOGIN)

        response = response.follow()

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

    def test_login_ok_came_from(self):

        test_came_from = '/_admin/users'

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

                                     came_from=test_came_from),

                                 {'username': TEST_USER_ADMIN_LOGIN,

                                  'password': TEST_USER_ADMIN_PASS})

        self.assertEqual(response.status, '302 Found')

        response = response.follow()

        self.assertEqual(response.status, '200 OK')

        response.mustcontain('Users Administration')

    def test_login_do_not_remember(self):

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

                                 {'username': TEST_USER_REGULAR_LOGIN,

                                  'password': TEST_USER_REGULAR_PASS,

                                  'remember': False})

        self.assertIn('Set-Cookie', response.headers)

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

            self.assertFalse(re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE),

                'Cookie %r has expiration date, but should be a session cookie' % cookie)

    def test_login_remember(self):

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

                                 {'username': TEST_USER_REGULAR_LOGIN,

                                  'password': TEST_USER_REGULAR_PASS,

                                  'remember': True})

        self.assertIn('Set-Cookie', response.headers)

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

            self.assertTrue(re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE),

                'Cookie %r should have expiration date, but is a session cookie' % cookie)

    def test_logout(self):

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

                                 {'username': TEST_USER_REGULAR_LOGIN,

                                  'password': TEST_USER_REGULAR_PASS})

        # Verify that a login session has been established.

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

        response = response.follow()

        self.assertIn('authuser', response.session)

        response.click('Log Out')

        # Verify that the login session has been terminated.

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

        self.assertNotIn('authuser', response.session)

    @parameterized.expand([

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

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

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

          ('ftp://some.ftp.server',),

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

    ])

    def test_login_bad_came_froms(self, url_came_from):

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

                                     came_from=url_came_from),

                                 {'username': TEST_USER_ADMIN_LOGIN,

                                  'password': TEST_USER_ADMIN_PASS})

        self.assertEqual(response.status, '302 Found')

        self.assertEqual(response._environ['paste.testing_variables']

                         ['tmpl_context'].came_from, '/')

        response = response.follow()

        self.assertEqual(response.status, '200 OK')

    def test_login_short_password(self):

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

                                 {'username': TEST_USER_ADMIN_LOGIN,

                                  'password': 'as'})

        self.assertEqual(response.status, '200 OK')

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

    def test_login_wrong_username_password(self):

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

                                 {'username': 'error',

                                  'password': 'test12'})

    # verify that get arguments are correctly passed along login redirection

    @parameterized.expand([

        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),

        ({'blue': u'blå'.encode('utf-8'), 'green':u'grøn'},

             ('blue=bl%C3%A5', 'green=gr%C3%B8n')),

    ])

    def test_redirection_to_login_form_preserves_get_args(self, args, args_encoded):

        with fixture.anon_access(False):

            response = self.app.get(url(controller='summary', action='index',

                                        repo_name=HG_REPO,

                                        **args))

            self.assertEqual(response.status, '302 Found')

            for encoded in args_encoded:

                self.assertIn(encoded, response.location)

    @parameterized.expand([

        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),

        ({'blue': u'blå'.encode('utf-8'), 'green':u'grøn'},

             ('blue=bl%C3%A5', 'green=gr%C3%B8n')),

    ])

    def test_login_form_preserves_get_args(self, args, args_encoded):

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

                                    came_from = '/_admin/users',

                                    **args))

        for encoded in args_encoded:

            self.assertIn(encoded, response.form.action)

    @parameterized.expand([

        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),

        ({'blue': u'blå'.encode('utf-8'), 'green':u'grøn'},

             ('blue=bl%C3%A5', 'green=gr%C3%B8n')),

    ])

    def test_redirection_after_successful_login_preserves_get_args(self, args, args_encoded):

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

                                     came_from = '/_admin/users',

                                     **args),

                                 {'username': TEST_USER_ADMIN_LOGIN,

                                  'password': TEST_USER_ADMIN_PASS})

        self.assertEqual(response.status, '302 Found')

        for encoded in args_encoded:

            self.assertIn(encoded, response.location)

    @parameterized.expand([

        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),

        ({'blue': u'blå'.encode('utf-8'), 'green':u'grøn'},

             ('blue=bl%C3%A5', 'green=gr%C3%B8n')),

    ])

    def test_login_form_after_incorrect_login_preserves_get_args(self, args, args_encoded):

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

                                     came_from = '/_admin/users',

                                     **args),

                                 {'username': 'error',

                                  'password': 'test12'})

        for encoded in args_encoded:

            self.assertIn(encoded, response.form.action)

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

    # REGISTRATIONS

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

    def test_register(self):

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

        response.mustcontain('Sign Up')

    def test_register_err_same_username(self):

        uname = TEST_USER_ADMIN_LOGIN

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

                                            {'username': uname,

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

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

                                             'firstname': 'test',

                                             'lastname': 'test'})

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

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

        response.mustcontain(msg)

    def test_register_err_same_email(self):

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

                                            {'username': 'test_admin_0',

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

                                             'email': TEST_USER_ADMIN_EMAIL,

                                             'firstname': 'test',

                                             'lastname': 'test'})

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

        response.mustcontain(msg)

    def test_register_err_same_email_case_sensitive(self):

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

                                            {'username': 'test_admin_1',

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

                                             'email': TEST_USER_ADMIN_EMAIL.title(),

                                             'firstname': 'test',

                                             'lastname': 'test'})

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

        response.mustcontain(msg)

    def test_register_err_wrong_data(self):

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

                                            {'username': 'xs',

                                             'password': 'test',

                                             'password_confirmation': 'test',

                                             'email': 'goodmailm',

                                             'firstname': 'test',

                                             'lastname': 'test'})

        self.assertEqual(response.status, '200 OK')

        response.mustcontain('An email address must contain a single @')

        response.mustcontain('Enter a value 6 characters long or more')

    def test_register_err_username(self):

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

                                            {'username': 'error user',

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

                                             'email': 'goodmailm',

                                             'firstname': 'test',

                                             'lastname': 'test'})

        response.mustcontain('An email address must contain a single @')

        response.mustcontain('Username may only contain '

                'alphanumeric characters underscores, '

                'periods or dashes and must begin with an '

                'alphanumeric character')

    def test_register_err_case_sensitive(self):

        usr = TEST_USER_ADMIN_LOGIN.title()

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

                                            {'username': usr,

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

                                             'email': 'goodmailm',

                                             'firstname': 'test',

                                             'lastname': 'test'})

        response.mustcontain('An email address must contain a single @')

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

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

        response.mustcontain(msg)

    def test_register_special_chars(self):

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

                                        {'username': 'xxxaxn',

                                         'password': 'ąćźżąśśśś',

                                         'password_confirmation': 'ąćźżąśśśś',

                                         'email': 'goodmailm@test.plx',

                                         'firstname': 'test',