changes = []

        diff_limit = safe_int(kallithea.CONFIG.get('rss_cut_off_limit', 32 * 1024))

        raw_diff = cs.diff()

        diff_processor = DiffProcessor(raw_diff,

                                       diff_limit=diff_limit,

                                       inline_diff=False)

        for st in diff_processor.parsed:

            st.update({'added': st['stats']['added'],

                       'removed': st['stats']['deleted']})

            changes.append('\n %(operation)s %(filename)s '

                           '(%(added)s lines added, %(removed)s lines removed)'

                            % st)

        if diff_processor.limited_diff:

            changes = changes + ['\n ' +

                                 _('Changeset was too big and was cut off...')]

        # rev link

        _url = webutils.canonical_url('changeset_home', repo_name=c.db_repo.repo_name,

                   revision=cs.raw_id)

        desc_msg.append('changeset: <a href="%s">%s</a>' % (_url, cs.raw_id[:8]))

        desc_msg.append('<pre>')

        desc_msg.append('\n')

        desc_msg.extend(changes)

        if asbool(kallithea.CONFIG.get('rss_include_diff', False)):

            desc_msg.append('\n\n')

            desc_msg.append(safe_str(raw_diff))

        desc_msg.append('</pre>')

        return desc_msg

    def _feed(self, repo_name, feeder):

        """Produce a simple feed"""

        @cache_region('long_term_file', '_get_feed_from_cache')

        def _get_feed_from_cache(*_cache_keys):  # parameters are not really used - only as caching key

            header = dict(

                title=_('%s %s feed') % (c.site_name, repo_name),

                link=webutils.canonical_url('summary_home', repo_name=repo_name),

                description=_('Changes on %s repository') % repo_name,

            )

            rss_items_per_page = safe_int(kallithea.CONFIG.get('rss_items_per_page', 20))

            entries=[]

            for cs in reversed(list(c.db_repo_scm_instance[-rss_items_per_page:])):

                entries.append(dict(

                    title=self._get_title(cs),

import urllib.parse

from beaker.cache import cache_region

from pygments import highlight as code_highlight

from pygments.formatters.html import HtmlFormatter

from tg import tmpl_context as c

from tg.i18n import ugettext as _

import kallithea

from kallithea.lib.annotate import annotate_highlight

#==============================================================================

# PERMS

#==============================================================================

from kallithea.lib.auth import HasPermissionAny, HasRepoGroupPermissionLevel, HasRepoPermissionLevel

from kallithea.lib.diffs import BIN_FILENODE, CHMOD_FILENODE, DEL_FILENODE, MOD_FILENODE, NEW_FILENODE, RENAMED_FILENODE

from kallithea.lib.pygmentsutils import get_custom_lexer

from kallithea.lib.utils2 import (AttributeDict, age, asbool, credentials_filter, fmt_date, link_to_ref, safe_bytes, safe_int, safe_str, shorter,

                                  time_to_datetime)

from kallithea.lib.vcs.backends.base import BaseChangeset, EmptyChangeset

from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError

#==============================================================================

# SCM FILTERS available via h.

#==============================================================================

from kallithea.lib.vcs.utils import author_email, author_name

from kallithea.model import db

from kallithea.model.changeset_status import ChangesetStatusModel

# mute pyflakes "imported but unused"

# from webutils

assert HTML

assert Option

assert canonical_url

assert checkbox

assert chop_at

assert end_form

assert form

assert format_byte_size

assert hidden

assert js

assert jshtml

assert password

assert pop_flash_messages

assert radio

assert reset

assert safeid

assert select

assert session_csrf_secret_name

assert session_csrf_secret_token

assert submit

assert text

assert textarea

assert wrap_paragraphs

# from kallithea.lib.auth

assert HasPermissionAny

assert HasRepoGroupPermissionLevel

assert HasRepoPermissionLevel

# from utils2

assert age

assert fmt_date

assert link_to_ref

assert shorter

assert time_to_datetime

# from vcs

assert EmptyChangeset

log = logging.getLogger(__name__)

def FID(raw_id, path):

    """

    Creates a unique ID for filenode based on it's hash of path and revision

    it's safe to use in urls

    """

    return 'C-%s-%s' % (short_id(raw_id), hashlib.md5(safe_bytes(path)).hexdigest()[:12])

    # needs > 9% of width to be visible or 0 to be hidden

    a_p = max(9, unit * a) if a > 0 else 0

    d_p = max(9, unit * d) if d > 0 else 0

    p_sum = a_p + d_p

    if p_sum > width:

        # adjust the percentage to be == 100% since we adjusted to 9

        if a_p > d_p:

            a_p = a_p - (p_sum - width)

        else:

            d_p = d_p - (p_sum - width)

    a_v = a if a > 0 else ''

    d_v = d if d > 0 else ''

    d_a = '<div class="added progress-bar" style="width:%s%%">%s</div>' % (

        a_p, a_v

    )

    d_d = '<div class="deleted progress-bar" style="width:%s%%">%s</div>' % (

        d_p, d_v

    )

    return literal('<div class="progress" style="width:%spx">%s%s</div>' % (width, d_a, d_d))

def changeset_status(repo, revision):

    return ChangesetStatusModel().get_status(repo, revision)

def changeset_status_lbl(changeset_status):

    return db.ChangesetStatus.get_status_lbl(changeset_status)

def get_permission_name(key):

    return dict(db.Permission.PERMS).get(key)

def journal_filter_help():

    return _(textwrap.dedent('''

        Example filter terms:

            repository:vcs

            username:developer

            action:*push*

            ip:127.0.0.1

            date:20120101

            date:[20120101100000 TO 20120102]

        Generate wildcards using '*' character:

            "repository:vcs*" - search everything starting with 'vcs'

    If in doubt, err on the side of using `jshtml` over `js`, since it's

    better to escape too much than too little.

    """

    return js(escape(val))

url_re = re.compile(r'''\bhttps?://(?:[\da-zA-Z0-9@:.-]+)'''

                    r'''(?:[/a-zA-Z0-9_=@#~&+%.,:;?!*()-]*[/a-zA-Z0-9_=@#~])?''')

# Must match regexp in kallithea/public/js/base.js MentionsAutoComplete()

# Check char before @ - it must not look like we are in an email addresses.

# Matching is greedy so we don't have to look beyond the end.

MENTIONS_REGEX = re.compile(r'(?:^|(?<=[^a-zA-Z0-9]))@([a-zA-Z0-9][-_.a-zA-Z0-9]*[a-zA-Z0-9])')

def extract_mentioned_usernames(text):

    r"""

    Returns list of (possible) usernames @mentioned in given text.

    >>> extract_mentioned_usernames('@1-2.a_X,@1234 not@not @ddd@not @n @ee @ff @gg, @gg;@hh @n\n@zz,')

    ['1-2.a_X', '1234', 'ddd', 'ee', 'ff', 'gg', 'gg', 'hh', 'zz']

    """

    return MENTIONS_REGEX.findall(text)

# You should have received a copy of the GNU General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""

kallithea.model.notification

~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Model for notifications

This file was forked by the Kallithea project in July 2014.

Original author and date, and relevant copyright and licensing information is below:

:created_on: Nov 20, 2011

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import datetime

import logging

from tg import app_globals

from tg import tmpl_context as c

from tg.i18n import ugettext as _

from kallithea.lib.utils2 import fmt_date

from kallithea.model import async_tasks, db

log = logging.getLogger(__name__)

class NotificationModel(object):

    TYPE_CHANGESET_COMMENT = 'cs_comment'

    TYPE_MESSAGE = 'message'

    TYPE_MENTION = 'mention' # not used

    TYPE_REGISTRATION = 'registration'

    TYPE_PULL_REQUEST = 'pull_request'

    TYPE_PULL_REQUEST_COMMENT = 'pull_request_comment'

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

               type_=TYPE_MESSAGE, with_email=True,

               email_kwargs=None, repo_name=None):

        """

        Creates notification of given type

        :param created_by: int, str or User instance. User who created this

            notification

        :param body:

        :param recipients: list of int, str or User objects, when None

            is given send to all admins

        :param type_: type of notification

        :param with_email: send email with this notification

        :param email_kwargs: additional dict to pass as args to email template

        """

        email_kwargs = email_kwargs or {}

        if recipients and not getattr(recipients, '__iter__', False):

            raise Exception('recipients must be a list or iterable')

        created_by_obj = db.User.guess_instance(created_by)

        recipients_objs = set()

        if recipients:

            for u in recipients:

                obj = db.User.guess_instance(u)

                if obj is not None:

                    recipients_objs.add(obj)

                else:

                    # TODO: inform user that requested operation couldn't be completed

                    log.error('cannot email unknown user %r', u)

            log.debug('sending notifications %s to %s',

                type_, recipients_objs

            )

        elif recipients is None:

            # empty recipients means to all admins

            recipients_objs = db.User.query().filter(db.User.admin == True).all()

            log.debug('sending notifications %s to admins: %s',

                type_, recipients_objs

            )

        #else: silently skip notification mails?

        if not with_email:

            return

        headers = {}

        headers['X-Kallithea-Notification-Type'] = type_

        if 'threading' in email_kwargs:

            headers['References'] = ' '.join('<%s>' % x for x in email_kwargs['threading'])

        # this is passed into template

        created_on = fmt_date(datetime.datetime.now())

        html_kwargs = {

                  'when': created_on,

                  'user': created_by_obj.username,

                  }

        txt_kwargs = {

                  'body': body,

                  'when': created_on,

                  'user': created_by_obj.username,

                  }

        html_kwargs.update(email_kwargs)

        txt_kwargs.update(email_kwargs)

        email_subject = EmailNotificationModel() \

                            .get_email_description(type_, **txt_kwargs)

        email_txt_body = EmailNotificationModel() \

                            .get_email_tmpl(type_, 'txt', **txt_kwargs)

        email_html_body = EmailNotificationModel() \

                            .get_email_tmpl(type_, 'html', **html_kwargs)

        # don't send email to the person who caused the notification, except for

        # notifications about new pull requests where the author is explicitly

        # added.

        rec_mails = set(obj.email for obj in recipients_objs)

        if type_ == NotificationModel.TYPE_PULL_REQUEST:

        import kallithea.lib.helpers as h

        def repo_lnk(name, rtype, rstate, private, fork_of):

            return _render('repo_name', name, rtype, rstate, private, fork_of,

                           short_name=short_name)

        def following(repo_id, is_following):

            return _render('following', repo_id, is_following)

        def last_change(last_change):

            return _render("last_change", last_change)

        def rss_lnk(repo_name):

            return _render("rss", repo_name)

        def atom_lnk(repo_name):

            return _render("atom", repo_name)

        def last_rev(repo_name, cs_cache):

            return _render('revision', repo_name, cs_cache.get('revision'),

                           cs_cache.get('raw_id'), cs_cache.get('author'),

                           cs_cache.get('message'))

        def desc(desc):

        def state(repo_state):

            return _render("repo_state", repo_state)

        def repo_actions(repo_name):

            return _render('repo_actions', repo_name)

        def owner_actions(owner_id, username):

            return _render('user_name', owner_id, username)

        repos_data = []

        for gr in repo_groups_list or []:

            repos_data.append(dict(

                raw_name='\0' + webutils.html_escape(gr.name),  # sort before repositories

                just_name=webutils.html_escape(gr.name),

                name=_render('group_name_html', group_name=gr.group_name, name=gr.name),

                desc=desc(gr.group_description)))

        for repo in repos_list:

            if not HasRepoPermissionLevel('read')(repo.repo_name, 'get_repos_as_dict check'):

                continue

            cs_cache = repo.changeset_cache

            row = {

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

import json

import mimetypes

import posixpath

from kallithea.model import db, meta

from kallithea.tests import base

from kallithea.tests.fixture import Fixture

fixture = Fixture()

ARCHIVE_SPECS = {

    '.tar.bz2': ('application/x-bzip2', 'tbz2', ''),

    '.tar.gz': ('application/x-gzip', 'tgz', ''),

    '.zip': ('application/zip', 'zip', ''),

}

HG_NODE_HISTORY = fixture.load_resource('hg_node_history_response.json')

GIT_NODE_HISTORY = fixture.load_resource('git_node_history_response.json')

def _set_downloads(repo_name, set_to):

    repo = db.Repository.get_by_repo_name(repo_name)

    repo.enable_downloads = set_to

    meta.Session().commit()

class TestFilesController(base.TestController):

                                    revision='97e8b885c04894463c51898e14387d80c30ed1ee',

                                    f_path='/'))

        response.mustcontain("""<option selected="selected" value="97e8b885c04894463c51898e14387d80c30ed1ee">git at 97e8b885c048</option>""")

    def test_index_paging(self):

        self.log_user()

        for r in [(73, 'a066b25d5df7016b45a41b7e2a78c33b57adc235'),

                  (92, 'cc66b61b8455b264a7a8a2d8ddc80fcfc58c221e'),

                  (109, '75feb4c33e81186c87eac740cee2447330288412'),

                  (1, '3d8f361e72ab303da48d799ff1ac40d5ac37c67e'),

                  (0, 'b986218ba1c9b0d6a259fac9b050b1724ed8e545')]:

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

                                    repo_name=base.HG_REPO,

                                    revision=r[1],

                                    f_path='/'))

            response.mustcontain("""@ r%s:%s""" % (r[0], r[1][:12]))

    def test_file_source(self):

        # Force the global cache to be populated now when we know the right .ini has been loaded.

        # (Without this, the test would fail.)

        self.log_user()

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

                                    repo_name=base.HG_REPO,

                                    revision='8911406ad776fdd3d0b9932a2e89677e57405a48',

                                    f_path='vcs/nodes.py'))