def test_show(self, create_test_user):
        self.log_user()
        with test_context(self.app):
            cur_user = self._get_logged_user()
            u1 = create_test_user(dict(username='u1', password='qweqwe',
                                       email='[email protected]',
                                       firstname=u'u1', lastname=u'u1',
                                       active=True))
            u2 = create_test_user(dict(username='u2', password='qweqwe',
                                       email='[email protected]',
                                       firstname=u'u2', lastname=u'u2',
                                       active=True))
            Session().commit()

            subject = u'test'
            notif_body = u'hi there'
            notification = NotificationModel().create(created_by=cur_user,
                                                      subject=subject,
                                                      body=notif_body,
                                                      recipients=[cur_user, u1, u2])

        response = self.app.get(url('notification',
                                    notification_id=notification.notification_id))

        response.mustcontain(subject)
        response.mustcontain(notif_body)

    def test_delete_ips(self, auto_clear_ip_permissions):
        self.log_user()
        default_user_id = User.get_default_user().user_id

        ## first add
        new_ip = '127.0.0.0/24'
        with test_context(self.app):
            user_model = UserModel()
            ip_obj = user_model.add_extra_ip(default_user_id, new_ip)
            Session().commit()

        ## double check that add worked
        # IP permissions are cached, need to invalidate this cache explicitly
        invalidate_all_caches()
        self.app.get(url('admin_permissions_ips'), status=302)
        # REMOTE_ADDR must match 127.0.0.0/24
        response = self.app.get(url('admin_permissions_ips'),
                                extra_environ={'REMOTE_ADDR': '127.0.0.1'})
        response.mustcontain('127.0.0.0/24')
        response.mustcontain('127.0.0.0 - 127.0.0.255')

        ## now delete
        response = self.app.post(url('edit_user_ips_delete', id=default_user_id),
                                 params=dict(del_ip_id=ip_obj.ip_id,
                                             _authentication_token=self.authentication_token()),
                                 extra_environ={'REMOTE_ADDR': '127.0.0.1'})

        # IP permissions are cached, need to invalidate this cache explicitly
        invalidate_all_caches()

        response = self.app.get(url('admin_permissions_ips'))
        response.mustcontain('All IP addresses are allowed')
        response.mustcontain(no=['127.0.0.0/24'])
        response.mustcontain(no=['127.0.0.0 - 127.0.0.255'])

def test_context_fixture(app_fixture):
    """
    Encompass the entire test using this fixture in a test_context,
    making sure that certain functionality still works even if no call to
    self.app.get/post has been made.
    The typical error message indicating you need a test_context is:
        TypeError: No object (name: context) has been registered for this thread

    The standard way to fix this is simply using the test_context context
    manager directly inside your test:
        with test_context(self.app):
            <actions>
    but if test setup code (xUnit-style or pytest fixtures) also needs to be
    executed inside the test context, that method is not possible.
    Even if there is no such setup code, the fixture may reduce code complexity
    if the entire test needs to run inside a test context.

    To apply this fixture (like any other fixture) to all test methods of a
    class, use the following class decorator:
        @pytest.mark.usefixtures("test_context_fixture")
        class TestFoo(TestController):
            ...
    """
    with test_context(app_fixture):
        yield

 def test_age_in_future(self, age_args, expected):
     from kallithea.lib.utils2 import age
     from dateutil import relativedelta
     with test_context(self.app):
         n = datetime.datetime(year=2012, month=5, day=17)
         delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
         assert age(n + delt(**age_args), now=n) == expected

    def test_create_err(self):
        self.log_user()
        username = 'new_user'
        password = ''
        name = u'name'
        lastname = u'lastname'
        email = 'errmail.example.com'

        response = self.app.post(url('new_user'),
            {'username': username,
             'password': password,
             'name': name,
             'active': False,
             'lastname': lastname,
             'email': email,
             '_authentication_token': self.authentication_token()})

        with test_context(self.app):
            msg = validators.ValidUsername(False, {})._messages['system_invalid_username']
        msg = h.html_escape(msg % {'username': 'new_user'})
        response.mustcontain("""<span class="error-message">%s</span>""" % msg)
        response.mustcontain("""<span class="error-message">Please enter a value</span>""")
        response.mustcontain("""<span class="error-message">An email address must contain a single @</span>""")

        def get_user():
            Session().query(User).filter(User.username == username).one()

        with pytest.raises(NoResultFound):
            get_user(), 'found user in database'

    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 == ['[email protected]']
                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_lurl():
    """url() can handle list parameters, with unicode too"""
    with test_context(None, '/'):
        value = lurl('/lurl')
        assert not isinstance(value, string_type)
        assert value.startswith('/lurl')
        assert str(value) == repr(value) == value.id == value.encode('utf-8').decode('utf-8') == value.__html__()

def test_unicode():
    """url() can handle unicode parameters"""
    with test_context(None, '/'):
        unicodestring =  u_('àèìòù')
        eq_(url('/', params=dict(x=unicodestring)),
            '/?x=%C3%A0%C3%A8%C3%AC%C3%B2%C3%B9'
            )

 def test_filename_template_not_found(self):
     try:
         with test_context(self.app):
             res = self.render('this_doesnt_exists/this_doesnt_exists.xhtml', {})
     except IOError as e:
         assert 'this_doesnt_exists.xhtml not found in template paths' in str(e)
     else:
         raise AssertionError('Should have raised IOError')

 def test_dotted_template_not_found(self):
     try:
         with test_context(self.app):
             res = self.render('tests.test_stack.rendering.templates.this_doesnt_exists', {})
     except IOError as e:
         assert 'this_doesnt_exists.xhtml not found' in str(e)
     else:
         raise AssertionError('Should have raised IOError')

def test_multi_values():
    with test_context(None, '/'):
        r = url("/foo", params=dict(bar=("asdf", "qwer")))
        assert r in \
                ["/foo?bar=qwer&bar=asdf", "/foo?bar=asdf&bar=qwer"], r
        r = url("/foo", params=dict(bar=[1,2]))
        assert r in \
                ["/foo?bar=1&bar=2", "/foo?bar=2&bar=1"], r

    def test_navigator_middle_page(self):
        with test_context(None, '/'):
            p = Page(range(100), items_per_page=10, page=5)
            pager = p.pager()

            assert '?page=1' in pager
            assert '?page=4' in pager
            assert '?page=6' in pager
            assert '?page=10' in pager

    def test_navigator_ajax(self):
        with test_context(None, '/'):
            p = Page(range(100), items_per_page=10, page=5)
            pager = p.pager(onclick='goto($page)')

            assert 'goto(1)' in pager
            assert 'goto(4)' in pager
            assert 'goto(6)' in pager
            assert 'goto(10)' in pager

def test_lurl_as_HTTPFound_location():
    with test_context(None, '/'):
        exc = HTTPFound(location=lurl('/lurl'))

        def _fake_start_response(*args, **kw):
            pass

        resp = exc({'PATH_INFO':'/',
                    'wsgi.url_scheme': 'HTTP',
                    'REQUEST_METHOD': 'GET',
                    'SERVER_NAME': 'localhost',
                    'SERVER_PORT': '80'}, _fake_start_response)
        assert b'resource was found at http://localhost:80/lurl' in resp[0]

    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 create_pullrequest(self, testcontroller, repo_name, pr_src_rev, pr_dst_rev, title=u'title'):
     org_ref = 'branch:stable:%s' % pr_src_rev
     other_ref = 'branch:default:%s' % pr_dst_rev
     with test_context(testcontroller.app): # needed to be able to mock request user
         org_repo = other_repo = Repository.get_by_repo_name(repo_name)
         owner_user = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         reviewers = [User.get_by_username(TEST_USER_REGULAR_LOGIN)]
         request.authuser = request.user = AuthUser(dbuser=owner_user)
         # creating a PR sends a message with an absolute URL - without routing that requires mocking
         with mock.patch.object(helpers, 'url', (lambda arg, qualified=False, **kwargs: ('https://localhost' if qualified else '') + '/fake/' + arg)):
             cmd = CreatePullRequestAction(org_repo, other_repo, org_ref, other_ref, title, u'No description', owner_user, reviewers)
             pull_request = cmd.execute()
         Session().commit()
     return pull_request.pull_request_id

    def test_unit__change_datetime_timezone__ok__with_naive_and_current_user(self):  # nopep8
        user_mock = MagicMock(timezone='America/Guadeloupe')

        with test_context(self.app):
            tmpl_context.current_user = user_mock
            naive_datetime = datetime.datetime(2000, 1, 1, 0, 0, 0)

            new_datetime = h.get_with_timezone(
                datetime_object=naive_datetime,
                default_from_timezone='UTC',
                to_timezone='',  # user_mock.timezone should be used
            )

            eq_(str(new_datetime), '1999-12-31 20:00:00-04:00')

    def test_description_with_datetime(self):
        self.log_user()
        with test_context(self.app):
            cur_user = self._get_logged_user()
            subject = u'test'
            notify_body = u'hi there'
            notification = NotificationModel().create(created_by = cur_user,
                                                      subject    = subject,
                                                      body       = notify_body)

            description = NotificationModel().make_description(notification, False)
            assert description == "{0} sent message at {1}".format(
                    cur_user.username,
                    h.fmt_date(notification.created_on)
                    )

    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 not notification 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