import os

import re

import mock

import routes.util

from kallithea.tests.base import *

from kallithea.lib import helpers as h

from kallithea.model.db import User, Notification, UserNotification

from kallithea.model.user import UserModel

from kallithea.model.meta import Session

from kallithea.model.notification import NotificationModel, EmailNotificationModel

import kallithea.lib.celerylib

import kallithea.lib.celerylib.tasks

from tg.util.webtest import test_context

class TestNotifications(TestController):

    def setup_method(self, method):

        Session.remove()

        u1 = UserModel().create_or_update(username=u'u1',

                                        password=u'qweqwe',

                                        email=u'u1@example.com',

                                        firstname=u'u1', lastname=u'u1')

        Session().commit()

        self.u1 = u1.user_id

        u2 = UserModel().create_or_update(username=u'u2',

                                        password=u'qweqwe',

                                        email=u'u2@example.com',

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

        Session().commit()

        self.u2 = u2.user_id

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

                                        password=u'qweqwe',

                                        email=u'u3@example.com',

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

        Session().commit()

        self.u3 = u3.user_id

        self.remove_all_notifications()

        assert [] == Notification.query().all()

        assert [] == UserNotification.query().all()

    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, author=None):

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

                assert subject == 'Test Message'

                assert body == u"hi there"

                assert '>hi there<' in html_body

                assert author.username == 'u1'

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

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

                                                   subject=u'subj', body=u'hi there',

                                                   recipients=usrs)

                Session().commit()

                u1 = User.get(self.u1)

                u2 = User.get(self.u2)

                u3 = User.get(self.u3)

                notifications = Notification.query().all()

                assert len(notifications) == 1

                assert notifications[0].recipients == [u1, u2]

                assert notification.notification_id == notifications[0].notification_id

                unotification = UserNotification.query() \

                    .filter(UserNotification.notification == notification).all()

                assert len(unotification) == len(usrs)

                assert set([x.user_id for x in unotification]) == set(usrs)

    def test_user_notifications(self):

        with test_context(self.app):

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

                                                subject=u'subj', body=u'hi there1',

                                                recipients=[self.u3])

            Session().commit()

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

                                                subject=u'subj', body=u'hi there2',

                                                recipients=[self.u3])

            Session().commit()

            u3 = Session().query(User).get(self.u3)

            assert sorted([x.notification for x in u3.notifications]) == sorted([notification2, notification1])

    def test_delete_notifications(self):

        with test_context(self.app):

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

                                               subject=u'title', body=u'hi there3',

                                        recipients=[self.u3, self.u1, self.u2])

            Session().commit()

            notifications = Notification.query().all()

            assert notification in notifications

            Notification.delete(notification.notification_id)

            Session().commit()

            notifications = Notification.query().all()

            assert notification not in notifications

            un = UserNotification.query().filter(UserNotification.notification

                                                 == notification).all()

            assert un == []

    def test_delete_association(self):

        with test_context(self.app):

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

                                               subject=u'title', body=u'hi there3',

                                        recipients=[self.u3, self.u1, self.u2])

            Session().commit()

            unotification = UserNotification.query() \

                                .filter(UserNotification.notification ==

                                        notification) \

                                .filter(UserNotification.user_id == self.u3) \

                                .scalar()

            assert unotification.user_id == self.u3

            NotificationModel().delete(self.u3,

                                       notification.notification_id)

            Session().commit()

            u3notification = UserNotification.query() \

                                .filter(UserNotification.notification ==

                                        notification) \

                                .filter(UserNotification.user_id == self.u3) \

                                .scalar()

            assert u3notification == None

            # notification object is still there

            assert Notification.query().all() == [notification]

            # u1 and u2 still have assignments

            u1notification = UserNotification.query() \

                                .filter(UserNotification.notification ==

                                        notification) \

                                .filter(UserNotification.user_id == self.u1) \

                                .scalar()

            assert u1notification != None

            u2notification = UserNotification.query() \

                                .filter(UserNotification.notification ==

                                        notification) \

                                .filter(UserNotification.user_id == self.u2) \

                                .scalar()

            assert u2notification != None

    def test_notification_counter(self):

        with test_context(self.app):

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

                                subject=u'title', body=u'hi there_delete',

                                recipients=[self.u3, self.u1])

            Session().commit()

            assert NotificationModel().get_unread_cnt_for_user(self.u1) == 0

            assert NotificationModel().get_unread_cnt_for_user(self.u2) == 0

            assert NotificationModel().get_unread_cnt_for_user(self.u3) == 1

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

                                               subject=u'title', body=u'hi there3',

                                        recipients=[self.u3, self.u1, self.u2])

            Session().commit()

            assert NotificationModel().get_unread_cnt_for_user(self.u1) == 0

            assert NotificationModel().get_unread_cnt_for_user(self.u2) == 1

            assert NotificationModel().get_unread_cnt_for_user(self.u3) == 2

    @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, author=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\n' % author.username)

            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(

                    pr_nice_id='#7',

                    pr_title='The Title',

                    pr_title_short='The Title',

                    pr_url='http://pr.org/7',

                    pr_target_repo='http://mainline.com/repo',

                    pr_target_branch='trunk',

                    pr_source_repo='https://dev.org/repo',

                    pr_source_branch='devbranch',

                    pr_owner=User.get(self.u2),

                    pr_owner_username='u2'

                    )

                for type_, body, kwargs in [

                    kwargs['repo_name'] = u'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

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

                                                           subject=u'unused', body=body, email_kwargs=kwargs,

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

                    author=User.get(self.u1))

        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)