kallithea Changeset - f65661179895

Changeset - f65661179895

Parent rev.

Child rev.

[Not reviewed]

default

0 2 1

Mads Kiilerich - 9 years ago 2016-06-29 16:52:07
madski@unity3d.com

tests: introduce tests and reference dump for notification mails

The mails are dumped to a tracked html file:
* changes shows up as diffs and are easy to spot and review
* all mails can easily can be investigated in a browser and checked for content
and consistency

The tests are mocking canonical_url because it has deep dependencies to
pylons.url which requires (thread local?) environment setup that the tests
doesn't have.

3 files changed with 718 insertions and 7 deletions:

kallithea/model/db.py

kallithea/tests/models/test_dump_html_mails.ref.html

629

kallithea/tests/models/test_notifications.py

0 comments (0 inline, 0 general)

kallithea/model/db.py

➞

Show inline comments

@@ @@ -2017,555 +2017,555 @@ class UserFollowing(Base, BaseModel): @@
+    )
     user_following_id = Column(Integer(), unique=True, primary_key=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     follows_repo_id = Column("follows_repository_id", Integer(), ForeignKey('repositories.repo_id'), nullable=True)
     follows_user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=True)
     follows_from = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
     user = relationship('User', primaryjoin='User.user_id==UserFollowing.user_id')
     follows_user = relationship('User', primaryjoin='User.user_id==UserFollowing.follows_user_id')
     follows_repository = relationship('Repository', order_by='Repository.repo_name')
     @classmethod
     def get_repo_followers(cls, repo_id):
         return cls.query().filter(cls.follows_repo_id == repo_id)
 class CacheInvalidation(Base, BaseModel):
     __tablename__ = 'cache_invalidation'
     __table_args__ = (
         Index('key_idx', 'cache_key'),
         _table_args_default_dict,
+    )
     # cache_id, not used
     cache_id = Column(Integer(), unique=True, primary_key=True)
     # cache_key as created by _get_cache_key
     cache_key = Column(Unicode(255), nullable=False, unique=True)
     # cache_args is a repo_name
     cache_args = Column(Unicode(255), nullable=False)
     # instance sets cache_active True when it is caching, other instances set
     # cache_active to False to indicate that this cache is invalid
     cache_active = Column(Boolean(), nullable=False, default=False)
     def __init__(self, cache_key, repo_name=''):
         self.cache_key = cache_key
         self.cache_args = repo_name
         self.cache_active = False
     def __unicode__(self):
         return u"<%s('%s:%s[%s]')>" % (
             self.__class__.__name__,
             self.cache_id, self.cache_key, self.cache_active)
     def _cache_key_partition(self):
         prefix, repo_name, suffix = self.cache_key.partition(self.cache_args)
         return prefix, repo_name, suffix
     def get_prefix(self):
         """
         get prefix that might have been used in _get_cache_key to
         generate self.cache_key. Only used for informational purposes
         in repo_edit.html.
         """
         # prefix, repo_name, suffix
         return self._cache_key_partition()[0]
     def get_suffix(self):
         """
         get suffix that might have been used in _get_cache_key to
         generate self.cache_key. Only used for informational purposes
         in repo_edit.html.
         """
         # prefix, repo_name, suffix
         return self._cache_key_partition()[2]
     @classmethod
     def clear_cache(cls):
         """
         Delete all cache keys from database.
         Should only be run when all instances are down and all entries thus stale.
         """
         cls.query().delete()
         Session().commit()
     @classmethod
     def _get_cache_key(cls, key):
         """
         Wrapper for generating a unique cache key for this instance and "key".
         key must / will start with a repo_name which will be stored in .cache_args .
         """
         import kallithea
         prefix = kallithea.CONFIG.get('instance_id', '')
         return "%s%s" % (prefix, key)
     @classmethod
     def set_invalidate(cls, repo_name):
         """
         Mark all caches of a repo as invalid in the database.
         """
         inv_objs = Session().query(cls).filter(cls.cache_args == repo_name).all()
         log.debug('for repo %s got %s invalidation objects',
                   safe_str(repo_name), inv_objs)
         for inv_obj in inv_objs:
             log.debug('marking %s key for invalidation based on repo_name=%s',
                       inv_obj, safe_str(repo_name))
             Session().delete(inv_obj)
         Session().commit()
     @classmethod
     def test_and_set_valid(cls, repo_name, kind, valid_cache_keys=None):
         """
         Mark this cache key as active and currently cached.
         Return True if the existing cache registration still was valid.
         Return False to indicate that it had been invalidated and caches should be refreshed.
         """
         key = (repo_name + '_' + kind) if kind else repo_name
         cache_key = cls._get_cache_key(key)
         if valid_cache_keys and cache_key in valid_cache_keys:
             return True
         inv_obj = cls.query().filter(cls.cache_key == cache_key).scalar()
         if inv_obj is None:
             inv_obj = cls(cache_key, repo_name)
         elif inv_obj.cache_active:
             return True
         inv_obj.cache_active = True
         Session().add(inv_obj)
         try:
             Session().commit()
         except sqlalchemy.exc.IntegrityError:
             log.error('commit of CacheInvalidation failed - retrying')
             Session().rollback()
             inv_obj = cls.query().filter(cls.cache_key == cache_key).scalar()
             if inv_obj is None:
                 log.error('failed to create CacheInvalidation entry')
                 # TODO: fail badly?
             # else: TOCTOU - another thread added the key at the same time; no further action required
         return False
     @classmethod
     def get_valid_cache_keys(cls):
         """
         Return opaque object with information of which caches still are valid
         and can be used without checking for invalidation.
         """
         return set(inv_obj.cache_key for inv_obj in cls.query().filter(cls.cache_active).all())
 class ChangesetComment(Base, BaseModel):
     __tablename__ = 'changeset_comments'
     __table_args__ = (
         Index('cc_revision_idx', 'revision'),
         Index('cc_pull_request_id_idx', 'pull_request_id'),
         _table_args_default_dict,
+    )
     comment_id = Column(Integer(), unique=True, primary_key=True)
     repo_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)
     revision = Column(String(40), nullable=True)
     pull_request_id = Column(Integer(), ForeignKey('pull_requests.pull_request_id'), nullable=True)
     line_no = Column(Unicode(10), nullable=True)
     f_path = Column(Unicode(1000), nullable=True)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     text = Column(UnicodeText(25000), 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)
     author = relationship('User')
     repo = relationship('Repository')
     # status_change is frequently used directly in templates - make it a lazy
     # join to avoid fetching each related ChangesetStatus on demand.
     # There will only be one ChangesetStatus referencing each comment so the join will not explode.
     status_change = relationship('ChangesetStatus',
                                  cascade="all, delete-orphan", lazy='joined')
     pull_request = relationship('PullRequest')
     @classmethod
     def get_users(cls, revision=None, pull_request_id=None):
         """
         Returns user associated with this ChangesetComment. ie those
         who actually commented
         :param cls:
         :param revision:
         """
         q = Session().query(User) \
                 .join(ChangesetComment.author)
         if revision is not None:
             q = q.filter(cls.revision == revision)
         elif pull_request_id is not None:
             q = q.filter(cls.pull_request_id == pull_request_id)
         return q.all()
     def url(self):
         anchor = "comment-%s" % self.comment_id
         import kallithea.lib.helpers as h
         if self.revision:
             return h.url('changeset_home', repo_name=self.repo.repo_name, revision=self.revision, anchor=anchor)
         elif self.pull_request_id is not None:
             return self.pull_request.url(anchor=anchor)
     def deletable(self):
         return self.created_on > datetime.datetime.now() - datetime.timedelta(minutes=5)
 class ChangesetStatus(Base, BaseModel):
     __tablename__ = 'changeset_statuses'
     __table_args__ = (
         Index('cs_revision_idx', 'revision'),
         Index('cs_version_idx', 'version'),
         Index('cs_pull_request_id_idx', 'pull_request_id'),
         Index('cs_changeset_comment_id_idx', 'changeset_comment_id'),
         Index('cs_pull_request_id_user_id_version_idx', 'pull_request_id', 'user_id', 'version'),
         Index('cs_repo_id_pull_request_id_idx', 'repo_id', 'pull_request_id'),
         UniqueConstraint('repo_id', 'revision', 'version'),
         _table_args_default_dict,
+    )
     STATUS_NOT_REVIEWED = DEFAULT = 'not_reviewed'
     STATUS_APPROVED = 'approved'
     STATUS_REJECTED = 'rejected' # is shown as "Not approved" - TODO: change database content / scheme
     STATUS_UNDER_REVIEW = 'under_review'
     STATUSES = [
         (STATUS_NOT_REVIEWED, _("Not reviewed")),  # (no icon) and default
         (STATUS_UNDER_REVIEW, _("Under review")),
         (STATUS_REJECTED, _("Not approved")),
         (STATUS_APPROVED, _("Approved")),
+    ]
     STATUSES_DICT = dict(STATUSES)
     changeset_status_id = Column(Integer(), unique=True, primary_key=True)
     repo_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     revision = Column(String(40), nullable=True)
     status = Column(String(128), nullable=False, default=DEFAULT)
     changeset_comment_id = Column(Integer(), ForeignKey('changeset_comments.comment_id'), nullable=False)
     modified_at = Column(DateTime(), nullable=False, default=datetime.datetime.now)
     version = Column(Integer(), nullable=False, default=0)
     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)
 class PullRequest(Base, BaseModel):
     __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(), unique=True, primary_key=True)
     title = Column(Unicode(255), nullable=False)
     description = Column(UnicodeText(10240), 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)
     user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)
     _revisions = Column('revisions', UnicodeText(20500), nullable=False)  # 500 revisions max
     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('PullRequestReviewers',
                              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")
     def get_reviewer_users(self):
         """Like .reviewers, but actually returning the users"""
         return User.query() \
             .join(PullRequestReviewers) \
             .filter(PullRequestReviewers.pull_request == self) \
             .order_by(PullRequestReviewers.pull_requests_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 __json__(self):
         return dict(
             revisions=self.revisions
+        )
     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 PullRequestReviewers(Base, BaseModel):
     __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_requests_reviewers_id = Column(Integer(), unique=True, 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')
 class Notification(Base, BaseModel):
     __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'
+    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(), unique=True, primary_key=True)
     subject = Column(Unicode(512), nullable=False)
     body = Column(UnicodeText(50000), 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, BaseModel):
     __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
         Session().add(self)
 class Gist(Base, BaseModel):
     __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(), unique=True, primary_key=True)
     gist_access_id = Column(Unicode(250), nullable=False)
     gist_description = Column(UnicodeText(1024), nullable=False)
     gist_owner = 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')
     def __repr__(self):
         return '<Gist:[%s]%s>' % (self.gist_type, self.gist_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])))
 class DbMigrateVersion(Base, BaseModel):
     __tablename__ = 'db_migrate_version'
     __table_args__ = (
         _table_args_default_dict,
+    )
     repository_id = Column(String(250), unique=True, primary_key=True)
     repository_path = Column(Text, nullable=False)
     version = Column(Integer, nullable=False)

kallithea/tests/models/test_dump_html_mails.ref.html

➞

Show inline comments

@@ new file 100644 @@
 <html><body>
 <h1>cs_comment, is_mention=False, status_change=None</h1>
 <pre>
 From: u1
 To: u2@example.com
 Subject: [Comment] repo/name changeset cafe on brunch
 --------------------
 Comment from Commenter Name on repo_target changeset c0ffeecafe:
 This is the new comment.
  - and here it ends indented.
 URL: http://comment.org
 Changeset: c0ffeecafe
 Description:
 This changeset did something clever which is hard to explain
 --
 This is an automatic notification. Don't reply to this mail.
 --------------------</pre>
 <p>Comment from Commenter Name on repo_target changeset c0ffeecafe:</p>
 <p><div class="formatted-fixed">This is the new comment.
  - and here it ends indented.</div></p>
 <p>URL: <a href="http://comment.org">http://comment.org</a></p>
 <p>Changeset: c0ffeecafe</p>
 <p>Description:<br/>
 This changeset did something clever which is hard to explain
 </p>
 <br/>
 <br/>
 -- <br/>
 This is an automatic notification. Don&#39;t reply to this mail.
 <pre>--------------------</pre>
 <h1>cs_comment, is_mention=True, status_change=None</h1>
 <pre>
 From: u1
 To: u2@example.com
 Subject: [Comment] repo/name changeset cafe on brunch
 --------------------
 Comment from Commenter Name on repo_target changeset c0ffeecafe mentioned you:
 This is the new comment.
  - and here it ends indented.
 URL: http://comment.org
 Changeset: c0ffeecafe
 Description:
 This changeset did something clever which is hard to explain
 --
 This is an automatic notification. Don't reply to this mail.
 --------------------</pre>
 <p>Comment from Commenter Name on repo_target changeset c0ffeecafe mentioned you:</p>
 <p><div class="formatted-fixed">This is the new comment.
  - and here it ends indented.</div></p>
 <p>URL: <a href="http://comment.org">http://comment.org</a></p>
 <p>Changeset: c0ffeecafe</p>
 <p>Description:<br/>
 This changeset did something clever which is hard to explain
 </p>
 <br/>
 <br/>
 -- <br/>
 This is an automatic notification. Don&#39;t reply to this mail.
 <pre>--------------------</pre>
 <h1>cs_comment, is_mention=False, status_change='Approved'</h1>
 <pre>
 From: u1
 To: u2@example.com
 Subject: [Approved: Comment] repo/name changeset cafe on brunch
 --------------------
 Comment from Commenter Name on repo_target changeset c0ffeecafe:
 This is the new comment.
  - and here it ends indented.
 The changeset status was changed to: Approved
 URL: http://comment.org
 Changeset: c0ffeecafe
 Description:
 This changeset did something clever which is hard to explain
 --
 This is an automatic notification. Don't reply to this mail.
 --------------------</pre>
 <p>Comment from Commenter Name on repo_target changeset c0ffeecafe:</p>
 <p><div class="formatted-fixed">This is the new comment.
  - and here it ends indented.</div></p>
     <p>The changeset status was changed to: <b>Approved</b></p>
 <p>URL: <a href="http://comment.org">http://comment.org</a></p>
 <p>Changeset: c0ffeecafe</p>
 <p>Description:<br/>
 This changeset did something clever which is hard to explain
 </p>
 <br/>
 <br/>
 -- <br/>
 This is an automatic notification. Don&#39;t reply to this mail.
 <pre>--------------------</pre>
 <h1>cs_comment, is_mention=True, status_change='Approved'</h1>
 <pre>
 From: u1
 To: u2@example.com
 Subject: [Approved: Comment] repo/name changeset cafe on brunch
 --------------------
 Comment from Commenter Name on repo_target changeset c0ffeecafe mentioned you:
 This is the new comment.
  - and here it ends indented.
 The changeset status was changed to: Approved
 URL: http://comment.org
 Changeset: c0ffeecafe
 Description:
 This changeset did something clever which is hard to explain
 --
 This is an automatic notification. Don't reply to this mail.
 --------------------</pre>
 <p>Comment from Commenter Name on repo_target changeset c0ffeecafe mentioned you:</p>
 <p><div class="formatted-fixed">This is the new comment.
  - and here it ends indented.</div></p>
     <p>The changeset status was changed to: <b>Approved</b></p>
 <p>URL: <a href="http://comment.org">http://comment.org</a></p>
 <p>Changeset: c0ffeecafe</p>
 <p>Description:<br/>
 This changeset did something clever which is hard to explain
 </p>
 <br/>
 <br/>
 -- <br/>
 This is an automatic notification. Don&#39;t reply to this mail.
 <pre>--------------------</pre>
 <h1>message</h1>
 <pre>
 From: u1
 To: u2@example.com
 Subject: Test Message
 --------------------
 This is the body of the test message
  - nothing interesting here except indentation.
 --
 This is an automatic notification. Don't reply to this mail.
 --------------------</pre>
 <div class="formatted-fixed">This is the body of the test message
  - nothing interesting here except indentation.</div>
 <br/>
 <br/>
 -- <br/>
 This is an automatic notification. Don&#39;t reply to this mail.
 <pre>--------------------</pre>
 <h1>registration</h1>
 <pre>
 From: u1
 To: u2@example.com
 Subject: New user newbie registered
 --------------------
 Registration body
 View this user here: http://newbie.org
 --
 This is an automatic notification. Don't reply to this mail.
 --------------------</pre>
 <div class="formatted-fixed">Registration body</div>
 View this user here: <a href="http://newbie.org">http://newbie.org</a>
 <br/>
 <br/>
 -- <br/>
 This is an automatic notification. Don&#39;t reply to this mail.
 <pre>--------------------</pre>
 <h1>pull_request, is_mention=False</h1>
 <pre>
 From: u1
 To: u2@example.com
 Subject: [Added] repo/name pull request 7 from ref
 --------------------
 Requester Name requested your review of repo/name pull request "The Title"
 URL: http://pr.org
 Description:
 This PR is awesome because it does stuff
  - please approve indented!
 Changesets:
 abc: http://changeset_home/?repo_name=repo_org&amp;revision=123abc
 Introduce one and two
 fed: http://changeset_home/?repo_name=repo_org&amp;revision=567fed
 Make one plus two equal tree
 --
 This is an automatic notification. Don't reply to this mail.
 --------------------</pre>
 <p>Requester Name requested your review of repo/name pull request &#34;The Title&#34;</p>
 <p>URL: <a href="http://pr.org">http://pr.org</a></p>
 <p>Description:</p>
 <p style="white-space: pre-wrap; font-family: monospace"><div class="formatted-fixed">This PR is awesome because it does stuff
  - please approve indented!</div></p>
 <p>Changesets:</p>
 <p style="white-space: pre-wrap">
 <i><a href="http://changeset_home/?repo_name=repo_org&amp;revision=123abc">123abc</a></i>:
 Introduce one and two
 <i><a href="http://changeset_home/?repo_name=repo_org&amp;revision=567fed">567fed</a></i>:
 Make one plus two equal tree
 </p>
 <br/>
 <br/>
 -- <br/>
 This is an automatic notification. Don&#39;t reply to this mail.
 <pre>--------------------</pre>
 <h1>pull_request, is_mention=True</h1>
 <pre>
 From: u1
 To: u2@example.com
 Subject: [Added] repo/name pull request 7 from ref
 --------------------
 Requester Name mentioned you on repo/name pull request "The Title"
 URL: http://pr.org
 Description:
 This PR is awesome because it does stuff
  - please approve indented!
 Changesets:
 abc: http://changeset_home/?repo_name=repo_org&amp;revision=123abc
 Introduce one and two
 fed: http://changeset_home/?repo_name=repo_org&amp;revision=567fed
 Make one plus two equal tree
 --
 This is an automatic notification. Don't reply to this mail.
 --------------------</pre>
 <p>Requester Name mentioned you on repo/name pull request &#34;The Title&#34;</p>
 <p>URL: <a href="http://pr.org">http://pr.org</a></p>
 <p>Description:</p>
 <p style="white-space: pre-wrap; font-family: monospace"><div class="formatted-fixed">This PR is awesome because it does stuff
  - please approve indented!</div></p>
 <p>Changesets:</p>
 <p style="white-space: pre-wrap">
 <i><a href="http://changeset_home/?repo_name=repo_org&amp;revision=123abc">123abc</a></i>:
 Introduce one and two
 <i><a href="http://changeset_home/?repo_name=repo_org&amp;revision=567fed">567fed</a></i>:
 Make one plus two equal tree
 </p>
 <br/>
 <br/>
 -- <br/>
 This is an automatic notification. Don&#39;t reply to this mail.
 <pre>--------------------</pre>
 <h1>pull_request_comment, status_change=None, closing_pr=False</h1>
 <pre>
 From: u1
 To: u2@example.com
 Subject: [Comment] repo/name pull request 7 from ref
 --------------------
 Comment from Commenter Name on repo/name pull request "The Title":
 Me too!
  - and indented on second line
 URL: http://pr.org/comment
 --
 This is an automatic notification. Don't reply to this mail.
 --------------------</pre>
 <p>Comment from Commenter Name on repo/name pull request &#34;The Title&#34;:</p>
 <p><div class="formatted-fixed">Me too!
  - and indented on second line</div></p>
 <p>URL: <a href="http://pr.org/comment">http://pr.org/comment</a></p>
 <br/>
 <br/>
 -- <br/>
 This is an automatic notification. Don&#39;t reply to this mail.
 <pre>--------------------</pre>
 <h1>pull_request_comment, status_change='Under Review', closing_pr=False</h1>
 <pre>
 From: u1
 To: u2@example.com
 Subject: [Under Review: Comment] repo/name pull request 7 from ref
 --------------------
 Comment from Commenter Name on repo/name pull request "The Title":
 Me too!
  - and indented on second line
 The comment was made with status: Under Review
 URL: http://pr.org/comment
 --
 This is an automatic notification. Don't reply to this mail.
 --------------------</pre>
 <p>Comment from Commenter Name on repo/name pull request &#34;The Title&#34;:</p>
 <p><div class="formatted-fixed">Me too!
  - and indented on second line</div></p>
        <p>The comment was made with status: <b>Under Review</b></p>
 <p>URL: <a href="http://pr.org/comment">http://pr.org/comment</a></p>
 <br/>
 <br/>
 -- <br/>
 This is an automatic notification. Don&#39;t reply to this mail.
 <pre>--------------------</pre>
 <h1>pull_request_comment, status_change=None, closing_pr=True</h1>
 <pre>
 From: u1
 To: u2@example.com
 Subject: [Closing: Comment] repo/name pull request 7 from ref
 --------------------
 Comment from Commenter Name on repo/name pull request "The Title":
 Me too!
  - and indented on second line
 URL: http://pr.org/comment
 --
 This is an automatic notification. Don't reply to this mail.
 --------------------</pre>
 <p>Comment from Commenter Name on repo/name pull request &#34;The Title&#34;:</p>
 <p><div class="formatted-fixed">Me too!
  - and indented on second line</div></p>
 <p>URL: <a href="http://pr.org/comment">http://pr.org/comment</a></p>
 <br/>
 <br/>
 -- <br/>
 This is an automatic notification. Don&#39;t reply to this mail.
 <pre>--------------------</pre>
 <h1>pull_request_comment, status_change='Under Review', closing_pr=True</h1>
 <pre>
 From: u1
 To: u2@example.com
 Subject: [Under Review, Closing: Comment] repo/name pull request 7 from ref
 --------------------
 Comment from Commenter Name on repo/name pull request "The Title":
 Me too!
  - and indented on second line
 The comment closed the pull request with status: Under Review
 URL: http://pr.org/comment
 --
 This is an automatic notification. Don't reply to this mail.
 --------------------</pre>
 <p>Comment from Commenter Name on repo/name pull request &#34;The Title&#34;:</p>
 <p><div class="formatted-fixed">Me too!
  - and indented on second line</div></p>
        <p>The comment closed the pull request with status: <b>Under Review</b></p>
 <p>URL: <a href="http://pr.org/comment">http://pr.org/comment</a></p>
 <br/>
 <br/>
 -- <br/>
 This is an automatic notification. Don&#39;t reply to this mail.
 <pre>--------------------</pre>
 <h1>TYPE_PASSWORD_RESET</h1>
 <pre>
 From: u1
 To: john@doe.com
 Subject: Password reset link
 --------------------
 Hello John Doe
 We have received a request to reset the password for your account.
 To set a new password, click the following link:
 http://reset.com/decbf64715098db5b0bd23eab44bd792670ab746
 Should you not be able to use the link above, please type the following code into the password reset form: decbf64715098db5b0bd23eab44bd792670ab746
 If it weren't you who requested the password reset, just disregard this message.
 --
 This is an automatic notification. Don't reply to this mail.
 --------------------</pre>
 <h4>Hello John Doe</h4>
 <p>We have received a request to reset the password for your account.</p>
 <p>To set a new password, click the following link:</p>
 <p><a href="http://reset.com/decbf64715098db5b0bd23eab44bd792670ab746">http://reset.com/decbf64715098db5b0bd23eab44bd792670ab746</a></p>
 <p>Should you not be able to use the link above, please type the following code into the password reset form: <code>decbf64715098db5b0bd23eab44bd792670ab746</code></p>
 <p>If it weren&#39;t you who requested the password reset, just disregard this message.</p>
 <br/>
 <br/>
 -- <br/>
 This is an automatic notification. Don&#39;t reply to this mail.
 <pre>--------------------</pre>
 </body></html>

kallithea/tests/models/test_notifications.py

➞

Show inline comments

 import os
 import mock
 import routes.util
 from kallithea.tests 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
 from kallithea.model.meta import Session
 from kallithea.model.notification import NotificationModel
 import kallithea.lib.celerylib
 import kallithea.lib.celerylib.tasks
 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):
         usrs = [self.u1, self.u2]
         notification = NotificationModel().create(created_by=self.u1,
                                            subject=u'subj', body=u'hi there',
                                            recipients=usrs)
         def send_email(recipients, subject, body='', html_body='', headers=None, author=None):
             assert recipients == ['u2@example.com']
             assert subject == 'Test Message'
             assert body == "\n\nhi there\n\n\n-- \nThis is an automatic notification. Don't reply to this mail.\n"
             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.user_id for x in unotification]) == set(usrs)
     def test_user_notifications(self):
         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):
         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 not notification in notifications
         un = UserNotification.query().filter(UserNotification.notification
                                              == notification).all()
         assert un == []
     def test_delete_association(self):
         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):
         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('\n\n<h1>%s</h1>\n' % desc) # desc is from outer scope
             l.append('<pre>\n\n')
             l.append('From: %s\n' % author.username)
             l.append('To: %s\n' % ' '.join(recipients))
             l.append('Subject: %s\n' % subject)
             l.append('\n--------------------\n%s\n--------------------' % body)
             l.append('</pre>\n')
             l.append('\n%s\n' % html_body)
             l.append('<pre>--------------------</pre>\n')
         l.append('<html><body>\n')
         with mock.patch.object(kallithea.lib.celerylib.tasks, 'send_email', send_email):
             pr_kwargs = dict(pr_nice_id='7', ref='ref', org_repo_name='repo_org', pr_title='The Title', pr_url='http://pr.org')
             for type_, body, kwargs in [
                 (Notification.TYPE_CHANGESET_COMMENT, 'This is the new comment.\n\n - and here it ends indented.', dict(short_id='cafe', raw_id='c0ffeecafe', branch='brunch', cs_comment_user='Commenter Name', cs_comment_url='http://comment.org', is_mention=[False, True], message='This changeset did something clever which is hard to explain', status_change=[None, 'Approved'], cs_target_repo='repo_target', cs_url='http://changeset.com')),
                 (Notification.TYPE_MESSAGE, '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, '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, 'This PR is awesome because it does stuff\n - please approve indented!', dict(pr_user_created='Requester Name', is_mention=[False, True], pr_revisions=[('123abc', 'Introduce one and two'), ('567fed', 'Make one plus two equal tree')], **pr_kwargs)),
                 (Notification.TYPE_PULL_REQUEST_COMMENT, 'Me too!\n\n - and indented on second line', dict(closing_pr=[False, True], pr_comment_user='Commenter Name', pr_comment_url='http://pr.org/comment', status_change=[None, 'Under Review'], pr_target_repo='http://target.com/repo', **pr_kwargs)),
                 ]:
                 kwargs['repo_name'] = 'repo/name'
                 params = [(type_, type_, body, kwargs)]
                 for param_name in ['is_mention', 'status_change', 'closing_pr']: # TODO: inline/general
                     if not isinstance(kwargs.get(param_name), list):
                         continue
                     new_params = []
                     for v in kwargs[param_name]:
                         for desc, type_, body, kwargs in params:
                             kwargs = dict(kwargs)
                             kwargs[param_name] = v
                             new_params.append(('%s, %s=%r' % (desc, param_name, v), type_, body, kwargs))
                     params = new_params
                 for desc, type_, body, kwargs in params:
                     # desc is used as "global" variable
                     notification = NotificationModel().create(created_by=self.u1,
                                                        subject='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.run_task(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))
         l.append('\n</body></html>\n')
         out = ''.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 file(outfn, 'w') as f:
             f.write(out)
         with file(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)