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

"""

kallithea.lib.celerylib.tasks

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

Kallithea task modules, containing all task that suppose to be run

by celery daemon

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: Oct 6, 2010

:author: marcink

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

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

"""

import email.utils

import os

import traceback

from collections import OrderedDict

from operator import itemgetter

from time import mktime

import celery.utils.log

from tg import config

import kallithea

from kallithea.lib import celerylib, ext_json

from kallithea.lib.helpers import person

from kallithea.lib.hooks import log_create_repository

from kallithea.lib.rcmail.smtp_mailer import SmtpMailer

from kallithea.lib.utils import action_logger

from kallithea.lib.utils2 import ascii_bytes, str2bool

from kallithea.lib.vcs.utils import author_email

from kallithea.model.db import RepoGroup, Repository, Statistics, User

__all__ = ['whoosh_index', 'get_commits_stats', 'send_email']

log = celery.utils.log.get_task_logger(__name__)

@celerylib.task

@celerylib.locked_task

@celerylib.dbsession

def whoosh_index(repo_location, full_index):

    from kallithea.lib.indexers.daemon import WhooshIndexingDaemon

    celerylib.get_session() # initialize database connection

    index_location = config['index_dir']

    WhooshIndexingDaemon(index_location=index_location,

                         repo_location=repo_location) \

                         .run(full_index=full_index)

# for js data compatibility cleans the key for person from '

def akc(k):

    return person(k).replace('"', '')

@celerylib.task

@celerylib.dbsession

def get_commits_stats(repo_name, ts_min_y, ts_max_y, recurse_limit=100):

    DBS = celerylib.get_session()

    lockkey = celerylib.__get_lockkey('get_commits_stats', repo_name, ts_min_y,

                            ts_max_y)

    lockkey_path = config.get('cache_dir') or config['app_conf']['cache_dir']  # Backward compatibility for TurboGears < 2.4

    log.info('running task with lockkey %s', lockkey)

    try:

        lock = celerylib.DaemonLock(os.path.join(lockkey_path, lockkey))

        co_day_auth_aggr = {}

        commits_by_day_aggregate = {}

        repo = Repository.get_by_repo_name(repo_name)

        if repo is None:

            return True

        repo = repo.scm_instance

        repo_size = repo.count()

        # return if repo have no revisions

        if repo_size < 1:

            lock.release()

            return True

        skip_date_limit = True

        parse_limit = int(config.get('commit_parse_limit'))

        last_rev = None

        last_cs = None

        timegetter = itemgetter('time')

        dbrepo = DBS.query(Repository) \

            .filter(Repository.repo_name == repo_name).scalar()

        cur_stats = DBS.query(Statistics) \

            .filter(Statistics.repository == dbrepo).scalar()

        if cur_stats is not None:

            last_rev = cur_stats.stat_on_revision

        if last_rev == repo.get_changeset().revision and repo_size > 1:

            # pass silently without any work if we're not on first revision or

            # current state of parsing revision(from db marker) is the

            # last revision

            lock.release()

            return True

        if cur_stats:

            commits_by_day_aggregate = OrderedDict(ext_json.loads(

                                        cur_stats.commit_activity_combined))

            co_day_auth_aggr = ext_json.loads(cur_stats.commit_activity)

        log.debug('starting parsing %s', parse_limit)

        last_rev = last_rev + 1 if last_rev and last_rev >= 0 else 0

        log.debug('Getting revisions from %s to %s',

             last_rev, last_rev + parse_limit

        )

        for cs in repo[last_rev:last_rev + parse_limit]:

            log.debug('parsing %s', cs)

            last_cs = cs  # remember last parsed changeset

            tt = cs.date.timetuple()

            k = mktime(tt[:3] + (0, 0, 0, 0, 0, 0))

            if akc(cs.author) in co_day_auth_aggr:

                try:

                    l = [timegetter(x) for x in

                         co_day_auth_aggr[akc(cs.author)]['data']]

                    time_pos = l.index(k)

                except ValueError:

                    time_pos = None

                if time_pos is not None and time_pos >= 0:

                    datadict = \

                        co_day_auth_aggr[akc(cs.author)]['data'][time_pos]

                    datadict["commits"] += 1

                    datadict["added"] += len(cs.added)

                    datadict["changed"] += len(cs.changed)

                    datadict["removed"] += len(cs.removed)

                else:

                    if k >= ts_min_y and k <= ts_max_y or skip_date_limit:

                        datadict = {"time": k,

                                    "commits": 1,

                                    "added": len(cs.added),

                                    "changed": len(cs.changed),

                                    "removed": len(cs.removed),

                                   }

                        co_day_auth_aggr[akc(cs.author)]['data'] \

                            .append(datadict)

            else:

                if k >= ts_min_y and k <= ts_max_y or skip_date_limit:

                    co_day_auth_aggr[akc(cs.author)] = {

                                        "label": akc(cs.author),

                                        "data": [{"time": k,

                                                 "commits": 1,

                                                 "added": len(cs.added),

                                                 "changed": len(cs.changed),

                                                 "removed": len(cs.removed),

                                                 }],

                                        "schema": ["commits"],

                                        }

            # gather all data by day

            if k in commits_by_day_aggregate:

                commits_by_day_aggregate[k] += 1

            else:

                commits_by_day_aggregate[k] = 1

        overview_data = sorted(commits_by_day_aggregate.items(),

                               key=itemgetter(0))

        if not co_day_auth_aggr:

            co_day_auth_aggr[akc(repo.contact)] = {

                "label": akc(repo.contact),

                "data": [0, 1],

                "schema": ["commits"],

            }

        stats = cur_stats if cur_stats else Statistics()

        stats.commit_activity = ascii_bytes(ext_json.dumps(co_day_auth_aggr))

        stats.commit_activity_combined = ascii_bytes(ext_json.dumps(overview_data))

        log.debug('last revision %s', last_rev)

        leftovers = len(repo.revisions[last_rev:])

        log.debug('revisions to parse %s', leftovers)

        if last_rev == 0 or leftovers < parse_limit:

            log.debug('getting code trending stats')

            stats.languages = ascii_bytes(ext_json.dumps(__get_codes_stats(repo_name)))

        try:

            stats.repository = dbrepo

            stats.stat_on_revision = last_cs.revision if last_cs else 0

            DBS.add(stats)

            DBS.commit()

        except:

            log.error(traceback.format_exc())

            DBS.rollback()

            lock.release()

            return False

        # final release

        lock.release()

        # execute another task if celery is enabled

        if len(repo.revisions) > 1 and kallithea.CELERY_APP and recurse_limit > 0:

            get_commits_stats(repo_name, ts_min_y, ts_max_y, recurse_limit - 1)

        elif recurse_limit <= 0:

            log.debug('Not recursing - limit has been reached')

        else:

            log.debug('Not recursing')

    except celerylib.LockHeld:

        log.info('Task with key %s already running', lockkey)

        return 'Task with key %s already running' % lockkey

@celerylib.task

@celerylib.dbsession

    """

    Sends an email with defined parameters from the .ini files.

    :param recipients: list of recipients, if this is None, the defined email

        address from field 'email_to' and all admins is used instead

    :param subject: subject of the mail

    :param body: body of the mail

    :param html_body: html version of body

    :param headers: dictionary of prepopulated e-mail headers

    """

    assert isinstance(recipients, list), recipients

    if headers is None:

        headers = {}

    else:

        # do not modify the original headers object passed by the caller

        headers = headers.copy()

    email_config = config

    email_prefix = email_config.get('email_prefix', '')

    if email_prefix:

        subject = "%s %s" % (email_prefix, subject)

    if not recipients:

        # if recipients are not defined we send to email_config + all admins

        recipients = [u.email for u in User.query()

                      .filter(User.admin == True).all()]

        if email_config.get('email_to') is not None:

            recipients += email_config.get('email_to').split(',')

        # If there are still no recipients, there are no admins and no address

        # configured in email_to, so return.

        if not recipients:

            log.error("No recipients specified and no fallback available.")

            return False

        log.warning("No recipients specified for '%s' - sending to admins %s", subject, ' '.join(recipients))

    # SMTP sender

    envelope_from = email_config.get('app_email_from', 'Kallithea')

    # 'From' header

        # In case app_email_from is in "Some Name <e-mail>" format, we first

        # extract the e-mail address.

        envelope_addr = author_email(envelope_from)

        headers['From'] = '"%s" <%s>' % (

            envelope_addr)

    user = email_config.get('smtp_username')

    passwd = email_config.get('smtp_password')

    mail_server = email_config.get('smtp_server')

    mail_port = email_config.get('smtp_port')

    tls = str2bool(email_config.get('smtp_use_tls'))

    ssl = str2bool(email_config.get('smtp_use_ssl'))

    debug = str2bool(email_config.get('debug'))

    smtp_auth = email_config.get('smtp_auth')

    logmsg = ("Mail details:\n"

              "recipients: %s\n"

              "headers: %s\n"

              "subject: %s\n"

              "body:\n%s\n"

              "html:\n%s\n"

              % (' '.join(recipients), headers, subject, body, html_body))

    if mail_server:

        log.debug("Sending e-mail. " + logmsg)

    else:

        log.error("SMTP mail server not configured - cannot send e-mail.")

        log.warning(logmsg)

        return False

    try:

        m = SmtpMailer(envelope_from, user, passwd, mail_server, smtp_auth,

                       mail_port, ssl, tls, debug=debug)

        m.send(recipients, subject, body, html_body, headers=headers)

    except:

        log.error('Mail sending failed')

        log.error(traceback.format_exc())

        return False

    return True

@celerylib.task

@celerylib.dbsession

def create_repo(form_data, cur_user):

    from kallithea.model.repo import RepoModel

    from kallithea.model.db import Setting

    DBS = celerylib.get_session()

    cur_user = User.guess_instance(cur_user)

    owner = cur_user

    repo_name = form_data['repo_name']

    repo_name_full = form_data['repo_name_full']

    repo_type = form_data['repo_type']

    description = form_data['repo_description']

    private = form_data['repo_private']

    clone_uri = form_data.get('clone_uri')

    repo_group = form_data['repo_group']

    landing_rev = form_data['repo_landing_rev']

    copy_fork_permissions = form_data.get('copy_permissions')

    copy_group_permissions = form_data.get('repo_copy_permissions')

    fork_of = form_data.get('fork_parent_id')

    state = form_data.get('repo_state', Repository.STATE_PENDING)

    # repo creation defaults, private and repo_type are filled in form

    defs = Setting.get_default_repo_settings(strip_prefix=True)

    enable_statistics = defs.get('repo_enable_statistics')

    enable_downloads = defs.get('repo_enable_downloads')

    try:

        repo = RepoModel()._create_repo(

            repo_name=repo_name_full,

            repo_type=repo_type,

            description=description,

            owner=owner,

            private=private,

            clone_uri=clone_uri,

            repo_group=repo_group,

            landing_rev=landing_rev,

            fork_of=fork_of,

            copy_fork_permissions=copy_fork_permissions,

            copy_group_permissions=copy_group_permissions,

            enable_statistics=enable_statistics,

            enable_downloads=enable_downloads,

            state=state

        )

        action_logger(cur_user, 'user_created_repo',

                      form_data['repo_name_full'], '')

        DBS.commit()

        # now create this repo on Filesystem

        RepoModel()._create_filesystem_repo(

            repo_name=repo_name,

            repo_type=repo_type,

            repo_group=RepoGroup.guess_instance(repo_group),

            clone_uri=clone_uri,

        )

        repo = Repository.get_by_repo_name(repo_name_full)

        log_create_repository(repo.get_dict(), created_by=owner.username)

        # update repo changeset caches initially

        repo.update_changeset_cache()

        # set new created state

        repo.set_state(Repository.STATE_CREATED)

        DBS.commit()

    except Exception as e:

        log.warning('Exception %s occurred when forking repository, '

                    'doing cleanup...' % e)

        # rollback things manually !

        repo = Repository.get_by_repo_name(repo_name_full)

        if repo:

            Repository.delete(repo.repo_id)

            DBS.commit()

            RepoModel()._delete_filesystem_repo(repo)

        raise

    return True

@celerylib.task

@celerylib.dbsession

def create_repo_fork(form_data, cur_user):

    """

    Creates a fork of repository using interval VCS methods

    :param form_data:

    :param cur_user:

    """

    from kallithea.model.repo import RepoModel

    DBS = celerylib.get_session()

    base_path = Repository.base_path()

    cur_user = User.guess_instance(cur_user)

    repo_name = form_data['repo_name']  # fork in this case

    repo_name_full = form_data['repo_name_full']

    repo_type = form_data['repo_type']

    owner = cur_user

    private = form_data['private']

    clone_uri = form_data.get('clone_uri')

    repo_group = form_data['repo_group']

    landing_rev = form_data['landing_rev']

    copy_fork_permissions = form_data.get('copy_permissions')

    try:

        fork_of = Repository.guess_instance(form_data.get('fork_parent_id'))

        RepoModel()._create_repo(

            repo_name=repo_name_full,

            repo_type=repo_type,

            description=form_data['description'],

            owner=owner,

            private=private,

            clone_uri=clone_uri,

            repo_group=repo_group,

            landing_rev=landing_rev,

            fork_of=fork_of,

            copy_fork_permissions=copy_fork_permissions

        )

        action_logger(cur_user, 'user_forked_repo:%s' % repo_name_full,

                      fork_of.repo_name, '')

        DBS.commit()

        source_repo_path = os.path.join(base_path, fork_of.repo_name)

        # now create this repo on Filesystem

        RepoModel()._create_filesystem_repo(

            repo_name=repo_name,

            repo_type=repo_type,

            repo_group=RepoGroup.guess_instance(repo_group),

            clone_uri=source_repo_path,

        )

        repo = Repository.get_by_repo_name(repo_name_full)

        log_create_repository(repo.get_dict(), created_by=owner.username)

        # update repo changeset caches initially

        repo.update_changeset_cache()

        # set new created state

        repo.set_state(Repository.STATE_CREATED)

        DBS.commit()

    except Exception as e:

        log.warning('Exception %s occurred when forking repository, '

                    'doing cleanup...' % e)

        # rollback things manually !

        repo = Repository.get_by_repo_name(repo_name_full)

        if repo:

            Repository.delete(repo.repo_id)

            DBS.commit()

            RepoModel()._delete_filesystem_repo(repo)

        raise

    return True

def __get_codes_stats(repo_name):

    from kallithea.config.conf import LANGUAGES_EXTENSIONS_MAP

    repo = Repository.get_by_repo_name(repo_name).scm_instance

    tip = repo.get_changeset()

    code_stats = {}

    for _topnode, _dirnodes, filenodes in tip.walk('/'):

        for filenode in filenodes:

            ext = filenode.extension.lower()

            if ext in LANGUAGES_EXTENSIONS_MAP and not filenode.is_binary:

                if ext in code_stats:

                    code_stats[ext] += 1

                else:

                    code_stats[ext] = 1

    return code_stats or {}

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

"""

kallithea.model.notification

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

Model for notifications

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: Nov 20, 2011

:author: marcink

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

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

"""

import datetime

import logging

from tg import app_globals

from tg import tmpl_context as c

from tg.i18n import ugettext as _

from kallithea.lib import helpers as h

from kallithea.model.db import User

log = logging.getLogger(__name__)

class NotificationModel(object):

    TYPE_CHANGESET_COMMENT = 'cs_comment'

    TYPE_MESSAGE = 'message'

    TYPE_MENTION = 'mention' # not used

    TYPE_REGISTRATION = 'registration'

    TYPE_PULL_REQUEST = 'pull_request'

    TYPE_PULL_REQUEST_COMMENT = 'pull_request_comment'

    def create(self, created_by, subject, body, recipients=None,

               type_=TYPE_MESSAGE, with_email=True,

               email_kwargs=None, repo_name=None):

        """

        Creates notification of given type

        :param created_by: int, str or User instance. User who created this

            notification

        :param subject:

        :param body:

        :param recipients: list of int, str or User objects, when None

            is given send to all admins

        :param type_: type of notification

        :param with_email: send email with this notification

        :param email_kwargs: additional dict to pass as args to email template

        """

        from kallithea.lib.celerylib import tasks

        email_kwargs = email_kwargs or {}

        if recipients and not getattr(recipients, '__iter__', False):

            raise Exception('recipients must be a list or iterable')

        created_by_obj = User.guess_instance(created_by)

        recipients_objs = set()

        if recipients:

            for u in recipients:

                obj = User.guess_instance(u)

                if obj is not None:

                    recipients_objs.add(obj)

                else:

                    # TODO: inform user that requested operation couldn't be completed

                    log.error('cannot email unknown user %r', u)

            log.debug('sending notifications %s to %s',

                type_, recipients_objs

            )

        elif recipients is None:

            # empty recipients means to all admins

            recipients_objs = User.query().filter(User.admin == True).all()

            log.debug('sending notifications %s to admins: %s',

                type_, recipients_objs

            )

        #else: silently skip notification mails?

        if not with_email:

            return

        headers = {}

        headers['X-Kallithea-Notification-Type'] = type_

        if 'threading' in email_kwargs:

            headers['References'] = ' '.join('<%s>' % x for x in email_kwargs['threading'])

        # this is passed into template

        created_on = h.fmt_date(datetime.datetime.now())

        html_kwargs = {

                  'subject': subject,

                  'body': h.render_w_mentions(body, repo_name),

                  'when': created_on,

                  'user': created_by_obj.username,

                  }

        txt_kwargs = {

                  'subject': subject,

                  'body': body,

                  'when': created_on,

                  'user': created_by_obj.username,

                  }

        html_kwargs.update(email_kwargs)

        txt_kwargs.update(email_kwargs)

        email_subject = EmailNotificationModel() \

                            .get_email_description(type_, **txt_kwargs)

        email_txt_body = EmailNotificationModel() \

                            .get_email_tmpl(type_, 'txt', **txt_kwargs)

        email_html_body = EmailNotificationModel() \

                            .get_email_tmpl(type_, 'html', **html_kwargs)

        # don't send email to person who created this comment

        rec_objs = set(recipients_objs).difference(set([created_by_obj]))

        # send email with notification to all other participants

        for rec in rec_objs:

            tasks.send_email([rec.email], email_subject, email_txt_body,

class EmailNotificationModel(object):

    TYPE_CHANGESET_COMMENT = NotificationModel.TYPE_CHANGESET_COMMENT

    TYPE_MESSAGE = NotificationModel.TYPE_MESSAGE # only used for testing

    # NotificationModel.TYPE_MENTION is not used

    TYPE_PASSWORD_RESET = 'password_link'

    TYPE_REGISTRATION = NotificationModel.TYPE_REGISTRATION

    TYPE_PULL_REQUEST = NotificationModel.TYPE_PULL_REQUEST

    TYPE_PULL_REQUEST_COMMENT = NotificationModel.TYPE_PULL_REQUEST_COMMENT

    TYPE_DEFAULT = 'default'

    def __init__(self):

        super(EmailNotificationModel, self).__init__()

        self._tmpl_lookup = app_globals.mako_lookup

        self.email_types = {

            self.TYPE_CHANGESET_COMMENT: 'changeset_comment',

            self.TYPE_PASSWORD_RESET: 'password_reset',

            self.TYPE_REGISTRATION: 'registration',

            self.TYPE_DEFAULT: 'default',

            self.TYPE_PULL_REQUEST: 'pull_request',

            self.TYPE_PULL_REQUEST_COMMENT: 'pull_request_comment',

        }

        self._subj_map = {

            self.TYPE_CHANGESET_COMMENT: _('[Comment] %(repo_name)s changeset %(short_id)s "%(message_short)s" on %(branch)s'),

            self.TYPE_MESSAGE: 'Test Message',

            # self.TYPE_PASSWORD_RESET

            self.TYPE_REGISTRATION: _('New user %(new_username)s registered'),

            # self.TYPE_DEFAULT

            self.TYPE_PULL_REQUEST: _('[Review] %(repo_name)s PR %(pr_nice_id)s "%(pr_title_short)s" from %(pr_source_branch)s by %(pr_owner_username)s'),

            self.TYPE_PULL_REQUEST_COMMENT: _('[Comment] %(repo_name)s PR %(pr_nice_id)s "%(pr_title_short)s" from %(pr_source_branch)s by %(pr_owner_username)s'),

        }

    def get_email_description(self, type_, **kwargs):

        """

        return subject for email based on given type

        """

        tmpl = self._subj_map[type_]

        try:

            subj = tmpl % kwargs

        except KeyError as e:

            log.error('error generating email subject for %r from %s: %s', type_, ', '.join(self._subj_map), e)

            raise

        # gmail doesn't do proper threading but will ignore leading square

        # bracket content ... so that is where we put status info

        bracket_tags = []

        status_change = kwargs.get('status_change')

        if status_change:

            bracket_tags.append(str(status_change))  # apply str to evaluate LazyString before .join

        if kwargs.get('closing_pr'):

            bracket_tags.append(_('Closing'))

        if bracket_tags:

            if subj.startswith('['):

                subj = '[' + ', '.join(bracket_tags) + ': ' + subj[1:]

            else:

                subj = '[' + ', '.join(bracket_tags) + '] ' + subj

        return subj

    def get_email_tmpl(self, type_, content_type, **kwargs):

        """

        return generated template for email based on given type

        """

        base = 'email_templates/' + self.email_types.get(type_, self.email_types[self.TYPE_DEFAULT]) + '.' + content_type

        email_template = self._tmpl_lookup.get_template(base)

        # translator and helpers inject

        _kwargs = {'_': _,

                   'h': h,

                   'c': c}

        _kwargs.update(kwargs)

        if content_type == 'html':

            _kwargs.update({

                "color_text": "#202020",

                "color_emph": "#395fa0",

                "color_link": "#395fa0",

                "color_border": "#ddd",

                "color_background_grey": "#f9f9f9",

                "color_button": "#395fa0",

                "monospace_style": "font-family:Lucida Console,Consolas,Monaco,Inconsolata,Liberation Mono,monospace",

                "sans_style": "font-family:Helvetica,Arial,sans-serif",

                })

            _kwargs.update({

                "default_style": "%(sans_style)s;font-weight:200;font-size:14px;line-height:17px;color:%(color_text)s" % _kwargs,

                "comment_style": "%(monospace_style)s;white-space:pre-wrap" % _kwargs,

                "data_style": "border:%(color_border)s 1px solid;background:%(color_background_grey)s" % _kwargs,

                "emph_style": "font-weight:600;color:%(color_emph)s" % _kwargs,

                "link_style": "color:%(color_link)s;text-decoration:none" % _kwargs,

                "link_text_style": "color:%(color_text)s;text-decoration:none;border:%(color_border)s 1px solid;background:%(color_background_grey)s" % _kwargs,

                })

        log.debug('rendering tmpl %s with kwargs %s', base, _kwargs)

        return email_template.render_unicode(**_kwargs)

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

import re

import time

import urllib.parse

import mock

from tg.util.webtest import test_context

import kallithea.lib.celerylib.tasks

from kallithea.lib import helpers as h

from kallithea.lib.auth import check_password

from kallithea.lib.utils2 import generate_api_key

from kallithea.model import validators

from kallithea.model.api_key import ApiKeyModel

from kallithea.model.db import User

from kallithea.model.meta import Session

from kallithea.model.user import UserModel

from kallithea.tests import base

from kallithea.tests.fixture import Fixture

fixture = Fixture()

class TestLoginController(base.TestController):

    def test_index(self):

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

        assert response.status == '200 OK'

        # Test response...

    def test_login_admin_ok(self):

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

                                 {'username': base.TEST_USER_ADMIN_LOGIN,

                                  'password': base.TEST_USER_ADMIN_PASS,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        assert response.status == '302 Found'

        self.assert_authenticated_user(response, base.TEST_USER_ADMIN_LOGIN)

        response = response.follow()

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

    def test_login_regular_ok(self):

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

                                 {'username': base.TEST_USER_REGULAR_LOGIN,

                                  'password': base.TEST_USER_REGULAR_PASS,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        assert response.status == '302 Found'

        self.assert_authenticated_user(response, base.TEST_USER_REGULAR_LOGIN)

        response = response.follow()

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

    def test_login_regular_email_ok(self):

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

                                 {'username': base.TEST_USER_REGULAR_EMAIL,

                                  'password': base.TEST_USER_REGULAR_PASS,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        assert response.status == '302 Found'

        self.assert_authenticated_user(response, base.TEST_USER_REGULAR_LOGIN)

        response = response.follow()

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

    def test_login_ok_came_from(self):

        test_came_from = '/_admin/users'

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

                                     came_from=test_came_from),

                                 {'username': base.TEST_USER_ADMIN_LOGIN,

                                  'password': base.TEST_USER_ADMIN_PASS,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        assert response.status == '302 Found'

        response = response.follow()

        assert response.status == '200 OK'

        response.mustcontain('Users Administration')

    def test_login_do_not_remember(self):

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

                                 {'username': base.TEST_USER_REGULAR_LOGIN,

                                  'password': base.TEST_USER_REGULAR_PASS,

                                  'remember': False,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        assert 'Set-Cookie' in response.headers

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

            assert not 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(base.url(controller='login', action='index'),

                                 {'username': base.TEST_USER_REGULAR_LOGIN,

                                  'password': base.TEST_USER_REGULAR_PASS,

                                  'remember': True,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        assert 'Set-Cookie' in response.headers

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

            assert 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(base.url(controller='login', action='index'),

                                 {'username': base.TEST_USER_REGULAR_LOGIN,

                                  'password': base.TEST_USER_REGULAR_PASS,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        # Verify that a login session has been established.

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

        response = response.follow()

        assert 'authuser' in response.session

        response.click('Log Out')

        # Verify that the login session has been terminated.

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

        assert 'authuser' not in response.session

    @base.parametrize('url_came_from', [

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

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

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

          ('ftp://ftp.example.com',),

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

          ('//evil.example.com/',),

          ('/\r\nX-Header-Injection: boo',),

          ('/invälid_url_bytes',),

          ('non-absolute-path',),

    ])

    def test_login_bad_came_froms(self, url_came_from):

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

                                     came_from=url_came_from),