Changeset - a33d1337db05
Parent rev.
Child rev.
[Not reviewed]
default
1 4 0
Thomas De Schampheleire - 7 years ago 2018-12-05 07:40:00
thomas.de_schampheleire@nokia.com
tests: remove tests of UI notifications

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.
5 files changed with 4 insertions and 378 deletions:
kallithea/tests/base.py
1
10
kallithea/tests/functional/test_admin_notifications.py
172
kallithea/tests/functional/test_changeset_pullrequests_comments.py
1
75
kallithea/tests/functional/test_login.py
1
4
kallithea/tests/models/test_notifications.py
1
117
0 comments (0 inline, 0 general)
kallithea/tests/base.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/>.
 

			




	
	
	
		
 
import datetime

			




	
	
	
		
 
import logging

			




	
	
	
		
 
import os

			




	
	
	
		
 
import pytest

			




	
	
	
		
 
import re

			




	
	
	
		
 
import tempfile

			




	
	
	
		
 
import time

			




	
	
	
		
 

			




	
	
	
		
 
from tg import config

			




	
	
	
		
 
from webtest import TestApp

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea import model

			




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

			




	
	
	
		
 
from kallithea.model.db import User

			




	
	
	
		
 
from kallithea.model.meta import Session

			




	
	
	
		
 
from kallithea.lib.utils2 import safe_str

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
log = logging.getLogger(__name__)

			




	
	
	
		
 

			




	
	
	
		
 
skipif = pytest.mark.skipif

			




	
	
	
		
 
parametrize = pytest.mark.parametrize

			




	
	
	
		
 

			




	
	
	
		
 
# Hack: These module global values MUST be set to actual values before running any tests. This is currently done by conftest.py.

			




	
	
	
		
 
url = None

			




	
	
	
		
 
testapp = None

			




	
	
	
		
 

			




	
	
	
		
 
__all__ = [

			




	
	
	
		
 
    'skipif', 'parametrize', 'url', 'TestController',

			




	
	
	
		
 
    'ldap_lib_installed', 'pam_lib_installed', 'invalidate_all_caches',

			




	
	
	
		
 
    'TESTS_TMP_PATH', 'HG_REPO', 'GIT_REPO', 'NEW_HG_REPO', 'NEW_GIT_REPO',

			




	
	
	
		
 
    'HG_FORK', 'GIT_FORK', 'TEST_USER_ADMIN_LOGIN', 'TEST_USER_ADMIN_PASS',

			




	
	
	
		
 
    'TEST_USER_ADMIN_EMAIL', 'TEST_USER_REGULAR_LOGIN', 'TEST_USER_REGULAR_PASS',

			




	
	
	
		
 
    'TEST_USER_REGULAR_EMAIL', 'TEST_USER_REGULAR2_LOGIN',

			




	
	
	
		
 
    'TEST_USER_REGULAR2_PASS', 'TEST_USER_REGULAR2_EMAIL', 'TEST_HG_REPO',

			




	
	
	
		
 
    'TEST_HG_REPO_CLONE', 'TEST_HG_REPO_PULL', 'TEST_GIT_REPO',

			




	
	
	
		
 
    'TEST_GIT_REPO_CLONE', 'TEST_GIT_REPO_PULL', 'HG_REMOTE_REPO',

			




	
	
	
		
 
    'GIT_REMOTE_REPO', 'HG_TEST_REVISION', 'GIT_TEST_REVISION',

			




	
	
	
		
 
]

			




	
	
	
		
 

			




	
	
	
		
 
## SOME GLOBALS FOR TESTS

			




	
	
	
		
 

			




	
	
	
		
 
TESTS_TMP_PATH = os.environ.get('KALLITHEA_TESTS_TMP_PATH', tempfile.mkdtemp(prefix='kallithea-test-'))

			




	
	
	
		
 

			




	
	
	
		
 
TEST_USER_ADMIN_LOGIN = 'test_admin'

			




	
	
	
		
 
TEST_USER_ADMIN_PASS = 'test12'

			




	
	
	
		
 
TEST_USER_ADMIN_EMAIL = 'test_admin@example.com'

			




	
	
	
		
 

			




	
	
	
		
 
TEST_USER_REGULAR_LOGIN = 'test_regular'

			




	
	
	
		
 
TEST_USER_REGULAR_PASS = 'test12'

			




	
	
	
		
 
TEST_USER_REGULAR_EMAIL = 'test_regular@example.com'

			




	
	
	
		
 

			




	
	
	
		
 
TEST_USER_REGULAR2_LOGIN = 'test_regular2'

			




	
	
	
		
 
TEST_USER_REGULAR2_PASS = 'test12'

			




	
	
	
		
 
TEST_USER_REGULAR2_EMAIL = 'test_regular2@example.com'

			




	
	
	
		
 

			




	
	
	
		
 
HG_REPO = u'vcs_test_hg'

			




	
	
	
		
 
GIT_REPO = u'vcs_test_git'

			




	
	
	
		
 

			




	
	
	
		
 
NEW_HG_REPO = u'vcs_test_hg_new'

			




	
	
	
		
 
NEW_GIT_REPO = u'vcs_test_git_new'

			




	
	
	
		
 

			




	
	
		
 
@@ -104,101 +104,92 @@ TEST_HG_REPO_PULL = os.path.join(TESTS_T

			




	
	
	
		
 
# HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'

			




	
	
	
		
 

			




	
	
	
		
 
# skip ldap tests if LDAP lib is not installed

			




	
	
	
		
 
ldap_lib_installed = False

			




	
	
	
		
 
try:

			




	
	
	
		
 
    import ldap

			




	
	
	
		
 
    ldap.API_VERSION

			




	
	
	
		
 
    ldap_lib_installed = True

			




	
	
	
		
 
except ImportError:

			




	
	
	
		
 
    # means that python-ldap is not installed

			




	
	
	
		
 
    pass

			




	
	
	
		
 

			




	
	
	
		
 
try:

			




	
	
	
		
 
    import pam

			




	
	
	
		
 
    pam.PAM_TEXT_INFO

			




	
	
	
		
 
    pam_lib_installed = True

			




	
	
	
		
 
except ImportError:

			




	
	
	
		
 
    pam_lib_installed = False

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def invalidate_all_caches():

			




	
	
	
		
 
    """Invalidate all beaker caches currently configured.

			




	
	
	
		
 
    Useful when manipulating IP permissions in a test and changes need to take

			




	
	
	
		
 
    effect immediately.

			




	
	
	
		
 
    Note: Any use of this function is probably a workaround - it should be

			




	
	
	
		
 
    replaced with a more specific cache invalidation in code or test."""

			




	
	
	
		
 
    from beaker.cache import cache_managers

			




	
	
	
		
 
    for cache in cache_managers.values():

			




	
	
	
		
 
        cache.clear()

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class NullHandler(logging.Handler):

			




	
	
	
		
 
    def emit(self, record):

			




	
	
	
		
 
        pass

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TestController(object):

			




	
	
	
		
 
    """Pytest-style test controller"""

			




	
	
	
		
 

			




	
	
	
		
 
    # Note: pytest base classes cannot have an __init__ method

			




	
	
	
		
 

			




	
	
	
		
 
    @pytest.fixture(autouse=True)

			




	
	
	
		
 
    def app_fixture(self):

			




	
	
	
		
 
        h = NullHandler()

			




	
	
	
		
 
        logging.getLogger("kallithea").addHandler(h)

			




	
	
	
		
 
        self.app = TestApp(testapp)

			




	
	
	
		
 
        return self.app

			




	
	
	
		
 

			




	
	
	
		
 
    def remove_all_notifications(self):

			




	
	
	
		
 
        # query().delete() does not (by default) trigger cascades

			




	
	
	
		
 
        # ( http://docs.sqlalchemy.org/en/rel_0_7/orm/collections.html#passive-deletes )

			




	
	
	
		
 
        # so delete the UserNotification first to ensure referential integrity.

			




	
	
	
		
 
        UserNotification.query().delete()

			




	
	
	
		
 

			




	
	
	
		
 
        Notification.query().delete()

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
    def log_user(self, username=TEST_USER_ADMIN_LOGIN,

			




	
	
	
		
 
                 password=TEST_USER_ADMIN_PASS):

			




	
	
	
		
 
        self._logged_username = username

			




	
	
	
		
 
        response = self.app.post(url(controller='login', action='index'),

			




	
	
	
		
 
                                 {'username': username,

			




	
	
	
		
 
                                  'password': password})

			




	
	
	
		
 

			




	
	
	
		
 
        if 'Invalid username or password' in response.body:

			




	
	
	
		
 
            pytest.fail('could not login using %s %s' % (username, password))

			




	
	
	
		
 

			




	
	
	
		
 
        assert response.status == '302 Found'

			




	
	
	
		
 
        self.assert_authenticated_user(response, username)

			




	
	
	
		
 

			




	
	
	
		
 
        response = response.follow()

			




	
	
	
		
 
        return response.session['authuser']

			




	
	
	
		
 

			




	
	
	
		
 
    def _get_logged_user(self):

			




	
	
	
		
 
        return User.get_by_username(self._logged_username)

			




	
	
	
		
 

			




	
	
	
		
 
    def assert_authenticated_user(self, response, expected_username):

			




	
	
	
		
 
        cookie = response.session.get('authuser')

			




	
	
	
		
 
        user = cookie and cookie.get('user_id')

			




	
	
	
		
 
        user = user and User.get(user)

			




	
	
	
		
 
        user = user and user.username

			




	
	
	
		
 
        assert user == expected_username

			




	
	
	
		
 

			




	
	
	
		
 
    def authentication_token(self):

			




	
	
	
		
 
        return self.app.get(url('authentication_token')).body

			




	
	
	
		
 

			




	
	
	
		
 
    def checkSessionFlash(self, response, msg=None, skip=0, _matcher=lambda msg, m: msg in m):

			




	
	
	
		
 
        if 'flash' not in response.session:

			




	
	
	
		
 
            pytest.fail(safe_str(u'msg `%s` not found - session has no flash:\n%s' % (msg, response)))

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            level, m = response.session['flash'][-1 - skip]

			




	
	
	
		
 
            if _matcher(msg, m):

			




	
	
	
		
 
                return

			




	
	
	
		
 
        except IndexError:

			




	
	
	
		
 
            pass

			




	
	
	
		
 
        pytest.fail(safe_str(u'msg `%s` not found in session flash (skipping %s): %s' %

			




	
	
	
		
 
                           (msg, skip,

			




	
	
	
		
 
                            ', '.join('`%s`' % m for level, m in response.session['flash']))))

			




	
	
	
		
 

			




	
	
	
		
 
    def checkSessionFlashRegex(self, response, regex, skip=0):

			




	
	
	
		
 
        self.checkSessionFlash(response, regex, skip=skip, _matcher=re.search)

			




        

    


    
    

    

        

                

                    kallithea/tests/functional/test_admin_notifications.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
deleted file

			




        

    


    
    

    

        

                

                    kallithea/tests/functional/test_changeset_pullrequests_comments.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 
import re

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.tests.base import *

			




	
	
	
		
 
from kallithea.model.changeset_status import ChangesetStatusModel

			




	
	
	
		
 
from kallithea.model.db import ChangesetComment, Notification, \

			




	
	
	
		
 
    UserNotification

			




	
	
	
		
 
from kallithea.model.db import ChangesetComment

			




	
	
	
		
 
from kallithea.model.meta import Session

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TestChangeSetCommentsController(TestController):

			




	
	
	
		
 

			




	
	
	
		
 
    def setup_method(self, method):

			




	
	
	
		
 
        for x in ChangesetComment.query().all():

			




	
	
	
		
 
            Session().delete(x)

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
        self.remove_all_notifications()

			




	
	
	
		
 

			




	
	
	
		
 
    def test_create(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'

			




	
	
	
		
 
        text = u'general comment on changeset'

			




	
	
	
		
 

			




	
	
	
		
 
        params = {'text': text, '_authentication_token': self.authentication_token()}

			




	
	
	
		
 
        response = self.app.post(url(controller='changeset', action='comment',

			




	
	
	
		
 
                                     repo_name=HG_REPO, revision=rev),

			




	
	
	
		
 
                                     params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})

			




	
	
	
		
 
        # Test response...

			




	
	
	
		
 
        assert response.status == '200 OK'

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.get(url(controller='changeset', action='index',

			




	
	
	
		
 
                                repo_name=HG_REPO, revision=rev))

			




	
	
	
		
 
        response.mustcontain(

			




	
	
	
		
 
            '''<div class="comments-number">'''

			




	
	
	
		
 
            ''' 1 comment (0 inline, 1 general)'''

			




	
	
	
		
 
        )

			




	
	
	
		
 
        response.mustcontain(text)

			




	
	
	
		
 

			




	
	
	
		
 
        # test DB

			




	
	
	
		
 
        assert ChangesetComment.query().count() == 1

			




	
	
	
		
 
        assert Notification.query().count() == 1

			




	
	
	
		
 

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        comment_id = ChangesetComment.query().first().comment_id

			




	
	
	
		
 
        assert notification.type_ == Notification.TYPE_CHANGESET_COMMENT

			




	
	
	
		
 
        sbj = (u'/%s/changeset/'

			




	
	
	
		
 
               '27cd5cce30c96924232dffcd24178a07ffeb5dfc#comment-%s'

			




	
	
	
		
 
               % (HG_REPO, comment_id))

			




	
	
	
		
 
        print "%s vs %s" % (sbj, notification.subject)

			




	
	
	
		
 
        assert sbj in notification.subject

			




	
	
	
		
 

			




	
	
	
		
 
    def test_create_inline(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'

			




	
	
	
		
 
        text = u'inline comment on changeset'

			




	
	
	
		
 
        f_path = 'vcs/web/simplevcs/views/repository.py'

			




	
	
	
		
 
        line = 'n1'

			




	
	
	
		
 

			




	
	
	
		
 
        params = {'text': text, 'f_path': f_path, 'line': line, '_authentication_token': self.authentication_token()}

			




	
	
	
		
 
        response = self.app.post(url(controller='changeset', action='comment',

			




	
	
	
		
 
                                     repo_name=HG_REPO, revision=rev),

			




	
	
	
		
 
                                     params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})

			




	
	
	
		
 
        # Test response...

			




	
	
	
		
 
        assert response.status == '200 OK'

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.get(url(controller='changeset', action='index',

			




	
	
	
		
 
                                repo_name=HG_REPO, revision=rev))

			




	
	
	
		
 
        response.mustcontain(

			




	
	
	
		
 
            '''<div class="comments-number">'''

			




	
	
	
		
 
            ''' 1 comment (1 inline, 0 general)'''

			




	
	
	
		
 
        )

			




	
	
	
		
 
        response.mustcontain(

			




	
	
	
		
 
            '''<div class="comments-list-chunk" '''

			




	
	
	
		
 
            '''data-f_path="vcs/web/simplevcs/views/repository.py" '''

			




	
	
	
		
 
            '''data-line_no="n1" data-target-id="vcswebsimplevcsviewsrepositorypy_n1">'''

			




	
	
	
		
 
        )

			




	
	
	
		
 
        response.mustcontain(text)

			




	
	
	
		
 

			




	
	
	
		
 
        # test DB

			




	
	
	
		
 
        assert ChangesetComment.query().count() == 1

			




	
	
	
		
 
        assert Notification.query().count() == 1

			




	
	
	
		
 

			




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

			




	
	
	
		
 
        comment_id = ChangesetComment.query().first().comment_id

			




	
	
	
		
 
        assert notification.type_ == Notification.TYPE_CHANGESET_COMMENT

			




	
	
	
		
 
        sbj = (u'/%s/changeset/'

			




	
	
	
		
 
               '27cd5cce30c96924232dffcd24178a07ffeb5dfc#comment-%s'

			




	
	
	
		
 
               % (HG_REPO, comment_id))

			




	
	
	
		
 
        print "%s vs %s" % (sbj, notification.subject)

			




	
	
	
		
 
        assert sbj in notification.subject

			




	
	
	
		
 

			




	
	
	
		
 
    def test_create_with_mention(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 

			




	
	
	
		
 
        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'

			




	
	
	
		
 
        text = u'@%s check CommentOnRevision' % TEST_USER_REGULAR_LOGIN

			




	
	
	
		
 

			




	
	
	
		
 
        params = {'text': text, '_authentication_token': self.authentication_token()}

			




	
	
	
		
 
        response = self.app.post(url(controller='changeset', action='comment',

			




	
	
	
		
 
                                     repo_name=HG_REPO, revision=rev),

			




	
	
	
		
 
                                     params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})

			




	
	
	
		
 
        # Test response...

			




	
	
	
		
 
        assert response.status == '200 OK'

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.get(url(controller='changeset', action='index',

			




	
	
	
		
 
                                repo_name=HG_REPO, revision=rev))

			




	
	
	
		
 
        response.mustcontain(

			




	
	
	
		
 
            '''<div class="comments-number">'''

			




	
	
	
		
 
            ''' 1 comment (0 inline, 1 general)'''

			




	
	
	
		
 
        )

			




	
	
	
		
 
        response.mustcontain('<b>@%s</b> check CommentOnRevision' % TEST_USER_REGULAR_LOGIN)

			




	
	
	
		
 

			




	
	
	
		
 
        # test DB

			




	
	
	
		
 
        assert ChangesetComment.query().count() == 1

			




	
	
	
		
 
        assert Notification.query().count() == 2

			




	
	
	
		
 
        users = [x.user.username for x in UserNotification.query().all()]

			




	
	
	
		
 

			




	
	
	
		
 
        # test_regular gets notification by @mention

			




	
	
	
		
 
        assert sorted(users) == [TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN]

			




	
	
	
		
 

			




	
	
	
		
 
    def test_create_status_change(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'

			




	
	
	
		
 
        text = u'general comment on changeset'

			




	
	
	
		
 

			




	
	
	
		
 
        params = {'text': text, 'changeset_status': 'rejected',

			




	
	
	
		
 
                '_authentication_token': self.authentication_token()}

			




	
	
	
		
 
        response = self.app.post(url(controller='changeset', action='comment',

			




	
	
	
		
 
                                     repo_name=HG_REPO, revision=rev),

			




	
	
	
		
 
                                     params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})

			




	
	
	
		
 
        # Test response...

			




	
	
	
		
 
        assert response.status == '200 OK'

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.get(url(controller='changeset', action='index',

			




	
	
	
		
 
                                repo_name=HG_REPO, revision=rev))

			




	
	
	
		
 
        response.mustcontain(

			




	
	
	
		
 
            '''<div class="comments-number">'''

			




	
	
	
		
 
            ''' 1 comment (0 inline, 1 general)'''

			




	
	
	
		
 
        )

			




	
	
	
		
 
        response.mustcontain(text)

			




	
	
	
		
 

			




	
	
	
		
 
        # test DB

			




	
	
	
		
 
        assert ChangesetComment.query().count() == 1

			




	
	
	
		
 
        assert Notification.query().count() == 1

			




	
	
	
		
 

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        comment_id = ChangesetComment.query().first().comment_id

			




	
	
	
		
 
        assert notification.type_ == Notification.TYPE_CHANGESET_COMMENT

			




	
	
	
		
 
        sbj = (u'/%s/changeset/'

			




	
	
	
		
 
               '27cd5cce30c96924232dffcd24178a07ffeb5dfc#comment-%s'

			




	
	
	
		
 
               % (HG_REPO, comment_id))

			




	
	
	
		
 
        print "%s vs %s" % (sbj, notification.subject)

			




	
	
	
		
 
        assert sbj in notification.subject

			




	
	
	
		
 

			




	
	
	
		
 
        # check status

			




	
	
	
		
 
        status = ChangesetStatusModel().get_status(repo=HG_REPO, revision=rev)

			




	
	
	
		
 
        assert status == 'rejected'

			




	
	
	
		
 

			




	
	
	
		
 
    def test_delete(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'

			




	
	
	
		
 
        text = u'general comment on changeset to be deleted'

			




	
	
	
		
 

			




	
	
	
		
 
        params = {'text': text, '_authentication_token': self.authentication_token()}

			




	
	
	
		
 
        response = self.app.post(url(controller='changeset', action='comment',

			




	
	
	
		
 
                                     repo_name=HG_REPO, revision=rev),

			




	
	
	
		
 
                                     params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})

			




	
	
	
		
 

			




	
	
	
		
 
        comments = ChangesetComment.query().all()

			




	
	
	
		
 
        assert len(comments) == 1

			




	
	
	
		
 
        comment_id = comments[0].comment_id

			




	
	
	
		
 

			




	
	
	
		
 
        self.app.post(url("changeset_comment_delete",

			




	
	
	
		
 
                                    repo_name=HG_REPO,

			




	
	
	
		
 
                                    comment_id=comment_id),

			




	
	
	
		
 
            params={'_authentication_token': self.authentication_token()})

			




	
	
	
		
 

			




	
	
	
		
 
        comments = ChangesetComment.query().all()

			




	
	
	
		
 
        assert len(comments) == 0

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.get(url(controller='changeset', action='index',

			




	
	
	
		
 
                                repo_name=HG_REPO, revision=rev))

			




	
	
	
		
 
        response.mustcontain(

			




	
	
	
		
 
            '''<div class="comments-number">'''

			




	
	
	
		
 
            ''' 0 comments (0 inline, 0 general)'''

			




	
	
	
		
 
        )

			




	
	
	
		
 
        response.mustcontain(no=text)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TestPullrequestsCommentsController(TestController):

			




	
	
	
		
 

			




	
	
	
		
 
    def setup_method(self, method):

			




	
	
	
		
 
        for x in ChangesetComment.query().all():

			




	
	
	
		
 
            Session().delete(x)

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
        self.remove_all_notifications()

			




	
	
	
		
 

			




	
	
	
		
 
    def _create_pr(self):

			




	
	
	
		
 
        response = self.app.post(url(controller='pullrequests', action='create',

			




	
	
	
		
 
                                     repo_name=HG_REPO),

			




	
	
	
		
 
                                 {'org_repo': HG_REPO,

			




	
	
	
		
 
                                  'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',

			




	
	
	
		
 
                                  'other_repo': HG_REPO,

			




	
	
	
		
 
                                  'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',

			




	
	
	
		
 
                                  'pullrequest_title': 'title',

			




	
	
	
		
 
                                  'pullrequest_desc': 'description',

			




	
	
	
		
 
                                  '_authentication_token': self.authentication_token(),

			




	
	
	
		
 
                                 },

			




	
	
	
		
 
                                 status=302)

			




	
	
	
		
 
        pr_id = int(re.search('/pull-request/(\d+)/', response.location).group(1))

			




	
	
	
		
 
        return pr_id

			




	
	
	
		
 

			




	
	
	
		
 
    def test_create(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        pr_id = self._create_pr()

			




	
	
	
		
 

			




	
	
	
		
 
        text = u'general comment on pullrequest'

			




	
	
	
		
 
        params = {'text': text, '_authentication_token': self.authentication_token()}

			




	
	
	
		
 
        response = self.app.post(url(controller='pullrequests', action='comment',

			




	
	
	
		
 
                                     repo_name=HG_REPO, pull_request_id=pr_id),

			




	
	
	
		
 
                                     params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})

			




	
	
	
		
 
        # Test response...

			




	
	
	
		
 
        assert response.status == '200 OK'

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.get(url(controller='pullrequests', action='show',

			




	
	
	
		
 
                                repo_name=HG_REPO, pull_request_id=pr_id, extra=''))

			




	
	
	
		
 
        # PRs currently always have an initial 'Under Review' status change

			




	
	
	
		
 
        # that counts as a general comment, hence '2' in the test below. That

			




	
	
	
		
 
        # could be counted as a misfeature, to be reworked later.

			




	
	
	
		
 
        response.mustcontain(

			




	
	
	
		
 
            '''<div class="comments-number">'''

			




	
	
	
		
 
            ''' 2 comments (0 inline, 2 general)'''

			




	
	
	
		
 
        )

			




	
	
	
		
 
        response.mustcontain(text)

			




	
	
	
		
 

			




	
	
	
		
 
        # test DB

			




	
	
	
		
 
        assert ChangesetComment.query().count() == 2

			




	
	
	
		
 
        assert Notification.query().count() == 1

			




	
	
	
		
 

			




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

			




	
	
	
		
 
        comment_id = ChangesetComment.query().order_by(ChangesetComment.comment_id.desc()).first().comment_id

			




	
	
	
		
 
        assert notification.type_ == Notification.TYPE_PULL_REQUEST_COMMENT

			




	
	
	
		
 
        sbj = (u'/%s/pull-request/%s/_/stable#comment-%s'

			




	
	
	
		
 
               % (HG_REPO, pr_id, comment_id))

			




	
	
	
		
 
        print "%s vs %s" % (sbj, notification.subject)

			




	
	
	
		
 
        assert sbj in notification.subject

			




	
	
	
		
 

			




	
	
	
		
 
    def test_create_inline(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        pr_id = self._create_pr()

			




	
	
	
		
 

			




	
	
	
		
 
        text = u'inline comment on changeset'

			




	
	
	
		
 
        f_path = 'vcs/web/simplevcs/views/repository.py'

			




	
	
	
		
 
        line = 'n1'

			




	
	
	
		
 
        params = {'text': text, 'f_path': f_path, 'line': line, '_authentication_token': self.authentication_token()}

			




	
	
	
		
 
        response = self.app.post(url(controller='pullrequests', action='comment',

			




	
	
	
		
 
                                     repo_name=HG_REPO, pull_request_id=pr_id),

			




	
	
	
		
 
                                     params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})

			




	
	
	
		
 
        # Test response...

			




	
	
	
		
 
        assert response.status == '200 OK'

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.get(url(controller='pullrequests', action='show',

			




	
	
	
		
 
                                repo_name=HG_REPO, pull_request_id=pr_id, extra=''))

			




	
	
	
		
 
        response.mustcontain(

			




	
	
	
		
 
            '''<div class="comments-number">'''

			




	
	
	
		
 
            ''' 2 comments (1 inline, 1 general)'''

			




	
	
	
		
 
        )

			




	
	
	
		
 
        response.mustcontain(

			




	
	
	
		
 
            '''<div class="comments-list-chunk" '''

			




	
	
	
		
 
            '''data-f_path="vcs/web/simplevcs/views/repository.py" '''

			




	
	
	
		
 
            '''data-line_no="n1" data-target-id="vcswebsimplevcsviewsrepositorypy_n1">'''

			




	
	
	
		
 
        )

			




	
	
	
		
 
        response.mustcontain(text)

			




	
	
	
		
 

			




	
	
	
		
 
        # test DB

			




	
	
	
		
 
        assert ChangesetComment.query().count() == 2

			




	
	
	
		
 
        assert Notification.query().count() == 1

			




	
	
	
		
 

			




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

			




	
	
	
		
 
        comment_id = ChangesetComment.query().order_by(ChangesetComment.comment_id.desc()).first().comment_id

			




	
	
	
		
 
        assert notification.type_ == Notification.TYPE_PULL_REQUEST_COMMENT

			




	
	
	
		
 
        sbj = (u'/%s/pull-request/%s/_/stable#comment-%s'

			




	
	
	
		
 
               % (HG_REPO, pr_id, comment_id))

			




	
	
	
		
 
        print "%s vs %s" % (sbj, notification.subject)

			




	
	
	
		
 
        assert sbj in notification.subject

			




	
	
	
		
 

			




	
	
	
		
 
    def test_create_with_mention(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        pr_id = self._create_pr()

			




	
	
	
		
 

			




	
	
	
		
 
        text = u'@%s check CommentOnRevision' % TEST_USER_REGULAR_LOGIN

			




	
	
	
		
 
        params = {'text': text, '_authentication_token': self.authentication_token()}

			




	
	
	
		
 
        response = self.app.post(url(controller='pullrequests', action='comment',

			




	
	
	
		
 
                                     repo_name=HG_REPO, pull_request_id=pr_id),

			




	
	
	
		
 
                                     params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})

			




	
	
	
		
 
        # Test response...

			




	
	
	
		
 
        assert response.status == '200 OK'

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.get(url(controller='pullrequests', action='show',

			




	
	
	
		
 
                                repo_name=HG_REPO, pull_request_id=pr_id, extra=''))

			




	
	
	
		
 
        response.mustcontain(

			




	
	
	
		
 
            '''<div class="comments-number">'''

			




	
	
	
		
 
            ''' 2 comments (0 inline, 2 general)'''

			




	
	
	
		
 
        )

			




	
	
	
		
 
        response.mustcontain('<b>@%s</b> check CommentOnRevision' % TEST_USER_REGULAR_LOGIN)

			




	
	
	
		
 

			




	
	
	
		
 
        # test DB

			




	
	
	
		
 
        assert ChangesetComment.query().count() == 2

			




	
	
	
		
 
        assert Notification.query().count() == 2

			




	
	
	
		
 
        users = [x.user.username for x in UserNotification.query().all()]

			




	
	
	
		
 

			




	
	
	
		
 
        # test_regular gets notification by @mention

			




	
	
	
		
 
        assert sorted(users) == [TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN]

			




	
	
	
		
 

			




	
	
	
		
 
    def test_create_status_change(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        pr_id = self._create_pr()

			




	
	
	
		
 

			




	
	
	
		
 
        text = u'general comment on pullrequest'

			




	
	
	
		
 
        params = {'text': text, 'changeset_status': 'rejected',

			




	
	
	
		
 
                '_authentication_token': self.authentication_token()}

			




	
	
	
		
 
        response = self.app.post(url(controller='pullrequests', action='comment',

			




	
	
	
		
 
                                     repo_name=HG_REPO, pull_request_id=pr_id),

			




	
	
	
		
 
                                     params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})

			




	
	
	
		
 
        # Test response...

			




	
	
	
		
 
        assert response.status == '200 OK'

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.get(url(controller='pullrequests', action='show',

			




	
	
	
		
 
                                repo_name=HG_REPO, pull_request_id=pr_id, extra=''))

			




	
	
	
		
 
        # PRs currently always have an initial 'Under Review' status change

			




	
	
	
		
 
        # that counts as a general comment, hence '2' in the test below. That

			




	
	
	
		
 
        # could be counted as a misfeature, to be reworked later.

			




	
	
	
		
 
        response.mustcontain(

			




	
	
	
		
 
            '''<div class="comments-number">'''

			




	
	
	
		
 
            ''' 2 comments (0 inline, 2 general)'''

			




	
	
	
		
 
        )

			




	
	
	
		
 
        response.mustcontain(text)

			




	
	
	
		
 

			




	
	
	
		
 
        # test DB

			




	
	
	
		
 
        assert ChangesetComment.query().count() == 2

			




	
	
	
		
 
        assert Notification.query().count() == 1

			




	
	
	
		
 

			




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

			




	
	
	
		
 
        comment_id = ChangesetComment.query().order_by(ChangesetComment.comment_id.desc()).first().comment_id

			




	
	
	
		
 
        assert notification.type_ == Notification.TYPE_PULL_REQUEST_COMMENT

			




	
	
	
		
 
        sbj = (u'/%s/pull-request/%s/_/stable#comment-%s'

			




	
	
	
		
 
               % (HG_REPO, pr_id, comment_id))

			




	
	
	
		
 
        print "%s vs %s" % (sbj, notification.subject)

			




	
	
	
		
 
        assert sbj in notification.subject

			




	
	
	
		
 

			




	
	
	
		
 
        # check status

			




	
	
	
		
 
        status = ChangesetStatusModel().get_status(repo=HG_REPO, pull_request=pr_id)

			




	
	
	
		
 
        assert status == 'rejected'

			




	
	
	
		
 

			




	
	
	
		
 
    def test_delete(self):

			




	
	
	
		
 
        self.log_user()

			




	
	
	
		
 
        pr_id = self._create_pr()

			




	
	
	
		
 

			




	
	
	
		
 
        text = u'general comment on changeset to be deleted'

			




	
	
	
		
 
        params = {'text': text, '_authentication_token': self.authentication_token()}

			




	
	
	
		
 
        response = self.app.post(url(controller='pullrequests', action='comment',

			




	
	
	
		
 
                                     repo_name=HG_REPO, pull_request_id=pr_id),

			




	
	
	
		
 
                                     params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})

			




	
	
	
		
 

			




	
	
	
		
 
        comments = ChangesetComment.query().all()

			




	
	
	
		
 
        assert len(comments) == 2

			




	
	
	
		
 
        comment_id = comments[-1].comment_id

			




	
	
	
		
 

			




	
	
	
		
 
        self.app.post(url("pullrequest_comment_delete",

			




	
	
	
		
 
                                    repo_name=HG_REPO,

			




	
	
	
		
 
                                    comment_id=comment_id),

			




	
	
	
		
 
            params={'_authentication_token': self.authentication_token()})

			




	
	
	
		
 

			




	
	
	
		
 
        comments = ChangesetComment.query().all()

			




	
	
	
		
 
        assert len(comments) == 1

			




	
	
	
		
 

			




	
	
	
		
 
        response = self.app.get(url(controller='pullrequests', action='show',

			




	
	
	
		
 
                                repo_name=HG_REPO, pull_request_id=pr_id, extra=''))

			




	
	
	
		
 
        response.mustcontain(

			




	
	
	
		
 
            '''<div class="comments-number">'''

			




	
	
	
		
 
            ''' 1 comment (0 inline, 1 general)'''

			




	
	
	
		
 
        )

			




	
	
	
		
 
        response.mustcontain(no=text)

			




        

    


    
    

    

        

                

                    kallithea/tests/functional/test_login.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 
# -*- coding: utf-8 -*-

			




	
	
	
		
 
import re

			




	
	
	
		
 
import time

			




	
	
	
		
 
import urlparse

			




	
	
	
		
 

			




	
	
	
		
 
import mock

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.tests.base import *

			




	
	
	
		
 
from kallithea.tests.fixture import Fixture

			




	
	
	
		
 
from kallithea.lib.utils2 import generate_api_key

			




	
	
	
		
 
from kallithea.lib.auth import check_password

			




	
	
	
		
 
from kallithea.lib import helpers as h

			




	
	
	
		
 
from kallithea.model.api_key import ApiKeyModel

			




	
	
	
		
 
from kallithea.model import validators

			




	
	
	
		
 
from kallithea.model.db import User, Notification

			




	
	
	
		
 
from kallithea.model.db import User

			




	
	
	
		
 
from kallithea.model.meta import Session

			




	
	
	
		
 
from kallithea.model.user import UserModel

			




	
	
	
		
 

			




	
	
	
		
 
from tg.util.webtest import test_context

			




	
	
	
		
 

			




	
	
	
		
 
fixture = Fixture()

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class TestLoginController(TestController):

			




	
	
	
		
 
    def setup_method(self, method):

			




	
	
	
		
 
        self.remove_all_notifications()

			




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

			




	
	
	
		
 

			




	
	
	
		
 
    def test_index(self):

			




	
	
	
		
 
        response = self.app.get(url(controller='login', action='index'))

			




	
	
	
		
 
        assert response.status == '200 OK'

			




	
	
	
		
 
        # Test response...

			




	
	
	
		
 

			




	
	
	
		
 
    def test_login_admin_ok(self):

			




	
	
	
		
 
        response = self.app.post(url(controller='login', action='index'),

			




	
	
	
		
 
                                 {'username': TEST_USER_ADMIN_LOGIN,

			




	
	
	
		
 
                                  'password': TEST_USER_ADMIN_PASS})

			




	
	
	
		
 
        assert response.status == '302 Found'

			




	
	
	
		
 
        self.assert_authenticated_user(response, TEST_USER_ADMIN_LOGIN)

			




	
	
	
		
 

			




	
	
	
		
 
        response = response.follow()

			




	
	
	
		
 
        response.mustcontain('/%s' % HG_REPO)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_login_regular_ok(self):

			




	
	
	
		
 
        response = self.app.post(url(controller='login', action='index'),

			




	
	
	
		
 
                                 {'username': TEST_USER_REGULAR_LOGIN,

			




	
	
	
		
 
                                  'password': TEST_USER_REGULAR_PASS})

			




	
	
	
		
 

			




	
	
	
		
 
        assert response.status == '302 Found'

			




	
	
	
		
 
        self.assert_authenticated_user(response, TEST_USER_REGULAR_LOGIN)

			




	
	
	
		
 

			




	
	
	
		
 
        response = response.follow()

			




	
	
	
		
 
        response.mustcontain('/%s' % HG_REPO)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_login_regular_email_ok(self):

			




	
	
	
		
 
        response = self.app.post(url(controller='login', action='index'),

			




	
	
	
		
 
                                 {'username': TEST_USER_REGULAR_EMAIL,

			




	
	
	
		
 
                                  'password': TEST_USER_REGULAR_PASS})

			




	
	
	
		
 

			




	
	
	
		
 
        assert response.status == '302 Found'

			




	
	
	
		
 
        self.assert_authenticated_user(response, TEST_USER_REGULAR_LOGIN)

			




	
	
	
		
 

			




	
	
	
		
 
        response = response.follow()

			




	
	
	
		
 
        response.mustcontain('/%s' % HG_REPO)

			




	
	
	
		
 

			




	
	
	
		
 
    def test_login_ok_came_from(self):

			




	
	
	
		
 
        test_came_from = '/_admin/users'

			




	
	
	
		
 
        response = self.app.post(url(controller='login', action='index',

			




	
	
	
		
 
                                     came_from=test_came_from),

			




	
	
	
		
 
                                 {'username': TEST_USER_ADMIN_LOGIN,

			




	
	
	
		
 
                                  'password': TEST_USER_ADMIN_PASS})

			




	
	
	
		
 
        assert response.status == '302 Found'

			




	
	
	
		
 
        response = response.follow()

			




	
	
	
		
 

			




	
	
	
		
 
        assert response.status == '200 OK'

			




        

    


    
    

    

        

                

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

			




	
	
	
		
 
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

			




	
	
	
		
 

			




	
	
	
		
 
        self.remove_all_notifications()

			




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

			




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

			




	
	
	
		
 

			




	
	
	
		
 
    def test_create_notification(self):

			




	
	
	
		
 
        with test_context(self.app):

			




	
	
	
		
 
            usrs = [self.u1, self.u2]

			




	
	
	
		
 

			




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

			




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

			




	
	
	
		
 
                assert subject == 'Test Message'

			




	
	
	
		
 
                assert body == u"hi there"

			




	
	
	
		
 
                assert '>hi there<' in html_body

			




	
	
	
		
 
                assert author.username == 'u1'

			




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

			




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

			




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

			




	
	
	
		
 
                                                   recipients=usrs)

			




	
	
	
		
 
                Session().commit()

			




	
	
	
		
 
                u1 = User.get(self.u1)

			




	
	
	
		
 
                u2 = User.get(self.u2)

			




	
	
	
		
 
                u3 = User.get(self.u3)

			




	
	
	
		
 
                notifications = Notification.query().all()

			




	
	
	
		
 
                assert len(notifications) == 1

			




	
	
	
		
 

			




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

			




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

			




	
	
	
		
 

			




	
	
	
		
 
                unotification = UserNotification.query() \

			




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

			




	
	
	
		
 

			




	
	
	
		
 
                assert len(unotification) == len(usrs)

			




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

			




	
	
	
		
 

			




	
	
	
		
 
    def test_user_notifications(self):

			




	
	
	
		
 
        with test_context(self.app):

			




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

			




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

			




	
	
	
		
 
                                                recipients=[self.u3])

			




	
	
	
		
 
            Session().commit()

			




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

			




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

			




	
	
	
		
 
                                                recipients=[self.u3])

			




	
	
	
		
 
            Session().commit()

			




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

			




	
	
	
		
 

			




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

			




	
	
	
		
 

			




	
	
	
		
 
    def test_delete_notifications(self):

			




	
	
	
		
 
        with test_context(self.app):

			




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

			




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

			




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

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 
            notifications = Notification.query().all()

			




	
	
	
		
 
            assert notification in notifications

			




	
	
	
		
 

			




	
	
	
		
 
            Notification.delete(notification.notification_id)

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
            notifications = Notification.query().all()

			




	
	
	
		
 
            assert notification not in notifications

			




	
	
	
		
 

			




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

			




	
	
	
		
 
                                                 == notification).all()

			




	
	
	
		
 
            assert un == []

			




	
	
	
		
 

			




	
	
	
		
 
    def test_delete_association(self):

			




	
	
	
		
 
        with test_context(self.app):

			




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

			




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

			




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

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
            unotification = UserNotification.query() \

			




	
	
	
		
 
                                .filter(UserNotification.notification ==

			




	
	
	
		
 
                                        notification) \

			




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

			




	
	
	
		
 
                                .scalar()

			




	
	
	
		
 

			




	
	
	
		
 
            assert unotification.user_id == self.u3

			




	
	
	
		
 

			




	
	
	
		
 
            NotificationModel().delete(self.u3,

			




	
	
	
		
 
                                       notification.notification_id)

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 

			




	
	
	
		
 
            u3notification = UserNotification.query() \

			




	
	
	
		
 
                                .filter(UserNotification.notification ==

			




	
	
	
		
 
                                        notification) \

			




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

			




	
	
	
		
 
                                .scalar()

			




	
	
	
		
 

			




	
	
	
		
 
            assert u3notification == None

			




	
	
	
		
 

			




	
	
	
		
 
            # notification object is still there

			




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

			




	
	
	
		
 

			




	
	
	
		
 
            # u1 and u2 still have assignments

			




	
	
	
		
 
            u1notification = UserNotification.query() \

			




	
	
	
		
 
                                .filter(UserNotification.notification ==

			




	
	
	
		
 
                                        notification) \

			




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

			




	
	
	
		
 
                                .scalar()

			




	
	
	
		
 
            assert u1notification != None

			




	
	
	
		
 
            u2notification = UserNotification.query() \

			




	
	
	
		
 
                                .filter(UserNotification.notification ==

			




	
	
	
		
 
                                        notification) \

			




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

			




	
	
	
		
 
                                .scalar()

			




	
	
	
		
 
            assert u2notification != None

			




	
	
	
		
 

			




	
	
	
		
 
    def test_notification_counter(self):

			




	
	
	
		
 
        with test_context(self.app):

			




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

			




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

			




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

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 

			




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

			




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

			




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

			




	
	
	
		
 

			




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

			




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

			




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

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 

			




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

			




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

			




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

			




	
	
	
		
 

			




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

			




	
	
	
		
 
    def test_dump_html_mails(self):

			




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

			




	
	
	
		
 
        l = []

			




	
	
	
		
 

			




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

			




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

			




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

			




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

			




	
	
	
		
 
            l.append('From: %s\n' % author.username)

			




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

			




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

			




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

			




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

			




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

			




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

			




	
	
	
		
 
            l.append(html_body)

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        with test_context(self.app):

			




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

			




	
	
	
		
 
                pr_kwargs = dict(

			




	
	
	
		
 
                    pr_nice_id='#7',

			




	
	
	
		
 
                    pr_title='The Title',

			




	
	
	
		
 
                    pr_title_short='The Title',

			




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

			




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

			




	
	
	
		
 
                    pr_target_branch='trunk',

			




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

			




	
	
	
		
 
                    pr_source_branch='devbranch',

			




	
	
	
		
 
                    pr_owner=User.get(self.u2),

			




	
	
	
		
 
                    pr_owner_username='u2'

			




	
	
	
		
 
                    )

			




	
	
	
		
 

			




	
	
	
		
 
                for type_, body, kwargs in [

			




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

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








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