# -*- coding: utf-8 -*-

from __future__ import with_statement

import mock

from kallithea.tests import *

from kallithea.tests.fixture import Fixture

from kallithea.lib.utils2 import generate_api_key

from kallithea.lib.auth import check_password

from kallithea.lib import helpers as h

from kallithea.model.api_key import ApiKeyModel

from kallithea.model import validators

from kallithea.model.db import User, Notification

from kallithea.model.meta import Session

fixture = Fixture()

class TestLoginController(TestController):

    def tearDown(self):

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

            Session().delete(n)

        Session().commit()

        self.assertEqual(Notification.query().all(), [])

    def test_index(self):

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

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

        # Test response...

    def test_login_admin_ok(self):

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

                                 {'username': TEST_USER_ADMIN_LOGIN,

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

        self.assertEqual(response.session['authuser'].get('username'),

                         TEST_USER_ADMIN_LOGIN)

        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,

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

        self.assertEqual(response.session['authuser'].get('username'),

                         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,

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

        response = response.follow()

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

        response.mustcontain('Users Administration')

    def test_logout(self):

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

                                 {'username': TEST_USER_REGULAR_LOGIN,

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

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

        response.mustcontain('invalid user name')

        response.mustcontain('invalid password')

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

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

        response.mustcontain('invalid user name')

        response.mustcontain('invalid password')

        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,

                                  skip_if_exists=True)

        params = usr.get_api_data(True)  # current user data

        user_id = usr.user_id

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

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

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

        params.update({'extern_type': 'internal'})

        params.update({'extern_name': self.test_user_1})

        params.update({'_authentication_token': self.authentication_token()})

        params.update(attrs)

        response = self.app.post(url('my_account'), params)

        self.checkSessionFlash(response,

                               'Your account was updated successfully')

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

        params['last_login'] = updated_params['last_login']

        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'] = str(user_id)

        if name == 'active':

            #my account cannot deactivate account

            params['active'] = True

        if name == 'admin':

            #my account cannot make you an admin !

            params['admin'] = False

        params.pop('_authentication_token')

        self.assertEqual(params, updated_params)

    def test_my_account_update_err_email_exists(self):

        self.log_user()

        new_email = TEST_USER_REGULAR_EMAIL  # already existing email

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

                                params=dict(

                                    username=TEST_USER_ADMIN_LOGIN,

                                    password_confirmation='test122',

                                    firstname='NewName',

                                    lastname='NewLastname',

                                    email=new_email,

                                    _authentication_token=self.authentication_token())

                                )

        response.mustcontain('This e-mail address is already taken')

    def test_my_account_update_err(self):

        new_email = 'newmail.pl'

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

                                 params=dict(

                                            username=TEST_USER_ADMIN_LOGIN,

                                            password_confirmation='test122',

                                            firstname='NewName',

                                            lastname='NewLastname',

                                            email=new_email,

                                            _authentication_token=self.authentication_token()))

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

        from kallithea.model import validators

        msg = validators.ValidUsername(edit=False, old_data={})\

                ._messages['username_exists']

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

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

    def test_my_account_api_keys(self):

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

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

        response.mustcontain(user.api_key)

        response.mustcontain('Expires: Never')

    @parameterized.expand([

        ('forever', -1),

        ('5mins', 60*5),

        ('30days', 60*60*24*30),

    ])

    def test_my_account_add_api_keys(self, desc, lifetime):

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

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

                                 {'description': desc, 'lifetime': lifetime, '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'API key successfully created')

        try:

            response = response.follow()

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

            for api_key in user.api_keys:

                response.mustcontain(api_key)

        finally:

            for api_key in UserApiKeys.query().all():

                Session().delete(api_key)

                Session().commit()

    def test_my_account_remove_api_key(self):

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

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

                                 {'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().all()

        self.assertEqual(1, len(keys))

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

                 {'_method': 'delete', 'del_api_key': keys[0].api_key, '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'API key successfully deleted')

        keys = UserApiKeys.query().all()

        self.assertEqual(0, len(keys))

    def test_my_account_reset_main_api_key(self):

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

        api_key = user.api_key

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

        response.mustcontain(api_key)

        response.mustcontain('Expires: Never')

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

                 {'_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])

# 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/>.

"""

kallithea.tests.test_hg_operations

~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Test suite for making push/pull operations

This file was forked by the Kallithea project in July 2014.

Original author and date, and relevant copyright and licensing information is below:

:created_on: Dec 30, 2010

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import os

import sys

import shutil

import logging

from os.path import join as jn

from os.path import dirname as dn

from tempfile import _RandomNameSequence

from subprocess import Popen, PIPE

from paste.deploy import appconfig

from sqlalchemy import engine_from_config

from kallithea.lib.utils import add_cache

from kallithea.model import init_model

from kallithea.model import meta

from kallithea.model.db import User, Repository

from kallithea.lib.auth import get_crypt_password

from kallithea.tests import TESTS_TMP_PATH, HG_REPO

from kallithea.config.environment import load_environment

rel_path = dn(dn(dn(dn(os.path.abspath(__file__)))))

conf = appconfig('config:development.ini', relative_to=rel_path)

load_environment(conf.global_conf, conf.local_conf)

add_cache(conf)

USER = TEST_USER_ADMIN_LOGIN

HOST = 'server.local'

METHOD = 'pull'

DEBUG = True

log = logging.getLogger(__name__)

class Command(object):

    def __init__(self, cwd):

        self.cwd = cwd

    def execute(self, cmd, *args):

        """Runs command on the system with given ``args``.

        """

        command = cmd + ' ' + ' '.join(args)

        log.debug('Executing %s' % command)

        if DEBUG:

            print command

        p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd)

        stdout, stderr = p.communicate()

        if DEBUG:

            print stdout, stderr

        return stdout, stderr

def get_session():

    engine = engine_from_config(conf, 'sqlalchemy.db1.')

    init_model(engine)

    sa = meta.Session

    return sa

def create_test_user(force=True):

    print 'creating test user'

    sa = get_session()

    user = sa.query(User).filter(User.username == USER).scalar()

    if force and user is not None:

        print 'removing current user'

        for repo in sa.query(Repository).filter(Repository.user == user).all():

            sa.delete(repo)

        sa.delete(user)

        sa.commit()

    if user is None or force:

        print 'creating new one'