# -*- 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

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

__all__ = [

    '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('/', 'tmp', '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

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

        if 'Invalid username or password' in response.body:

            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)

    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:

            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 _matcher(msg, 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']))))

    def checkSessionFlashRegex(self, response, regex, skip=0):

        self.checkSessionFlash(response, regex, skip=skip, _matcher=re.search)

from kallithea.tests import *

from kallithea.model.db import Setting

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

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

        self.assertEqual(new_settings['auth_ldap_host'], u'dc.example.com',

                         'fail db write compare')

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

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

                           }

        )

        self.assertEqual(response.form['email'].value, 'john@example.org')

        self.assertEqual(response.form['firstname'].value, 'John')

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

        )

        self.assertNotIn('Log Out', 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()

        self.assertEqual(new_settings['auth_crowd_host'], u'hostname',

                         'fail db write compare')

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

        self.assertEqual(new_settings['auth_pam_service'], u'kallithea',

                         'fail db write compare')