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

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

                    '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

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

                                     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,

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

        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,

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

        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

        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],

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