Changeset - cd6c577ade97
Parent rev.
Child rev.
[Not reviewed]
default
0 4 0
Thomas De Schampheleire - 7 years ago 2018-12-05 21:27:58
thomas.de_schampheleire@nokia.com
model: remove UI notification feature

This commit is part of the removal of the UI notification feature from
Kallithea, which is not deemed useful in its current form. Only email
notifications are preserved.
4 files changed with 13 insertions and 213 deletions:
docs/api/models.rst
3
kallithea/model/db.py
3
71
kallithea/model/notification.py
8
137
kallithea/tests/models/test_notifications.py
2
2
0 comments (0 inline, 0 general)
docs/api/models.rst
➞
Show inline comments
 
.. _models:
 

			




	
	
	
		
 
========================

			




	
	
	
		
 
The :mod:`models` module

			




	
	
	
		
 
========================

			




	
	
	
		
 

			




	
	
	
		
 
.. automodule:: kallithea.model

			




	
	
	
		
 
   :members:

			




	
	
	
		
 

			




	
	
	
		
 
.. automodule:: kallithea.model.comment

			




	
	
	
		
 
   :members:

			




	
	
	
		
 

			




	
	
	
		
 
.. automodule:: kallithea.model.notification

			




	
	
	
		
 
   :members:

			




	
	
	
		
 

			




	
	
	
		
 
.. automodule:: kallithea.model.permission

			




	
	
	
		
 
   :members:

			




	
	
	
		
 

			




	
	
	
		
 
.. automodule:: kallithea.model.repo_permission

			




	
	
	
		
 
   :members:

			




	
	
	
		
 

			




	
	
	
		
 
.. automodule:: kallithea.model.repo

			




	
	
	
		
 
   :members:

			




	
	
	
		
 

			




	
	
	
		
 
.. automodule:: kallithea.model.repo_group

			




	
	
	
		
 
   :members:

			




	
	
	
		
 

			




        

    


    
    

    

        

                

                    kallithea/model/db.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -443,27 +443,24 @@ class User(Base, BaseDbModel):

			




	
	
	
		
 

			




	
	
	
		
 
    repositories = relationship('Repository')

			




	
	
	
		
 
    repo_groups = relationship('RepoGroup')

			




	
	
	
		
 
    user_groups = relationship('UserGroup')

			




	
	
	
		
 
    user_followers = relationship('UserFollowing', primaryjoin='UserFollowing.follows_user_id==User.user_id', cascade='all')

			




	
	
	
		
 
    followings = relationship('UserFollowing', primaryjoin='UserFollowing.user_id==User.user_id', cascade='all')

			




	
	
	
		
 

			




	
	
	
		
 
    repo_to_perm = relationship('UserRepoToPerm', primaryjoin='UserRepoToPerm.user_id==User.user_id', cascade='all')

			




	
	
	
		
 
    repo_group_to_perm = relationship('UserRepoGroupToPerm', primaryjoin='UserRepoGroupToPerm.user_id==User.user_id', cascade='all')

			




	
	
	
		
 

			




	
	
	
		
 
    group_member = relationship('UserGroupMember', cascade='all')

			




	
	
	
		
 

			




	
	
	
		
 
    notifications = relationship('UserNotification', cascade='all')

			




	
	
	
		
 
    # notifications assigned to this user

			




	
	
	
		
 
    user_created_notifications = relationship('Notification', cascade='all')

			




	
	
	
		
 
    # comments created by this user

			




	
	
	
		
 
    user_comments = relationship('ChangesetComment', cascade='all')

			




	
	
	
		
 
    # extra emails for this user

			




	
	
	
		
 
    user_emails = relationship('UserEmailMap', cascade='all')

			




	
	
	
		
 
    # extra API keys

			




	
	
	
		
 
    user_api_keys = relationship('UserApiKeys', cascade='all')

			




	
	
	
		
 

			




	
	
	
		
 
    @hybrid_property

			




	
	
	
		
 
    def email(self):

			




	
	
	
		
 
        return self._email

			




	
	
	
		
 

			




	
	
	
		
 
    @email.setter

			




	
	
		
 
@@ -2469,102 +2466,37 @@ class PullRequestReviewer(Base, BaseDbMo

			




	
	
	
		
 
    pull_request_id = Column(Integer(), ForeignKey('pull_requests.pull_request_id'), nullable=False)

			




	
	
	
		
 
    user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)

			




	
	
	
		
 

			




	
	
	
		
 
    user = relationship('User')

			




	
	
	
		
 
    pull_request = relationship('PullRequest')

			




	
	
	
		
 

			




	
	
	
		
 
    def __json__(self):

			




	
	
	
		
 
        return dict(

			




	
	
	
		
 
            username=self.user.username if self.user else None,

			




	
	
	
		
 
        )

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class Notification(Base, BaseDbModel):

			




	
	
	
		
 
class Notification(object):

			




	
	
	
		
 
    __tablename__ = 'notifications'

			




	
	
	
		
 
    __table_args__ = (

			




	
	
	
		
 
        Index('notification_type_idx', 'type'),

			




	
	
	
		
 
        _table_args_default_dict,

			




	
	
	
		
 
    )

			




	
	
	
		
 

			




	
	
	
		
 
    TYPE_CHANGESET_COMMENT = u'cs_comment'

			




	
	
	
		
 
    TYPE_MESSAGE = u'message'

			




	
	
	
		
 
    TYPE_MENTION = u'mention' # not used

			




	
	
	
		
 
    TYPE_REGISTRATION = u'registration'

			




	
	
	
		
 
    TYPE_PULL_REQUEST = u'pull_request'

			




	
	
	
		
 
    TYPE_PULL_REQUEST_COMMENT = u'pull_request_comment'

			




	
	
	
		
 

			




	
	
	
		
 
    notification_id = Column(Integer(), primary_key=True)

			




	
	
	
		
 
    subject = Column(Unicode(512), nullable=False)

			




	
	
	
		
 
    body = Column(UnicodeText(), nullable=False)

			




	
	
	
		
 
    created_by = Column(Integer(), ForeignKey('users.user_id'), nullable=False)

			




	
	
	
		
 
    created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

			




	
	
	
		
 
    type_ = Column('type', Unicode(255), nullable=False)

			




	
	
	
		
 

			




	
	
	
		
 
    created_by_user = relationship('User')

			




	
	
	
		
 
    notifications_to_users = relationship('UserNotification', cascade="all, delete-orphan")

			




	
	
	
		
 

			




	
	
	
		
 
    @property

			




	
	
	
		
 
    def recipients(self):

			




	
	
	
		
 
        return [x.user for x in UserNotification.query()

			




	
	
	
		
 
                .filter(UserNotification.notification == self)

			




	
	
	
		
 
                .order_by(UserNotification.user_id.asc()).all()]

			




	
	
	
		
 

			




	
	
	
		
 
    @classmethod

			




	
	
	
		
 
    def create(cls, created_by, subject, body, recipients, type_=None):

			




	
	
	
		
 
        if type_ is None:

			




	
	
	
		
 
            type_ = Notification.TYPE_MESSAGE

			




	
	
	
		
 

			




	
	
	
		
 
        notification = cls()

			




	
	
	
		
 
        notification.created_by_user = created_by

			




	
	
	
		
 
        notification.subject = subject

			




	
	
	
		
 
        notification.body = body

			




	
	
	
		
 
        notification.type_ = type_

			




	
	
	
		
 
        notification.created_on = datetime.datetime.now()

			




	
	
	
		
 

			




	
	
	
		
 
        for recipient in recipients:

			




	
	
	
		
 
            un = UserNotification()

			




	
	
	
		
 
            un.notification = notification

			




	
	
	
		
 
            un.user_id = recipient.user_id

			




	
	
	
		
 
            # Mark notifications to self "pre-read" - should perhaps just be skipped

			




	
	
	
		
 
            if recipient == created_by:

			




	
	
	
		
 
                un.read = True

			




	
	
	
		
 
            Session().add(un)

			




	
	
	
		
 

			




	
	
	
		
 
        Session().add(notification)

			




	
	
	
		
 
        Session().flush() # assign notification.notification_id

			




	
	
	
		
 
        return notification

			




	
	
	
		
 

			




	
	
	
		
 
    @property

			




	
	
	
		
 
    def description(self):

			




	
	
	
		
 
        from kallithea.model.notification import NotificationModel

			




	
	
	
		
 
        return NotificationModel().make_description(self)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class UserNotification(Base, BaseDbModel):

			




	
	
	
		
 

			




	
	
	
		
 
class UserNotification(object):

			




	
	
	
		
 
    __tablename__ = 'user_to_notification'

			




	
	
	
		
 
    __table_args__ = (

			




	
	
	
		
 
        UniqueConstraint('user_id', 'notification_id'),

			




	
	
	
		
 
        _table_args_default_dict,

			




	
	
	
		
 
    )

			




	
	
	
		
 

			




	
	
	
		
 
    user_id = Column(Integer(), ForeignKey('users.user_id'), primary_key=True)

			




	
	
	
		
 
    notification_id = Column(Integer(), ForeignKey('notifications.notification_id'), primary_key=True)

			




	
	
	
		
 
    read = Column(Boolean, nullable=False, default=False)

			




	
	
	
		
 
    sent_on = Column(DateTime(timezone=False), nullable=True) # FIXME: not nullable?

			




	
	
	
		
 

			




	
	
	
		
 
    user = relationship('User')

			




	
	
	
		
 
    notification = relationship('Notification')

			




	
	
	
		
 

			




	
	
	
		
 
    def mark_as_read(self):

			




	
	
	
		
 
        self.read = True

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class Gist(Base, BaseDbModel):

			




	
	
	
		
 
    __tablename__ = 'gists'

			




	
	
	
		
 
    __table_args__ = (

			




	
	
	
		
 
        Index('g_gist_access_id_idx', 'gist_access_id'),

			




	
	
	
		
 
        Index('g_created_on_idx', 'created_on'),

			




	
	
	
		
 
        _table_args_default_dict,

			




	
	
	
		
 
    )

			




	
	
	
		
 

			




	
	
	
		
 
    GIST_PUBLIC = u'public'

			




	
	
	
		
 
    GIST_PRIVATE = u'private'

			




        

    


    
    

    

        

                

                    kallithea/model/notification.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -17,35 +17,36 @@ 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

			




	
	
	
		
 
import traceback

			




	
	
	
		
 

			




	
	
	
		
 
from tg import tmpl_context as c, app_globals

			




	
	
	
		
 
from tg.i18n import ugettext as _

			




	
	
	
		
 
from sqlalchemy.orm import joinedload, subqueryload

			




	
	
	
		
 

			




	
	
	
		
 
import kallithea

			




	
	
	
		
 
from kallithea.lib import helpers as h

			




	
	
	
		
 
from kallithea.lib.utils2 import safe_unicode

			




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

			




	
	
	
		
 
from kallithea.model.db import Notification, User

			




	
	
	
		
 
from kallithea.model.meta import Session

			




	
	
	
		
 

			




	
	
	
		
 
log = logging.getLogger(__name__)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class NotificationModel(object):

			




	
	
	
		
 

			




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

			




	
	
	
		
 
               type_=Notification.TYPE_MESSAGE, with_email=True,

			




	
	
	
		
 
               email_kwargs=None, repo_name=None):

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




	
	
		
 
@@ -79,195 +80,65 @@ class NotificationModel(object):

			




	
	
	
		
 
                    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?

			




	
	
	
		
 

			




	
	
	
		
 
        # TODO: inform user who are notified

			




	
	
	
		
 
        notif = Notification.create(

			




	
	
	
		
 
            created_by=created_by_obj, subject=subject,

			




	
	
	
		
 
            body=body, recipients=recipients_objs, type_=type_

			




	
	
	
		
 
        )

			




	
	
	
		
 

			




	
	
	
		
 
        if not with_email:

			




	
	
	
		
 
            return notif

			




	
	
	
		
 
            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': h.fmt_date(notif.created_on),

			




	
	
	
		
 
                  'user': notif.created_by_user.username,

			




	
	
	
		
 
                  'when': created_on,

			




	
	
	
		
 
                  'user': created_by_obj.username,

			




	
	
	
		
 
                  }

			




	
	
	
		
 

			




	
	
	
		
 
        txt_kwargs = {

			




	
	
	
		
 
                  'subject': subject,

			




	
	
	
		
 
                  'body': body,

			




	
	
	
		
 
                  'when': h.fmt_date(notif.created_on),

			




	
	
	
		
 
                  'user': notif.created_by_user.username,

			




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

			




	
	
	
		
 
                     email_html_body, headers, author=created_by_obj)

			




	
	
	
		
 

			




	
	
	
		
 
        return notif

			




	
	
	
		
 

			




	
	
	
		
 
    def delete(self, user, notification):

			




	
	
	
		
 
        # we don't want to remove actual notification just the assignment

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            notification = Notification.guess_instance(notification)

			




	
	
	
		
 
            user = User.guess_instance(user)

			




	
	
	
		
 
            if notification and user:

			




	
	
	
		
 
                obj = UserNotification.query() \

			




	
	
	
		
 
                        .filter(UserNotification.user == user) \

			




	
	
	
		
 
                        .filter(UserNotification.notification

			




	
	
	
		
 
                                == notification) \

			




	
	
	
		
 
                        .one()

			




	
	
	
		
 
                Session().delete(obj)

			




	
	
	
		
 
                return True

			




	
	
	
		
 
        except Exception:

			




	
	
	
		
 
            log.error(traceback.format_exc())

			




	
	
	
		
 
            raise

			




	
	
	
		
 

			




	
	
	
		
 
    def query_for_user(self, user, filter_=None):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Get notifications for given user, filter them if filter dict is given

			




	
	
	
		
 

			




	
	
	
		
 
        :param user:

			




	
	
	
		
 
        :param filter:

			




	
	
	
		
 
        """

			




	
	
	
		
 
        user = User.guess_instance(user)

			




	
	
	
		
 

			




	
	
	
		
 
        q = UserNotification.query() \

			




	
	
	
		
 
            .filter(UserNotification.user == user) \

			




	
	
	
		
 
            .join((Notification, UserNotification.notification_id ==

			




	
	
	
		
 
                                 Notification.notification_id)) \

			




	
	
	
		
 
            .options(joinedload('notification')) \

			




	
	
	
		
 
            .options(subqueryload('notification.created_by_user')) \

			




	
	
	
		
 
            .order_by(Notification.created_on.desc())

			




	
	
	
		
 

			




	
	
	
		
 
        if filter_:

			




	
	
	
		
 
            q = q.filter(Notification.type_.in_(filter_))

			




	
	
	
		
 

			




	
	
	
		
 
        return q

			




	
	
	
		
 

			




	
	
	
		
 
    def mark_read(self, user, notification):

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            notification = Notification.guess_instance(notification)

			




	
	
	
		
 
            user = User.guess_instance(user)

			




	
	
	
		
 
            if notification and user:

			




	
	
	
		
 
                obj = UserNotification.query() \

			




	
	
	
		
 
                        .filter(UserNotification.user == user) \

			




	
	
	
		
 
                        .filter(UserNotification.notification

			




	
	
	
		
 
                                == notification) \

			




	
	
	
		
 
                        .one()

			




	
	
	
		
 
                obj.read = True

			




	
	
	
		
 
                return True

			




	
	
	
		
 
        except Exception:

			




	
	
	
		
 
            log.error(traceback.format_exc())

			




	
	
	
		
 
            raise

			




	
	
	
		
 

			




	
	
	
		
 
    def mark_all_read_for_user(self, user, filter_=None):

			




	
	
	
		
 
        user = User.guess_instance(user)

			




	
	
	
		
 
        q = UserNotification.query() \

			




	
	
	
		
 
            .filter(UserNotification.user == user) \

			




	
	
	
		
 
            .filter(UserNotification.read == False) \

			




	
	
	
		
 
            .join((Notification, UserNotification.notification_id ==

			




	
	
	
		
 
                                 Notification.notification_id))

			




	
	
	
		
 
        if filter_:

			




	
	
	
		
 
            q = q.filter(Notification.type_.in_(filter_))

			




	
	
	
		
 

			




	
	
	
		
 
        # this is a little inefficient but sqlalchemy doesn't support

			




	
	
	
		
 
        # update on joined tables :(

			




	
	
	
		
 
        for obj in q:

			




	
	
	
		
 
            obj.read = True

			




	
	
	
		
 

			




	
	
	
		
 
    def get_unread_cnt_for_user(self, user):

			




	
	
	
		
 
        user = User.guess_instance(user)

			




	
	
	
		
 
        return UserNotification.query() \

			




	
	
	
		
 
                .filter(UserNotification.read == False) \

			




	
	
	
		
 
                .filter(UserNotification.user == user).count()

			




	
	
	
		
 

			




	
	
	
		
 
    def get_unread_for_user(self, user):

			




	
	
	
		
 
        user = User.guess_instance(user)

			




	
	
	
		
 
        return [x.notification for x in UserNotification.query() \

			




	
	
	
		
 
                .filter(UserNotification.read == False) \

			




	
	
	
		
 
                .filter(UserNotification.user == user).all()]

			




	
	
	
		
 

			




	
	
	
		
 
    def get_user_notification(self, user, notification):

			




	
	
	
		
 
        user = User.guess_instance(user)

			




	
	
	
		
 
        notification = Notification.guess_instance(notification)

			




	
	
	
		
 

			




	
	
	
		
 
        return UserNotification.query() \

			




	
	
	
		
 
            .filter(UserNotification.notification == notification) \

			




	
	
	
		
 
            .filter(UserNotification.user == user).scalar()

			




	
	
	
		
 

			




	
	
	
		
 
    def make_description(self, notification, show_age=True):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Creates a human readable description based on properties

			




	
	
	
		
 
        of notification object

			




	
	
	
		
 
        """

			




	
	
	
		
 
        # alias

			




	
	
	
		
 
        _n = notification

			




	
	
	
		
 

			




	
	
	
		
 
        if show_age:

			




	
	
	
		
 
            return {

			




	
	
	
		
 
                    _n.TYPE_CHANGESET_COMMENT: _('%(user)s commented on changeset %(age)s'),

			




	
	
	
		
 
                    _n.TYPE_MESSAGE: _('%(user)s sent message %(age)s'),

			




	
	
	
		
 
                    _n.TYPE_MENTION: _('%(user)s mentioned you %(age)s'),

			




	
	
	
		
 
                    _n.TYPE_REGISTRATION: _('%(user)s registered in Kallithea %(age)s'),

			




	
	
	
		
 
                    _n.TYPE_PULL_REQUEST: _('%(user)s opened new pull request %(age)s'),

			




	
	
	
		
 
                    _n.TYPE_PULL_REQUEST_COMMENT: _('%(user)s commented on pull request %(age)s'),

			




	
	
	
		
 
                }[notification.type_] % dict(

			




	
	
	
		
 
                    user=notification.created_by_user.username,

			




	
	
	
		
 
                    age=h.age(notification.created_on),

			




	
	
	
		
 
                )

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            return {

			




	
	
	
		
 
                    _n.TYPE_CHANGESET_COMMENT: _('%(user)s commented on changeset at %(when)s'),

			




	
	
	
		
 
                    _n.TYPE_MESSAGE: _('%(user)s sent message at %(when)s'),

			




	
	
	
		
 
                    _n.TYPE_MENTION: _('%(user)s mentioned you at %(when)s'),

			




	
	
	
		
 
                    _n.TYPE_REGISTRATION: _('%(user)s registered in Kallithea at %(when)s'),

			




	
	
	
		
 
                    _n.TYPE_PULL_REQUEST: _('%(user)s opened new pull request at %(when)s'),

			




	
	
	
		
 
                    _n.TYPE_PULL_REQUEST_COMMENT: _('%(user)s commented on pull request at %(when)s'),

			




	
	
	
		
 
                }[notification.type_] % dict(

			




	
	
	
		
 
                    user=notification.created_by_user.username,

			




	
	
	
		
 
                    when=h.fmt_date(notification.created_on),

			




	
	
	
		
 
                )

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class EmailNotificationModel(object):

			




	
	
	
		
 

			




	
	
	
		
 
    TYPE_CHANGESET_COMMENT = Notification.TYPE_CHANGESET_COMMENT

			




	
	
	
		
 
    TYPE_MESSAGE = Notification.TYPE_MESSAGE # only used for testing

			




	
	
	
		
 
    # Notification.TYPE_MENTION is not used

			




	
	
	
		
 
    TYPE_PASSWORD_RESET = 'password_link'

			




	
	
	
		
 
    TYPE_REGISTRATION = Notification.TYPE_REGISTRATION

			




	
	
	
		
 
    TYPE_PULL_REQUEST = Notification.TYPE_PULL_REQUEST

			




	
	
	
		
 
    TYPE_PULL_REQUEST_COMMENT = Notification.TYPE_PULL_REQUEST_COMMENT

			




	
	
	
		
 
    TYPE_DEFAULT = 'default'

			




	
	
	
		
 

			




        

    


    
    

    

        

                

                    kallithea/tests/models/test_notifications.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -44,25 +44,25 @@ class TestNotifications(TestController):

			




	
	
	
		
 

			




	
	
	
		
 
    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,

			




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

			




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

			




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

			




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

			




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

			




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

			




	
	
		
 
@@ -142,25 +142,25 @@ class TestNotifications(TestController):

			




	
	
	
		
 
                        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,

			




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

			




	
	
	
		
 

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








    
        
    
    
        This site is powered by
            Kallithea 0.7.0,
        which is
        © 2010–2025 by various authors & licensed under GPLv3.