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

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

comments model for Kallithea

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 11, 2011

:author: marcink

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

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

"""

import logging

from collections import defaultdict

from kallithea.lib import webutils

from kallithea.lib.utils import extract_mentioned_users

from kallithea.lib.utils2 import shorter

from kallithea.model import db, meta, notification

log = logging.getLogger(__name__)

def _list_changeset_commenters(revision):

    return (meta.Session().query(db.User)

        .join(db.ChangesetComment.author)

        .filter(db.ChangesetComment.revision == revision)

        .all())

def _list_pull_request_commenters(pull_request):

    return (meta.Session().query(db.User)

        .join(db.ChangesetComment.author)

        .filter(db.ChangesetComment.pull_request_id == pull_request.pull_request_id)

        .all())

class ChangesetCommentsModel(object):

    def create_notification(self, repo, comment, author, comment_text,

                                line_no=None, revision=None, pull_request=None,

                                status_change=None, closing_pr=False):

        # changeset

        if revision:

            notification_type = notification.NotificationModel.TYPE_CHANGESET_COMMENT

            cs = repo.scm_instance.get_changeset(revision)

            threading = ['%s-rev-%s@%s' % (repo.repo_name, revision, webutils.canonical_hostname())]

            if line_no: # TODO: url to file _and_ line number

                threading.append('%s-rev-%s-line-%s@%s' % (repo.repo_name, revision, line_no,

                                                           webutils.canonical_hostname()))

            comment_url = webutils.canonical_url('changeset_home',

                repo_name=repo.repo_name,

                revision=revision,

                anchor='comment-%s' % comment.comment_id)

            # get the current participants of this changeset

            recipients = _list_changeset_commenters(revision)

            # add changeset author if it's known locally

            cs_author = db.User.get_from_cs_author(cs.author)

            if not cs_author:

                # use repo owner if we cannot extract the author correctly

                # FIXME: just use committer name even if not a user

                cs_author = repo.owner

            recipients.append(cs_author)

            email_kwargs = {

                'status_change': status_change,

                'cs_comment_user': author.full_name_and_username,

                'cs_target_repo': webutils.canonical_url('summary_home', repo_name=repo.repo_name),

                'cs_comment_url': comment_url,

                'cs_url': webutils.canonical_url('changeset_home', repo_name=repo.repo_name, revision=revision),

                'raw_id': revision,

                'message': cs.message,

                'message_short': shorter(cs.message, 50, firstline=True),

                'cs_author': cs_author,

                'cs_author_username': cs_author.username,

                'repo_name': repo.repo_name,

                'short_id': revision[:12],

                'branch': cs.branch,

                'comment_username': author.username,

                'threading': threading,

            }

        # pull request

        elif pull_request:

            notification_type = notification.NotificationModel.TYPE_PULL_REQUEST_COMMENT

            _org_ref_type, org_ref_name, _org_rev = comment.pull_request.org_ref.split(':')

            _other_ref_type, other_ref_name, _other_rev = comment.pull_request.other_ref.split(':')

            threading = ['%s-pr-%s@%s' % (pull_request.other_repo.repo_name,

                                          pull_request.pull_request_id,

                                          webutils.canonical_hostname())]

            if line_no: # TODO: url to file _and_ line number

                threading.append('%s-pr-%s-line-%s@%s' % (pull_request.other_repo.repo_name,

                                                          pull_request.pull_request_id, line_no,

                                                          webutils.canonical_hostname()))

            comment_url = pull_request.url(canonical=True,

                anchor='comment-%s' % comment.comment_id)

            # get the current participants of this pull request

            recipients = _list_pull_request_commenters(pull_request)

            recipients.append(pull_request.owner)

            recipients += pull_request.get_reviewer_users()

            # set some variables for email notification

            email_kwargs = {

                'pr_title': pull_request.title,

                'pr_title_short': shorter(pull_request.title, 50),

                'pr_nice_id': pull_request.nice_id(),

                'status_change': status_change,

                'closing_pr': closing_pr,

                'pr_comment_url': comment_url,

                'pr_url': pull_request.url(canonical=True),

                'pr_comment_user': author.full_name_and_username,

                'pr_target_repo': webutils.canonical_url('summary_home',

                                   repo_name=pull_request.other_repo.repo_name),

                'pr_target_branch': other_ref_name,

                'pr_source_repo': webutils.canonical_url('summary_home',

                                   repo_name=pull_request.org_repo.repo_name),

                'pr_source_branch': org_ref_name,

                'pr_owner': pull_request.owner,

                'pr_owner_username': pull_request.owner.username,

                'repo_name': pull_request.other_repo.repo_name,

                'comment_username': author.username,

                'threading': threading,

            }

        email_kwargs['is_mention'] = False

        # create notification objects, and emails

        notification.NotificationModel().create(

            recipients=recipients, type_=notification_type,

            email_kwargs=email_kwargs,

        )

        mention_recipients = extract_mentioned_users(comment_text).difference(recipients)

        if mention_recipients:

            email_kwargs['is_mention'] = True

            notification.NotificationModel().create(

                recipients=mention_recipients,

                type_=notification_type,

                email_kwargs=email_kwargs

            )

    def create(self, text, repo, author, revision=None, pull_request=None,

               f_path=None, line_no=None, status_change=None, closing_pr=False,

               send_email=True):

        """

        Creates a new comment for either a changeset or a pull request.

        status_change and closing_pr is only for the optional email.

        Returns the created comment.

        """

        if not status_change and not text:

            log.warning('Missing text for comment, skipping...')

            return None

        repo = db.Repository.guess_instance(repo)

        author = db.User.guess_instance(author)

        comment = db.ChangesetComment()

        comment.repo = repo

        comment.author = author

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.utils2 import fmt_date

from kallithea.model import db

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'

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

        """

        import kallithea.lib.helpers as h

        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 = db.User.guess_instance(created_by)

        recipients_objs = set()

        if recipients:

            for u in recipients:

                obj = db.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 = db.User.query().filter(db.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 = fmt_date(datetime.datetime.now())

        html_kwargs = {

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

                  'when': created_on,

                  'user': created_by_obj.username,

                  }

        txt_kwargs = {

                  '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 the person who caused the notification, except for

        # notifications about new pull requests where the author is explicitly

        # added.

        rec_mails = set(obj.email for obj in recipients_objs)

        if type_ == NotificationModel.TYPE_PULL_REQUEST:

            rec_mails.add(created_by_obj.email)

        else:

            rec_mails.discard(created_by_obj.email)

        # send email with notification to participants

            mention_recipients = set(mention_recipients) - reviewers

            _assert_valid_reviewers(mention_recipients)

        redundant_reviewers = set(db.User.query() \

            .join(db.PullRequestReviewer) \

            .filter(db.PullRequestReviewer.pull_request == pr) \

            .filter(db.PullRequestReviewer.user_id.in_(r.user_id for r in reviewers))

            .all())

        if redundant_reviewers:

            log.debug('Following reviewers were already part of pull request %s: %s', pr.pull_request_id, redundant_reviewers)

            reviewers -= redundant_reviewers

        log.debug('Adding reviewers to pull request %s: %s', pr.pull_request_id, reviewers)

        for reviewer in reviewers:

            prr = db.PullRequestReviewer(reviewer, pr)

            meta.Session().add(prr)

        # notification to reviewers

        pr_url = pr.url(canonical=True)

        threading = ['%s-pr-%s@%s' % (pr.other_repo.repo_name,

                                      pr.pull_request_id,

                                      webutils.canonical_hostname())]

        body = pr.description

        _org_ref_type, org_ref_name, _org_rev = pr.org_ref.split(':')

        _other_ref_type, other_ref_name, _other_rev = pr.other_ref.split(':')

        revision_data = [(x.raw_id, x.message)

                         for x in map(pr.org_repo.get_changeset, pr.revisions)]

        email_kwargs = {

            'pr_title': pr.title,

            'pr_title_short': shorter(pr.title, 50),

            'pr_user_created': user.full_name_and_username,

            'pr_repo_url': webutils.canonical_url('summary_home', repo_name=pr.other_repo.repo_name),

            'pr_url': pr_url,

            'pr_revisions': revision_data,

            'repo_name': pr.other_repo.repo_name,

            'org_repo_name': pr.org_repo.repo_name,

            'pr_nice_id': pr.nice_id(),

            'pr_target_repo': webutils.canonical_url('summary_home',

                               repo_name=pr.other_repo.repo_name),

            'pr_target_branch': other_ref_name,

            'pr_source_repo': webutils.canonical_url('summary_home',

                               repo_name=pr.org_repo.repo_name),

            'pr_source_branch': org_ref_name,

            'pr_owner': pr.owner,

            'pr_owner_username': pr.owner.username,

            'pr_username': user.username,

            'threading': threading,

            'is_mention': False,

            }

        if reviewers:

                                       recipients=reviewers,

                                       type_=notification.NotificationModel.TYPE_PULL_REQUEST,

                                       email_kwargs=email_kwargs)

        if mention_recipients:

            email_kwargs['is_mention'] = True

                                       recipients=mention_recipients,

                                       type_=notification.NotificationModel.TYPE_PULL_REQUEST,

                                       email_kwargs=email_kwargs)

        return reviewers, redundant_reviewers

    def mention_from_description(self, user, pr, old_description=''):

        mention_recipients = (extract_mentioned_users(pr.description) -

                              extract_mentioned_users(old_description))

        log.debug("Mentioning %s", mention_recipients)

        self.add_reviewers(user, pr, set(), mention_recipients)

    def remove_reviewers(self, user, pull_request, reviewers):

        """Remove specified users from being reviewers of the PR."""

        if not reviewers:

            return # avoid SQLAlchemy warning about empty sequence for IN-predicate

        db.PullRequestReviewer.query() \

            .filter_by(pull_request=pull_request) \

            .filter(db.PullRequestReviewer.user_id.in_(r.user_id for r in reviewers)) \

            .delete(synchronize_session='fetch') # the default of 'evaluate' is not available

    def delete(self, pull_request):

                new_user.password = get_crypt_password(password) \

                    if password else ''

            if user is None:

                meta.Session().add(new_user)

                meta.Session().flush() # make database assign new_user.user_id

            if not edit:

                hooks.log_create_user(new_user.get_dict(), cur_user)

            return new_user

        except (DatabaseError,):

            log.error(traceback.format_exc())

            raise

    def create_registration(self, form_data):

        from kallithea.model import notification

        form_data['admin'] = False

        form_data['extern_type'] = db.User.DEFAULT_AUTH_TYPE

        form_data['extern_name'] = ''

        new_user = self.create(form_data)

        # notification to admins

        body = (

            'New user registration\n'

            '---------------------\n'

            '- Username: {user.username}\n'

            '- Full Name: {user.full_name}\n'

            '- Email: {user.email}\n'

            ).format(user=new_user)

        edit_url = webutils.canonical_url('edit_user', id=new_user.user_id)

        email_kwargs = {

            'registered_user_url': edit_url,

            'new_username': new_user.username,

            'new_email': new_user.email,

            'new_full_name': new_user.full_name}

                                   body=body, recipients=None,

                                   type_=notification.NotificationModel.TYPE_REGISTRATION,

                                   email_kwargs=email_kwargs)

    def update(self, user_id, form_data, skip_attrs=None):

        skip_attrs = skip_attrs or []

        user = self.get(user_id)

        if user.is_default_user:

            raise DefaultUserException(

                            _("You can't edit this user since it's "

                              "crucial for entire application"))

        for k, v in form_data.items():

            if k in skip_attrs:

                continue

            if k == 'new_password' and v:

                user.password = get_crypt_password(v)

            else:

                # old legacy thing orm models store firstname as name,

                # need proper refactor to username

                if k == 'firstname':

                    k = 'name'

                setattr(user, k, v)

                                        email='u2@example.com',

                                        firstname='u2', lastname='u3')

        meta.Session().commit()

        self.u2 = u2.user_id

        u3 = UserModel().create_or_update(username='u3',

                                        password='qweqwe',

                                        email='u3@example.com',

                                        firstname='u3', lastname='u3')

        meta.Session().commit()

        self.u3 = u3.user_id

    def test_create_notification(self):

        with test_context(self.app):

            usrs = [self.u1, self.u2]

            def send_email(recipients, subject, body='', html_body='', headers=None, from_name=None):

                assert recipients == ['u2@example.com']

                assert subject == 'Test Message'

                assert body == "hi there"

                assert '>hi there<' in html_body

                assert from_name == 'u1 u1'

            with mock.patch.object(kallithea.lib.celerylib.tasks, 'send_email', send_email):

                NotificationModel().create(created_by=self.u1,

                                                   recipients=usrs)

    @mock.patch.object(h, 'canonical_url', (lambda arg, **kwargs: 'http://%s/?%s' % (arg, '&'.join('%s=%s' % (k, v) for (k, v) in sorted(kwargs.items())))))

    def test_dump_html_mails(self):

        # Exercise all notification types and dump them to one big html file

        l = []

        def send_email(recipients, subject, body='', html_body='', headers=None, from_name=None):

            l.append('<hr/>\n')

            l.append('<h1>%s</h1>\n' % desc) # desc is from outer scope

            l.append('<pre>\n')

            l.append('From: %s <name@example.com>\n' % from_name)

            l.append('To: %s\n' % ' '.join(recipients))

            l.append('Subject: %s\n' % subject)

            l.append('</pre>\n')

            l.append('<hr/>\n')

            l.append('<pre>%s</pre>\n' % body)

            l.append('<hr/>\n')

            l.append(html_body)

            l.append('<hr/>\n')

        with test_context(self.app):

            with mock.patch.object(kallithea.lib.celerylib.tasks, 'send_email', send_email):

                pr_kwargs = dict(

                         dict(

                            closing_pr=[False, True],

                            is_mention=[False, True],

                            pr_comment_user='Opinionated User (jsmith)',

                            pr_comment_url='http://pr.org/comment',

                            status_change=[None, 'Under Review'],

                            **pr_kwargs)),

                        ]:

                    kwargs['repo_name'] = 'repo/name'

                    params = [(type_, type_, body, kwargs)]

                    for param_name in ['is_mention', 'status_change', 'closing_pr']: # TODO: inline/general

                        if not isinstance(kwargs.get(param_name), list):

                            continue

                        new_params = []

                        for v in kwargs[param_name]:

                            for desc, type_, body, kwargs in params:

                                kwargs = dict(kwargs)

                                kwargs[param_name] = v

                                new_params.append(('%s, %s=%r' % (desc, param_name, v), type_, body, kwargs))

                        params = new_params

                    for desc, type_, body, kwargs in params:

                        # desc is used as "global" variable

                        NotificationModel().create(created_by=self.u1,

                                                           recipients=[self.u2], type_=type_)

                # Email type TYPE_PASSWORD_RESET has no corresponding notification type - test it directly:

                desc = 'TYPE_PASSWORD_RESET'

                kwargs = dict(user='John Doe', reset_token='decbf64715098db5b0bd23eab44bd792670ab746', reset_url='http://reset.com/decbf64715098db5b0bd23eab44bd792670ab746')

                kallithea.lib.celerylib.tasks.send_email(['john@doe.com'],

                    "Password reset link",

                    EmailNotificationModel().get_email_tmpl(EmailNotificationModel.TYPE_PASSWORD_RESET, 'txt', **kwargs),

                    EmailNotificationModel().get_email_tmpl(EmailNotificationModel.TYPE_PASSWORD_RESET, 'html', **kwargs),

                    from_name=db.User.get(self.u1).full_name_or_username)

        out = '<!doctype html>\n<html lang="en">\n<head><title>Notifications</title><meta http-equiv="Content-Type" content="text/html; charset=UTF-8"></head>\n<body>\n%s\n</body>\n</html>\n' % \

            re.sub(r'<(/?(?:!doctype|html|head|title|meta|body)\b[^>]*)>', r'<!--\1-->', ''.join(l))

        outfn = os.path.join(os.path.dirname(__file__), 'test_dump_html_mails.out.html')

        reffn = os.path.join(os.path.dirname(__file__), 'test_dump_html_mails.ref.html')

        with open(outfn, 'w') as f:

            f.write(out)

        with open(reffn) as f:

            ref = f.read()

        assert ref == out # copy test_dump_html_mails.out.html to test_dump_html_mails.ref.html to update expectations

        os.unlink(outfn)