Changeset - e2519d2e74c2
[Not reviewed]
default
0 12 0
Mads Kiilerich - 7 years ago 2018-06-09 16:30:22
mads@kiilerich.com
unicode: consistently use the preferred Python spelling 'utf-8' instead of the alias 'utf8'
12 files changed with 14 insertions and 14 deletions:
0 comments (0 inline, 0 general)
development.ini
Show inline comments
 
################################################################################
 
################################################################################
 
# Kallithea - config file generated with kallithea-config                      #
 
#                                                                              #
 
# The %(here)s variable will be replaced with the parent directory of this file#
 
################################################################################
 
################################################################################
 

	
 
[DEFAULT]
 

	
 
################################################################################
 
## Email settings                                                             ##
 
##                                                                            ##
 
## Refer to the documentation ("Email settings") for more details.            ##
 
##                                                                            ##
 
## It is recommended to use a valid sender address that passes access         ##
 
## validation and spam filtering in mail servers.                             ##
 
################################################################################
 

	
 
## 'From' header for application emails. You can optionally add a name.
 
## Default:
 
#app_email_from = Kallithea
 
## Examples:
 
#app_email_from = Kallithea <kallithea-noreply@example.com>
 
#app_email_from = kallithea-noreply@example.com
 

	
 
## Subject prefix for application emails.
 
## A space between this prefix and the real subject is automatically added.
 
## Default:
 
#email_prefix =
 
## Example:
 
#email_prefix = [Kallithea]
 

	
 
## Recipients for error emails and fallback recipients of application mails.
 
## Multiple addresses can be specified, comma-separated.
 
## Only addresses are allowed, do not add any name part.
 
## Default:
 
#email_to =
 
## Examples:
 
#email_to = admin@example.com
 
#email_to = admin@example.com,another_admin@example.com
 
email_to =
 

	
 
## 'From' header for error emails. You can optionally add a name.
 
## Default: (none)
 
## Examples:
 
#error_email_from = Kallithea Errors <kallithea-noreply@example.com>
 
#error_email_from = kallithea_errors@example.com
 
error_email_from =
 

	
 
## SMTP server settings
 
## If specifying credentials, make sure to use secure connections.
 
## Default: Send unencrypted unauthenticated mails to the specified smtp_server.
 
## For "SSL", use smtp_use_ssl = true and smtp_port = 465.
 
## For "STARTTLS", use smtp_use_tls = true and smtp_port = 587.
 
smtp_server =
 
#smtp_username =
 
#smtp_password =
 
smtp_port =
 
#smtp_use_ssl = false
 
#smtp_use_tls = false
 

	
 
## Entry point for 'gearbox serve'
 
[server:main]
 
#host = 127.0.0.1
 
host = 0.0.0.0
 
port = 5000
 

	
 
## WAITRESS ##
 
use = egg:waitress#main
 
## number of worker threads
 
threads = 1
 
## MAX BODY SIZE 100GB
 
max_request_body_size = 107374182400
 
## use poll instead of select, fixes fd limits, may not work on old
 
## windows systems.
 
#asyncore_use_poll = True
 

	
 
## middleware for hosting the WSGI application under a URL prefix
 
#[filter:proxy-prefix]
 
#use = egg:PasteDeploy#prefix
 
#prefix = /<your-prefix>
 

	
 
[app:main]
 
use = egg:kallithea
 
## enable proxy prefix middleware
 
#filter-with = proxy-prefix
 

	
 
full_stack = true
 
static_files = true
 

	
 
## Internationalization (see setup documentation for details)
 
## By default, the language requested by the browser is used if available.
 
#i18n.enable = false
 
## Fallback language, empty for English (valid values are the names of subdirectories in kallithea/i18n):
 
i18n.lang =
 

	
 
cache_dir = %(here)s/data
 
index_dir = %(here)s/data/index
 

	
 
## uncomment and set this path to use archive download cache
 
archive_cache_dir = %(here)s/tarballcache
 

	
 
## change this to unique ID for security
 
#app_instance_uuid = VERY-SECRET
 
app_instance_uuid = development-not-secret
 

	
 
## cut off limit for large diffs (size in bytes)
 
cut_off_limit = 256000
 

	
 
## force https in Kallithea, fixes https redirects, assumes it's always https
 
force_https = false
 

	
 
## use Strict-Transport-Security headers
 
use_htsts = false
 

	
 
## number of commits stats will parse on each iteration
 
commit_parse_limit = 25
 

	
 
## path to git executable
 
git_path = git
 

	
 
## git rev filter option, --all is the default filter, if you need to
 
## hide all refs in changelog switch this to --branches --tags
 
#git_rev_filter = --branches --tags
 

	
 
## RSS feed options
 
rss_cut_off_limit = 256000
 
rss_items_per_page = 10
 
rss_include_diff = false
 

	
 
## options for showing and identifying changesets
 
show_sha_length = 12
 
show_revision_number = false
 

	
 
## Canonical URL to use when creating full URLs in UI and texts.
 
## Useful when the site is available under different names or protocols.
 
## Defaults to what is provided in the WSGI environment.
 
#canonical_url = https://kallithea.example.com/repos
 

	
 
## gist URL alias, used to create nicer urls for gist. This should be an
 
## url that does rewrites to _admin/gists/<gistid>.
 
## example: http://gist.example.com/{gistid}. Empty means use the internal
 
## Kallithea url, ie. http[s]://kallithea.example.com/_admin/gists/<gistid>
 
gist_alias_url =
 

	
 
## white list of API enabled controllers. This allows to add list of
 
## controllers to which access will be enabled by api_key. eg: to enable
 
## api access to raw_files put `FilesController:raw`, to enable access to patches
 
## add `ChangesetController:changeset_patch`. This list should be "," separated
 
## Syntax is <ControllerClass>:<function>. Check debug logs for generated names
 
## Recommended settings below are commented out:
 
api_access_controllers_whitelist =
 
#    ChangesetController:changeset_patch,
 
#    ChangesetController:changeset_raw,
 
#    FilesController:raw,
 
#    FilesController:archivefile
 

	
 
## default encoding used to convert from and to unicode
 
## can be also a comma separated list of encoding in case of mixed encodings
 
default_encoding = utf8
 
default_encoding = utf-8
 

	
 
## Set Mercurial encoding, similar to setting HGENCODING before launching Kallithea
 
hgencoding = utf-8
 

	
 
## issue tracker for Kallithea (leave blank to disable, absent for default)
 
#bugtracker = https://bitbucket.org/conservancy/kallithea/issues
 

	
 
## issue tracking mapping for commit messages, comments, PR descriptions, ...
 
## Refer to the documentation ("Integration with issue trackers") for more details.
 

	
 
## regular expression to match issue references
 
## This pattern may/should contain parenthesized groups, that can
 
## be referred to in issue_server_link or issue_sub using Python backreferences
 
## (e.g. \1, \2, ...). You can also create named groups with '(?P<groupname>)'.
 
## To require mandatory whitespace before the issue pattern, use:
 
## (?:^|(?<=\s)) before the actual pattern, and for mandatory whitespace
 
## behind the issue pattern, use (?:$|(?=\s)) after the actual pattern.
 

	
 
issue_pat = #(\d+)
 

	
 
## server url to the issue
 
## This pattern may/should contain backreferences to parenthesized groups in issue_pat.
 
## A backreference can be \1, \2, ... or \g<groupname> if you specified a named group
 
## called 'groupname' in issue_pat.
 
## The special token {repo} is replaced with the full repository name
 
## including repository groups, while {repo_name} is replaced with just
 
## the name of the repository.
 

	
 
issue_server_link = https://issues.example.com/{repo}/issue/\1
 

	
 
## substitution pattern to use as the link text
 
## If issue_sub is empty, the text matched by issue_pat is retained verbatim
 
## for the link text. Otherwise, the link text is that of issue_sub, with any
 
## backreferences to groups in issue_pat replaced.
 

	
 
issue_sub =
 

	
 
## issue_pat, issue_server_link and issue_sub can have suffixes to specify
 
## multiple patterns, to other issues server, wiki or others
 
## below an example how to create a wiki pattern
 
# wiki-some-id -> https://wiki.example.com/some-id
 

	
 
#issue_pat_wiki = wiki-(\S+)
 
#issue_server_link_wiki = https://wiki.example.com/\1
 
#issue_sub_wiki = WIKI-\1
 

	
 
## alternative return HTTP header for failed authentication. Default HTTP
 
## response is 401 HTTPUnauthorized. Currently Mercurial clients have trouble with
 
## handling that. Set this variable to 403 to return HTTPForbidden
 
auth_ret_code =
 

	
 
## locking return code. When repository is locked return this HTTP code. 2XX
 
## codes don't break the transactions while 4XX codes do
 
lock_ret_code = 423
 

	
 
## allows to change the repository location in settings page
 
allow_repo_location_change = True
 

	
 
## allows to setup custom hooks in settings page
 
allow_custom_hooks_settings = True
 

	
 
## extra extensions for indexing, space separated and without the leading '.'.
 
# index.extensions =
 
#    gemfile
 
#    lock
 

	
 
## extra filenames for indexing, space separated
 
# index.filenames =
 
#    .dockerignore
 
#    .editorconfig
 
#    INSTALL
 
#    CHANGELOG
 

	
 
####################################
 
###        CELERY CONFIG        ####
 
####################################
 

	
 
use_celery = false
 

	
 
## Example: connect to the virtual host 'rabbitmqhost' on localhost as rabbitmq:
 
broker.url = amqp://rabbitmq:qewqew@localhost:5672/rabbitmqhost
 

	
 
celery.imports = kallithea.lib.celerylib.tasks
 
celery.accept.content = pickle
 
celery.result.backend = amqp
 
celery.result.dburi = amqp://
 
celery.result.serialier = json
 

	
 
#celery.send.task.error.emails = true
 
#celery.amqp.task.result.expires = 18000
 

	
 
celeryd.concurrency = 2
 
celeryd.max.tasks.per.child = 1
 

	
 
## If true, tasks will never be sent to the queue, but executed locally instead.
 
celery.always.eager = false
 

	
 
####################################
 
###         BEAKER CACHE        ####
 
####################################
 

	
 
beaker.cache.data_dir = %(here)s/data/cache/data
 
beaker.cache.lock_dir = %(here)s/data/cache/lock
 

	
 
beaker.cache.regions = short_term,long_term,sql_cache_short
 

	
 
beaker.cache.short_term.type = memory
 
beaker.cache.short_term.expire = 60
 
beaker.cache.short_term.key_length = 256
 

	
 
beaker.cache.long_term.type = memory
 
beaker.cache.long_term.expire = 36000
 
beaker.cache.long_term.key_length = 256
 

	
 
beaker.cache.sql_cache_short.type = memory
 
beaker.cache.sql_cache_short.expire = 10
 
beaker.cache.sql_cache_short.key_length = 256
 

	
 
####################################
 
###       BEAKER SESSION        ####
 
####################################
 

	
 
## Name of session cookie. Should be unique for a given host and path, even when running
 
## on different ports. Otherwise, cookie sessions will be shared and messed up.
 
beaker.session.key = kallithea
 
## Sessions should always only be accessible by the browser, not directly by JavaScript.
 
beaker.session.httponly = true
 
## Session lifetime. 2592000 seconds is 30 days.
 
beaker.session.timeout = 2592000
 

	
 
## Server secret used with HMAC to ensure integrity of cookies.
 
#beaker.session.secret = VERY-SECRET
 
beaker.session.secret = development-not-secret
 
## Further, encrypt the data with AES.
 
#beaker.session.encrypt_key = <key_for_encryption>
 
#beaker.session.validate_key = <validation_key>
 

	
 
## Type of storage used for the session, current types are
 
## dbm, file, memcached, database, and memory.
 

	
 
## File system storage of session data. (default)
 
#beaker.session.type = file
 

	
 
## Cookie only, store all session data inside the cookie. Requires secure secrets.
 
#beaker.session.type = cookie
 

	
 
## Database storage of session data.
 
#beaker.session.type = ext:database
 
#beaker.session.sa.url = postgresql://postgres:qwe@localhost/kallithea
 
#beaker.session.table_name = db_session
 

	
 
################################################################################
 
## WARNING: *DEBUG MODE MUST BE OFF IN A PRODUCTION ENVIRONMENT*              ##
 
## Debug mode will enable the interactive debugging tool, allowing ANYONE to  ##
 
## execute malicious code after an exception is raised.                       ##
 
################################################################################
 
#debug = false
 
debug = true
 

	
 
##################################
 
###       LOGVIEW CONFIG       ###
 
##################################
 

	
 
logview.sqlalchemy = #faa
 
logview.pylons.templating = #bfb
 
logview.pylons.util = #eee
 

	
 
#########################################################
 
### DB CONFIGS - EACH DB WILL HAVE IT'S OWN CONFIG    ###
 
#########################################################
 

	
 
# SQLITE [default]
 
sqlalchemy.url = sqlite:///%(here)s/kallithea.db?timeout=60
 

	
 
# see sqlalchemy docs for others
 

	
 
sqlalchemy.pool_recycle = 3600
 

	
 
################################
 
### ALEMBIC CONFIGURATION   ####
 
################################
 

	
 
[alembic]
 
script_location = kallithea:alembic
 

	
 
################################
 
### LOGGING CONFIGURATION   ####
 
################################
 

	
 
[loggers]
 
keys = root, routes, kallithea, sqlalchemy, tg, gearbox, beaker, templates, whoosh_indexer, werkzeug, backlash
 

	
 
[handlers]
 
keys = console, console_sql
 

	
 
[formatters]
 
keys = generic, color_formatter, color_formatter_sql
 

	
 
#############
 
## LOGGERS ##
 
#############
 

	
 
[logger_root]
 
level = NOTSET
 
handlers = console
 

	
 
[logger_routes]
 
#level = WARN
 
level = DEBUG
 
handlers =
 
qualname = routes.middleware
 
## "level = DEBUG" logs the route matched and routing variables.
 
propagate = 1
 

	
 
[logger_beaker]
 
#level = WARN
 
level = DEBUG
 
handlers =
 
qualname = beaker.container
 
propagate = 1
 

	
 
[logger_templates]
 
#level = WARN
 
level = INFO
 
handlers =
 
qualname = pylons.templating
 
propagate = 1
 

	
 
[logger_kallithea]
 
#level = WARN
 
level = DEBUG
 
handlers =
 
qualname = kallithea
 
propagate = 1
 

	
 
[logger_tg]
 
#level = WARN
 
level = DEBUG
 
handlers =
 
qualname = tg
 
propagate = 1
 

	
 
[logger_gearbox]
 
#level = WARN
 
level = DEBUG
 
handlers =
 
qualname = gearbox
 
propagate = 1
 

	
 
[logger_sqlalchemy]
 
level = WARN
 
handlers = console_sql
 
qualname = sqlalchemy.engine
 
propagate = 0
 

	
 
[logger_whoosh_indexer]
 
#level = WARN
 
level = DEBUG
 
handlers =
 
qualname = whoosh_indexer
 
propagate = 1
 

	
 
[logger_werkzeug]
 
level = WARN
 
handlers =
 
qualname = werkzeug
 
propagate = 1
 

	
 
[logger_backlash]
 
level = WARN
 
handlers =
 
qualname = backlash
 
propagate = 1
 

	
 
##############
 
## HANDLERS ##
 
##############
 

	
 
[handler_console]
 
class = StreamHandler
 
args = (sys.stderr,)
 
#formatter = generic
 
formatter = color_formatter
 

	
 
[handler_console_sql]
 
class = StreamHandler
 
args = (sys.stderr,)
 
#formatter = generic
 
formatter = color_formatter_sql
 

	
 
################
 
## FORMATTERS ##
 
################
 

	
 
[formatter_generic]
 
format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 
datefmt = %Y-%m-%d %H:%M:%S
 

	
 
[formatter_color_formatter]
 
class = kallithea.lib.colored_formatter.ColorFormatter
 
format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 
datefmt = %Y-%m-%d %H:%M:%S
 

	
 
[formatter_color_formatter_sql]
 
class = kallithea.lib.colored_formatter.ColorFormatterSql
 
format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 
datefmt = %Y-%m-%d %H:%M:%S
kallithea/lib/helpers.py
Show inline comments
 
@@ -94,769 +94,769 @@ def js(value):
 
    This is necessary to safely insert arbitrary values into HTML <script>
 
    sections e.g. using Mako template expression substitution.
 

	
 
    Note: Rather than using this function, it's preferable to avoid the
 
    insertion of values into HTML <script> sections altogether. Instead,
 
    data should (to the extent possible) be passed to JavaScript using
 
    data attributes or AJAX calls, eliminating the need for JS specific
 
    escaping.
 

	
 
    Note: This is not safe for use in attributes (e.g. onclick), because
 
    quotes are not escaped.
 

	
 
    Because the rules for parsing <script> varies between XHTML (where
 
    normal rules apply for any special characters) and HTML (where
 
    entities are not interpreted, but the literal string "</script>"
 
    is forbidden), the function ensures that the result never contains
 
    '&', '<' and '>', thus making it safe in both those contexts (but
 
    not in attributes).
 
    """
 
    return literal(
 
        ('(' + json.dumps(value) + ')')
 
        # In JSON, the following can only appear in string literals.
 
        .replace('&', r'\x26')
 
        .replace('<', r'\x3c')
 
        .replace('>', r'\x3e')
 
    )
 

	
 

	
 
def jshtml(val):
 
    """HTML escapes a string value, then converts the resulting string
 
    to its corresponding JavaScript representation (see `js`).
 

	
 
    This is used when a plain-text string (possibly containing special
 
    HTML characters) will be used by a script in an HTML context (e.g.
 
    element.innerHTML or jQuery's 'html' method).
 

	
 
    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))
 

	
 

	
 
def shorter(s, size=20, firstline=False, postfix='...'):
 
    """Truncate s to size, including the postfix string if truncating.
 
    If firstline, truncate at newline.
 
    """
 
    if firstline:
 
        s = s.split('\n', 1)[0].rstrip()
 
    if len(s) > size:
 
        return s[:size - len(postfix)] + postfix
 
    return s
 

	
 

	
 
def _reset(name, value=None, id=NotGiven, type="reset", **attrs):
 
    """
 
    Reset button
 
    """
 
    _set_input_attrs(attrs, type, name, value)
 
    _set_id_attr(attrs, id, name)
 
    convert_boolean_attrs(attrs, ["disabled"])
 
    return HTML.input(**attrs)
 

	
 

	
 
reset = _reset
 
safeid = _make_safe_id_component
 

	
 

	
 
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
 

	
 
    :param raw_id:
 
    :param path:
 
    """
 

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

	
 

	
 
class _FilesBreadCrumbs(object):
 

	
 
    def __call__(self, repo_name, rev, paths):
 
        if isinstance(paths, str):
 
            paths = safe_unicode(paths)
 
        url_l = [link_to(repo_name, url('files_home',
 
                                        repo_name=repo_name,
 
                                        revision=rev, f_path=''),
 
                         class_='ypjax-link')]
 
        paths_l = paths.split('/')
 
        for cnt, p in enumerate(paths_l):
 
            if p != '':
 
                url_l.append(link_to(p,
 
                                     url('files_home',
 
                                         repo_name=repo_name,
 
                                         revision=rev,
 
                                         f_path='/'.join(paths_l[:cnt + 1])
 
                                         ),
 
                                     class_='ypjax-link'
 
                                     )
 
                             )
 

	
 
        return literal('/'.join(url_l))
 

	
 

	
 
files_breadcrumbs = _FilesBreadCrumbs()
 

	
 

	
 
class CodeHtmlFormatter(HtmlFormatter):
 
    """
 
    My code Html Formatter for source codes
 
    """
 

	
 
    def wrap(self, source, outfile):
 
        return self._wrap_div(self._wrap_pre(self._wrap_code(source)))
 

	
 
    def _wrap_code(self, source):
 
        for cnt, it in enumerate(source):
 
            i, t = it
 
            t = '<span id="L%s">%s</span>' % (cnt + 1, t)
 
            yield i, t
 

	
 
    def _wrap_tablelinenos(self, inner):
 
        dummyoutfile = StringIO.StringIO()
 
        lncount = 0
 
        for t, line in inner:
 
            if t:
 
                lncount += 1
 
            dummyoutfile.write(line)
 

	
 
        fl = self.linenostart
 
        mw = len(str(lncount + fl - 1))
 
        sp = self.linenospecial
 
        st = self.linenostep
 
        la = self.lineanchors
 
        aln = self.anchorlinenos
 
        nocls = self.noclasses
 
        if sp:
 
            lines = []
 

	
 
            for i in range(fl, fl + lncount):
 
                if i % st == 0:
 
                    if i % sp == 0:
 
                        if aln:
 
                            lines.append('<a href="#%s%d" class="special">%*d</a>' %
 
                                         (la, i, mw, i))
 
                        else:
 
                            lines.append('<span class="special">%*d</span>' % (mw, i))
 
                    else:
 
                        if aln:
 
                            lines.append('<a href="#%s%d">%*d</a>' % (la, i, mw, i))
 
                        else:
 
                            lines.append('%*d' % (mw, i))
 
                else:
 
                    lines.append('')
 
            ls = '\n'.join(lines)
 
        else:
 
            lines = []
 
            for i in range(fl, fl + lncount):
 
                if i % st == 0:
 
                    if aln:
 
                        lines.append('<a href="#%s%d">%*d</a>' % (la, i, mw, i))
 
                    else:
 
                        lines.append('%*d' % (mw, i))
 
                else:
 
                    lines.append('')
 
            ls = '\n'.join(lines)
 

	
 
        # in case you wonder about the seemingly redundant <div> here: since the
 
        # content in the other cell also is wrapped in a div, some browsers in
 
        # some configurations seem to mess up the formatting...
 
        if nocls:
 
            yield 0, ('<table class="%stable">' % self.cssclass +
 
                      '<tr><td><div class="linenodiv">'
 
                      '<pre>' + ls + '</pre></div></td>'
 
                      '<td id="hlcode" class="code">')
 
        else:
 
            yield 0, ('<table class="%stable">' % self.cssclass +
 
                      '<tr><td class="linenos"><div class="linenodiv">'
 
                      '<pre>' + ls + '</pre></div></td>'
 
                      '<td id="hlcode" class="code">')
 
        yield 0, dummyoutfile.getvalue()
 
        yield 0, '</td></tr></table>'
 

	
 

	
 
_whitespace_re = re.compile(r'(\t)|( )(?=\n|</div>)')
 

	
 

	
 
def _markup_whitespace(m):
 
    groups = m.groups()
 
    if groups[0]:
 
        return '<u>\t</u>'
 
    if groups[1]:
 
        return ' <i></i>'
 

	
 

	
 
def markup_whitespace(s):
 
    return _whitespace_re.sub(_markup_whitespace, s)
 

	
 

	
 
def pygmentize(filenode, **kwargs):
 
    """
 
    pygmentize function using pygments
 

	
 
    :param filenode:
 
    """
 
    lexer = get_custom_lexer(filenode.extension) or filenode.lexer
 
    return literal(markup_whitespace(
 
        code_highlight(filenode.content, lexer, CodeHtmlFormatter(**kwargs))))
 

	
 

	
 
def pygmentize_annotation(repo_name, filenode, **kwargs):
 
    """
 
    pygmentize function for annotation
 

	
 
    :param filenode:
 
    """
 

	
 
    color_dict = {}
 

	
 
    def gen_color(n=10000):
 
        """generator for getting n of evenly distributed colors using
 
        hsv color and golden ratio. It always return same order of colors
 

	
 
        :returns: RGB tuple
 
        """
 

	
 
        def hsv_to_rgb(h, s, v):
 
            if s == 0.0:
 
                return v, v, v
 
            i = int(h * 6.0)  # XXX assume int() truncates!
 
            f = (h * 6.0) - i
 
            p = v * (1.0 - s)
 
            q = v * (1.0 - s * f)
 
            t = v * (1.0 - s * (1.0 - f))
 
            i = i % 6
 
            if i == 0:
 
                return v, t, p
 
            if i == 1:
 
                return q, v, p
 
            if i == 2:
 
                return p, v, t
 
            if i == 3:
 
                return p, q, v
 
            if i == 4:
 
                return t, p, v
 
            if i == 5:
 
                return v, p, q
 

	
 
        golden_ratio = 0.618033988749895
 
        h = 0.22717784590367374
 

	
 
        for _unused in xrange(n):
 
            h += golden_ratio
 
            h %= 1
 
            HSV_tuple = [h, 0.95, 0.95]
 
            RGB_tuple = hsv_to_rgb(*HSV_tuple)
 
            yield map(lambda x: str(int(x * 256)), RGB_tuple)
 

	
 
    cgenerator = gen_color()
 

	
 
    def get_color_string(cs):
 
        if cs in color_dict:
 
            col = color_dict[cs]
 
        else:
 
            col = color_dict[cs] = cgenerator.next()
 
        return "color: rgb(%s)! important;" % (', '.join(col))
 

	
 
    def url_func(repo_name):
 

	
 
        def _url_func(changeset):
 
            author = escape(changeset.author)
 
            date = changeset.date
 
            message = escape(changeset.message)
 
            tooltip_html = ("<b>Author:</b> %s<br/>"
 
                            "<b>Date:</b> %s</b><br/>"
 
                            "<b>Message:</b> %s") % (author, date, message)
 

	
 
            lnk_format = show_id(changeset)
 
            uri = link_to(
 
                    lnk_format,
 
                    url('changeset_home', repo_name=repo_name,
 
                        revision=changeset.raw_id),
 
                    style=get_color_string(changeset.raw_id),
 
                    **{'data-toggle': 'popover',
 
                       'data-content': tooltip_html}
 
                  )
 

	
 
            uri += '\n'
 
            return uri
 
        return _url_func
 

	
 
    return literal(markup_whitespace(annotate_highlight(filenode, url_func(repo_name), **kwargs)))
 

	
 

	
 
class _Message(object):
 
    """A message returned by ``Flash.pop_messages()``.
 

	
 
    Converting the message to a string returns the message text. Instances
 
    also have the following attributes:
 

	
 
    * ``message``: the message text.
 
    * ``category``: the category specified when the message was created.
 
    """
 

	
 
    def __init__(self, category, message):
 
        self.category = category
 
        self.message = message
 

	
 
    def __str__(self):
 
        return self.message
 

	
 
    __unicode__ = __str__
 

	
 
    def __html__(self):
 
        return escape(safe_unicode(self.message))
 

	
 

	
 
class Flash(_Flash):
 

	
 
    def __call__(self, message, category=None, ignore_duplicate=False, logf=None):
 
        """
 
        Show a message to the user _and_ log it through the specified function
 

	
 
        category: notice (default), warning, error, success
 
        logf: a custom log function - such as log.debug
 

	
 
        logf defaults to log.info, unless category equals 'success', in which
 
        case logf defaults to log.debug.
 
        """
 
        if logf is None:
 
            logf = log.info
 
            if category == 'success':
 
                logf = log.debug
 

	
 
        logf('Flash %s: %s', category, message)
 

	
 
        super(Flash, self).__call__(message, category, ignore_duplicate)
 

	
 
    def pop_messages(self):
 
        """Return all accumulated messages and delete them from the session.
 

	
 
        The return value is a list of ``Message`` objects.
 
        """
 
        from tg import session
 
        messages = session.pop(self.session_key, [])
 
        session.save()
 
        return [_Message(*m) for m in messages]
 

	
 

	
 
flash = Flash()
 

	
 
#==============================================================================
 
# SCM FILTERS available via h.
 
#==============================================================================
 
from kallithea.lib.vcs.utils import author_name, author_email
 
from kallithea.lib.utils2 import credentials_filter, age as _age
 

	
 
age = lambda x, y=False: _age(x, y)
 
capitalize = lambda x: x.capitalize()
 
email = author_email
 
short_id = lambda x: x[:12]
 
hide_credentials = lambda x: ''.join(credentials_filter(x))
 

	
 

	
 
def show_id(cs):
 
    """
 
    Configurable function that shows ID
 
    by default it's r123:fffeeefffeee
 

	
 
    :param cs: changeset instance
 
    """
 
    from kallithea import CONFIG
 
    def_len = safe_int(CONFIG.get('show_sha_length', 12))
 
    show_rev = str2bool(CONFIG.get('show_revision_number', False))
 

	
 
    raw_id = cs.raw_id[:def_len]
 
    if show_rev:
 
        return 'r%s:%s' % (cs.revision, raw_id)
 
    else:
 
        return raw_id
 

	
 

	
 
def fmt_date(date):
 
    if date:
 
        return date.strftime("%Y-%m-%d %H:%M:%S").decode('utf8')
 
        return date.strftime("%Y-%m-%d %H:%M:%S").decode('utf-8')
 

	
 
    return ""
 

	
 

	
 
def is_git(repository):
 
    if hasattr(repository, 'alias'):
 
        _type = repository.alias
 
    elif hasattr(repository, 'repo_type'):
 
        _type = repository.repo_type
 
    else:
 
        _type = repository
 
    return _type == 'git'
 

	
 

	
 
def is_hg(repository):
 
    if hasattr(repository, 'alias'):
 
        _type = repository.alias
 
    elif hasattr(repository, 'repo_type'):
 
        _type = repository.repo_type
 
    else:
 
        _type = repository
 
    return _type == 'hg'
 

	
 

	
 
@cache_region('long_term', 'user_or_none')
 
def user_or_none(author):
 
    """Try to match email part of VCS committer string with a local user - or return None"""
 
    from kallithea.model.db import User
 
    email = author_email(author)
 
    if email:
 
        return User.get_by_email(email, cache=True) # cache will only use sql_cache_short
 
    return None
 

	
 

	
 
def email_or_none(author):
 
    """Try to match email part of VCS committer string with a local user.
 
    Return primary email of user, email part of the specified author name, or None."""
 
    if not author:
 
        return None
 
    user = user_or_none(author)
 
    if user is not None:
 
        return user.email # always use main email address - not necessarily the one used to find user
 

	
 
    # extract email from the commit string
 
    email = author_email(author)
 
    if email:
 
        return email
 

	
 
    # No valid email, not a valid user in the system, none!
 
    return None
 

	
 

	
 
def person(author, show_attr="username"):
 
    """Find the user identified by 'author', return one of the users attributes,
 
    default to the username attribute, None if there is no user"""
 
    from kallithea.model.db import User
 
    # attr to return from fetched user
 
    person_getter = lambda usr: getattr(usr, show_attr)
 

	
 
    # if author is already an instance use it for extraction
 
    if isinstance(author, User):
 
        return person_getter(author)
 

	
 
    user = user_or_none(author)
 
    if user is not None:
 
        return person_getter(user)
 

	
 
    # Still nothing?  Just pass back the author name if any, else the email
 
    return author_name(author) or email(author)
 

	
 

	
 
def person_by_id(id_, show_attr="username"):
 
    from kallithea.model.db import User
 
    # attr to return from fetched user
 
    person_getter = lambda usr: getattr(usr, show_attr)
 

	
 
    # maybe it's an ID ?
 
    if str(id_).isdigit() or isinstance(id_, int):
 
        id_ = int(id_)
 
        user = User.get(id_)
 
        if user is not None:
 
            return person_getter(user)
 
    return id_
 

	
 

	
 
def boolicon(value):
 
    """Returns boolean value of a value, represented as small html image of true/false
 
    icons
 

	
 
    :param value: value
 
    """
 

	
 
    if value:
 
        return HTML.tag('i', class_="icon-ok")
 
    else:
 
        return HTML.tag('i', class_="icon-minus-circled")
 

	
 

	
 
def action_parser(user_log, feed=False, parse_cs=False):
 
    """
 
    This helper will action_map the specified string action into translated
 
    fancy names with icons and links
 

	
 
    :param user_log: user log instance
 
    :param feed: use output for feeds (no html and fancy icons)
 
    :param parse_cs: parse Changesets into VCS instances
 
    """
 

	
 
    action = user_log.action
 
    action_params = ' '
 

	
 
    x = action.split(':')
 

	
 
    if len(x) > 1:
 
        action, action_params = x
 

	
 
    def get_cs_links():
 
        revs_limit = 3  # display this amount always
 
        revs_top_limit = 50  # show upto this amount of changesets hidden
 
        revs_ids = action_params.split(',')
 
        deleted = user_log.repository is None
 
        if deleted:
 
            return ','.join(revs_ids)
 

	
 
        repo_name = user_log.repository.repo_name
 

	
 
        def lnk(rev, repo_name):
 
            lazy_cs = False
 
            title_ = None
 
            url_ = '#'
 
            if isinstance(rev, BaseChangeset) or isinstance(rev, AttributeDict):
 
                if rev.op and rev.ref_name:
 
                    if rev.op == 'delete_branch':
 
                        lbl = _('Deleted branch: %s') % rev.ref_name
 
                    elif rev.op == 'tag':
 
                        lbl = _('Created tag: %s') % rev.ref_name
 
                    else:
 
                        lbl = 'Unknown operation %s' % rev.op
 
                else:
 
                    lazy_cs = True
 
                    lbl = rev.short_id[:8]
 
                    url_ = url('changeset_home', repo_name=repo_name,
 
                               revision=rev.raw_id)
 
            else:
 
                # changeset cannot be found - it might have been stripped or removed
 
                lbl = rev[:12]
 
                title_ = _('Changeset %s not found') % lbl
 
            if parse_cs:
 
                return link_to(lbl, url_, title=title_, **{'data-toggle': 'tooltip'})
 
            return link_to(lbl, url_, class_='lazy-cs' if lazy_cs else '',
 
                           **{'data-raw_id': rev.raw_id, 'data-repo_name': repo_name})
 

	
 
        def _get_op(rev_txt):
 
            _op = None
 
            _name = rev_txt
 
            if len(rev_txt.split('=>')) == 2:
 
                _op, _name = rev_txt.split('=>')
 
            return _op, _name
 

	
 
        revs = []
 
        if len(filter(lambda v: v != '', revs_ids)) > 0:
 
            repo = None
 
            for rev in revs_ids[:revs_top_limit]:
 
                _op, _name = _get_op(rev)
 

	
 
                # we want parsed changesets, or new log store format is bad
 
                if parse_cs:
 
                    try:
 
                        if repo is None:
 
                            repo = user_log.repository.scm_instance
 
                        _rev = repo.get_changeset(rev)
 
                        revs.append(_rev)
 
                    except ChangesetDoesNotExistError:
 
                        log.error('cannot find revision %s in this repo', rev)
 
                        revs.append(rev)
 
                else:
 
                    _rev = AttributeDict({
 
                        'short_id': rev[:12],
 
                        'raw_id': rev,
 
                        'message': '',
 
                        'op': _op,
 
                        'ref_name': _name
 
                    })
 
                    revs.append(_rev)
 
        cs_links = [" " + ', '.join(
 
            [lnk(rev, repo_name) for rev in revs[:revs_limit]]
 
        )]
 
        _op1, _name1 = _get_op(revs_ids[0])
 
        _op2, _name2 = _get_op(revs_ids[-1])
 

	
 
        _rev = '%s...%s' % (_name1, _name2)
 

	
 
        compare_view = (
 
            ' <div class="compare_view" data-toggle="tooltip" title="%s">'
 
            '<a href="%s">%s</a> </div>' % (
 
                _('Show all combined changesets %s->%s') % (
 
                    revs_ids[0][:12], revs_ids[-1][:12]
 
                ),
 
                url('changeset_home', repo_name=repo_name,
 
                    revision=_rev
 
                ),
 
                _('Compare view')
 
            )
 
        )
 

	
 
        # if we have exactly one more than normally displayed
 
        # just display it, takes less space than displaying
 
        # "and 1 more revisions"
 
        if len(revs_ids) == revs_limit + 1:
 
            cs_links.append(", " + lnk(revs[revs_limit], repo_name))
 

	
 
        # hidden-by-default ones
 
        if len(revs_ids) > revs_limit + 1:
 
            uniq_id = revs_ids[0]
 
            html_tmpl = (
 
                '<span> %s <a class="show_more" id="_%s" '
 
                'href="#more">%s</a> %s</span>'
 
            )
 
            if not feed:
 
                cs_links.append(html_tmpl % (
 
                      _('and'),
 
                      uniq_id, _('%s more') % (len(revs_ids) - revs_limit),
 
                      _('revisions')
 
                    )
 
                )
 

	
 
            if not feed:
 
                html_tmpl = '<span id="%s" style="display:none">, %s </span>'
 
            else:
 
                html_tmpl = '<span id="%s"> %s </span>'
 

	
 
            morelinks = ', '.join(
 
              [lnk(rev, repo_name) for rev in revs[revs_limit:]]
 
            )
 

	
 
            if len(revs_ids) > revs_top_limit:
 
                morelinks += ', ...'
 

	
 
            cs_links.append(html_tmpl % (uniq_id, morelinks))
 
        if len(revs) > 1:
 
            cs_links.append(compare_view)
 
        return ''.join(cs_links)
 

	
 
    def get_fork_name():
 
        repo_name = action_params
 
        url_ = url('summary_home', repo_name=repo_name)
 
        return _('Fork name %s') % link_to(action_params, url_)
 

	
 
    def get_user_name():
 
        user_name = action_params
 
        return user_name
 

	
 
    def get_users_group():
 
        group_name = action_params
 
        return group_name
 

	
 
    def get_pull_request():
 
        from kallithea.model.db import PullRequest
 
        pull_request_id = action_params
 
        nice_id = PullRequest.make_nice_id(pull_request_id)
 

	
 
        deleted = user_log.repository is None
 
        if deleted:
 
            repo_name = user_log.repository_name
 
        else:
 
            repo_name = user_log.repository.repo_name
 

	
 
        return link_to(_('Pull request %s') % nice_id,
 
                    url('pullrequest_show', repo_name=repo_name,
 
                    pull_request_id=pull_request_id))
 

	
 
    def get_archive_name():
 
        archive_name = action_params
 
        return archive_name
 

	
 
    # action : translated str, callback(extractor), icon
 
    action_map = {
 
    'user_deleted_repo':           (_('[deleted] repository'),
 
                                    None, 'icon-trashcan'),
 
    'user_created_repo':           (_('[created] repository'),
 
                                    None, 'icon-plus'),
 
    'user_created_fork':           (_('[created] repository as fork'),
 
                                    None, 'icon-fork'),
 
    'user_forked_repo':            (_('[forked] repository'),
 
                                    get_fork_name, 'icon-fork'),
 
    'user_updated_repo':           (_('[updated] repository'),
 
                                    None, 'icon-pencil'),
 
    'user_downloaded_archive':      (_('[downloaded] archive from repository'),
 
                                    get_archive_name, 'icon-download-cloud'),
 
    'admin_deleted_repo':          (_('[delete] repository'),
 
                                    None, 'icon-trashcan'),
 
    'admin_created_repo':          (_('[created] repository'),
 
                                    None, 'icon-plus'),
 
    'admin_forked_repo':           (_('[forked] repository'),
 
                                    None, 'icon-fork'),
 
    'admin_updated_repo':          (_('[updated] repository'),
 
                                    None, 'icon-pencil'),
 
    'admin_created_user':          (_('[created] user'),
 
                                    get_user_name, 'icon-user'),
 
    'admin_updated_user':          (_('[updated] user'),
 
                                    get_user_name, 'icon-user'),
 
    'admin_created_users_group':   (_('[created] user group'),
 
                                    get_users_group, 'icon-pencil'),
 
    'admin_updated_users_group':   (_('[updated] user group'),
 
                                    get_users_group, 'icon-pencil'),
 
    'user_commented_revision':     (_('[commented] on revision in repository'),
 
                                    get_cs_links, 'icon-comment'),
 
    'user_commented_pull_request': (_('[commented] on pull request for'),
 
                                    get_pull_request, 'icon-comment'),
 
    'user_closed_pull_request':    (_('[closed] pull request for'),
 
                                    get_pull_request, 'icon-ok'),
 
    'push':                        (_('[pushed] into'),
 
                                    get_cs_links, 'icon-move-up'),
 
    'push_local':                  (_('[committed via Kallithea] into repository'),
 
                                    get_cs_links, 'icon-pencil'),
 
    'push_remote':                 (_('[pulled from remote] into repository'),
 
                                    get_cs_links, 'icon-move-up'),
 
    'pull':                        (_('[pulled] from'),
 
                                    None, 'icon-move-down'),
 
    'started_following_repo':      (_('[started following] repository'),
 
                                    None, 'icon-heart'),
 
    'stopped_following_repo':      (_('[stopped following] repository'),
 
                                    None, 'icon-heart-empty'),
 
    }
 

	
 
    action_str = action_map.get(action, action)
 
    if feed:
 
        action = action_str[0].replace('[', '').replace(']', '')
 
    else:
 
        action = action_str[0] \
 
            .replace('[', '<b>') \
 
            .replace(']', '</b>')
 

	
 
    action_params_func = lambda: ""
 

	
 
    if callable(action_str[1]):
 
        action_params_func = action_str[1]
 

	
 
    def action_parser_icon():
 
        action = user_log.action
 
        action_params = None
 
        x = action.split(':')
 

	
 
        if len(x) > 1:
 
            action, action_params = x
 

	
 
        ico = action_map.get(action, ['', '', ''])[2]
 
        html = """<i class="%s"></i>""" % ico
 
        return literal(html)
 

	
 
    # returned callbacks we need to call to get
 
    return [lambda: literal(action), action_params_func, action_parser_icon]
 

	
 

	
 

	
 
#==============================================================================
 
# PERMS
 
#==============================================================================
 
from kallithea.lib.auth import HasPermissionAny, \
 
    HasRepoPermissionLevel, HasRepoGroupPermissionLevel
 

	
 

	
 
#==============================================================================
 
# GRAVATAR URL
 
#==============================================================================
 
def gravatar_div(email_address, cls='', size=30, **div_attributes):
 
    """Return an html literal with a span around a gravatar if they are enabled.
 
    Extra keyword parameters starting with 'div_' will get the prefix removed
 
    and '_' changed to '-' and be used as attributes on the div. The default
 
    class is 'gravatar'.
 
    """
 
    from tg import tmpl_context as c
 
    if not c.visual.use_gravatar:
 
        return ''
 
    if 'div_class' not in div_attributes:
 
        div_attributes['div_class'] = "gravatar"
 
    attributes = []
 
    for k, v in sorted(div_attributes.items()):
 
        assert k.startswith('div_'), k
 
        attributes.append(' %s="%s"' % (k[4:].replace('_', '-'), escape(v)))
 
    return literal("""<span%s>%s</span>""" %
 
                   (''.join(attributes),
 
                    gravatar(email_address, cls=cls, size=size)))
 

	
kallithea/lib/middleware/pygrack.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
# This program is free software: you can redistribute it and/or modify
 
# it under the terms of the GNU General Public License as published by
 
# the Free Software Foundation, either version 3 of the License, or
 
# (at your option) any later version.
 
#
 
# This program is distributed in the hope that it will be useful,
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
# GNU General Public License for more details.
 
#
 
# You should have received a copy of the GNU General Public License
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
"""
 
kallithea.lib.middleware.pygrack
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

	
 
Python implementation of git-http-backend's Smart HTTP protocol
 

	
 
Based on original code from git_http_backend.py project.
 

	
 
Copyright (c) 2010 Daniel Dotsenko <dotsa@hotmail.com>
 
Copyright (c) 2012 Marcin Kuzminski <marcin@python-works.com>
 

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

	
 
import os
 
import socket
 
import logging
 
import traceback
 

	
 
from webob import Request, Response, exc
 

	
 
import kallithea
 
from kallithea.lib.vcs import subprocessio
 
from kallithea.lib.utils2 import safe_unicode
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
class FileWrapper(object):
 

	
 
    def __init__(self, fd, content_length):
 
        self.fd = fd
 
        self.content_length = content_length
 
        self.remain = content_length
 

	
 
    def read(self, size):
 
        if size <= self.remain:
 
            try:
 
                data = self.fd.read(size)
 
            except socket.error:
 
                raise IOError(self)
 
            self.remain -= size
 
        elif self.remain:
 
            data = self.fd.read(self.remain)
 
            self.remain = 0
 
        else:
 
            data = None
 
        return data
 

	
 
    def __repr__(self):
 
        return '<FileWrapper %s len: %s, read: %s>' % (
 
            self.fd, self.content_length, self.content_length - self.remain
 
        )
 

	
 

	
 
class GitRepository(object):
 
    git_folder_signature = set(['config', 'head', 'info', 'objects', 'refs'])
 
    commands = ['git-upload-pack', 'git-receive-pack']
 

	
 
    def __init__(self, repo_name, content_path, extras):
 
        files = set([f.lower() for f in os.listdir(content_path)])
 
        if not (self.git_folder_signature.intersection(files)
 
                == self.git_folder_signature):
 
            raise OSError('%s missing git signature' % content_path)
 
        self.content_path = content_path
 
        self.valid_accepts = ['application/x-%s-result' %
 
                              c for c in self.commands]
 
        self.repo_name = repo_name
 
        self.extras = extras
 

	
 
    def _get_fixedpath(self, path):
 
        """
 
        Small fix for repo_path
 

	
 
        :param path:
 
        """
 
        path = safe_unicode(path)
 
        assert path.startswith('/' + self.repo_name + '/')
 
        return path[len(self.repo_name) + 2:].strip('/')
 

	
 
    def inforefs(self, req, environ):
 
        """
 
        WSGI Response producer for HTTP GET Git Smart
 
        HTTP /info/refs request.
 
        """
 

	
 
        git_command = req.GET.get('service')
 
        if git_command not in self.commands:
 
            log.debug('command %s not allowed', git_command)
 
            return exc.HTTPMethodNotAllowed()
 

	
 
        # From Documentation/technical/http-protocol.txt shipped with Git:
 
        #
 
        # Clients MUST verify the first pkt-line is `# service=$servicename`.
 
        # Servers MUST set $servicename to be the request parameter value.
 
        # Servers SHOULD include an LF at the end of this line.
 
        # Clients MUST ignore an LF at the end of the line.
 
        #
 
        #  smart_reply     =  PKT-LINE("# service=$servicename" LF)
 
        #                     ref_list
 
        #                     "0000"
 
        server_advert = '# service=%s\n' % git_command
 
        packet_len = str(hex(len(server_advert) + 4)[2:].rjust(4, '0')).lower()
 
        _git_path = kallithea.CONFIG.get('git_path', 'git')
 
        cmd = [_git_path, git_command[4:],
 
               '--stateless-rpc', '--advertise-refs', self.content_path]
 
        log.debug('handling cmd %s', cmd)
 
        try:
 
            out = subprocessio.SubprocessIOChunker(cmd,
 
                starting_values=[packet_len + server_advert + '0000']
 
            )
 
        except EnvironmentError as e:
 
            log.error(traceback.format_exc())
 
            raise exc.HTTPExpectationFailed()
 
        resp = Response()
 
        resp.content_type = 'application/x-%s-advertisement' % str(git_command)
 
        resp.charset = None
 
        resp.app_iter = out
 
        return resp
 

	
 
    def backend(self, req, environ):
 
        """
 
        WSGI Response producer for HTTP POST Git Smart HTTP requests.
 
        Reads commands and data from HTTP POST's body.
 
        returns an iterator obj with contents of git command's
 
        response to stdout
 
        """
 
        _git_path = kallithea.CONFIG.get('git_path', 'git')
 
        git_command = self._get_fixedpath(req.path_info)
 
        if git_command not in self.commands:
 
            log.debug('command %s not allowed', git_command)
 
            return exc.HTTPMethodNotAllowed()
 

	
 
        if 'CONTENT_LENGTH' in environ:
 
            inputstream = FileWrapper(environ['wsgi.input'],
 
                                      req.content_length)
 
        else:
 
            inputstream = environ['wsgi.input']
 

	
 
        gitenv = dict(os.environ)
 
        # forget all configs
 
        gitenv['GIT_CONFIG_NOGLOBAL'] = '1'
 
        cmd = [_git_path, git_command[4:], '--stateless-rpc', self.content_path]
 
        log.debug('handling cmd %s', cmd)
 
        try:
 
            out = subprocessio.SubprocessIOChunker(
 
                cmd,
 
                inputstream=inputstream,
 
                env=gitenv,
 
                cwd=self.content_path,
 
            )
 
        except EnvironmentError as e:
 
            log.error(traceback.format_exc())
 
            raise exc.HTTPExpectationFailed()
 

	
 
        if git_command in [u'git-receive-pack']:
 
            # updating refs manually after each push.
 
            # Needed for pre-1.7.0.4 git clients using regular HTTP mode.
 
            from kallithea.lib.vcs import get_repo
 
            from dulwich.server import update_server_info
 
            repo = get_repo(self.content_path)
 
            if repo:
 
                update_server_info(repo._repo)
 

	
 
        resp = Response()
 
        resp.content_type = 'application/x-%s-result' % git_command.encode('utf8')
 
        resp.content_type = 'application/x-%s-result' % git_command.encode('utf-8')
 
        resp.charset = None
 
        resp.app_iter = out
 
        return resp
 

	
 
    def __call__(self, environ, start_response):
 
        req = Request(environ)
 
        _path = self._get_fixedpath(req.path_info)
 
        if _path.startswith('info/refs'):
 
            app = self.inforefs
 
        elif [a for a in self.valid_accepts if a in req.accept]:
 
            app = self.backend
 
        try:
 
            resp = app(req, environ)
 
        except exc.HTTPException as e:
 
            resp = e
 
            log.error(traceback.format_exc())
 
        except Exception as e:
 
            log.error(traceback.format_exc())
 
            resp = exc.HTTPInternalServerError()
 
        return resp(environ, start_response)
 

	
 

	
 
class GitDirectory(object):
 

	
 
    def __init__(self, repo_root, repo_name, extras):
 
        repo_location = os.path.join(repo_root, repo_name)
 
        if not os.path.isdir(repo_location):
 
            raise OSError(repo_location)
 

	
 
        self.content_path = repo_location
 
        self.repo_name = repo_name
 
        self.repo_location = repo_location
 
        self.extras = extras
 

	
 
    def __call__(self, environ, start_response):
 
        content_path = self.content_path
 
        try:
 
            app = GitRepository(self.repo_name, content_path, self.extras)
 
        except (AssertionError, OSError):
 
            content_path = os.path.join(content_path, '.git')
 
            if os.path.isdir(content_path):
 
                app = GitRepository(self.repo_name, content_path, self.extras)
 
            else:
 
                return exc.HTTPNotFound()(environ, start_response)
 
        return app(environ, start_response)
 

	
 

	
 
def make_wsgi_app(repo_name, repo_root, extras):
 
    from dulwich.web import LimitedInputFilter, GunzipFilter
 
    app = GitDirectory(repo_root, repo_name, extras)
 
    return GunzipFilter(LimitedInputFilter(app))
kallithea/lib/paster_commands/template.ini.mako
Show inline comments
 
## -*- coding: utf-8 -*-
 
<%text>################################################################################</%text>
 
<%text>################################################################################</%text>
 
# Kallithea - config file generated with kallithea-config                      #
 
#                                                                              #
 
# The %(here)s variable will be replaced with the parent directory of this file#
 
<%text>################################################################################</%text>
 
<%text>################################################################################</%text>
 

	
 
[DEFAULT]
 

	
 
<%text>################################################################################</%text>
 
<%text>## Email settings                                                             ##</%text>
 
<%text>##                                                                            ##</%text>
 
<%text>## Refer to the documentation ("Email settings") for more details.            ##</%text>
 
<%text>##                                                                            ##</%text>
 
<%text>## It is recommended to use a valid sender address that passes access         ##</%text>
 
<%text>## validation and spam filtering in mail servers.                             ##</%text>
 
<%text>################################################################################</%text>
 

	
 
<%text>## 'From' header for application emails. You can optionally add a name.</%text>
 
<%text>## Default:</%text>
 
#app_email_from = Kallithea
 
<%text>## Examples:</%text>
 
#app_email_from = Kallithea <kallithea-noreply@example.com>
 
#app_email_from = kallithea-noreply@example.com
 

	
 
<%text>## Subject prefix for application emails.</%text>
 
<%text>## A space between this prefix and the real subject is automatically added.</%text>
 
<%text>## Default:</%text>
 
#email_prefix =
 
<%text>## Example:</%text>
 
#email_prefix = [Kallithea]
 

	
 
<%text>## Recipients for error emails and fallback recipients of application mails.</%text>
 
<%text>## Multiple addresses can be specified, comma-separated.</%text>
 
<%text>## Only addresses are allowed, do not add any name part.</%text>
 
<%text>## Default:</%text>
 
#email_to =
 
<%text>## Examples:</%text>
 
#email_to = admin@example.com
 
#email_to = admin@example.com,another_admin@example.com
 
email_to =
 

	
 
<%text>## 'From' header for error emails. You can optionally add a name.</%text>
 
<%text>## Default: (none)</%text>
 
<%text>## Examples:</%text>
 
#error_email_from = Kallithea Errors <kallithea-noreply@example.com>
 
#error_email_from = kallithea_errors@example.com
 
error_email_from =
 

	
 
<%text>## SMTP server settings</%text>
 
<%text>## If specifying credentials, make sure to use secure connections.</%text>
 
<%text>## Default: Send unencrypted unauthenticated mails to the specified smtp_server.</%text>
 
<%text>## For "SSL", use smtp_use_ssl = true and smtp_port = 465.</%text>
 
<%text>## For "STARTTLS", use smtp_use_tls = true and smtp_port = 587.</%text>
 
smtp_server =
 
#smtp_username =
 
#smtp_password =
 
smtp_port =
 
#smtp_use_ssl = false
 
#smtp_use_tls = false
 

	
 
%if http_server != 'uwsgi':
 
<%text>## Entry point for 'gearbox serve'</%text>
 
[server:main]
 
host = ${host}
 
port = ${port}
 

	
 
%if http_server == 'gearbox':
 
<%text>## Gearbox default web server ##</%text>
 
use = egg:gearbox#wsgiref
 
<%text>## nr of worker threads to spawn</%text>
 
threadpool_workers = 1
 
<%text>## max request before thread respawn</%text>
 
threadpool_max_requests = 100
 
<%text>## option to use threads of process</%text>
 
use_threadpool = true
 

	
 
%elif http_server == 'gevent':
 
<%text>## Gearbox gevent web server ##</%text>
 
use = egg:gearbox#gevent
 

	
 
%elif http_server == 'waitress':
 
<%text>## WAITRESS ##</%text>
 
use = egg:waitress#main
 
<%text>## number of worker threads</%text>
 
threads = 1
 
<%text>## MAX BODY SIZE 100GB</%text>
 
max_request_body_size = 107374182400
 
<%text>## use poll instead of select, fixes fd limits, may not work on old</%text>
 
<%text>## windows systems.</%text>
 
#asyncore_use_poll = True
 

	
 
%elif http_server == 'gunicorn':
 
<%text>## GUNICORN ##</%text>
 
use = egg:gunicorn#main
 
<%text>## number of process workers. You must set `instance_id = *` when this option</%text>
 
<%text>## is set to more than one worker</%text>
 
workers = 4
 
<%text>## process name</%text>
 
proc_name = kallithea
 
<%text>## type of worker class, one of sync, eventlet, gevent, tornado</%text>
 
<%text>## recommended for bigger setup is using of of other than sync one</%text>
 
worker_class = sync
 
max_requests = 1000
 
<%text>## amount of time a worker can handle request before it gets killed and</%text>
 
<%text>## restarted</%text>
 
timeout = 3600
 

	
 
%endif
 
%else:
 
<%text>## UWSGI ##</%text>
 
<%text>## run with uwsgi --ini-paste-logged <inifile.ini></%text>
 
[uwsgi]
 
socket = /tmp/uwsgi.sock
 
master = true
 
http = ${host}:${port}
 

	
 
<%text>## set as daemon and redirect all output to file</%text>
 
#daemonize = ./uwsgi_kallithea.log
 

	
 
<%text>## master process PID</%text>
 
pidfile = ./uwsgi_kallithea.pid
 

	
 
<%text>## stats server with workers statistics, use uwsgitop</%text>
 
<%text>## for monitoring, `uwsgitop 127.0.0.1:1717`</%text>
 
stats = 127.0.0.1:1717
 
memory-report = true
 

	
 
<%text>## log 5XX errors</%text>
 
log-5xx = true
 

	
 
<%text>## Set the socket listen queue size.</%text>
 
listen = 128
 

	
 
<%text>## Gracefully Reload workers after the specified amount of managed requests</%text>
 
<%text>## (avoid memory leaks).</%text>
 
max-requests = 1000
 

	
 
<%text>## enable large buffers</%text>
 
buffer-size = 65535
 

	
 
<%text>## socket and http timeouts ##</%text>
 
http-timeout = 3600
 
socket-timeout = 3600
 

	
 
<%text>## Log requests slower than the specified number of milliseconds.</%text>
 
log-slow = 10
 

	
 
<%text>## Exit if no app can be loaded.</%text>
 
need-app = true
 

	
 
<%text>## Set lazy mode (load apps in workers instead of master).</%text>
 
lazy = true
 

	
 
<%text>## scaling ##</%text>
 
<%text>## set cheaper algorithm to use, if not set default will be used</%text>
 
cheaper-algo = spare
 

	
 
<%text>## minimum number of workers to keep at all times</%text>
 
cheaper = 1
 

	
 
<%text>## number of workers to spawn at startup</%text>
 
cheaper-initial = 1
 

	
 
<%text>## maximum number of workers that can be spawned</%text>
 
workers = 4
 

	
 
<%text>## how many workers should be spawned at a time</%text>
 
cheaper-step = 1
 

	
 
%endif
 
<%text>## middleware for hosting the WSGI application under a URL prefix</%text>
 
#[filter:proxy-prefix]
 
#use = egg:PasteDeploy#prefix
 
#prefix = /<your-prefix>
 

	
 
[app:main]
 
use = egg:kallithea
 
<%text>## enable proxy prefix middleware</%text>
 
#filter-with = proxy-prefix
 

	
 
full_stack = true
 
static_files = true
 

	
 
<%text>## Internationalization (see setup documentation for details)</%text>
 
<%text>## By default, the language requested by the browser is used if available.</%text>
 
#i18n.enable = false
 
<%text>## Fallback language, empty for English (valid values are the names of subdirectories in kallithea/i18n):</%text>
 
i18n.lang =
 

	
 
cache_dir = %(here)s/data
 
index_dir = %(here)s/data/index
 

	
 
<%text>## uncomment and set this path to use archive download cache</%text>
 
archive_cache_dir = %(here)s/tarballcache
 

	
 
<%text>## change this to unique ID for security</%text>
 
app_instance_uuid = ${uuid()}
 

	
 
<%text>## cut off limit for large diffs (size in bytes)</%text>
 
cut_off_limit = 256000
 

	
 
<%text>## force https in Kallithea, fixes https redirects, assumes it's always https</%text>
 
force_https = false
 

	
 
<%text>## use Strict-Transport-Security headers</%text>
 
use_htsts = false
 

	
 
<%text>## number of commits stats will parse on each iteration</%text>
 
commit_parse_limit = 25
 

	
 
<%text>## path to git executable</%text>
 
git_path = git
 

	
 
<%text>## git rev filter option, --all is the default filter, if you need to</%text>
 
<%text>## hide all refs in changelog switch this to --branches --tags</%text>
 
#git_rev_filter = --branches --tags
 

	
 
<%text>## RSS feed options</%text>
 
rss_cut_off_limit = 256000
 
rss_items_per_page = 10
 
rss_include_diff = false
 

	
 
<%text>## options for showing and identifying changesets</%text>
 
show_sha_length = 12
 
show_revision_number = false
 

	
 
<%text>## Canonical URL to use when creating full URLs in UI and texts.</%text>
 
<%text>## Useful when the site is available under different names or protocols.</%text>
 
<%text>## Defaults to what is provided in the WSGI environment.</%text>
 
#canonical_url = https://kallithea.example.com/repos
 

	
 
<%text>## gist URL alias, used to create nicer urls for gist. This should be an</%text>
 
<%text>## url that does rewrites to _admin/gists/<gistid>.</%text>
 
<%text>## example: http://gist.example.com/{gistid}. Empty means use the internal</%text>
 
<%text>## Kallithea url, ie. http[s]://kallithea.example.com/_admin/gists/<gistid></%text>
 
gist_alias_url =
 

	
 
<%text>## white list of API enabled controllers. This allows to add list of</%text>
 
<%text>## controllers to which access will be enabled by api_key. eg: to enable</%text>
 
<%text>## api access to raw_files put `FilesController:raw`, to enable access to patches</%text>
 
<%text>## add `ChangesetController:changeset_patch`. This list should be "," separated</%text>
 
<%text>## Syntax is <ControllerClass>:<function>. Check debug logs for generated names</%text>
 
<%text>## Recommended settings below are commented out:</%text>
 
api_access_controllers_whitelist =
 
#    ChangesetController:changeset_patch,
 
#    ChangesetController:changeset_raw,
 
#    FilesController:raw,
 
#    FilesController:archivefile
 

	
 
<%text>## default encoding used to convert from and to unicode</%text>
 
<%text>## can be also a comma separated list of encoding in case of mixed encodings</%text>
 
default_encoding = utf8
 
default_encoding = utf-8
 

	
 
<%text>## Set Mercurial encoding, similar to setting HGENCODING before launching Kallithea</%text>
 
hgencoding = utf-8
 

	
 
<%text>## issue tracker for Kallithea (leave blank to disable, absent for default)</%text>
 
#bugtracker = https://bitbucket.org/conservancy/kallithea/issues
 

	
 
<%text>## issue tracking mapping for commit messages, comments, PR descriptions, ...</%text>
 
<%text>## Refer to the documentation ("Integration with issue trackers") for more details.</%text>
 

	
 
<%text>## regular expression to match issue references</%text>
 
<%text>## This pattern may/should contain parenthesized groups, that can</%text>
 
<%text>## be referred to in issue_server_link or issue_sub using Python backreferences</%text>
 
<%text>## (e.g. \1, \2, ...). You can also create named groups with '(?P<groupname>)'.</%text>
 
<%text>## To require mandatory whitespace before the issue pattern, use:</%text>
 
<%text>## (?:^|(?<=\s)) before the actual pattern, and for mandatory whitespace</%text>
 
<%text>## behind the issue pattern, use (?:$|(?=\s)) after the actual pattern.</%text>
 

	
 
issue_pat = #(\d+)
 

	
 
<%text>## server url to the issue</%text>
 
<%text>## This pattern may/should contain backreferences to parenthesized groups in issue_pat.</%text>
 
<%text>## A backreference can be \1, \2, ... or \g<groupname> if you specified a named group</%text>
 
<%text>## called 'groupname' in issue_pat.</%text>
 
<%text>## The special token {repo} is replaced with the full repository name</%text>
 
<%text>## including repository groups, while {repo_name} is replaced with just</%text>
 
<%text>## the name of the repository.</%text>
 

	
 
issue_server_link = https://issues.example.com/{repo}/issue/\1
 

	
 
<%text>## substitution pattern to use as the link text</%text>
 
<%text>## If issue_sub is empty, the text matched by issue_pat is retained verbatim</%text>
 
<%text>## for the link text. Otherwise, the link text is that of issue_sub, with any</%text>
 
<%text>## backreferences to groups in issue_pat replaced.</%text>
 

	
 
issue_sub =
 

	
 
<%text>## issue_pat, issue_server_link and issue_sub can have suffixes to specify</%text>
 
<%text>## multiple patterns, to other issues server, wiki or others</%text>
 
<%text>## below an example how to create a wiki pattern</%text>
 
# wiki-some-id -> https://wiki.example.com/some-id
 

	
 
#issue_pat_wiki = wiki-(\S+)
 
#issue_server_link_wiki = https://wiki.example.com/\1
 
#issue_sub_wiki = WIKI-\1
 

	
 
<%text>## alternative return HTTP header for failed authentication. Default HTTP</%text>
 
<%text>## response is 401 HTTPUnauthorized. Currently Mercurial clients have trouble with</%text>
 
<%text>## handling that. Set this variable to 403 to return HTTPForbidden</%text>
 
auth_ret_code =
 

	
 
<%text>## locking return code. When repository is locked return this HTTP code. 2XX</%text>
 
<%text>## codes don't break the transactions while 4XX codes do</%text>
 
lock_ret_code = 423
 

	
 
<%text>## allows to change the repository location in settings page</%text>
 
allow_repo_location_change = True
 

	
 
<%text>## allows to setup custom hooks in settings page</%text>
 
allow_custom_hooks_settings = True
 

	
 
<%text>## extra extensions for indexing, space separated and without the leading '.'.</%text>
 
# index.extensions =
 
#    gemfile
 
#    lock
 

	
 
<%text>## extra filenames for indexing, space separated</%text>
 
# index.filenames =
 
#    .dockerignore
 
#    .editorconfig
 
#    INSTALL
 
#    CHANGELOG
 

	
 
<%text>####################################</%text>
 
<%text>###        CELERY CONFIG        ####</%text>
 
<%text>####################################</%text>
 

	
 
use_celery = false
 

	
 
<%text>## Example: connect to the virtual host 'rabbitmqhost' on localhost as rabbitmq:</%text>
 
broker.url = amqp://rabbitmq:qewqew@localhost:5672/rabbitmqhost
 

	
 
celery.imports = kallithea.lib.celerylib.tasks
 
celery.accept.content = pickle
 
celery.result.backend = amqp
 
celery.result.dburi = amqp://
 
celery.result.serialier = json
 

	
 
#celery.send.task.error.emails = true
 
#celery.amqp.task.result.expires = 18000
 

	
 
celeryd.concurrency = 2
 
celeryd.max.tasks.per.child = 1
 

	
 
<%text>## If true, tasks will never be sent to the queue, but executed locally instead.</%text>
 
celery.always.eager = false
 

	
 
<%text>####################################</%text>
 
<%text>###         BEAKER CACHE        ####</%text>
 
<%text>####################################</%text>
 

	
 
beaker.cache.data_dir = %(here)s/data/cache/data
 
beaker.cache.lock_dir = %(here)s/data/cache/lock
 

	
 
beaker.cache.regions = short_term,long_term,sql_cache_short
 

	
 
beaker.cache.short_term.type = memory
 
beaker.cache.short_term.expire = 60
 
beaker.cache.short_term.key_length = 256
 

	
 
beaker.cache.long_term.type = memory
 
beaker.cache.long_term.expire = 36000
 
beaker.cache.long_term.key_length = 256
 

	
 
beaker.cache.sql_cache_short.type = memory
 
beaker.cache.sql_cache_short.expire = 10
 
beaker.cache.sql_cache_short.key_length = 256
 

	
 
<%text>####################################</%text>
 
<%text>###       BEAKER SESSION        ####</%text>
 
<%text>####################################</%text>
 

	
 
<%text>## Name of session cookie. Should be unique for a given host and path, even when running</%text>
 
<%text>## on different ports. Otherwise, cookie sessions will be shared and messed up.</%text>
 
beaker.session.key = kallithea
 
<%text>## Sessions should always only be accessible by the browser, not directly by JavaScript.</%text>
 
beaker.session.httponly = true
 
<%text>## Session lifetime. 2592000 seconds is 30 days.</%text>
 
beaker.session.timeout = 2592000
 

	
 
<%text>## Server secret used with HMAC to ensure integrity of cookies.</%text>
 
beaker.session.secret = ${uuid()}
 
<%text>## Further, encrypt the data with AES.</%text>
 
#beaker.session.encrypt_key = <key_for_encryption>
 
#beaker.session.validate_key = <validation_key>
 

	
 
<%text>## Type of storage used for the session, current types are</%text>
 
<%text>## dbm, file, memcached, database, and memory.</%text>
 

	
 
<%text>## File system storage of session data. (default)</%text>
 
#beaker.session.type = file
 

	
 
<%text>## Cookie only, store all session data inside the cookie. Requires secure secrets.</%text>
 
#beaker.session.type = cookie
 

	
 
<%text>## Database storage of session data.</%text>
 
#beaker.session.type = ext:database
 
#beaker.session.sa.url = postgresql://postgres:qwe@localhost/kallithea
 
#beaker.session.table_name = db_session
 

	
 
%if error_aggregation_service == 'appenlight':
 
<%text>############################</%text>
 
<%text>## ERROR HANDLING SYSTEMS ##</%text>
 
<%text>############################</%text>
 

	
 
# Propagate email settings to ErrorReporter of TurboGears2
 
# You do not normally need to change these lines
 
get trace_errors.error_email = email_to
 
get trace_errors.smtp_server = smtp_server
 
get trace_errors.smtp_port = smtp_port
 
get trace_errors.from_address = error_email_from
 

	
 
<%text>####################</%text>
 
<%text>### [appenlight] ###</%text>
 
<%text>####################</%text>
 

	
 
<%text>## AppEnlight is tailored to work with Kallithea, see</%text>
 
<%text>## http://appenlight.com for details how to obtain an account</%text>
 
<%text>## you must install python package `appenlight_client` to make it work</%text>
 

	
 
<%text>## appenlight enabled</%text>
 
appenlight = false
 

	
 
appenlight.server_url = https://api.appenlight.com
 
appenlight.api_key = YOUR_API_KEY
 

	
 
<%text>## TWEAK AMOUNT OF INFO SENT HERE</%text>
 

	
 
<%text>## enables 404 error logging (default False)</%text>
 
appenlight.report_404 = false
 

	
 
<%text>## time in seconds after request is considered being slow (default 1)</%text>
 
appenlight.slow_request_time = 1
 

	
 
<%text>## record slow requests in application</%text>
 
<%text>## (needs to be enabled for slow datastore recording and time tracking)</%text>
 
appenlight.slow_requests = true
 

	
 
<%text>## enable hooking to application loggers</%text>
 
#appenlight.logging = true
 

	
 
<%text>## minimum log level for log capture</%text>
 
#appenlight.logging.level = WARNING
 

	
 
<%text>## send logs only from erroneous/slow requests</%text>
 
<%text>## (saves API quota for intensive logging)</%text>
 
appenlight.logging_on_error = false
 

	
 
<%text>## list of additional keywords that should be grabbed from environ object</%text>
 
<%text>## can be string with comma separated list of words in lowercase</%text>
 
<%text>## (by default client will always send following info:</%text>
 
<%text>## 'REMOTE_USER', 'REMOTE_ADDR', 'SERVER_NAME', 'CONTENT_TYPE' + all keys that</%text>
 
<%text>## start with HTTP* this list be extended with additional keywords here</%text>
 
appenlight.environ_keys_whitelist =
 

	
 
<%text>## list of keywords that should be blanked from request object</%text>
 
<%text>## can be string with comma separated list of words in lowercase</%text>
 
<%text>## (by default client will always blank keys that contain following words</%text>
 
<%text>## 'password', 'passwd', 'pwd', 'auth_tkt', 'secret', 'csrf'</%text>
 
<%text>## this list be extended with additional keywords set here</%text>
 
appenlight.request_keys_blacklist =
 

	
 
<%text>## list of namespaces that should be ignores when gathering log entries</%text>
 
<%text>## can be string with comma separated list of namespaces</%text>
 
<%text>## (by default the client ignores own entries: appenlight_client.client)</%text>
 
appenlight.log_namespace_blacklist =
 

	
 
%elif error_aggregation_service == 'sentry':
 
<%text>################</%text>
 
<%text>### [sentry] ###</%text>
 
<%text>################</%text>
 

	
 
<%text>## sentry is a alternative open source error aggregator</%text>
 
<%text>## you must install python packages `sentry` and `raven` to enable</%text>
 

	
 
sentry.dsn = YOUR_DNS
 
sentry.servers =
 
sentry.name =
 
sentry.key =
 
sentry.public_key =
 
sentry.secret_key =
 
sentry.project =
 
sentry.site =
 
sentry.include_paths =
 
sentry.exclude_paths =
 

	
 
%endif
 
<%text>################################################################################</%text>
 
<%text>## WARNING: *DEBUG MODE MUST BE OFF IN A PRODUCTION ENVIRONMENT*              ##</%text>
 
<%text>## Debug mode will enable the interactive debugging tool, allowing ANYONE to  ##</%text>
 
<%text>## execute malicious code after an exception is raised.                       ##</%text>
 
<%text>################################################################################</%text>
 
debug = false
 

	
 
<%text>##################################</%text>
 
<%text>###       LOGVIEW CONFIG       ###</%text>
 
<%text>##################################</%text>
 

	
 
logview.sqlalchemy = #faa
 
logview.pylons.templating = #bfb
 
logview.pylons.util = #eee
 

	
 
<%text>#########################################################</%text>
 
<%text>### DB CONFIGS - EACH DB WILL HAVE IT'S OWN CONFIG    ###</%text>
 
<%text>#########################################################</%text>
 

	
 
%if database_engine == 'sqlite':
 
# SQLITE [default]
 
sqlalchemy.url = sqlite:///%(here)s/kallithea.db?timeout=60
 

	
 
%elif database_engine == 'postgres':
 
# POSTGRESQL
 
sqlalchemy.url = postgresql://user:pass@localhost/kallithea
 

	
 
%elif database_engine == 'mysql':
 
# MySQL
 
sqlalchemy.url = mysql://user:pass@localhost/kallithea?charset=utf8
 

	
 
%endif
 
# see sqlalchemy docs for others
 

	
 
sqlalchemy.pool_recycle = 3600
 

	
 
<%text>################################</%text>
 
<%text>### ALEMBIC CONFIGURATION   ####</%text>
 
<%text>################################</%text>
 

	
 
[alembic]
 
script_location = kallithea:alembic
 

	
 
<%text>################################</%text>
 
<%text>### LOGGING CONFIGURATION   ####</%text>
 
<%text>################################</%text>
 

	
 
[loggers]
 
keys = root, routes, kallithea, sqlalchemy, tg, gearbox, beaker, templates, whoosh_indexer, werkzeug, backlash
 

	
 
[handlers]
 
keys = console, console_sql
 

	
 
[formatters]
 
keys = generic, color_formatter, color_formatter_sql
 

	
 
<%text>#############</%text>
 
<%text>## LOGGERS ##</%text>
 
<%text>#############</%text>
 

	
 
[logger_root]
 
level = NOTSET
 
handlers = console
 

	
 
[logger_routes]
 
level = WARN
 
handlers =
 
qualname = routes.middleware
 
<%text>## "level = DEBUG" logs the route matched and routing variables.</%text>
 
propagate = 1
 

	
 
[logger_beaker]
 
level = WARN
 
handlers =
 
qualname = beaker.container
 
propagate = 1
 

	
 
[logger_templates]
 
level = WARN
 
handlers =
 
qualname = pylons.templating
 
propagate = 1
 

	
 
[logger_kallithea]
 
level = WARN
 
handlers =
 
qualname = kallithea
 
propagate = 1
 

	
 
[logger_tg]
 
level = WARN
 
handlers =
 
qualname = tg
 
propagate = 1
 

	
 
[logger_gearbox]
 
level = WARN
 
handlers =
 
qualname = gearbox
 
propagate = 1
 

	
 
[logger_sqlalchemy]
 
level = WARN
 
handlers = console_sql
 
qualname = sqlalchemy.engine
 
propagate = 0
 

	
 
[logger_whoosh_indexer]
 
level = WARN
 
handlers =
 
qualname = whoosh_indexer
 
propagate = 1
 

	
 
[logger_werkzeug]
 
level = WARN
 
handlers =
 
qualname = werkzeug
 
propagate = 1
 

	
 
[logger_backlash]
 
level = WARN
 
handlers =
 
qualname = backlash
 
propagate = 1
 

	
 
<%text>##############</%text>
 
<%text>## HANDLERS ##</%text>
 
<%text>##############</%text>
 

	
 
[handler_console]
 
class = StreamHandler
 
args = (sys.stderr,)
 
formatter = generic
 

	
 
[handler_console_sql]
 
class = StreamHandler
 
args = (sys.stderr,)
 
formatter = generic
 

	
 
<%text>################</%text>
 
<%text>## FORMATTERS ##</%text>
 
<%text>################</%text>
 

	
 
[formatter_generic]
 
format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 
datefmt = %Y-%m-%d %H:%M:%S
 

	
kallithea/lib/utils.py
Show inline comments
 
@@ -54,657 +54,657 @@ log = logging.getLogger(__name__)
 
REMOVED_REPO_PAT = re.compile(r'rm__\d{8}_\d{6}_\d{6}_.*')
 

	
 

	
 
def recursive_replace(str_, replace=' '):
 
    """
 
    Recursive replace of given sign to just one instance
 

	
 
    :param str_: given string
 
    :param replace: char to find and replace multiple instances
 

	
 
    Examples::
 
    >>> recursive_replace("Mighty---Mighty-Bo--sstones",'-')
 
    'Mighty-Mighty-Bo-sstones'
 
    """
 

	
 
    if str_.find(replace * 2) == -1:
 
        return str_
 
    else:
 
        str_ = str_.replace(replace * 2, replace)
 
        return recursive_replace(str_, replace)
 

	
 

	
 
def repo_name_slug(value):
 
    """
 
    Return slug of name of repository
 
    This function is called on each creation/modification
 
    of repository to prevent bad names in repo
 
    """
 

	
 
    slug = remove_formatting(value)
 
    slug = strip_tags(slug)
 

	
 
    for c in """`?=[]\;'"<>,/~!@#$%^&*()+{}|: """:
 
        slug = slug.replace(c, '-')
 
    slug = recursive_replace(slug, '-')
 
    slug = collapse(slug, '-')
 
    return slug
 

	
 

	
 
#==============================================================================
 
# PERM DECORATOR HELPERS FOR EXTRACTING NAMES FOR PERM CHECKS
 
#==============================================================================
 
def get_repo_slug(request):
 
    _repo = request.environ['pylons.routes_dict'].get('repo_name')
 
    if _repo:
 
        _repo = _repo.rstrip('/')
 
    return _repo
 

	
 

	
 
def get_repo_group_slug(request):
 
    _group = request.environ['pylons.routes_dict'].get('group_name')
 
    if _group:
 
        _group = _group.rstrip('/')
 
    return _group
 

	
 

	
 
def get_user_group_slug(request):
 
    _group = request.environ['pylons.routes_dict'].get('id')
 
    _group = UserGroup.get(_group)
 
    if _group:
 
        return _group.users_group_name
 
    return None
 

	
 

	
 
def _extract_id_from_repo_name(repo_name):
 
    if repo_name.startswith('/'):
 
        repo_name = repo_name.lstrip('/')
 
    by_id_match = re.match(r'^_(\d{1,})', repo_name)
 
    if by_id_match:
 
        return by_id_match.groups()[0]
 

	
 

	
 
def get_repo_by_id(repo_name):
 
    """
 
    Extracts repo_name by id from special urls. Example url is _11/repo_name
 

	
 
    :param repo_name:
 
    :return: repo_name if matched else None
 
    """
 
    _repo_id = _extract_id_from_repo_name(repo_name)
 
    if _repo_id:
 
        from kallithea.model.db import Repository
 
        repo = Repository.get(_repo_id)
 
        if repo:
 
            # TODO: return repo instead of reponame? or would that be a layering violation?
 
            return repo.repo_name
 
    return None
 

	
 

	
 
def action_logger(user, action, repo, ipaddr='', commit=False):
 
    """
 
    Action logger for various actions made by users
 

	
 
    :param user: user that made this action, can be a unique username string or
 
        object containing user_id attribute
 
    :param action: action to log, should be on of predefined unique actions for
 
        easy translations
 
    :param repo: string name of repository or object containing repo_id,
 
        that action was made on
 
    :param ipaddr: optional IP address from what the action was made
 

	
 
    """
 

	
 
    # if we don't get explicit IP address try to get one from registered user
 
    # in tmpl context var
 
    if not ipaddr:
 
        ipaddr = getattr(get_current_authuser(), 'ip_addr', '')
 

	
 
    if getattr(user, 'user_id', None):
 
        user_obj = User.get(user.user_id)
 
    elif isinstance(user, basestring):
 
        user_obj = User.get_by_username(user)
 
    else:
 
        raise Exception('You have to provide a user object or a username')
 

	
 
    if getattr(repo, 'repo_id', None):
 
        repo_obj = Repository.get(repo.repo_id)
 
        repo_name = repo_obj.repo_name
 
    elif isinstance(repo, basestring):
 
        repo_name = repo.lstrip('/')
 
        repo_obj = Repository.get_by_repo_name(repo_name)
 
    else:
 
        repo_obj = None
 
        repo_name = u''
 

	
 
    user_log = UserLog()
 
    user_log.user_id = user_obj.user_id
 
    user_log.username = user_obj.username
 
    user_log.action = safe_unicode(action)
 

	
 
    user_log.repository = repo_obj
 
    user_log.repository_name = repo_name
 

	
 
    user_log.action_date = datetime.datetime.now()
 
    user_log.user_ip = ipaddr
 
    meta.Session().add(user_log)
 

	
 
    log.info('Logging action:%s on %s by user:%s ip:%s',
 
             action, safe_unicode(repo), user_obj, ipaddr)
 
    if commit:
 
        meta.Session().commit()
 

	
 

	
 
def get_filesystem_repos(path):
 
    """
 
    Scans given path for repos and return (name,(type,path)) tuple
 

	
 
    :param path: path to scan for repositories
 
    :param recursive: recursive search and return names with subdirs in front
 
    """
 

	
 
    # remove ending slash for better results
 
    path = safe_str(path.rstrip(os.sep))
 
    log.debug('now scanning in %s', path)
 

	
 
    def isdir(*n):
 
        return os.path.isdir(os.path.join(*n))
 

	
 
    for root, dirs, _files in os.walk(path):
 
        recurse_dirs = []
 
        for subdir in dirs:
 
            # skip removed repos
 
            if REMOVED_REPO_PAT.match(subdir):
 
                continue
 

	
 
            # skip .<something> dirs TODO: rly? then we should prevent creating them ...
 
            if subdir.startswith('.'):
 
                continue
 

	
 
            cur_path = os.path.join(root, subdir)
 
            if isdir(cur_path, '.git'):
 
                log.warning('ignoring non-bare Git repo: %s', cur_path)
 
                continue
 

	
 
            if (isdir(cur_path, '.hg') or
 
                isdir(cur_path, '.svn') or
 
                isdir(cur_path, 'objects') and (isdir(cur_path, 'refs') or
 
                                                os.path.isfile(os.path.join(cur_path, 'packed-refs')))):
 

	
 
                if not os.access(cur_path, os.R_OK) or not os.access(cur_path, os.X_OK):
 
                    log.warning('ignoring repo path without access: %s', cur_path)
 
                    continue
 

	
 
                if not os.access(cur_path, os.W_OK):
 
                    log.warning('repo path without write access: %s', cur_path)
 

	
 
                try:
 
                    scm_info = get_scm(cur_path)
 
                    assert cur_path.startswith(path)
 
                    repo_path = cur_path[len(path) + 1:]
 
                    yield repo_path, scm_info
 
                    continue # no recursion
 
                except VCSError:
 
                    # We should perhaps ignore such broken repos, but especially
 
                    # the bare git detection is unreliable so we dive into it
 
                    pass
 

	
 
            recurse_dirs.append(subdir)
 

	
 
        dirs[:] = recurse_dirs
 

	
 

	
 
def is_valid_repo_uri(repo_type, url, ui):
 
    """Check if the url seems like a valid remote repo location - raise an Exception if any problems"""
 
    if repo_type == 'hg':
 
        from kallithea.lib.vcs.backends.hg.repository import MercurialRepository
 
        if url.startswith('http') or url.startswith('ssh'):
 
            # initially check if it's at least the proper URL
 
            # or does it pass basic auth
 
            MercurialRepository._check_url(url, ui)
 
        elif url.startswith('svn+http'):
 
            try:
 
                from hgsubversion.svnrepo import svnremoterepo
 
            except ImportError:
 
                raise HgsubversionImportError(_('Unable to activate hgsubversion support. '
 
                                                'The "hgsubversion" library is missing'))
 
            svnremoterepo(ui, url).svn.uuid
 
        elif url.startswith('git+http'):
 
            raise NotImplementedError()
 
        else:
 
            raise Exception('URI %s not allowed' % (url,))
 

	
 
    elif repo_type == 'git':
 
        from kallithea.lib.vcs.backends.git.repository import GitRepository
 
        if url.startswith('http') or url.startswith('git'):
 
            # initially check if it's at least the proper URL
 
            # or does it pass basic auth
 
            GitRepository._check_url(url)
 
        elif url.startswith('svn+http'):
 
            raise NotImplementedError()
 
        elif url.startswith('hg+http'):
 
            raise NotImplementedError()
 
        else:
 
            raise Exception('URI %s not allowed' % (url))
 

	
 

	
 
def is_valid_repo(repo_name, base_path, scm=None):
 
    """
 
    Returns True if given path is a valid repository False otherwise.
 
    If scm param is given also compare if given scm is the same as expected
 
    from scm parameter
 

	
 
    :param repo_name:
 
    :param base_path:
 
    :param scm:
 

	
 
    :return True: if given path is a valid repository
 
    """
 
    full_path = os.path.join(safe_str(base_path), safe_str(repo_name))
 

	
 
    try:
 
        scm_ = get_scm(full_path)
 
        if scm:
 
            return scm_[0] == scm
 
        return True
 
    except VCSError:
 
        return False
 

	
 

	
 
def is_valid_repo_group(repo_group_name, base_path, skip_path_check=False):
 
    """
 
    Returns True if given path is a repository group False otherwise
 

	
 
    :param repo_name:
 
    :param base_path:
 
    """
 
    full_path = os.path.join(safe_str(base_path), safe_str(repo_group_name))
 

	
 
    # check if it's not a repo
 
    if is_valid_repo(repo_group_name, base_path):
 
        return False
 

	
 
    try:
 
        # we need to check bare git repos at higher level
 
        # since we might match branches/hooks/info/objects or possible
 
        # other things inside bare git repo
 
        get_scm(os.path.dirname(full_path))
 
        return False
 
    except VCSError:
 
        pass
 

	
 
    # check if it's a valid path
 
    if skip_path_check or os.path.isdir(full_path):
 
        return True
 

	
 
    return False
 

	
 

	
 
# propagated from mercurial documentation
 
ui_sections = ['alias', 'auth',
 
                'decode/encode', 'defaults',
 
                'diff', 'email',
 
                'extensions', 'format',
 
                'merge-patterns', 'merge-tools',
 
                'hooks', 'http_proxy',
 
                'smtp', 'patch',
 
                'paths', 'profiling',
 
                'server', 'trusted',
 
                'ui', 'web', ]
 

	
 

	
 
def make_ui(read_from='file', path=None, clear_session=True):
 
    """
 
    A function that will read python rc files or database
 
    and make an mercurial ui object from read options
 

	
 
    :param path: path to mercurial config file
 
    :param read_from: read from 'file' or 'db'
 
    """
 

	
 
    baseui = ui.ui()
 

	
 
    # clean the baseui object
 
    baseui._ocfg = config.config()
 
    baseui._ucfg = config.config()
 
    baseui._tcfg = config.config()
 

	
 
    if read_from == 'file':
 
        if not os.path.isfile(path):
 
            log.debug('hgrc file is not present at %s, skipping...', path)
 
            return False
 
        log.debug('reading hgrc from %s', path)
 
        cfg = config.config()
 
        cfg.read(path)
 
        for section in ui_sections:
 
            for k, v in cfg.items(section):
 
                log.debug('settings ui from file: [%s] %s=%s', section, k, v)
 
                baseui.setconfig(safe_str(section), safe_str(k), safe_str(v))
 

	
 
    elif read_from == 'db':
 
        sa = meta.Session()
 
        ret = sa.query(Ui).all()
 

	
 
        hg_ui = ret
 
        for ui_ in hg_ui:
 
            if ui_.ui_active:
 
                ui_val = '' if ui_.ui_value is None else safe_str(ui_.ui_value)
 
                log.debug('settings ui from db: [%s] %s=%r', ui_.ui_section,
 
                          ui_.ui_key, ui_val)
 
                baseui.setconfig(safe_str(ui_.ui_section), safe_str(ui_.ui_key),
 
                                 ui_val)
 
        if clear_session:
 
            meta.Session.remove()
 

	
 
        # force set push_ssl requirement to False, Kallithea handles that
 
        baseui.setconfig('web', 'push_ssl', False)
 
        baseui.setconfig('web', 'allow_push', '*')
 
        # prevent interactive questions for ssh password / passphrase
 
        ssh = baseui.config('ui', 'ssh', default='ssh')
 
        baseui.setconfig('ui', 'ssh', '%s -oBatchMode=yes -oIdentitiesOnly=yes' % ssh)
 

	
 
    return baseui
 

	
 

	
 
def set_app_settings(config):
 
    """
 
    Updates app config with new settings from database
 

	
 
    :param config:
 
    """
 
    try:
 
        hgsettings = Setting.get_app_settings()
 
        for k, v in hgsettings.items():
 
            config[k] = v
 
    finally:
 
        meta.Session.remove()
 

	
 

	
 
def set_vcs_config(config):
 
    """
 
    Patch VCS config with some Kallithea specific stuff
 

	
 
    :param config: kallithea.CONFIG
 
    """
 
    from kallithea.lib.vcs import conf
 
    from kallithea.lib.utils2 import aslist
 
    conf.settings.BACKENDS = {
 
        'hg': 'kallithea.lib.vcs.backends.hg.MercurialRepository',
 
        'git': 'kallithea.lib.vcs.backends.git.GitRepository',
 
    }
 

	
 
    conf.settings.GIT_EXECUTABLE_PATH = config.get('git_path', 'git')
 
    conf.settings.GIT_REV_FILTER = config.get('git_rev_filter', '--all').strip()
 
    conf.settings.DEFAULT_ENCODINGS = aslist(config.get('default_encoding',
 
                                                        'utf8'), sep=',')
 
                                                        'utf-8'), sep=',')
 

	
 

	
 
def set_indexer_config(config):
 
    """
 
    Update Whoosh index mapping
 

	
 
    :param config: kallithea.CONFIG
 
    """
 
    from kallithea.config import conf
 

	
 
    log.debug('adding extra into INDEX_EXTENSIONS')
 
    conf.INDEX_EXTENSIONS.extend(re.split('\s+', config.get('index.extensions', '')))
 

	
 
    log.debug('adding extra into INDEX_FILENAMES')
 
    conf.INDEX_FILENAMES.extend(re.split('\s+', config.get('index.filenames', '')))
 

	
 

	
 
def map_groups(path):
 
    """
 
    Given a full path to a repository, create all nested groups that this
 
    repo is inside. This function creates parent-child relationships between
 
    groups and creates default perms for all new groups.
 

	
 
    :param paths: full path to repository
 
    """
 
    sa = meta.Session()
 
    groups = path.split(Repository.url_sep())
 
    parent = None
 
    group = None
 

	
 
    # last element is repo in nested groups structure
 
    groups = groups[:-1]
 
    rgm = RepoGroupModel()
 
    owner = User.get_first_admin()
 
    for lvl, group_name in enumerate(groups):
 
        group_name = u'/'.join(groups[:lvl] + [group_name])
 
        group = RepoGroup.get_by_group_name(group_name)
 
        desc = '%s group' % group_name
 

	
 
        # skip folders that are now removed repos
 
        if REMOVED_REPO_PAT.match(group_name):
 
            break
 

	
 
        if group is None:
 
            log.debug('creating group level: %s group_name: %s',
 
                      lvl, group_name)
 
            group = RepoGroup(group_name, parent)
 
            group.group_description = desc
 
            group.owner = owner
 
            sa.add(group)
 
            rgm._create_default_perms(group)
 
            sa.flush()
 

	
 
        parent = group
 
    return group
 

	
 

	
 
def repo2db_mapper(initial_repo_list, remove_obsolete=False,
 
                   install_git_hooks=False, user=None, overwrite_git_hooks=False):
 
    """
 
    maps all repos given in initial_repo_list, non existing repositories
 
    are created, if remove_obsolete is True it also check for db entries
 
    that are not in initial_repo_list and removes them.
 

	
 
    :param initial_repo_list: list of repositories found by scanning methods
 
    :param remove_obsolete: check for obsolete entries in database
 
    :param install_git_hooks: if this is True, also check and install git hook
 
        for a repo if missing
 
    :param overwrite_git_hooks: if this is True, overwrite any existing git hooks
 
        that may be encountered (even if user-deployed)
 
    """
 
    from kallithea.model.repo import RepoModel
 
    from kallithea.model.scm import ScmModel
 
    sa = meta.Session()
 
    repo_model = RepoModel()
 
    if user is None:
 
        user = User.get_first_admin()
 
    added = []
 

	
 
    # creation defaults
 
    defs = Setting.get_default_repo_settings(strip_prefix=True)
 
    enable_statistics = defs.get('repo_enable_statistics')
 
    enable_locking = defs.get('repo_enable_locking')
 
    enable_downloads = defs.get('repo_enable_downloads')
 
    private = defs.get('repo_private')
 

	
 
    for name, repo in initial_repo_list.items():
 
        group = map_groups(name)
 
        unicode_name = safe_unicode(name)
 
        db_repo = repo_model.get_by_repo_name(unicode_name)
 
        # found repo that is on filesystem not in Kallithea database
 
        if not db_repo:
 
            log.info('repository %s not found, creating now', name)
 
            added.append(name)
 
            desc = (repo.description
 
                    if repo.description != 'unknown'
 
                    else '%s repository' % name)
 

	
 
            new_repo = repo_model._create_repo(
 
                repo_name=name,
 
                repo_type=repo.alias,
 
                description=desc,
 
                repo_group=getattr(group, 'group_id', None),
 
                owner=user,
 
                enable_locking=enable_locking,
 
                enable_downloads=enable_downloads,
 
                enable_statistics=enable_statistics,
 
                private=private,
 
                state=Repository.STATE_CREATED
 
            )
 
            sa.commit()
 
            # we added that repo just now, and make sure it has githook
 
            # installed, and updated server info
 
            if new_repo.repo_type == 'git':
 
                git_repo = new_repo.scm_instance
 
                ScmModel().install_git_hooks(git_repo)
 
                # update repository server-info
 
                log.debug('Running update server info')
 
                git_repo._update_server_info()
 
            new_repo.update_changeset_cache()
 
        elif install_git_hooks:
 
            if db_repo.repo_type == 'git':
 
                ScmModel().install_git_hooks(db_repo.scm_instance, force_create=overwrite_git_hooks)
 

	
 
    removed = []
 
    # remove from database those repositories that are not in the filesystem
 
    unicode_initial_repo_list = set(safe_unicode(name) for name in initial_repo_list)
 
    for repo in sa.query(Repository).all():
 
        if repo.repo_name not in unicode_initial_repo_list:
 
            if remove_obsolete:
 
                log.debug("Removing non-existing repository found in db `%s`",
 
                          repo.repo_name)
 
                try:
 
                    RepoModel().delete(repo, forks='detach', fs_remove=False)
 
                    sa.commit()
 
                except Exception:
 
                    #don't hold further removals on error
 
                    log.error(traceback.format_exc())
 
                    sa.rollback()
 
            removed.append(repo.repo_name)
 
    return added, removed
 

	
 

	
 
def load_rcextensions(root_path):
 
    import kallithea
 
    from kallithea.config import conf
 

	
 
    path = os.path.join(root_path, 'rcextensions', '__init__.py')
 
    if os.path.isfile(path):
 
        rcext = create_module('rc', path)
 
        EXT = kallithea.EXTENSIONS = rcext
 
        log.debug('Found rcextensions now loading %s...', rcext)
 

	
 
        # Additional mappings that are not present in the pygments lexers
 
        conf.LANGUAGES_EXTENSIONS_MAP.update(getattr(EXT, 'EXTRA_MAPPINGS', {}))
 

	
 
        # OVERRIDE OUR EXTENSIONS FROM RC-EXTENSIONS (if present)
 

	
 
        if getattr(EXT, 'INDEX_EXTENSIONS', []):
 
            log.debug('settings custom INDEX_EXTENSIONS')
 
            conf.INDEX_EXTENSIONS = getattr(EXT, 'INDEX_EXTENSIONS', [])
 

	
 
        # ADDITIONAL MAPPINGS
 
        log.debug('adding extra into INDEX_EXTENSIONS')
 
        conf.INDEX_EXTENSIONS.extend(getattr(EXT, 'EXTRA_INDEX_EXTENSIONS', []))
 

	
 
        # auto check if the module is not missing any data, set to default if is
 
        # this will help autoupdate new feature of rcext module
 
        #from kallithea.config import rcextensions
 
        #for k in dir(rcextensions):
 
        #    if not k.startswith('_') and not hasattr(EXT, k):
 
        #        setattr(EXT, k, getattr(rcextensions, k))
 

	
 

	
 
#==============================================================================
 
# MISC
 
#==============================================================================
 

	
 
def check_git_version():
 
    """
 
    Checks what version of git is installed in system, and issues a warning
 
    if it's too old for Kallithea to work properly.
 
    """
 
    from kallithea import BACKENDS
 
    from kallithea.lib.vcs.backends.git.repository import GitRepository
 
    from kallithea.lib.vcs.conf import settings
 
    from distutils.version import StrictVersion
 

	
 
    if 'git' not in BACKENDS:
 
        return None
 

	
 
    stdout, stderr = GitRepository._run_git_command(['--version'], _bare=True,
 
                                                    _safe=True)
 

	
 
    m = re.search("\d+.\d+.\d+", stdout)
 
    if m:
 
        ver = StrictVersion(m.group(0))
 
    else:
 
        ver = StrictVersion('0.0.0')
 

	
 
    req_ver = StrictVersion('1.7.4')
 

	
 
    log.debug('Git executable: "%s" version %s detected: %s',
 
              settings.GIT_EXECUTABLE_PATH, ver, stdout)
 
    if stderr:
 
        log.warning('Error detecting git version: %r', stderr)
 
    elif ver < req_ver:
 
        log.warning('Kallithea detected git version %s, which is too old '
 
                    'for the system to function properly. '
 
                    'Please upgrade to version %s or later.' % (ver, req_ver))
 
    return ver
 

	
 

	
 
#===============================================================================
 
# CACHE RELATED METHODS
 
#===============================================================================
 

	
 
# set cache regions for beaker so celery can utilise it
 
def setup_cache_regions(settings):
 
    # Create dict with just beaker cache configs with prefix stripped
 
    cache_settings = {'regions': None}
 
    prefix = 'beaker.cache.'
 
    for key in settings:
 
        if key.startswith(prefix):
 
            name = key[len(prefix):]
 
            cache_settings[name] = settings[key]
 
    # Find all regions, apply defaults, and apply to beaker
 
    if cache_settings['regions']:
 
        for region in cache_settings['regions'].split(','):
 
            region = region.strip()
 
            prefix = region + '.'
 
            region_settings = {}
 
            for key in cache_settings:
 
                if key.startswith(prefix):
 
                    name = key[len(prefix):]
 
                    region_settings[name] = cache_settings[key]
 
            region_settings.setdefault('expire',
 
                                       cache_settings.get('expire', '60'))
 
            region_settings.setdefault('lock_dir',
 
                                       cache_settings.get('lock_dir'))
 
            region_settings.setdefault('data_dir',
 
                                       cache_settings.get('data_dir'))
 
            region_settings.setdefault('type',
 
                                       cache_settings.get('type', 'memory'))
 
            beaker.cache.cache_regions[region] = region_settings
 

	
 

	
 
def conditional_cache(region, prefix, condition, func):
 
    """
 

	
 
    Conditional caching function use like::
 
        def _c(arg):
 
            #heavy computation function
 
            return data
 

	
 
        # depending from condition the compute is wrapped in cache or not
 
        compute = conditional_cache('short_term', 'cache_desc', condition=True, func=func)
 
        return compute(arg)
 

	
 
    :param region: name of cache region
 
    :param prefix: cache region prefix
 
    :param condition: condition for cache to be triggered, and return data cached
 
    :param func: wrapped heavy function to compute
 

	
 
    """
 
    wrapped = func
 
    if condition:
 
        log.debug('conditional_cache: True, wrapping call of '
 
                  'func: %s into %s region cache' % (region, func))
 
        wrapped = _cache_decorate((prefix,), None, None, region)(func)
 

	
 
    return wrapped
kallithea/lib/utils2.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
# This program is free software: you can redistribute it and/or modify
 
# it under the terms of the GNU General Public License as published by
 
# the Free Software Foundation, either version 3 of the License, or
 
# (at your option) any later version.
 
#
 
# This program is distributed in the hope that it will be useful,
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
# GNU General Public License for more details.
 
#
 
# You should have received a copy of the GNU General Public License
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
"""
 
kallithea.lib.utils2
 
~~~~~~~~~~~~~~~~~~~~
 

	
 
Some simple helper functions
 

	
 
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: Jan 5, 2011
 
:author: marcink
 
:copyright: (c) 2013 RhodeCode GmbH, and others.
 
:license: GPLv3, see LICENSE.md for more details.
 
"""
 

	
 

	
 
import os
 
import re
 
import sys
 
import time
 
import uuid
 
import datetime
 
import urllib
 
import binascii
 

	
 
import webob
 
import urlobject
 

	
 
from tg.i18n import ugettext as _, ungettext
 
from kallithea.lib.vcs.utils.lazy import LazyProperty
 
from kallithea.lib.compat import json
 

	
 

	
 
def str2bool(_str):
 
    """
 
    returns True/False value from given string, it tries to translate the
 
    string into boolean
 

	
 
    :param _str: string value to translate into boolean
 
    :rtype: boolean
 
    :returns: boolean from given string
 
    """
 
    if _str is None:
 
        return False
 
    if _str in (True, False):
 
        return _str
 
    _str = str(_str).strip().lower()
 
    return _str in ('t', 'true', 'y', 'yes', 'on', '1')
 

	
 

	
 
def aslist(obj, sep=None, strip=True):
 
    """
 
    Returns given string separated by sep as list
 

	
 
    :param obj:
 
    :param sep:
 
    :param strip:
 
    """
 
    if isinstance(obj, (basestring)):
 
        lst = obj.split(sep)
 
        if strip:
 
            lst = [v.strip() for v in lst]
 
        return lst
 
    elif isinstance(obj, (list, tuple)):
 
        return obj
 
    elif obj is None:
 
        return []
 
    else:
 
        return [obj]
 

	
 

	
 
def convert_line_endings(line, mode):
 
    """
 
    Converts a given line  "line end" according to given mode
 

	
 
    Available modes are::
 
        0 - Unix
 
        1 - Mac
 
        2 - DOS
 

	
 
    :param line: given line to convert
 
    :param mode: mode to convert to
 
    :rtype: str
 
    :return: converted line according to mode
 
    """
 
    from string import replace
 

	
 
    if mode == 0:
 
            line = replace(line, '\r\n', '\n')
 
            line = replace(line, '\r', '\n')
 
    elif mode == 1:
 
            line = replace(line, '\r\n', '\r')
 
            line = replace(line, '\n', '\r')
 
    elif mode == 2:
 
            line = re.sub("\r(?!\n)|(?<!\r)\n", "\r\n", line)
 
    return line
 

	
 

	
 
def detect_mode(line, default):
 
    """
 
    Detects line break for given line, if line break couldn't be found
 
    given default value is returned
 

	
 
    :param line: str line
 
    :param default: default
 
    :rtype: int
 
    :return: value of line end on of 0 - Unix, 1 - Mac, 2 - DOS
 
    """
 
    if line.endswith('\r\n'):
 
        return 2
 
    elif line.endswith('\n'):
 
        return 0
 
    elif line.endswith('\r'):
 
        return 1
 
    else:
 
        return default
 

	
 

	
 
def generate_api_key():
 
    """
 
    Generates a random (presumably unique) API key.
 

	
 
    This value is used in URLs and "Bearer" HTTP Authorization headers,
 
    which in practice means it should only contain URL-safe characters
 
    (RFC 3986):
 

	
 
        unreserved = ALPHA / DIGIT / "-" / "." / "_" / "~"
 
    """
 
    # Hexadecimal certainly qualifies as URL-safe.
 
    return binascii.hexlify(os.urandom(20))
 

	
 

	
 
def safe_int(val, default=None):
 
    """
 
    Returns int() of val if val is not convertable to int use default
 
    instead
 

	
 
    :param val:
 
    :param default:
 
    """
 

	
 
    try:
 
        val = int(val)
 
    except (ValueError, TypeError):
 
        val = default
 

	
 
    return val
 

	
 

	
 
def safe_unicode(str_, from_encoding=None):
 
    """
 
    safe unicode function. Does few trick to turn str_ into unicode
 

	
 
    In case of UnicodeDecode error we try to return it with encoding detected
 
    by chardet library if it fails fallback to unicode with errors replaced
 

	
 
    :param str_: string to decode
 
    :rtype: unicode
 
    :returns: unicode object
 
    """
 
    if isinstance(str_, unicode):
 
        return str_
 

	
 
    if not from_encoding:
 
        import kallithea
 
        DEFAULT_ENCODINGS = aslist(kallithea.CONFIG.get('default_encoding',
 
                                                        'utf8'), sep=',')
 
                                                        'utf-8'), sep=',')
 
        from_encoding = DEFAULT_ENCODINGS
 

	
 
    if not isinstance(from_encoding, (list, tuple)):
 
        from_encoding = [from_encoding]
 

	
 
    try:
 
        return unicode(str_)
 
    except UnicodeDecodeError:
 
        pass
 

	
 
    for enc in from_encoding:
 
        try:
 
            return unicode(str_, enc)
 
        except UnicodeDecodeError:
 
            pass
 

	
 
    try:
 
        import chardet
 
        encoding = chardet.detect(str_)['encoding']
 
        if encoding is None:
 
            raise Exception()
 
        return str_.decode(encoding)
 
    except (ImportError, UnicodeDecodeError, Exception):
 
        return unicode(str_, from_encoding[0], 'replace')
 

	
 

	
 
def safe_str(unicode_, to_encoding=None):
 
    """
 
    safe str function. Does few trick to turn unicode_ into string
 

	
 
    In case of UnicodeEncodeError we try to return it with encoding detected
 
    by chardet library if it fails fallback to string with errors replaced
 

	
 
    :param unicode_: unicode to encode
 
    :rtype: str
 
    :returns: str object
 
    """
 

	
 
    # if it's not basestr cast to str
 
    if not isinstance(unicode_, basestring):
 
        return str(unicode_)
 

	
 
    if isinstance(unicode_, str):
 
        return unicode_
 

	
 
    if not to_encoding:
 
        import kallithea
 
        DEFAULT_ENCODINGS = aslist(kallithea.CONFIG.get('default_encoding',
 
                                                        'utf8'), sep=',')
 
                                                        'utf-8'), sep=',')
 
        to_encoding = DEFAULT_ENCODINGS
 

	
 
    if not isinstance(to_encoding, (list, tuple)):
 
        to_encoding = [to_encoding]
 

	
 
    for enc in to_encoding:
 
        try:
 
            return unicode_.encode(enc)
 
        except UnicodeEncodeError:
 
            pass
 

	
 
    try:
 
        import chardet
 
        encoding = chardet.detect(unicode_)['encoding']
 
        if encoding is None:
 
            raise UnicodeEncodeError()
 

	
 
        return unicode_.encode(encoding)
 
    except (ImportError, UnicodeEncodeError):
 
        return unicode_.encode(to_encoding[0], 'replace')
 

	
 

	
 
def remove_suffix(s, suffix):
 
    if s.endswith(suffix):
 
        s = s[:-1 * len(suffix)]
 
    return s
 

	
 

	
 
def remove_prefix(s, prefix):
 
    if s.startswith(prefix):
 
        s = s[len(prefix):]
 
    return s
 

	
 

	
 
def age(prevdate, show_short_version=False, now=None):
 
    """
 
    turns a datetime into an age string.
 
    If show_short_version is True, then it will generate a not so accurate but shorter string,
 
    example: 2days ago, instead of 2 days and 23 hours ago.
 

	
 
    :param prevdate: datetime object
 
    :param show_short_version: if it should approximate the date and return a shorter string
 
    :rtype: unicode
 
    :returns: unicode words describing age
 
    """
 
    now = now or datetime.datetime.now()
 
    order = ['year', 'month', 'day', 'hour', 'minute', 'second']
 
    deltas = {}
 
    future = False
 

	
 
    if prevdate > now:
 
        now, prevdate = prevdate, now
 
        future = True
 
    if future:
 
        prevdate = prevdate.replace(microsecond=0)
 
    # Get date parts deltas
 
    from dateutil import relativedelta
 
    for part in order:
 
        d = relativedelta.relativedelta(now, prevdate)
 
        deltas[part] = getattr(d, part + 's')
 

	
 
    # Fix negative offsets (there is 1 second between 10:59:59 and 11:00:00,
 
    # not 1 hour, -59 minutes and -59 seconds)
 
    for num, length in [(5, 60), (4, 60), (3, 24)]:  # seconds, minutes, hours
 
        part = order[num]
 
        carry_part = order[num - 1]
 

	
 
        if deltas[part] < 0:
 
            deltas[part] += length
 
            deltas[carry_part] -= 1
 

	
 
    # Same thing for days except that the increment depends on the (variable)
 
    # number of days in the month
 
    month_lengths = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
 
    if deltas['day'] < 0:
 
        if prevdate.month == 2 and (prevdate.year % 4 == 0 and
 
            (prevdate.year % 100 != 0 or prevdate.year % 400 == 0)):
 
            deltas['day'] += 29
 
        else:
 
            deltas['day'] += month_lengths[prevdate.month - 1]
 

	
 
        deltas['month'] -= 1
 

	
 
    if deltas['month'] < 0:
 
        deltas['month'] += 12
 
        deltas['year'] -= 1
 

	
 
    # In short version, we want nicer handling of ages of more than a year
 
    if show_short_version:
 
        if deltas['year'] == 1:
 
            # ages between 1 and 2 years: show as months
 
            deltas['month'] += 12
 
            deltas['year'] = 0
 
        if deltas['year'] >= 2:
 
            # ages 2+ years: round
 
            if deltas['month'] > 6:
 
                deltas['year'] += 1
 
                deltas['month'] = 0
 

	
 
    # Format the result
 
    fmt_funcs = {
 
        'year': lambda d: ungettext(u'%d year', '%d years', d) % d,
 
        'month': lambda d: ungettext(u'%d month', '%d months', d) % d,
 
        'day': lambda d: ungettext(u'%d day', '%d days', d) % d,
 
        'hour': lambda d: ungettext(u'%d hour', '%d hours', d) % d,
 
        'minute': lambda d: ungettext(u'%d minute', '%d minutes', d) % d,
 
        'second': lambda d: ungettext(u'%d second', '%d seconds', d) % d,
 
    }
 

	
 
    for i, part in enumerate(order):
 
        value = deltas[part]
 
        if value == 0:
 
            continue
 

	
 
        if i < 5:
 
            sub_part = order[i + 1]
 
            sub_value = deltas[sub_part]
 
        else:
 
            sub_value = 0
 

	
 
        if sub_value == 0 or show_short_version:
 
            if future:
 
                return _('in %s') % fmt_funcs[part](value)
 
            else:
 
                return _('%s ago') % fmt_funcs[part](value)
 
        if future:
 
            return _('in %s and %s') % (fmt_funcs[part](value),
 
                fmt_funcs[sub_part](sub_value))
 
        else:
 
            return _('%s and %s ago') % (fmt_funcs[part](value),
 
                fmt_funcs[sub_part](sub_value))
 

	
 
    return _('just now')
 

	
 

	
 
def uri_filter(uri):
 
    """
 
    Removes user:password from given url string
 

	
 
    :param uri:
 
    :rtype: unicode
 
    :returns: filtered list of strings
 
    """
 
    if not uri:
 
        return ''
 

	
 
    proto = ''
 

	
 
    for pat in ('https://', 'http://', 'git://'):
 
        if uri.startswith(pat):
 
            uri = uri[len(pat):]
 
            proto = pat
 
            break
 

	
 
    # remove passwords and username
 
    uri = uri[uri.find('@') + 1:]
 

	
 
    # get the port
 
    cred_pos = uri.find(':')
 
    if cred_pos == -1:
 
        host, port = uri, None
 
    else:
 
        host, port = uri[:cred_pos], uri[cred_pos + 1:]
 

	
 
    return filter(None, [proto, host, port])
 

	
 

	
 
def credentials_filter(uri):
 
    """
 
    Returns a url with removed credentials
 

	
 
    :param uri:
 
    """
 

	
 
    uri = uri_filter(uri)
 
    # check if we have port
 
    if len(uri) > 2 and uri[2]:
 
        uri[2] = ':' + uri[2]
 

	
 
    return ''.join(uri)
 

	
 

	
 
def get_clone_url(uri_tmpl, qualified_home_url, repo_name, repo_id, **override):
 
    parsed_url = urlobject.URLObject(qualified_home_url)
 
    decoded_path = safe_unicode(urllib.unquote(parsed_url.path.rstrip('/')))
 
    args = {
 
        'scheme': parsed_url.scheme,
 
        'user': '',
 
        'netloc': parsed_url.netloc+decoded_path,  # path if we use proxy-prefix
 
        'prefix': decoded_path,
 
        'repo': repo_name,
 
        'repoid': str(repo_id)
 
    }
 
    args.update(override)
 
    args['user'] = urllib.quote(safe_str(args['user']))
 

	
 
    for k, v in args.items():
 
        uri_tmpl = uri_tmpl.replace('{%s}' % k, v)
 

	
 
    # remove leading @ sign if it's present. Case of empty user
 
    url_obj = urlobject.URLObject(uri_tmpl)
 
    url = url_obj.with_netloc(url_obj.netloc.lstrip('@'))
 

	
 
    return safe_unicode(url)
 

	
 

	
 
def get_changeset_safe(repo, rev):
 
    """
 
    Safe version of get_changeset if this changeset doesn't exists for a
 
    repo it returns a Dummy one instead
 

	
 
    :param repo:
 
    :param rev:
 
    """
 
    from kallithea.lib.vcs.backends.base import BaseRepository
 
    from kallithea.lib.vcs.exceptions import RepositoryError
 
    from kallithea.lib.vcs.backends.base import EmptyChangeset
 
    if not isinstance(repo, BaseRepository):
 
        raise Exception('You must pass an Repository '
 
                        'object as first argument got %s', type(repo))
 

	
 
    try:
 
        cs = repo.get_changeset(rev)
 
    except (RepositoryError, LookupError):
 
        cs = EmptyChangeset(requested_revision=rev)
 
    return cs
 

	
 

	
 
def datetime_to_time(dt):
 
    if dt:
 
        return time.mktime(dt.timetuple())
 

	
 

	
 
def time_to_datetime(tm):
 
    if tm:
 
        if isinstance(tm, basestring):
 
            try:
 
                tm = float(tm)
 
            except ValueError:
 
                return
 
        return datetime.datetime.fromtimestamp(tm)
 

	
 

	
 
# 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', 'hh', 'zz']
 
    """
 
    return MENTIONS_REGEX.findall(text)
 

	
 

	
 
def extract_mentioned_users(text):
 
    """ Returns set of actual database Users @mentioned in given text. """
 
    from kallithea.model.db import User
 
    result = set()
 
    for name in extract_mentioned_usernames(text):
 
        user = User.get_by_username(name, case_insensitive=True)
 
        if user is not None and not user.is_default_user:
 
            result.add(user)
 
    return result
 

	
 

	
 
class AttributeDict(dict):
 
    def __getattr__(self, attr):
 
        return self.get(attr, None)
 
    __setattr__ = dict.__setitem__
 
    __delattr__ = dict.__delitem__
 

	
 

	
 
def fix_PATH(os_=None):
 
    """
 
    Get current active python path, and append it to PATH variable to fix issues
 
    of subprocess calls and different python versions
 
    """
 
    if os_ is None:
 
        import os
 
    else:
 
        os = os_
 

	
 
    cur_path = os.path.split(sys.executable)[0]
 
    if not os.environ['PATH'].startswith(cur_path):
 
        os.environ['PATH'] = '%s:%s' % (cur_path, os.environ['PATH'])
 

	
 

	
 
def obfuscate_url_pw(engine):
 
    from sqlalchemy.engine import url as sa_url
 
    from sqlalchemy.exc import ArgumentError
 
    try:
 
        _url = sa_url.make_url(engine or '')
 
    except ArgumentError:
 
        return engine
 
    if _url.password:
 
        _url.password = 'XXXXX'
 
    return str(_url)
 

	
 

	
 
def get_server_url(environ):
 
    req = webob.Request(environ)
 
    return req.host_url + req.script_name
 

	
 

	
 
def _extract_extras(env=None):
 
    """
 
    Extracts the Kallithea extras data from os.environ, and wraps it into named
 
    AttributeDict object
 
    """
 
    if not env:
 
        env = os.environ
 

	
 
    try:
 
        extras = json.loads(env['KALLITHEA_EXTRAS'])
 
    except KeyError:
 
        extras = {}
 

	
 
    try:
 
        for k in ['username', 'repository', 'locked_by', 'scm', 'make_lock',
 
                  'action', 'ip']:
 
            extras[k]
 
    except KeyError as e:
 
        raise Exception('Missing key %s in os.environ %s' % (e, extras))
 

	
 
    return AttributeDict(extras)
 

	
 

	
 
def _set_extras(extras):
 
    # RC_SCM_DATA can probably be removed in the future, but for compatibility now...
 
    os.environ['KALLITHEA_EXTRAS'] = os.environ['RC_SCM_DATA'] = json.dumps(extras)
 

	
 

	
 
def get_current_authuser():
 
    """
 
    Gets kallithea user from threadlocal tmpl_context variable if it's
 
    defined, else returns None.
 
    """
 
    from tg import tmpl_context
 
    if hasattr(tmpl_context, 'authuser'):
 
        return tmpl_context.authuser
 

	
 
    return None
 

	
 

	
 
class OptionalAttr(object):
 
    """
 
    Special Optional Option that defines other attribute. Example::
 

	
 
        def test(apiuser, userid=Optional(OAttr('apiuser')):
 
            user = Optional.extract(userid)
 
            # calls
 

	
 
    """
 

	
 
    def __init__(self, attr_name):
 
        self.attr_name = attr_name
 

	
 
    def __repr__(self):
 
        return '<OptionalAttr:%s>' % self.attr_name
 

	
 
    def __call__(self):
 
        return self
 

	
 

	
 
# alias
 
OAttr = OptionalAttr
 

	
 

	
 
class Optional(object):
 
    """
 
    Defines an optional parameter::
 

	
 
        param = param.getval() if isinstance(param, Optional) else param
 
        param = param() if isinstance(param, Optional) else param
 

	
 
    is equivalent of::
 

	
 
        param = Optional.extract(param)
 

	
kallithea/lib/vcs/backends/hg/inmemory.py
Show inline comments
 
import datetime
 
import errno
 

	
 
from kallithea.lib.vcs.backends.base import BaseInMemoryChangeset
 
from kallithea.lib.vcs.exceptions import RepositoryError
 

	
 
from kallithea.lib.vcs.utils.hgcompat import memfilectx, memctx, hex, tolocal
 

	
 

	
 
class MercurialInMemoryChangeset(BaseInMemoryChangeset):
 

	
 
    def commit(self, message, author, parents=None, branch=None, date=None,
 
            **kwargs):
 
        """
 
        Performs in-memory commit (doesn't check workdir in any way) and
 
        returns newly created ``Changeset``. Updates repository's
 
        ``revisions``.
 

	
 
        :param message: message of the commit
 
        :param author: full username, i.e. "Joe Doe <joe.doe@example.com>"
 
        :param parents: single parent or sequence of parents from which commit
 
          would be derived
 
        :param date: ``datetime.datetime`` instance. Defaults to
 
          ``datetime.datetime.now()``.
 
        :param branch: branch name, as string. If none given, default backend's
 
          branch would be used.
 

	
 
        :raises ``CommitError``: if any error occurs while committing
 
        """
 
        self.check_integrity(parents)
 

	
 
        from .repository import MercurialRepository
 
        if not isinstance(message, unicode) or not isinstance(author, unicode):
 
            raise RepositoryError('Given message and author needs to be '
 
                                  'an <unicode> instance got %r & %r instead'
 
                                  % (type(message), type(author)))
 

	
 
        if branch is None:
 
            branch = MercurialRepository.DEFAULT_BRANCH_NAME
 
        kwargs['branch'] = branch
 

	
 
        def filectxfn(_repo, memctx, path):
 
            """
 
            Marks given path as added/changed/removed in a given _repo. This is
 
            for internal mercurial commit function.
 
            """
 

	
 
            # check if this path is removed
 
            if path in (node.path for node in self.removed):
 
                return None
 

	
 
            # check if this path is added
 
            for node in self.added:
 
                if node.path == path:
 
                    return memfilectx(_repo, memctx, path=node.path,
 
                        data=(node.content.encode('utf8')
 
                        data=(node.content.encode('utf-8')
 
                              if not node.is_binary else node.content),
 
                        islink=False,
 
                        isexec=node.is_executable,
 
                        copied=False)
 

	
 
            # or changed
 
            for node in self.changed:
 
                if node.path == path:
 
                    return memfilectx(_repo, memctx, path=node.path,
 
                        data=(node.content.encode('utf8')
 
                        data=(node.content.encode('utf-8')
 
                              if not node.is_binary else node.content),
 
                        islink=False,
 
                        isexec=node.is_executable,
 
                        copied=False)
 

	
 
            raise RepositoryError("Given path haven't been marked as added,"
 
                                  "changed or removed (%s)" % path)
 

	
 
        parents = [None, None]
 
        for i, parent in enumerate(self.parents):
 
            if parent is not None:
 
                parents[i] = parent._ctx.node()
 

	
 
        if date and isinstance(date, datetime.datetime):
 
            date = date.strftime('%a, %d %b %Y %H:%M:%S')
 

	
 
        commit_ctx = memctx(repo=self.repository._repo,
 
            parents=parents,
 
            text='',
 
            files=self.get_paths(),
 
            filectxfn=filectxfn,
 
            user=author,
 
            date=date,
 
            extra=kwargs)
 

	
 
        loc = lambda u: tolocal(u.encode('utf-8'))
 

	
 
        # injecting given _repo params
 
        commit_ctx._text = loc(message)
 
        commit_ctx._user = loc(author)
 
        commit_ctx._date = date
 

	
 
        # TODO: Catch exceptions!
 
        n = self.repository._repo.commitctx(commit_ctx)
 
        # Returns mercurial node
 
        self._commit_ctx = commit_ctx  # For reference
 
        # Update vcs repository object & recreate mercurial _repo
 
        # new_ctx = self.repository._repo[node]
 
        # new_tip = self.repository.get_changeset(new_ctx.hex())
 
        new_id = hex(n)
 
        self.repository.revisions.append(new_id)
 
        self._repo = self.repository._get_repo(create=False)
 
        self.repository.branches = self.repository._get_branches()
 
        tip = self.repository.get_changeset()
 
        self.reset()
 
        return tip
kallithea/lib/vcs/conf/settings.py
Show inline comments
 
import os
 
import tempfile
 
from kallithea.lib.vcs.utils import aslist
 
from kallithea.lib.vcs.utils.paths import get_user_home
 

	
 
abspath = lambda * p: os.path.abspath(os.path.join(*p))
 

	
 
VCSRC_PATH = os.environ.get('VCSRC_PATH')
 

	
 
if not VCSRC_PATH:
 
    HOME_ = get_user_home()
 
    if not HOME_:
 
        HOME_ = tempfile.gettempdir()
 

	
 
VCSRC_PATH = VCSRC_PATH or abspath(HOME_, '.vcsrc')
 
if os.path.isdir(VCSRC_PATH):
 
    VCSRC_PATH = os.path.join(VCSRC_PATH, '__init__.py')
 

	
 
# list of default encoding used in safe_unicode/safe_str methods
 
DEFAULT_ENCODINGS = aslist('utf8')
 
DEFAULT_ENCODINGS = aslist('utf-8')
 

	
 
# path to git executable run by run_git_command function
 
GIT_EXECUTABLE_PATH = 'git'
 
# can be also --branches --tags
 
GIT_REV_FILTER = '--all'
 

	
 
BACKENDS = {
 
    'hg': 'kallithea.lib.vcs.backends.hg.MercurialRepository',
 
    'git': 'kallithea.lib.vcs.backends.git.GitRepository',
 
}
 

	
 
ARCHIVE_SPECS = {
 
    'tar': ('application/x-tar', '.tar'),
 
    'tbz2': ('application/x-bzip2', '.tar.bz2'),
 
    'tgz': ('application/x-gzip', '.tar.gz'),
 
    'zip': ('application/zip', '.zip'),
 
}
kallithea/tests/vcs/test_changesets.py
Show inline comments
 
# encoding: utf8
 
# encoding: utf-8
 

	
 
import time
 
import datetime
 

	
 
import pytest
 

	
 
from kallithea.lib import vcs
 

	
 
from kallithea.lib.vcs.backends.base import BaseChangeset
 
from kallithea.lib.vcs.nodes import (
 
    FileNode, AddedFileNodesGenerator,
 
    ChangedFileNodesGenerator, RemovedFileNodesGenerator
 
)
 
from kallithea.lib.vcs.exceptions import (
 
    BranchDoesNotExistError, ChangesetDoesNotExistError,
 
    RepositoryError, EmptyRepositoryError
 
)
 

	
 
from kallithea.tests.vcs.base import _BackendTestMixin
 
from kallithea.tests.vcs.conf import get_new_dir
 

	
 

	
 
class TestBaseChangeset(object):
 

	
 
    def test_as_dict(self):
 
        changeset = BaseChangeset()
 
        changeset.id = 'ID'
 
        changeset.raw_id = 'RAW_ID'
 
        changeset.short_id = 'SHORT_ID'
 
        changeset.revision = 1009
 
        changeset.date = datetime.datetime(2011, 1, 30, 1, 45)
 
        changeset.message = 'Message of a commit'
 
        changeset.author = 'Joe Doe <joe.doe@example.com>'
 
        changeset.added = [FileNode('foo/bar/baz'), FileNode(u'foobar'), FileNode(u'blåbærgrød')]
 
        changeset.changed = []
 
        changeset.removed = []
 
        assert changeset.as_dict() == {
 
            'id': 'ID',
 
            'raw_id': 'RAW_ID',
 
            'short_id': 'SHORT_ID',
 
            'revision': 1009,
 
            'date': datetime.datetime(2011, 1, 30, 1, 45),
 
            'message': 'Message of a commit',
 
            'author': {
 
                'name': 'Joe Doe',
 
                'email': 'joe.doe@example.com',
 
            },
 
            'added': ['foo/bar/baz', 'foobar', u'bl\xe5b\xe6rgr\xf8d'],
 
            'changed': [],
 
            'removed': [],
 
        }
 

	
 

	
 
class _ChangesetsWithCommitsTestCaseixin(_BackendTestMixin):
 

	
 
    @classmethod
 
    def _get_commits(cls):
 
        start_date = datetime.datetime(2010, 1, 1, 20)
 
        for x in xrange(5):
 
            yield {
 
                'message': 'Commit %d' % x,
 
                'author': 'Joe Doe <joe.doe@example.com>',
 
                'date': start_date + datetime.timedelta(hours=12 * x),
 
                'added': [
 
                    FileNode('file_%d.txt' % x, content='Foobar %d' % x),
 
                ],
 
            }
 

	
 
    def test_new_branch(self):
 
        self.imc.add(vcs.nodes.FileNode('docs/index.txt',
 
            content='Documentation\n'))
 
        foobar_tip = self.imc.commit(
 
            message=u'New branch: foobar',
 
            author=u'joe',
 
            branch='foobar',
 
        )
 
        assert 'foobar' in self.repo.branches
 
        assert foobar_tip.branch == 'foobar'
 
        assert foobar_tip.branches == ['foobar']
 
        # 'foobar' should be the only branch that contains the new commit
 
        branch_tips = self.repo.branches.values()
 
        assert branch_tips.count(str(foobar_tip.raw_id)) == 1
 

	
 
    def test_new_head_in_default_branch(self):
 
        tip = self.repo.get_changeset()
 
        self.imc.add(vcs.nodes.FileNode('docs/index.txt',
 
            content='Documentation\n'))
 
        foobar_tip = self.imc.commit(
 
            message=u'New branch: foobar',
 
            author=u'joe',
 
            branch='foobar',
 
            parents=[tip],
 
        )
 
        self.imc.change(vcs.nodes.FileNode('docs/index.txt',
 
            content='Documentation\nand more...\n'))
 
        newtip = self.imc.commit(
 
            message=u'At default branch',
 
            author=u'joe',
 
            branch=foobar_tip.branch,
 
            parents=[foobar_tip],
 
        )
 

	
 
        newest_tip = self.imc.commit(
 
            message=u'Merged with %s' % foobar_tip.raw_id,
 
            author=u'joe',
 
            branch=self.backend_class.DEFAULT_BRANCH_NAME,
 
            parents=[newtip, foobar_tip],
 
        )
 

	
 
        assert newest_tip.branch == self.backend_class.DEFAULT_BRANCH_NAME
 
        assert newest_tip.branches == [self.backend_class.DEFAULT_BRANCH_NAME]
 

	
 
    def test_get_changesets_respects_branch_name(self):
 
        tip = self.repo.get_changeset()
 
        self.imc.add(vcs.nodes.FileNode('docs/index.txt',
 
            content='Documentation\n'))
 
        doc_changeset = self.imc.commit(
 
            message=u'New branch: docs',
 
            author=u'joe',
 
            branch='docs',
 
        )
 
        self.imc.add(vcs.nodes.FileNode('newfile', content=''))
 
        self.imc.commit(
 
            message=u'Back in default branch',
 
            author=u'joe',
 
            parents=[tip],
 
        )
 
        default_branch_changesets = self.repo.get_changesets(
 
            branch_name=self.repo.DEFAULT_BRANCH_NAME)
 
        assert doc_changeset not in default_branch_changesets
 

	
 
    def test_get_changeset_by_branch(self):
 
        for branch, sha in self.repo.branches.iteritems():
 
            assert sha == self.repo.get_changeset(branch).raw_id
 

	
 
    def test_get_changeset_by_tag(self):
 
        for tag, sha in self.repo.tags.iteritems():
 
            assert sha == self.repo.get_changeset(tag).raw_id
 

	
 
    def test_get_changeset_parents(self):
 
        for test_rev in [1, 2, 3]:
 
            sha = self.repo.get_changeset(test_rev-1)
 
            assert [sha] == self.repo.get_changeset(test_rev).parents
 

	
 
    def test_get_changeset_children(self):
 
        for test_rev in [1, 2, 3]:
 
            sha = self.repo.get_changeset(test_rev+1)
 
            assert [sha] == self.repo.get_changeset(test_rev).children
 

	
 

	
 
class _ChangesetsTestCaseMixin(_BackendTestMixin):
 
    recreate_repo_per_test = False
 

	
 
    @classmethod
 
    def _get_commits(cls):
 
        start_date = datetime.datetime(2010, 1, 1, 20)
 
        for x in xrange(5):
 
            yield {
 
                'message': u'Commit %d' % x,
 
                'author': u'Joe Doe <joe.doe@example.com>',
 
                'date': start_date + datetime.timedelta(hours=12 * x),
 
                'added': [
 
                    FileNode('file_%d.txt' % x, content='Foobar %d' % x),
 
                ],
 
            }
 

	
 
    def test_simple(self):
 
        tip = self.repo.get_changeset()
 
        assert tip.date == datetime.datetime(2010, 1, 3, 20)
 

	
 
    def test_get_changesets_is_ordered_by_date(self):
 
        changesets = list(self.repo.get_changesets())
 
        ordered_by_date = sorted(changesets,
 
            key=lambda cs: cs.date)
 

	
 
        assert changesets == ordered_by_date
 

	
 
    def test_get_changesets_respects_start(self):
 
        second_id = self.repo.revisions[1]
 
        changesets = list(self.repo.get_changesets(start=second_id))
 
        assert len(changesets) == 4
 

	
 
    def test_get_changesets_numerical_id_respects_start(self):
 
        second_id = 1
 
        changesets = list(self.repo.get_changesets(start=second_id))
 
        assert len(changesets) == 4
 

	
 
    def test_get_changesets_includes_start_changeset(self):
 
        second_id = self.repo.revisions[1]
 
        changesets = list(self.repo.get_changesets(start=second_id))
 
        assert changesets[0].raw_id == second_id
 

	
 
    def test_get_changesets_respects_end(self):
 
        second_id = self.repo.revisions[1]
 
        changesets = list(self.repo.get_changesets(end=second_id))
 
        assert changesets[-1].raw_id == second_id
 
        assert len(changesets) == 2
 

	
 
    def test_get_changesets_numerical_id_respects_end(self):
 
        second_id = 1
 
        changesets = list(self.repo.get_changesets(end=second_id))
 
        assert changesets.index(changesets[-1]) == second_id
 
        assert len(changesets) == 2
 

	
 
    def test_get_changesets_respects_both_start_and_end(self):
 
        second_id = self.repo.revisions[1]
 
        third_id = self.repo.revisions[2]
 
        changesets = list(self.repo.get_changesets(start=second_id,
 
            end=third_id))
 
        assert len(changesets) == 2
 

	
 
    def test_get_changesets_numerical_id_respects_both_start_and_end(self):
 
        changesets = list(self.repo.get_changesets(start=2, end=3))
 
        assert len(changesets) == 2
 

	
 
    def test_get_changesets_on_empty_repo_raises_EmptyRepository_error(self):
 
        repo = self.setup_empty_repo(self.backend_class)
 
        with pytest.raises(EmptyRepositoryError):
 
            list(repo.get_changesets(start='foobar'))
 

	
 
    def test_get_changesets_includes_end_changeset(self):
 
        second_id = self.repo.revisions[1]
 
        changesets = list(self.repo.get_changesets(end=second_id))
 
        assert changesets[-1].raw_id == second_id
 

	
 
    def test_get_changesets_respects_start_date(self):
 
        start_date = datetime.datetime(2010, 2, 1)
 
        for cs in self.repo.get_changesets(start_date=start_date):
 
            assert cs.date >= start_date
 

	
 
    def test_get_changesets_respects_end_date(self):
 
        start_date = datetime.datetime(2010, 1, 1)
 
        end_date = datetime.datetime(2010, 2, 1)
 
        for cs in self.repo.get_changesets(start_date=start_date,
 
                                           end_date=end_date):
 
            assert cs.date >= start_date
 
            assert cs.date <= end_date
 

	
 
    def test_get_changesets_respects_start_date_and_end_date(self):
 
        end_date = datetime.datetime(2010, 2, 1)
 
        for cs in self.repo.get_changesets(end_date=end_date):
 
            assert cs.date <= end_date
 

	
 
    def test_get_changesets_respects_reverse(self):
 
        changesets_id_list = [cs.raw_id for cs in
 
            self.repo.get_changesets(reverse=True)]
 
        assert changesets_id_list == list(reversed(self.repo.revisions))
 

	
 
    def test_get_filenodes_generator(self):
 
        tip = self.repo.get_changeset()
 
        filepaths = [node.path for node in tip.get_filenodes_generator()]
 
        assert filepaths == ['file_%d.txt' % x for x in xrange(5)]
 

	
 
    def test_size(self):
 
        tip = self.repo.get_changeset()
 
        size = 5 * len('Foobar N') # Size of 5 files
 
        assert tip.size == size
 

	
 
    def test_author(self):
 
        tip = self.repo.get_changeset()
 
        assert tip.author == u'Joe Doe <joe.doe@example.com>'
 

	
 
    def test_author_name(self):
 
        tip = self.repo.get_changeset()
 
        assert tip.author_name == u'Joe Doe'
 

	
 
    def test_author_email(self):
 
        tip = self.repo.get_changeset()
 
        assert tip.author_email == u'joe.doe@example.com'
 

	
 
    def test_get_changesets_raise_changesetdoesnotexist_for_wrong_start(self):
 
        with pytest.raises(ChangesetDoesNotExistError):
 
            list(self.repo.get_changesets(start='foobar'))
 

	
 
    def test_get_changesets_raise_changesetdoesnotexist_for_wrong_end(self):
 
        with pytest.raises(ChangesetDoesNotExistError):
 
            list(self.repo.get_changesets(end='foobar'))
 

	
 
    def test_get_changesets_raise_branchdoesnotexist_for_wrong_branch_name(self):
 
        with pytest.raises(BranchDoesNotExistError):
 
            list(self.repo.get_changesets(branch_name='foobar'))
 

	
 
    def test_get_changesets_raise_repositoryerror_for_wrong_start_end(self):
 
        start = self.repo.revisions[-1]
 
        end = self.repo.revisions[0]
 
        with pytest.raises(RepositoryError):
 
            list(self.repo.get_changesets(start=start, end=end))
 

	
 
    def test_get_changesets_numerical_id_reversed(self):
 
        with pytest.raises(RepositoryError):
 
            [x for x in self.repo.get_changesets(start=3, end=2)]
 

	
 
    def test_get_changesets_numerical_id_respects_both_start_and_end_last(self):
 
        with pytest.raises(RepositoryError):
 
            last = len(self.repo.revisions)
 
            list(self.repo.get_changesets(start=last-1, end=last-2))
 

	
 
    def test_get_changesets_numerical_id_last_zero_error(self):
 
        with pytest.raises(RepositoryError):
 
            last = len(self.repo.revisions)
 
            list(self.repo.get_changesets(start=last-1, end=0))
 

	
 

	
 
class _ChangesetsChangesTestCaseMixin(_BackendTestMixin):
 
    recreate_repo_per_test = False
 

	
 
    @classmethod
 
    def _get_commits(cls):
 
        return [
 
            {
 
                'message': u'Initial',
 
                'author': u'Joe Doe <joe.doe@example.com>',
 
                'date': datetime.datetime(2010, 1, 1, 20),
 
                'added': [
 
                    FileNode('foo/bar', content='foo'),
 
                    FileNode('foo/bał', content='foo'),
 
                    FileNode('foobar', content='foo'),
 
                    FileNode('qwe', content='foo'),
 
                ],
 
            },
 
            {
 
                'message': u'Massive changes',
 
                'author': u'Joe Doe <joe.doe@example.com>',
 
                'date': datetime.datetime(2010, 1, 1, 22),
 
                'added': [FileNode('fallout', content='War never changes')],
 
                'changed': [
 
                    FileNode('foo/bar', content='baz'),
 
                    FileNode('foobar', content='baz'),
 
                ],
 
                'removed': [FileNode('qwe')],
 
            },
 
        ]
 

	
 
    def test_initial_commit(self):
 
        changeset = self.repo.get_changeset(0)
 
        assert sorted(list(changeset.added)) == sorted([
 
            changeset.get_node('foo/bar'),
 
            changeset.get_node('foo/bał'),
 
            changeset.get_node('foobar'),
 
            changeset.get_node('qwe'),
 
        ])
 
        assert list(changeset.changed) == []
 
        assert list(changeset.removed) == []
 
        assert u'foo/ba\u0142' in changeset.as_dict()['added']
 
        assert u'foo/ba\u0142' in changeset.__json__(with_file_list=True)['added']
 

	
 
    def test_head_added(self):
 
        changeset = self.repo.get_changeset()
 
        assert isinstance(changeset.added, AddedFileNodesGenerator)
 
        assert list(changeset.added) == [
 
            changeset.get_node('fallout'),
 
        ]
 
        assert isinstance(changeset.changed, ChangedFileNodesGenerator)
 
        assert list(changeset.changed) == [
 
            changeset.get_node('foo/bar'),
 
            changeset.get_node('foobar'),
 
        ]
 
        assert isinstance(changeset.removed, RemovedFileNodesGenerator)
 
        assert len(changeset.removed) == 1
 
        assert list(changeset.removed)[0].path == 'qwe'
 

	
 
    def test_get_filemode(self):
 
        changeset = self.repo.get_changeset()
 
        assert 33188 == changeset.get_file_mode('foo/bar')
 

	
 
    def test_get_filemode_non_ascii(self):
 
        changeset = self.repo.get_changeset()
 
        assert 33188 == changeset.get_file_mode('foo/bał')
 
        assert 33188 == changeset.get_file_mode(u'foo/bał')
 

	
 

	
 
class TestGitChangesetsWithCommits(_ChangesetsWithCommitsTestCaseixin):
 
    backend_alias = 'git'
 

	
 

	
 
class TestGitChangesets(_ChangesetsTestCaseMixin):
 
    backend_alias = 'git'
 

	
 

	
 
class TestGitChangesetsChanges(_ChangesetsChangesTestCaseMixin):
 
    backend_alias = 'git'
 

	
 

	
 
class TestHgChangesetsWithCommits(_ChangesetsWithCommitsTestCaseixin):
kallithea/tests/vcs/test_filenodes_unicode_path.py
Show inline comments
 
# encoding: utf8
 
# encoding: utf-8
 

	
 
import datetime
 

	
 
from kallithea.lib.vcs.nodes import FileNode
 
from kallithea.tests.vcs.base import _BackendTestMixin
 

	
 

	
 
class FileNodeUnicodePathTestsMixin(_BackendTestMixin):
 

	
 
    fname = 'ąśðąęłąć.txt'
 
    ufname = (fname).decode('utf-8')
 

	
 
    @classmethod
 
    def _get_commits(cls):
 
        cls.nodes = [
 
            FileNode(cls.fname, content='Foobar'),
 
        ]
 

	
 
        commits = [
 
            {
 
                'message': 'Initial commit',
 
                'author': 'Joe Doe <joe.doe@example.com>',
 
                'date': datetime.datetime(2010, 1, 1, 20),
 
                'added': cls.nodes,
 
            },
 
        ]
 
        return commits
 

	
 
    def test_filenode_path(self):
 
        node = self.tip.get_node(self.fname)
 
        unode = self.tip.get_node(self.ufname)
 
        assert node == unode
 

	
 

	
 
class TestGitFileNodeUnicodePath(FileNodeUnicodePathTestsMixin):
 
    backend_alias = 'git'
 

	
 

	
 
class TestHgFileNodeUnicodePath(FileNodeUnicodePathTestsMixin):
 
    backend_alias = 'hg'
kallithea/tests/vcs/test_inmemchangesets.py
Show inline comments
 
# encoding: utf8
 
# encoding: utf-8
 
"""
 
Tests so called "in memory changesets" commit API of vcs.
 
"""
 

	
 
import time
 
import datetime
 

	
 
import pytest
 

	
 
from kallithea.lib import vcs
 
from kallithea.lib.vcs.exceptions import EmptyRepositoryError
 
from kallithea.lib.vcs.exceptions import NodeAlreadyAddedError
 
from kallithea.lib.vcs.exceptions import NodeAlreadyExistsError
 
from kallithea.lib.vcs.exceptions import NodeAlreadyRemovedError
 
from kallithea.lib.vcs.exceptions import NodeAlreadyChangedError
 
from kallithea.lib.vcs.exceptions import NodeDoesNotExistError
 
from kallithea.lib.vcs.exceptions import NodeNotChangedError
 
from kallithea.lib.vcs.nodes import DirNode
 
from kallithea.lib.vcs.nodes import FileNode
 
from kallithea.lib.vcs.utils import safe_unicode
 

	
 
from kallithea.tests.vcs.base import _BackendTestMixin
 

	
 

	
 
class InMemoryChangesetTestMixin(_BackendTestMixin):
 

	
 
    @classmethod
 
    def _get_commits(cls):
 
        # Note: this is slightly different than the regular _get_commits methods
 
        # as we don't actually return any commits. The creation of commits is
 
        # handled in the tests themselves.
 
        cls.nodes = [
 
            FileNode('foobar', content='Foo & bar'),
 
            FileNode('foobar2', content='Foo & bar, doubled!'),
 
            FileNode('foo bar with spaces', content=''),
 
            FileNode('foo/bar/baz', content='Inside'),
 
            FileNode('foo/bar/file.bin', content='\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00;\x00\x03\x00\xfe\xff\t\x00\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x1a\x00\x00\x00\x00\x00\x00\x00\x00\x10\x00\x00\x18\x00\x00\x00\x01\x00\x00\x00\xfe\xff\xff\xff\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff'),
 
        ]
 
        commits = []
 
        return commits
 

	
 
    def test_add(self):
 
        rev_count = len(self.repo.revisions)
 
        to_add = [FileNode(node.path, content=node.content)
 
            for node in self.nodes]
 
        for node in to_add:
 
            self.imc.add(node)
 
        message = u'Added: %s' % ', '.join((node.path for node in self.nodes))
 
        author = unicode(self.__class__)
 
        changeset = self.imc.commit(message=message, author=author)
 

	
 
        newtip = self.repo.get_changeset()
 
        assert changeset == newtip
 
        assert rev_count + 1 == len(self.repo.revisions)
 
        assert newtip.message == message
 
        assert newtip.author == author
 
        assert not any((
 
            self.imc.added,
 
            self.imc.changed,
 
            self.imc.removed
 
        ))
 
        for node in to_add:
 
            assert newtip.get_node(node.path).content == node.content
 

	
 
    def test_add_in_bulk(self):
 
        rev_count = len(self.repo.revisions)
 
        to_add = [FileNode(node.path, content=node.content)
 
            for node in self.nodes]
 
        self.imc.add(*to_add)
 
        message = u'Added: %s' % ', '.join((node.path for node in self.nodes))
 
        author = unicode(self.__class__)
 
        changeset = self.imc.commit(message=message, author=author)
 

	
 
        newtip = self.repo.get_changeset()
 
        assert changeset == newtip
 
        assert rev_count + 1 == len(self.repo.revisions)
 
        assert newtip.message == message
 
        assert newtip.author == author
 
        assert not any((
 
            self.imc.added,
 
            self.imc.changed,
 
            self.imc.removed
 
        ))
 
        for node in to_add:
 
            assert newtip.get_node(node.path).content == node.content
 

	
 
    def test_add_actually_adds_all_nodes_at_second_commit_too(self):
 
        self.imc.add(FileNode('foo/bar/image.png', content='\0'))
 
        self.imc.add(FileNode('foo/README.txt', content='readme!'))
 
        changeset = self.imc.commit(u'Initial', u'joe.doe@example.com')
 
        assert isinstance(changeset.get_node('foo'), DirNode)
 
        assert isinstance(changeset.get_node('foo/bar'), DirNode)
 
        assert changeset.get_node('foo/bar/image.png').content == '\0'
 
        assert changeset.get_node('foo/README.txt').content == 'readme!'
 

	
 
        # commit some more files again
 
        to_add = [
 
            FileNode('foo/bar/foobaz/bar', content='foo'),
 
            FileNode('foo/bar/another/bar', content='foo'),
 
            FileNode('foo/baz.txt', content='foo'),
 
            FileNode('foobar/foobaz/file', content='foo'),
 
            FileNode('foobar/barbaz', content='foo'),
 
        ]
 
        self.imc.add(*to_add)
 
        changeset = self.imc.commit(u'Another', u'joe.doe@example.com')
 
        changeset.get_node('foo/bar/foobaz/bar').content == 'foo'
 
        changeset.get_node('foo/bar/another/bar').content == 'foo'
 
        changeset.get_node('foo/baz.txt').content == 'foo'
 
        changeset.get_node('foobar/foobaz/file').content == 'foo'
 
        changeset.get_node('foobar/barbaz').content == 'foo'
 

	
 
    def test_add_non_ascii_files(self):
 
        rev_count = len(self.repo.revisions)
 
        to_add = [
 
            FileNode('żółwik/zwierzątko', content='ćććć'),
 
            FileNode(u'żółwik/zwierzątko_uni', content=u'ćććć'),
 
        ]
 
        for node in to_add:
 
            self.imc.add(node)
 
        message = u'Added: %s' % ', '.join((node.path for node in self.nodes))
 
        author = unicode(self.__class__)
 
        changeset = self.imc.commit(message=message, author=author)
 

	
 
        newtip = self.repo.get_changeset()
 
        assert changeset == newtip
 
        assert rev_count + 1 == len(self.repo.revisions)
 
        assert newtip.message == message
 
        assert newtip.author == author
 
        assert not any((
 
            self.imc.added,
 
            self.imc.changed,
 
            self.imc.removed
 
        ))
 
        for node in to_add:
 
            assert newtip.get_node(node.path).content == node.content
 

	
 
    def test_add_raise_already_added(self):
 
        node = FileNode('foobar', content='baz')
 
        self.imc.add(node)
 
        with pytest.raises(NodeAlreadyAddedError):
 
            self.imc.add(node)
 

	
 
    def test_check_integrity_raise_already_exist(self):
 
        node = FileNode('foobar', content='baz')
 
        self.imc.add(node)
 
        self.imc.commit(message=u'Added foobar', author=unicode(self))
 
        self.imc.add(node)
 
        with pytest.raises(NodeAlreadyExistsError):
 
            self.imc.commit(message='new message',
 
                            author=str(self))
 

	
 
    def test_change(self):
 
        self.imc.add(FileNode('foo/bar/baz', content='foo'))
 
        self.imc.add(FileNode('foo/fbar', content='foobar'))
 
        tip = self.imc.commit(u'Initial', u'joe.doe@example.com')
 

	
 
        # Change node's content
 
        node = FileNode('foo/bar/baz', content='My **changed** content')
 
        self.imc.change(node)
 
        self.imc.commit(u'Changed %s' % node.path, u'joe.doe@example.com')
 

	
 
        newtip = self.repo.get_changeset()
 
        assert tip != newtip
 
        assert tip.id != newtip.id
 
        assert newtip.get_node('foo/bar/baz').content == 'My **changed** content'
 

	
 
    def test_change_non_ascii(self):
 
        to_add = [
 
            FileNode('żółwik/zwierzątko', content='ćććć'),
 
            FileNode(u'żółwik/zwierzątko_uni', content=u'ćććć'),
 
        ]
 
        for node in to_add:
 
            self.imc.add(node)
 

	
 
        tip = self.imc.commit(u'Initial', u'joe.doe@example.com')
 

	
 
        # Change node's content
 
        node = FileNode('żółwik/zwierzątko', content='My **changed** content')
 
        self.imc.change(node)
 
        self.imc.commit(u'Changed %s' % safe_unicode(node.path),
 
                        u'joe.doe@example.com')
 

	
 
        node = FileNode(u'żółwik/zwierzątko_uni', content=u'My **changed** content')
 
        self.imc.change(node)
 
        self.imc.commit(u'Changed %s' % safe_unicode(node.path),
 
                        u'joe.doe@example.com')
 

	
 
        newtip = self.repo.get_changeset()
 
        assert tip != newtip
 
        assert tip.id != newtip.id
 

	
 
        assert newtip.get_node('żółwik/zwierzątko').content == 'My **changed** content'
 
        assert newtip.get_node('żółwik/zwierzątko_uni').content == 'My **changed** content'
 

	
 
    def test_change_raise_empty_repository(self):
 
        node = FileNode('foobar')
 
        with pytest.raises(EmptyRepositoryError):
 
            self.imc.change(node)
 

	
 
    def test_check_integrity_change_raise_node_does_not_exist(self):
 
        node = FileNode('foobar', content='baz')
 
        self.imc.add(node)
 
        self.imc.commit(message=u'Added foobar', author=unicode(self))
 
        node = FileNode('not-foobar', content='')
 
        self.imc.change(node)
 
        with pytest.raises(NodeDoesNotExistError):
 
            self.imc.commit(message='Changed not existing node', author=str(self))
 

	
 
    def test_change_raise_node_already_changed(self):
 
        node = FileNode('foobar', content='baz')
 
        self.imc.add(node)
 
        self.imc.commit(message=u'Added foobar', author=unicode(self))
 
        node = FileNode('foobar', content='more baz')
 
        self.imc.change(node)
 
        with pytest.raises(NodeAlreadyChangedError):
 
            self.imc.change(node)
 

	
 
    def test_check_integrity_change_raise_node_not_changed(self):
 
        self.test_add()  # Performs first commit
 

	
 
        node = FileNode(self.nodes[0].path, content=self.nodes[0].content)
 
        self.imc.change(node)
 
        with pytest.raises(NodeNotChangedError):
 
            self.imc.commit(
 
                message=u'Trying to mark node as changed without touching it',
 
                author=unicode(self)
 
            )
 

	
 
    def test_change_raise_node_already_removed(self):
 
        node = FileNode('foobar', content='baz')
 
        self.imc.add(node)
 
        self.imc.commit(message=u'Added foobar', author=unicode(self))
 
        self.imc.remove(FileNode('foobar'))
 
        with pytest.raises(NodeAlreadyRemovedError):
 
            self.imc.change(node)
 

	
 
    def test_remove(self):
 
        self.test_add()  # Performs first commit
 

	
 
        tip = self.repo.get_changeset()
 
        node = self.nodes[0]
 
        assert node.content == tip.get_node(node.path).content
 
        self.imc.remove(node)
 
        self.imc.commit(message=u'Removed %s' % node.path, author=unicode(self))
 

	
 
        newtip = self.repo.get_changeset()
 
        assert tip != newtip
 
        assert tip.id != newtip.id
 
        with pytest.raises(NodeDoesNotExistError):
 
            newtip.get_node(node.path)
 

	
 
    def test_remove_last_file_from_directory(self):
 
        node = FileNode('omg/qwe/foo/bar', content='foobar')
 
        self.imc.add(node)
 
        self.imc.commit(u'added', u'joe doe')
 

	
 
        self.imc.remove(node)
 
        tip = self.imc.commit(u'removed', u'joe doe')
 
        with pytest.raises(NodeDoesNotExistError):
 
            tip.get_node('omg/qwe/foo/bar')
 

	
 
    def test_remove_raise_node_does_not_exist(self):
 
        self.imc.remove(self.nodes[0])
 
        with pytest.raises(NodeDoesNotExistError):
 
            self.imc.commit(
 
                message='Trying to remove node at empty repository',
 
                author=str(self)
 
            )
 

	
 
    def test_check_integrity_remove_raise_node_does_not_exist(self):
 
        self.test_add()  # Performs first commit
 

	
 
        node = FileNode('no-such-file')
 
        self.imc.remove(node)
 
        with pytest.raises(NodeDoesNotExistError):
 
            self.imc.commit(
 
                message=u'Trying to remove not existing node',
 
                author=unicode(self)
 
            )
 

	
 
    def test_remove_raise_node_already_removed(self):
 
        self.test_add() # Performs first commit
 

	
 
        node = FileNode(self.nodes[0].path)
 
        self.imc.remove(node)
 
        with pytest.raises(NodeAlreadyRemovedError):
 
            self.imc.remove(node)
 

	
 
    def test_remove_raise_node_already_changed(self):
 
        self.test_add()  # Performs first commit
 

	
 
        node = FileNode(self.nodes[0].path, content='Bending time')
 
        self.imc.change(node)
 
        with pytest.raises(NodeAlreadyChangedError):
 
            self.imc.remove(node)
 

	
 
    def test_reset(self):
 
        self.imc.add(FileNode('foo', content='bar'))
 
        #self.imc.change(FileNode('baz', content='new'))
 
        #self.imc.remove(FileNode('qwe'))
 
        self.imc.reset()
 
        assert not any((
 
            self.imc.added,
 
            self.imc.changed,
 
            self.imc.removed
 
        ))
 

	
 
    def test_multiple_commits(self):
 
        N = 3  # number of commits to perform
 
        last = None
 
        for x in xrange(N):
 
            fname = 'file%s' % str(x).rjust(5, '0')
 
            content = 'foobar\n' * x
 
            node = FileNode(fname, content=content)
 
            self.imc.add(node)
 
            commit = self.imc.commit(u"Commit no. %s" % (x + 1), author=u'vcs')
 
            assert last != commit
 
            last = commit
 

	
 
        # Check commit number for same repo
 
        assert len(self.repo.revisions) == N
 

	
 
        # Check commit number for recreated repo
 
        assert len(self.repo.revisions) == N
 

	
 
    def test_date_attr(self):
 
        node = FileNode('foobar.txt', content='Foobared!')
 
        self.imc.add(node)
 
        date = datetime.datetime(1985, 1, 30, 1, 45)
 
        commit = self.imc.commit(u"Committed at time when I was born ;-)",
 
            author=u'lb <lb@example.com>', date=date)
 

	
 
        assert commit.date == date
 

	
 

	
 
class TestGitInMemoryChangeset(InMemoryChangesetTestMixin):
 
    backend_alias = 'git'
 

	
 

	
 
class TestHgInMemoryChangeset(InMemoryChangesetTestMixin):
 
    backend_alias = 'hg'
scripts/update-copyrights.py
Show inline comments
 
#!/usr/bin/env python2
 
# -*- coding: utf-8 -*-
 

	
 
"""
 
Kallithea script for maintaining contributor lists from version control
 
history.
 

	
 
This script and the data in it is a best effort attempt at reverse engineering
 
previous attributions and correlate that with version control history while
 
preserving all existing copyright statements and attribution. This script is
 
processing and summarizing information found elsewhere - it is not by itself
 
making any claims. Comments in the script are an attempt at reverse engineering
 
possible explanations - they are not showing any intent or confirming it is
 
correct.
 

	
 
Three files are generated / modified by this script:
 

	
 
kallithea/templates/about.html claims to show copyright holders, and the GPL
 
license requires such existing "legal notices" to be preserved. We also try to
 
keep it updated with copyright holders, but do not claim it is a correct list.
 

	
 
CONTRIBUTORS has the purpose of giving credit where credit is due and list all
 
the contributor names in the source.
 

	
 
kallithea/templates/base/base.html contains the copyright years in the page
 
footer.
 

	
 
Both make a best effort of listing all copyright holders, but revision control
 
history might be a better and more definitive source.
 

	
 
Contributors are sorted "fairly" by copyright year and amount of
 
contribution.
 

	
 
New contributors are listed, without considering if the contribution contains
 
copyrightable work.
 

	
 
When the copyright might belong to a different legal entity than the
 
contributor, the legal entity is given credit too.
 
"""
 

	
 

	
 
# Some committers are so wrong that it doesn't point at any contributor:
 
total_ignore = set()
 
total_ignore.add('*** failed to import extension hggit: No module named hggit')
 
total_ignore.add('<>')
 

	
 
# Normalize some committer names where people have contributed under different
 
# names or email addresses:
 
name_fixes = {}
 
name_fixes['Andrew Shadura'] = "Andrew Shadura <andrew@shadura.me>"
 
name_fixes['aparkar'] = "Aparkar <aparkar@icloud.com>"
 
name_fixes['Aras Pranckevicius'] = "Aras Pranckevičius <aras@unity3d.com>"
 
name_fixes['Augosto Hermann'] = "Augusto Herrmann <augusto.herrmann@planejamento.gov.br>"
 
name_fixes['"Bradley M. Kuhn" <bkuhn@ebb.org>'] = "Bradley M. Kuhn <bkuhn@sfconservancy.org>"
 
name_fixes['dmitri.kuznetsov'] = "Dmitri Kuznetsov"
 
name_fixes['Dmitri Kuznetsov'] = "Dmitri Kuznetsov"
 
name_fixes['domruf'] = "Dominik Ruf <dominikruf@gmail.com>"
 
name_fixes['Ingo von borstel'] = "Ingo von Borstel <kallithea@planetmaker.de>"
 
name_fixes['Jan Heylen'] = "Jan Heylen <heyleke@gmail.com>"
 
name_fixes['Jason F. Harris'] = "Jason Harris <jason@jasonfharris.com>"
 
name_fixes['Jelmer Vernooij'] = "Jelmer Vernooij <jelmer@samba.org>"
 
name_fixes['jfh <jason@jasonfharris.com>'] = "Jason Harris <jason@jasonfharris.com>"
 
name_fixes['Leonardo Carneiro<leonardo@unity3d.com>'] = "Leonardo Carneiro <leonardo@unity3d.com>"
 
name_fixes['leonardo'] = "Leonardo Carneiro <leonardo@unity3d.com>"
 
name_fixes['Leonardo <leo@unity3d.com>'] = "Leonardo Carneiro <leonardo@unity3d.com>"
 
name_fixes['Les Peabody'] = "Les Peabody <lpeabody@gmail.com>"
 
name_fixes['"Lorenzo M. Catucci" <lorenzo@sancho.ccd.uniroma2.it>'] = "Lorenzo M. Catucci <lorenzo@sancho.ccd.uniroma2.it>"
 
name_fixes['Lukasz Balcerzak'] = "Łukasz Balcerzak <lukaszbalcerzak@gmail.com>"
 
name_fixes['mao <mao@lins.fju.edu.tw>'] = "Ching-Chen Mao <mao@lins.fju.edu.tw>"
 
name_fixes['marcink'] = "Marcin Kuźmiński <marcin@python-works.com>"
 
name_fixes['Marcin Kuzminski'] = "Marcin Kuźmiński <marcin@python-works.com>"
 
name_fixes['nansenat16@null.tw'] = "nansenat16 <nansenat16@null.tw>"
 
name_fixes['Peter Vitt'] = "Peter Vitt <petervitt@web.de>"
 
name_fixes['philip.j@hostdime.com'] = "Philip Jameson <philip.j@hostdime.com>"
 
name_fixes['Søren Løvborg'] = "Søren Løvborg <sorenl@unity3d.com>"
 
name_fixes['Thomas De Schampheleire'] = "Thomas De Schampheleire <thomas.de_schampheleire@nokia.com>"
 
name_fixes['Weblate'] = "<>"
 
name_fixes['xpol'] = "xpol <xpolife@gmail.com>"
 

	
 

	
 
# Some committer email address domains that indicate that another entity might
 
# hold some copyright too:
 
domain_extra = {}
 
domain_extra['unity3d.com'] = "Unity Technologies"
 
domain_extra['rhodecode.com'] = "RhodeCode GmbH"
 

	
 
# Repository history show some old contributions that traditionally hasn't been
 
# listed in about.html - preserve that:
 
no_about = set(total_ignore)
 
# The following contributors were traditionally not listed in about.html and it
 
# seems unclear if the copyright is personal or belongs to a company.
 
no_about.add(('Thayne Harbaugh <thayne@fusionio.com>', '2011'))
 
no_about.add(('Dies Koper <diesk@fast.au.fujitsu.com>', '2012'))
 
no_about.add(('Erwin Kroon <e.kroon@smartmetersolutions.nl>', '2012'))
 
no_about.add(('Vincent Caron <vcaron@bearstech.com>', '2012'))
 
# These contributors' contributions might be too small to be copyrightable:
 
no_about.add(('philip.j@hostdime.com', '2012'))
 
no_about.add(('Stefan Engel <mail@engel-stefan.de>', '2012'))
 
no_about.add(('Ton Plomp <tcplomp@gmail.com>', '2013'))
 
# Was reworked and contributed later and shadowed by other contributions:
 
no_about.add(('Sean Farley <sean.michael.farley@gmail.com>', '2013'))
 

	
 
# Preserve contributors listed in about.html but not appearing in repository
 
# history:
 
other_about = [
 
    ("2011", "Aparkar <aparkar@icloud.com>"),
 
    ("2010", "RhodeCode GmbH"),
 
    ("2011", "RhodeCode GmbH"),
 
    ("2012", "RhodeCode GmbH"),
 
    ("2013", "RhodeCode GmbH"),
 
]
 

	
 
# Preserve contributors listed in CONTRIBUTORS but not appearing in repository
 
# history:
 
other_contributors = [
 
    ("", "Andrew Kesterson <andrew@aklabs.net>"),
 
    ("", "cejones"),
 
    ("", "David A. Sjøen <david.sjoen@westcon.no>"),
 
    ("", "James Rhodes <jrhodes@redpointsoftware.com.au>"),
 
    ("", "Jonas Oberschweiber <jonas.oberschweiber@d-velop.de>"),
 
    ("", "larikale"),
 
    ("", "RhodeCode GmbH"),
 
    ("", "Sebastian Kreutzberger <sebastian@rhodecode.com>"),
 
    ("", "Steve Romanow <slestak989@gmail.com>"),
 
    ("", "SteveCohen"),
 
    ("", "Thomas <thomas@rhodecode.com>"),
 
    ("", "Thomas Waldmann <tw-public@gmx.de>"),
 
]
 

	
 

	
 
import os
 
import re
 
from collections import defaultdict
 

	
 

	
 
def sortkey(x):
 
    """Return key for sorting contributors "fairly":
 
    * latest contribution
 
    * first contribution
 
    * number of contribution years
 
    * name (with some unicode normalization)
 
    The entries must be 2-tuples of a list of string years and the unicode name"""
 
    return (x[0] and -int(x[0][-1]),
 
            x[0] and int(x[0][0]),
 
            -len(x[0]),
 
            x[1].decode('utf8').lower().replace(u'\xe9', u'e').replace(u'\u0142', u'l')
 
            x[1].decode('utf-8').lower().replace(u'\xe9', u'e').replace(u'\u0142', u'l')
 
        )
 

	
 

	
 
def nice_years(l, dash='-', join=' '):
 
    """Convert a list of years into brief range like '1900-1901, 1921'."""
 
    if not l:
 
        return ''
 
    start = end = int(l[0])
 
    ranges = []
 
    for year in l[1:] + [0]:
 
        year = int(year)
 
        if year == end + 1:
 
            end = year
 
            continue
 
        if start == end:
 
            ranges.append('%s' % start)
 
        else:
 
            ranges.append('%s%s%s' % (start, dash, end))
 
        start = end = year
 
    assert start == 0 and end == 0, (start, end)
 
    return join.join(ranges)
 

	
 

	
 
def insert_entries(
 
        filename,
 
        all_entries,
 
        no_entries,
 
        domain_extra,
 
        split_re,
 
        normalize_name,
 
        format_f):
 
    """Update file with contributor information.
 
    all_entries: list of tuples with year and name
 
    no_entries: set of names or name and year tuples to ignore
 
    domain_extra: map domain name to extra credit name
 
    split_re: regexp matching the part of file to rewrite
 
    normalize_name: function to normalize names for grouping and display
 
    format_f: function formatting year list and name to a string
 
    """
 
    name_years = defaultdict(set)
 

	
 
    for year, name in all_entries:
 
        if name in no_entries or (name, year) in no_entries:
 
            continue
 
        domain = name.split('@', 1)[-1].rstrip('>')
 
        if domain in domain_extra:
 
            name_years[domain_extra[domain]].add(year)
 
        name_years[normalize_name(name)].add(year)
 

	
 
    l = [(list(sorted(year for year in years if year)), name)
 
         for name, years in name_years.items()]
 
    l.sort(key=sortkey)
 

	
 
    with open(filename) as f:
 
        pre, post = re.split(split_re, f.read())
 

	
 
    with open(filename, 'w') as f:
 
        f.write(pre +
 
                ''.join(format_f(years, name) for years, name in l) +
 
                post)
 

	
 

	
 
def main():
 
    repo_entries = [
 
        (year, name_fixes.get(name) or name_fixes.get(name.rsplit('<', 1)[0].strip()) or name)
 
        for year, name in
 
        (line.strip().split(' ', 1)
 
         for line in os.popen("""hg log -r '::.' -T '{date(date,"%Y")} {author}\n'""").readlines())
 
        ]
 

	
 
    insert_entries(
 
        filename='kallithea/templates/about.html',
 
        all_entries=repo_entries + other_about,
 
        no_entries=no_about,
 
        domain_extra=domain_extra,
 
        split_re=r'(?:  <li>Copyright &copy; [^\n]*</li>\n)*',
 
        normalize_name=lambda name: name.split('<', 1)[0].strip(),
 
        format_f=lambda years, name: '  <li>Copyright &copy; %s, %s</li>\n' % (nice_years(years, '&ndash;', ', '), name),
 
        )
 

	
 
    insert_entries(
 
        filename='CONTRIBUTORS',
 
        all_entries=repo_entries + other_contributors,
 
        no_entries=total_ignore,
 
        domain_extra=domain_extra,
 
        split_re=r'(?:    [^\n]*\n)*',
 
        normalize_name=lambda name: name,
 
        format_f=lambda years, name: ('    %s%s%s\n' % (name, ' ' if years else '', nice_years(years))),
 
        )
 

	
 
    insert_entries(
 
        filename='kallithea/templates/base/base.html',
 
        all_entries=repo_entries,
 
        no_entries=total_ignore,
 
        domain_extra={},
 
        split_re=r'(?<=&copy;) .* (?=by various authors)',
 
        normalize_name=lambda name: '',
 
        format_f=lambda years, name: ' ' + nice_years(years, '&ndash;', ', ') + ' ',
 
        )
 

	
 

	
 
if __name__ == '__main__':
 
    main()
 

	
 

	
 
# To list new contributors since last tagging:
 
# { hg log -r '::tagged()' -T '    {author}\n    {author}\n'; hg log -r '::.' -T '    {author}\n' | sort | uniq; } | sort | uniq -u
0 comments (0 inline, 0 general)