nosetests --pdb --pdb-failures

nosetests --with-coverage --cover-package=rhodecode.model.validators rhodecode.tests.test_validators

optional FLAGS:

    RC_WHOOSH_TEST_DISABLE=1 - skip whoosh index building and tests

    RC_NO_TMP_PATH=1 - disable new temp path for tests, used mostly for test_vcs_operations

"""

import os

import time

import logging

import datetime

import hashlib

import tempfile

from os.path import join as jn

from unittest import TestCase

from tempfile import _RandomNameSequence

from paste.deploy import loadapp

from paste.script.appinstall import SetupCommand

from pylons import config, url

from routes.util import URLGenerator

from webtest import TestApp

from rhodecode import is_windows

from rhodecode.model.meta import Session

from rhodecode.model.db import User

from rhodecode.tests.nose_parametrized import parameterized

import pylons.test

from rhodecode.lib.utils2 import safe_unicode, safe_str

os.environ['TZ'] = 'UTC'

if not is_windows:

    time.tzset()

log = logging.getLogger(__name__)

__all__ = [

    'parameterized', 'environ', 'url', 'get_new_dir', 'TestController',

    '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_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', '_get_repo_create_params',

    '_get_group_create_params'

]

# Invoke websetup with the current config file

# SetupCommand('setup-app').run([config_file])

environ = {}

#SOME GLOBALS FOR TESTS

TESTS_TMP_PATH = jn('/', 'tmp', 'rc_test_%s' % _RandomNameSequence().next())

TEST_USER_ADMIN_LOGIN = 'test_admin'

TEST_USER_ADMIN_PASS = 'test12'

TEST_USER_ADMIN_EMAIL = 'test_admin@mail.com'

TEST_USER_REGULAR_LOGIN = 'test_regular'

from rhodecode.tests import *

from rhodecode.model.db import RhodeCodeSetting

skip_ldap_test = False

try:

    import ldap

except ImportError:

    # means that python-ldap is not installed

    skip_ldap_test = True

    pass

class TestLdapSettingsController(TestController):

    def test_index(self):

        self.log_user()

        response = self.app.get(url(controller='admin/ldap_settings',

                                    action='index'))

    def test_ldap_save_settings(self):

        self.log_user()

        if skip_ldap_test:

            raise SkipTest('skipping due to missing ldap lib')

        test_url = url(controller='admin/ldap_settings',

                       action='ldap_settings')

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

            params={'ldap_host' : u'dc.example.com',

                    'ldap_port' : '999',

                    'ldap_tls_kind' : 'PLAIN',

                    'ldap_tls_reqcert' : 'NEVER',

                    'ldap_dn_user':'test_user',

                    'ldap_dn_pass':'test_pass',

                    'ldap_base_dn':'test_base_dn',

                    'ldap_filter':'test_filter',

                    'ldap_search_scope':'BASE',

                    'ldap_attr_login':'test_attr_login',

                    'ldap_attr_firstname':'ima',

                    'ldap_attr_lastname':'tester',

                    'ldap_attr_email':'test@example.com' })

    def test_ldap_error_form(self):

        self.log_user()

        if skip_ldap_test:

            raise SkipTest('skipping due to missing ldap lib')

        test_url = url(controller='admin/ldap_settings',

                       action='ldap_settings')

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

            params={'ldap_host' : '',

                    'ldap_port' : 'i-should-be-number',

                    'ldap_tls_kind' : 'PLAIN',

                    'ldap_tls_reqcert' : 'NEVER',

                    'ldap_dn_user':'',

                    'ldap_dn_pass':'',

                    'ldap_base_dn':'',

                    'ldap_filter':'',

                    'ldap_search_scope':'BASE',

                    'ldap_attr_login':'', #  <----- missing required input

                    'ldap_attr_firstname':'',

                    'ldap_attr_lastname':'',

                    'ldap_attr_email':'' })

    def test_ldap_login(self):

        pass

    def test_ldap_login_incorrect(self):

        pass

from rhodecode.tests import *

from rhodecode.model.db import Notification, User

from rhodecode.model.user import UserModel

from rhodecode.model.notification import NotificationModel

class TestNotificationsController(TestController):

    def tearDown(self):

        for n in Notification.query().all():

            inst = Notification.get(n.notification_id)

            self.Session().delete(inst)

        self.Session().commit()

    def test_index(self):

        self.log_user()

        u1 = UserModel().create_or_update(username='u1', password='qweqwe',

                                               email='u1@rhodecode.org',

                                               firstname='u1', lastname='u1')

        u1 = u1.user_id

        response = self.app.get(url('notifications'))

        cur_user = self._get_logged_user()

        NotificationModel().create(created_by=u1, subject=u'test_notification_1',

                                   body=u'notification_1',

                                   recipients=[cur_user])

        self.Session().commit()

        response = self.app.get(url('notifications'))

#    def test_index_as_xml(self):

#        response = self.app.get(url('formatted_notifications', format='xml'))

#

#    def test_create(self):

#        response = self.app.post(url('notifications'))

#

#    def test_new(self):

#        response = self.app.get(url('new_notification'))

#

#    def test_new_as_xml(self):

#        response = self.app.get(url('formatted_new_notification', format='xml'))

#

#    def test_update(self):

#        response = self.app.put(url('notification', notification_id=1))

#

#    def test_update_browser_fakeout(self):

#        response = self.app.post(url('notification', notification_id=1), params=dict(_method='put'))

    def test_delete(self):

        self.log_user()

        cur_user = self._get_logged_user()

        u1 = UserModel().create_or_update(username='u1', password='qweqwe',

        self.assertEqual(RhodeCodeSetting

                         .get_app_settings()['rhodecode_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 = 'RhodeCode'

        old_realm = 'RhodeCode authentication'

        new_ga_code = ''

        response = self.app.post(url('admin_setting', setting_id='global'),

                                     params=dict(

                                                 _method='put',

                                                 rhodecode_title=old_title,

                                                 rhodecode_realm=old_realm,

                                                 rhodecode_ga_code=new_ga_code

                                                 ))

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

        self.assertEqual(RhodeCodeSetting

                        .get_app_settings()['rhodecode_ga_code'], new_ga_code)

        response = response.follow()

    def test_title_change(self):

        self.log_user()

        old_title = 'RhodeCode'

        new_title = old_title + '_changed'

        old_realm = 'RhodeCode authentication'

        for new_title in ['Changed', 'Żółwik', old_title]:

            response = self.app.post(url('admin_setting', setting_id='global'),

                                         params=dict(

                                                     _method='put',

                                                     rhodecode_title=new_title,

                                                     rhodecode_realm=old_realm,

                                                     rhodecode_ga_code=''

                                                     ))

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

            self.assertEqual(RhodeCodeSetting

                             .get_app_settings()['rhodecode_title'],

                             new_title.decode('utf-8'))

            response = response.follow()

            response.mustcontain("""<h1><a href="/">%s</a></h1>""" % new_title)

    def test_my_account(self):

        self.log_user()

        response = self.app.get(url('admin_settings_my_account'))

    @parameterized.expand([('firstname', 'new_username'),

                           ('lastname', 'new_username'),

                           ('admin', True),

                           ('admin', False),

                           ('ldap_dn', 'test'),

                           ('ldap_dn', None),

                           ('active', False),

                           ('active', True),

                           ('email', 'some@email.com'),

                           ])

    def test_my_account_update(self, name, expected):

        uname = 'testme'

        usr = UserModel().create_or_update(username=uname, password='qweqwe',

                                           email='testme@rhodecod.org')

        self.Session().commit()

        params = usr.get_api_data()

        user_id = usr.user_id

        self.log_user(username=uname, password='qweqwe')

        params.update({name: expected})

        params.update({'password_confirmation': ''})

        params.update({'new_password': ''})

        try:

        new_email = 'newmail.pl'

        response = self.app.post(url('admin_settings_my_account_update'),

                                 params=dict(

                                            _method='put',

                                            username='test_admin',

                                            new_password='test12',

                                            password_confirmation='test122',

                                            firstname='NewName',

                                            lastname='NewLastname',

                                            email=new_email,)

                                 )

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

        from rhodecode.model import validators

        msg = validators.ValidUsername(edit=False,

                                    old_data={})._messages['username_exists']

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

        response.mustcontain(u"%s" % msg)

    def test_set_repo_fork_has_no_self_id(self):

        self.log_user()

        repo = Repository.get_by_repo_name(HG_REPO)

        response = self.app.get(url('edit_repo', repo_name=HG_REPO))

        opt = """<option value="%s">vcs_test_git</option>""" % repo.repo_id

    def test_set_fork_of_repo(self):

        self.log_user()

        repo = Repository.get_by_repo_name(HG_REPO)

        repo2 = Repository.get_by_repo_name(GIT_REPO)

        response = self.app.put(url('repo_as_fork', repo_name=HG_REPO),

                                 params=dict(

                                    id_fork_of=repo2.repo_id

                                 ))

        repo = Repository.get_by_repo_name(HG_REPO)

        repo2 = Repository.get_by_repo_name(GIT_REPO)

        self.checkSessionFlash(response,

        'Marked repo %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)

        # clean session flash

        #response = self.app.get(url('edit_repo', repo_name=HG_REPO))

from rhodecode.tests import *

class TestFollowersController(TestController):

        self.log_user()

        repo_name = HG_REPO

        response = self.app.get(url(controller='followers',

                                    action='followers',

                                    repo_name=repo_name))

from rhodecode.model.meta import Session

class TestForksController(TestController):

    def setUp(self):

        self.username = u'forkuser'

        self.password = u'qweqwe'

        self.u1 = UserModel().create_or_update(

            username=self.username, password=self.password,

            email=u'fork_king@rhodecode.org', firstname=u'u1', lastname=u'u1'

        )

        Session().commit()

    def tearDown(self):

        Session().delete(self.u1)

        Session().commit()

    def test_index(self):

        self.log_user()

        repo_name = HG_REPO

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

                                    repo_name=repo_name))

    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 = HG_REPO

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

                          repo_name=repo_name), {}, status=403)

    def test_index_with_fork_hg(self):

        self.log_user()

        # create a fork

        fork_name = HG_FORK

        description = 'fork of vcs test'

        repo_name = HG_REPO

        org_repo = Repository.get_by_repo_name(repo_name)

        response = self.app.post(url(controller='forks',

                                     'fork_parent_id':org_repo.repo_id,

                                     'repo_type':'hg',

                                     'description':description,

                                     'private':'False',

                                     'landing_rev': 'tip'})

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

        self.assertEqual(fork_repo.repo_name, fork_name)

        self.assertEqual(fork_repo.fork.repo_name, repo_name)

        #test if fork is visible in the list ?

        response = response.follow()

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

                                    repo_name=fork_name))

    def test_zz_fork_permission_page(self):

        usr = self.log_user(self.username, self.password)['user_id']

        repo_name = HG_REPO

        forks = Session().query(Repository)\

            .filter(Repository.fork_id != None)\

            .all()

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

import os

from rhodecode.tests import *

class TestSearchController(TestController):

    def test_index(self):

        self.log_user()

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

        # Test response...

    def test_empty_search(self):

        if os.path.isdir(self.index_location):

            raise SkipTest('skipped due to existing index')

        else:

            self.log_user()

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

                                    {'q': HG_REPO})

    def test_normal_search(self):

        self.log_user()

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

                                {'q': 'def repo'})

        response.mustcontain('39 results')

    def test_repo_search(self):

        self.log_user()

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

                                {'q': 'repository:%s def test' % HG_REPO})

        response.mustcontain('4 results')

    def test_search_last(self):

        self.log_user()

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

                                {'q': 'last:t', 'type': 'commit'})

        response.mustcontain('2 results')

    def test_search_commit_message(self):

        self.log_user()

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