kallithea Changeset - cd6c577ade97

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

kallithea/model/db.py

kallithea/model/notification.py

137

kallithea/tests/models/test_notifications.py

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:
 .. automodule:: kallithea.model.scm
    :members:
 .. automodule:: kallithea.model.user
    :members:
 .. automodule:: kallithea.model.user_group
    :members:

kallithea/model/db.py

➞

Show inline comments

@@ @@ -263,387 +263,384 @@ class Setting(Base, BaseDbModel): @@
         :param key:
         :param val:
         :param type:
         :return:
         """
         res = cls.get_by_name(key)
         if res is None:
             val = Optional.extract(val)
             type = Optional.extract(type)
             res = cls(key, val, type)
             Session().add(res)
         else:
             res.app_settings_name = key
             if not isinstance(val, Optional):
                 # update if set
                 res.app_settings_value = val
             if not isinstance(type, Optional):
                 # update if set
                 res.app_settings_type = type
         return res
     @classmethod
     def get_app_settings(cls, cache=False):
         ret = cls.query()
         if cache:
             ret = ret.options(FromCache("sql_cache_short", "get_hg_settings"))
         if ret is None:
             raise Exception('Could not get application settings !')
         settings = {}
         for each in ret:
             settings[each.app_settings_name] = \
                 each.app_settings_value
         return settings
     @classmethod
     def get_auth_settings(cls, cache=False):
         ret = cls.query() \
                 .filter(cls.app_settings_name.startswith('auth_')).all()
         fd = {}
         for row in ret:
             fd[row.app_settings_name] = row.app_settings_value
         return fd
     @classmethod
     def get_default_repo_settings(cls, cache=False, strip_prefix=False):
         ret = cls.query() \
                 .filter(cls.app_settings_name.startswith('default_')).all()
         fd = {}
         for row in ret:
             key = row.app_settings_name
             if strip_prefix:
                 key = remove_prefix(key, prefix='default_')
             fd.update({key: row.app_settings_value})
         return fd
     @classmethod
     def get_server_info(cls):
         import pkg_resources
         import platform
         import kallithea
         from kallithea.lib.utils import check_git_version
         mods = [(p.project_name, p.version) for p in pkg_resources.working_set]
         info = {
             'modules': sorted(mods, key=lambda k: k[0].lower()),
             'py_version': platform.python_version(),
             'platform': safe_unicode(platform.platform()),
             'kallithea_version': kallithea.__version__,
             'git_version': safe_unicode(check_git_version()),
             'git_path': kallithea.CONFIG.get('git_path')
+        }
         return info
 class Ui(Base, BaseDbModel):
     __tablename__ = 'ui'
     __table_args__ = (
         # FIXME: ui_key as key is wrong and should be removed when the corresponding
         # Ui.get_by_key has been replaced by the composite key
         UniqueConstraint('ui_key'),
         UniqueConstraint('ui_section', 'ui_key'),
         _table_args_default_dict,
+    )
     HOOK_UPDATE = 'changegroup.update'
     HOOK_REPO_SIZE = 'changegroup.repo_size'
     HOOK_PUSH_LOG = 'changegroup.push_logger'
     HOOK_PUSH_LOCK = 'prechangegroup.push_lock_handling'
     HOOK_PULL_LOG = 'outgoing.pull_logger'
     HOOK_PULL_LOCK = 'preoutgoing.pull_lock_handling'
     ui_id = Column(Integer(), primary_key=True)
     ui_section = Column(String(255), nullable=False)
     ui_key = Column(String(255), nullable=False)
     ui_value = Column(String(255), nullable=True) # FIXME: not nullable?
     ui_active = Column(Boolean(), nullable=False, default=True)
     @classmethod
     def get_by_key(cls, section, key):
         """ Return specified Ui object, or None if not found. """
         return cls.query().filter_by(ui_section=section, ui_key=key).scalar()
     @classmethod
     def get_or_create(cls, section, key):
         """ Return specified Ui object, creating it if necessary. """
         setting = cls.get_by_key(section, key)
         if setting is None:
             setting = cls(ui_section=section, ui_key=key)
             Session().add(setting)
         return setting
     @classmethod
     def get_builtin_hooks(cls):
         q = cls.query()
         q = q.filter(cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE,
                                      cls.HOOK_PUSH_LOG, cls.HOOK_PUSH_LOCK,
                                      cls.HOOK_PULL_LOG, cls.HOOK_PULL_LOCK]))
         q = q.filter(cls.ui_section == 'hooks')
         return q.all()
     @classmethod
     def get_custom_hooks(cls):
         q = cls.query()
         q = q.filter(~cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE,
                                       cls.HOOK_PUSH_LOG, cls.HOOK_PUSH_LOCK,
                                       cls.HOOK_PULL_LOG, cls.HOOK_PULL_LOCK]))
         q = q.filter(cls.ui_section == 'hooks')
         return q.all()
     @classmethod
     def get_repos_location(cls):
         return cls.get_by_key('paths', '/').ui_value
     @classmethod
     def create_or_update_hook(cls, key, val):
         new_ui = cls.get_or_create('hooks', key)
         new_ui.ui_active = True
         new_ui.ui_value = val
     def __repr__(self):
         return '<%s[%s]%s=>%s]>' % (self.__class__.__name__, self.ui_section,
                                     self.ui_key, self.ui_value)
 class User(Base, BaseDbModel):
     __tablename__ = 'users'
     __table_args__ = (
         Index('u_username_idx', 'username'),
         Index('u_email_idx', 'email'),
         _table_args_default_dict,
+    )
     DEFAULT_USER = 'default'
     DEFAULT_GRAVATAR_URL = 'https://secure.gravatar.com/avatar/{md5email}?d=identicon&s={size}'
     # The name of the default auth type in extern_type, 'internal' lives in auth_internal.py
     DEFAULT_AUTH_TYPE = 'internal'
     user_id = Column(Integer(), primary_key=True)
     username = Column(String(255), nullable=False, unique=True)
     password = Column(String(255), nullable=False)
     active = Column(Boolean(), nullable=False, default=True)
     admin = Column(Boolean(), nullable=False, default=False)
     name = Column("firstname", Unicode(255), nullable=False)
     lastname = Column(Unicode(255), nullable=False)
     _email = Column("email", String(255), nullable=True, unique=True) # FIXME: not nullable?
     last_login = Column(DateTime(timezone=False), nullable=True)
     extern_type = Column(String(255), nullable=True) # FIXME: not nullable?
     extern_name = Column(String(255), nullable=True) # FIXME: not nullable?
     api_key = Column(String(255), nullable=False)
     inherit_default_permissions = Column(Boolean(), nullable=False, default=True)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     _user_data = Column("user_data", LargeBinary(), nullable=True)  # JSON data # FIXME: not nullable?
     user_log = relationship('UserLog')
     user_perms = relationship('UserToPerm', primaryjoin="User.user_id==UserToPerm.user_id", cascade='all')
     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
     def email(self, val):
         self._email = val.lower() if val else None
     @property
     def firstname(self):
         # alias for future
         return self.name
     @property
     def emails(self):
         other = UserEmailMap.query().filter(UserEmailMap.user == self).all()
         return [self.email] + [x.email for x in other]
     @property
     def api_keys(self):
         other = UserApiKeys.query().filter(UserApiKeys.user == self).all()
         return [self.api_key] + [x.api_key for x in other]
     @property
     def ip_addresses(self):
         ret = UserIpMap.query().filter(UserIpMap.user == self).all()
         return [x.ip_addr for x in ret]
     @property
     def full_name(self):
         return '%s %s' % (self.firstname, self.lastname)
     @property
     def full_name_or_username(self):
         """
         Show full name.
         If full name is not set, fall back to username.
         """
         return ('%s %s' % (self.firstname, self.lastname)
                 if (self.firstname and self.lastname) else self.username)
     @property
     def full_name_and_username(self):
         """
         Show full name and username as 'Firstname Lastname (username)'.
         If full name is not set, fall back to username.
         """
         return ('%s %s (%s)' % (self.firstname, self.lastname, self.username)
                 if (self.firstname and self.lastname) else self.username)
     @property
     def full_contact(self):
         return '%s %s <%s>' % (self.firstname, self.lastname, self.email)
     @property
     def short_contact(self):
         return '%s %s' % (self.firstname, self.lastname)
     @property
     def is_admin(self):
         return self.admin
     @hybrid_property
     def is_default_user(self):
         return self.username == User.DEFAULT_USER
     @property
     def AuthUser(self):
         """
         Returns instance of AuthUser for this user
         """
         from kallithea.lib.auth import AuthUser
         return AuthUser(dbuser=self)
     @hybrid_property
     def user_data(self):
         if not self._user_data:
             return {}
         try:
             return json.loads(self._user_data)
         except TypeError:
             return {}
     @user_data.setter
     def user_data(self, val):
         try:
             self._user_data = json.dumps(val)
         except Exception:
             log.error(traceback.format_exc())
     def __unicode__(self):
         return u"<%s('id:%s:%s')>" % (self.__class__.__name__,
                                       self.user_id, self.username)
     @classmethod
     def guess_instance(cls, value):
         return super(User, cls).guess_instance(value, User.get_by_username)
     @classmethod
     def get_or_404(cls, id_, allow_default=True):
         '''
         Overridden version of BaseDbModel.get_or_404, with an extra check on
         the default user.
         '''
         user = super(User, cls).get_or_404(id_)
         if not allow_default and user.is_default_user:
             raise DefaultUserException()
         return user
     @classmethod
     def get_by_username_or_email(cls, username_or_email, case_insensitive=False, cache=False):
         """
         For anything that looks like an email address, look up by the email address (matching
         case insensitively).
         For anything else, try to look up by the user name.
         This assumes no normal username can have '@' symbol.
         """
         if '@' in username_or_email:
             return User.get_by_email(username_or_email, cache=cache)
         else:
             return User.get_by_username(username_or_email, case_insensitive=case_insensitive, cache=cache)
     @classmethod
     def get_by_username(cls, username, case_insensitive=False, cache=False):
         if case_insensitive:
             q = cls.query().filter(func.lower(cls.username) == func.lower(username))
         else:
             q = cls.query().filter(cls.username == username)
         if cache:
             q = q.options(FromCache(
                             "sql_cache_short",
                             "get_user_%s" % _hash_key(username)
+                          )
+            )
         return q.scalar()
     @classmethod
     def get_by_api_key(cls, api_key, cache=False, fallback=True):
         if len(api_key) != 40 or not api_key.isalnum():
             return None
         q = cls.query().filter(cls.api_key == api_key)
         if cache:
             q = q.options(FromCache("sql_cache_short",
                                     "get_api_key_%s" % api_key))
         res = q.scalar()
         if fallback and not res:
             # fallback to additional keys
             _res = UserApiKeys.query().filter_by(api_key=api_key, is_expired=False).first()
             if _res:
                 res = _res.user
         return res
     @classmethod
     def get_by_email(cls, email, cache=False):
         q = cls.query().filter(func.lower(cls.email) == func.lower(email))
         if cache:
             q = q.options(FromCache("sql_cache_short",
                                     "get_email_key_%s" % email))
         ret = q.scalar()
         if ret is None:
             q = UserEmailMap.query()
             # try fetching in alternate email map
             q = q.filter(func.lower(UserEmailMap.email) == func.lower(email))
             q = q.options(joinedload(UserEmailMap.user))
             if cache:
                 q = q.options(FromCache("sql_cache_short",
                                         "get_email_map_key_%s" % email))
             ret = getattr(q.scalar(), 'user', None)
         return ret
     @classmethod
     def get_from_cs_author(cls, author):
         """
         Tries to get User objects out of commit author string
         :param author:
@@ @@ -2289,368 +2286,303 @@ class ChangesetStatus(Base, BaseDbModel) @@
     pull_request_id = Column(Integer(), ForeignKey('pull_requests.pull_request_id'), nullable=True)
     author = relationship('User')
     repo = relationship('Repository')
     comment = relationship('ChangesetComment')
     pull_request = relationship('PullRequest')
     def __unicode__(self):
         return u"<%s('%s:%s')>" % (
             self.__class__.__name__,
             self.status, self.author
+        )
     @classmethod
     def get_status_lbl(cls, value):
         return cls.STATUSES_DICT.get(value)
     @property
     def status_lbl(self):
         return ChangesetStatus.get_status_lbl(self.status)
     def __json__(self):
         return dict(
             status=self.status,
             modified_at=self.modified_at.replace(microsecond=0),
             reviewer=self.author.username,
+            )
 class PullRequest(Base, BaseDbModel):
     __tablename__ = 'pull_requests'
     __table_args__ = (
         Index('pr_org_repo_id_idx', 'org_repo_id'),
         Index('pr_other_repo_id_idx', 'other_repo_id'),
         _table_args_default_dict,
+    )
     # values for .status
     STATUS_NEW = u'new'
     STATUS_CLOSED = u'closed'
     pull_request_id = Column(Integer(), primary_key=True)
     title = Column(Unicode(255), nullable=False)
     description = Column(UnicodeText(), nullable=False)
     status = Column(Unicode(255), nullable=False, default=STATUS_NEW) # only for closedness, not approve/reject/etc
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     updated_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     owner_id = Column('user_id', Integer(), ForeignKey('users.user_id'), nullable=False)
     _revisions = Column('revisions', UnicodeText(), nullable=False)
     org_repo_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)
     org_ref = Column(Unicode(255), nullable=False)
     other_repo_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)
     other_ref = Column(Unicode(255), nullable=False)
     @hybrid_property
     def revisions(self):
         return self._revisions.split(':')
     @revisions.setter
     def revisions(self, val):
         self._revisions = safe_unicode(':'.join(val))
     @property
     def org_ref_parts(self):
         return self.org_ref.split(':')
     @property
     def other_ref_parts(self):
         return self.other_ref.split(':')
     owner = relationship('User')
     reviewers = relationship('PullRequestReviewer',
                              cascade="all, delete-orphan")
     org_repo = relationship('Repository', primaryjoin='PullRequest.org_repo_id==Repository.repo_id')
     other_repo = relationship('Repository', primaryjoin='PullRequest.other_repo_id==Repository.repo_id')
     statuses = relationship('ChangesetStatus', order_by='ChangesetStatus.changeset_status_id')
     comments = relationship('ChangesetComment', order_by='ChangesetComment.comment_id',
                              cascade="all, delete-orphan")
     @classmethod
     def query(cls, reviewer_id=None, include_closed=True, sorted=False):
         """Add PullRequest-specific helpers for common query constructs.
         reviewer_id: only PRs with the specified user added as reviewer.
         include_closed: if False, do not include closed PRs.
         sorted: if True, apply the default ordering (newest first).
         """
         q = super(PullRequest, cls).query()
         if reviewer_id is not None:
             q = q.join(PullRequestReviewer).filter(PullRequestReviewer.user_id == reviewer_id)
         if not include_closed:
             q = q.filter(PullRequest.status != PullRequest.STATUS_CLOSED)
         if sorted:
             q = q.order_by(PullRequest.created_on.desc())
         return q
     def get_reviewer_users(self):
         """Like .reviewers, but actually returning the users"""
         return User.query() \
             .join(PullRequestReviewer) \
             .filter(PullRequestReviewer.pull_request == self) \
             .order_by(PullRequestReviewer.pull_request_reviewers_id) \
             .all()
     def is_closed(self):
         return self.status == self.STATUS_CLOSED
     def user_review_status(self, user_id):
         """Return the user's latest status votes on PR"""
         # note: no filtering on repo - that would be redundant
         status = ChangesetStatus.query() \
             .filter(ChangesetStatus.pull_request == self) \
             .filter(ChangesetStatus.user_id == user_id) \
             .order_by(ChangesetStatus.version) \
             .first()
         return str(status.status) if status else ''
     @classmethod
     def make_nice_id(cls, pull_request_id):
         '''Return pull request id nicely formatted for displaying'''
         return '#%s' % pull_request_id
     def nice_id(self):
         '''Return the id of this pull request, nicely formatted for displaying'''
         return self.make_nice_id(self.pull_request_id)
     def get_api_data(self):
         return self.__json__()
     def __json__(self):
         return dict(
             pull_request_id=self.pull_request_id,
             url=self.url(),
             reviewers=self.reviewers,
             revisions=self.revisions,
             owner=self.owner.username,
             title=self.title,
             description=self.description,
             org_repo_url=self.org_repo.clone_url(),
             org_ref_parts=self.org_ref_parts,
             other_ref_parts=self.other_ref_parts,
             status=self.status,
             comments=self.comments,
             statuses=self.statuses,
+        )
     def url(self, **kwargs):
         canonical = kwargs.pop('canonical', None)
         import kallithea.lib.helpers as h
         b = self.org_ref_parts[1]
         if b != self.other_ref_parts[1]:
             s = '/_/' + b
         else:
             s = '/_/' + self.title
         kwargs['extra'] = urlreadable(s)
         if canonical:
             return h.canonical_url('pullrequest_show', repo_name=self.other_repo.repo_name,
                                    pull_request_id=self.pull_request_id, **kwargs)
         return h.url('pullrequest_show', repo_name=self.other_repo.repo_name,
                      pull_request_id=self.pull_request_id, **kwargs)
 class PullRequestReviewer(Base, BaseDbModel):
     __tablename__ = 'pull_request_reviewers'
     __table_args__ = (
         Index('pull_request_reviewers_user_id_idx', 'user_id'),
         _table_args_default_dict,
+    )
     def __init__(self, user=None, pull_request=None):
         self.user = user
         self.pull_request = pull_request
     pull_request_reviewers_id = Column('pull_requests_reviewers_id', Integer(), primary_key=True)
     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'
     DEFAULT_FILENAME = u'gistfile1.txt'
     gist_id = Column(Integer(), primary_key=True)
     gist_access_id = Column(Unicode(250), nullable=False)
     gist_description = Column(UnicodeText(), nullable=False)
     owner_id = Column('user_id', Integer(), ForeignKey('users.user_id'), nullable=False)
     gist_expires = Column(Float(53), nullable=False)
     gist_type = Column(Unicode(128), nullable=False)
     created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     modified_at = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     owner = relationship('User')
     @hybrid_property
     def is_expired(self):
         return (self.gist_expires != -1) & (time.time() > self.gist_expires)
     def __repr__(self):
         return '<Gist:[%s]%s>' % (self.gist_type, self.gist_access_id)
     @classmethod
     def guess_instance(cls, value):
         return super(Gist, cls).guess_instance(value, Gist.get_by_access_id)
     @classmethod
     def get_or_404(cls, id_):
         res = cls.query().filter(cls.gist_access_id == id_).scalar()
         if res is None:
             raise HTTPNotFound
         return res
     @classmethod
     def get_by_access_id(cls, gist_access_id):
         return cls.query().filter(cls.gist_access_id == gist_access_id).scalar()
     def gist_url(self):
         import kallithea
         alias_url = kallithea.CONFIG.get('gist_alias_url')
         if alias_url:
             return alias_url.replace('{gistid}', self.gist_access_id)
         import kallithea.lib.helpers as h
         return h.canonical_url('gist', gist_id=self.gist_access_id)
     @classmethod
     def base_path(cls):
         """
         Returns base path where all gists are stored
         :param cls:
         """
         from kallithea.model.gist import GIST_STORE_LOC
         q = Session().query(Ui) \
             .filter(Ui.ui_key == URL_SEP)
         q = q.options(FromCache("sql_cache_short", "repository_repo_path"))
         return os.path.join(q.one().ui_value, GIST_STORE_LOC)
     def get_api_data(self):
         """
         Common function for generating gist related data for API
         """
         gist = self
         data = dict(
             gist_id=gist.gist_id,
             type=gist.gist_type,
             access_id=gist.gist_access_id,
             description=gist.gist_description,
             url=gist.gist_url(),
             expires=gist.gist_expires,
             created_on=gist.created_on,
+        )
         return data
     def __json__(self):
         data = dict(
+        )
         data.update(self.get_api_data())
         return data
     ## SCM functions
     @property
     def scm_instance(self):
         from kallithea.lib.vcs import get_repo
         base_path = self.base_path()
         return get_repo(os.path.join(*map(safe_str,
                                           [base_path, self.gist_access_id])))

kallithea/model/notification.py

➞

Show inline comments

 # -*- 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
 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):
         """
         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?
         # 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'
     def __init__(self):
         super(EmailNotificationModel, self).__init__()
         self._template_root = kallithea.CONFIG['paths']['templates'][0]
         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.keys()), e)
             raise
         l = [safe_unicode(x) for x in [kwargs.get('status_change'), kwargs.get('closing_pr') and _('Closing')] if x]
         if l:
             if subj.startswith('['):
                 subj = '[' + ', '.join(l) + ': ' + subj[1:]
             else:
                 subj = '[' + ', '.join(l) + '] ' + 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)

kallithea/tests/models/test_notifications.py

➞

Show inline comments

 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
 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
     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')
             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 [
                         (Notification.TYPE_CHANGESET_COMMENT,
                          u'This is the new \'comment\'.\n\n - and here it ends indented.',
                          dict(
                             short_id='cafe1234',
                             raw_id='cafe1234c0ffeecafe',
                             branch='brunch',
                             cs_comment_user='Opinionated User (jsmith)',
                             cs_comment_url='http://comment.org',
                             is_mention=[False, True],
                             message='This changeset did something clever which is hard to explain',
                             message_short='This changeset did something cl...',
                             status_change=[None, 'Approved'],
                             cs_target_repo='http://example.com/repo_target',
                             cs_url='http://changeset.com',
                             cs_author=User.get(self.u2))),
                         (Notification.TYPE_MESSAGE,
                          u'This is the \'body\' of the "test" message\n - nothing interesting here except indentation.',
                          dict()),
                         #(Notification.TYPE_MENTION, '$body', None), # not used
                         (Notification.TYPE_REGISTRATION,
                          u'Registration body',
                          dict(
                             new_username='newbie',
                             registered_user_url='http://newbie.org',
                             new_email='new@email.com',
                             new_full_name='New Full Name')),
                         (Notification.TYPE_PULL_REQUEST,
                          u'This PR is \'awesome\' because it does <stuff>\n - please approve indented!',
                          dict(
                             pr_user_created='Requesting User (root)', # pr_owner should perhaps be used for @mention in description ...
                             is_mention=[False, True],
                             pr_revisions=[('123abc'*7, "Introduce one and two\n\nand that's it"), ('567fed'*7, 'Make one plus two equal tree')],
                             org_repo_name='repo_org',
                             **pr_kwargs)),
                         (Notification.TYPE_PULL_REQUEST_COMMENT,
                          u'Me too!\n\n - and indented on second line',
                          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'] = 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,
                         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)

0 comments (0 inline, 0 general)