################################################################################
# RhodeCode - Pylons environment configuration #
# #
# The %(here)s variable will be replaced with the parent directory of this file#
[DEFAULT]
debug = true
pdebug = false
## Uncomment and replace with the address which should receive ##
## any error reports after application crash ##
## Additionally those settings will be used by RhodeCode mailing system ##
#email_to = admin@localhost
#error_email_from = paste_error@localhost
#app_email_from = rhodecode-noreply@localhost
#error_message =
#email_prefix = [RhodeCode]
#smtp_server = mail.server.com
#smtp_username =
#smtp_password =
#smtp_port =
#smtp_use_tls = false
#smtp_use_ssl = true
# Specify available auth parameters here (e.g. LOGIN PLAIN CRAM-MD5, etc.)
#smtp_auth =
[server:main]
##nr of threads to spawn
#threadpool_workers = 5
##max request before thread respawn
#threadpool_max_requests = 10
##option to use threads of process
#use_threadpool = true
#use = egg:Paste#http
use = egg:waitress#main
host = 0.0.0.0
port = 5000
[filter:proxy-prefix]
# prefix middleware for rc
use = egg:PasteDeploy#prefix
prefix = /<your-prefix>
[app:main]
use = egg:rhodecode
#filter-with = proxy-prefix
full_stack = true
static_files = true
# Optional Languages
# en, fr, ja, pt_BR, zh_CN, zh_TW
lang = en
cache_dir = %(here)s/data
index_dir = %(here)s/data/index
app_instance_uuid = rc-develop
cut_off_limit = 256000
force_https = false
commit_parse_limit = 25
use_gravatar = true
## alternative_gravatar_url allows you to use your own avatar server application
## the following parts of the URL will be replaced
## {email} user email
## {md5email} md5 hash of the user email (like at gravatar.com)
## {size} size of the image that is expected from the server application
## {scheme} http/https from RhodeCode server
## {netloc} network location from RhodeCode server
#alternative_gravatar_url = http://myavatarserver.com/getbyemail/{email}/{size}
#alternative_gravatar_url = http://myavatarserver.com/getbymd5/{md5email}?s={size}
container_auth_enabled = false
proxypass_auth_enabled = false
default_encoding = utf8
## overwrite schema of clone url
## available vars:
## scheme - http/https
## user - current user
## pass - password
## netloc - network location
## path - usually repo_name
#clone_uri = {scheme}://{user}{pass}{netloc}{path}
## issue tracking mapping for commits messages
## comment out issue_pat, issue_server, issue_prefix to enable
## pattern to get the issues from commit messages
## default one used here is #<numbers> with a regex passive group for `#`
## {id} will be all groups matched from this pattern
issue_pat = (?:\s*#)(\d+)
## server url to the issue, each {id} will be replaced with match
## fetched from the regex and {repo} is replaced with full repository name
## including groups {repo_name} is replaced with just name of repo
issue_server_link = https://myissueserver.com/{repo}/issue/{id}
## prefix to add to link to indicate it's an url
## #314 will be replaced by <issue_prefix><id>
issue_prefix = #
## instance-id prefix
## a prefix key for this instance used for cache invalidation when running
## multiple instances of rhodecode, make sure it's globally unique for
## all running rhodecode instances. Leave empty if you don't use it
instance_id =
## alternative return HTTP header for failed authentication. Default HTTP
## response is 401 HTTPUnauthorized. Currently HG clients have troubles with
## handling that. Set this variable to 403 to return HTTPForbidden
auth_ret_code =
####################################
### CELERY CONFIG ####
use_celery = false
broker.host = localhost
broker.vhost = rabbitmqhost
broker.port = 5672
broker.user = rabbitmq
broker.password = qweqwe
celery.imports = rhodecode.lib.celerylib.tasks
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.log.file = celeryd.log
celeryd.log.level = debug
celeryd.max.tasks.per.child = 1
#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=super_short_term,short_term,long_term,sql_cache_short,sql_cache_med,sql_cache_long
beaker.cache.super_short_term.type=memory
beaker.cache.super_short_term.expire=10
beaker.cache.super_short_term.key_length = 256
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
host = 127.0.0.1
port = 8001
app_instance_uuid = rc-production
commit_parse_limit = 50
app_instance_uuid = ${app_instance_uuid}
"""Helper functions
Consists of functions to typically be used within templates, but also
available to Controllers. This module is available to both as 'h'.
"""
import random
import hashlib
import StringIO
import urllib
import math
import logging
import re
import urlparse
from datetime import datetime
from pygments.formatters.html import HtmlFormatter
from pygments import highlight as code_highlight
from pylons import url, request, config
from pylons.i18n.translation import _, ungettext
from hashlib import md5
from webhelpers.html import literal, HTML, escape
from webhelpers.html.tools import *
from webhelpers.html.builder import make_tag
from webhelpers.html.tags import auto_discovery_link, checkbox, css_classes, \
end_form, file, form, hidden, image, javascript_link, link_to, \
link_to_if, link_to_unless, ol, required_legend, select, stylesheet_link, \
submit, text, password, textarea, title, ul, xml_declaration, radio
from webhelpers.html.tools import auto_link, button_to, highlight, \
js_obfuscate, mail_to, strip_links, strip_tags, tag_re
from webhelpers.number import format_byte_size, format_bit_size
from webhelpers.pylonslib import Flash as _Flash
from webhelpers.pylonslib.secure_form import secure_form
from webhelpers.text import chop_at, collapse, convert_accented_entities, \
convert_misc_entities, lchop, plural, rchop, remove_formatting, \
replace_whitespace, urlify, truncate, wrap_paragraphs
from webhelpers.date import time_ago_in_words
from webhelpers.paginate import Page
from webhelpers.html.tags import _set_input_attrs, _set_id_attr, \
convert_boolean_attrs, NotGiven, _make_safe_id_component
from rhodecode.lib.annotate import annotate_highlight
from rhodecode.lib.utils import repo_name_slug
from rhodecode.lib.utils2 import str2bool, safe_unicode, safe_str, \
get_changeset_safe, datetime_to_time, time_to_datetime
from rhodecode.lib.markup_renderer import MarkupRenderer
from rhodecode.lib.vcs.exceptions import ChangesetDoesNotExistError
from rhodecode.lib.vcs.backends.base import BaseChangeset
from rhodecode.config.conf import DATE_FORMAT, DATETIME_FORMAT
from rhodecode.model.changeset_status import ChangesetStatusModel
from rhodecode.model.db import URL_SEP, Permission
log = logging.getLogger(__name__)
html_escape_table = {
"&": "&",
'"': """,
"'": "'",
">": ">",
"<": "<",
}
def html_escape(text):
"""Produce entities within text."""
return "".join(html_escape_table.get(c,c) for c in text)
def shorter(text, size=20):
postfix = '...'
if len(text) > size:
return text[:size - len(postfix)] + postfix
return text
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 uniqe 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), md5(safe_str(path)).hexdigest()[:12])
def get_token():
"""Return the current authentication token, creating one if one doesn't
already exist.
token_key = "_authentication_token"
from pylons import session
if not token_key in session:
try:
@@ -618,197 +619,201 @@ def action_parser(user_log, feed=False):
pull_request_id = action_params
repo_name = user_log.repository.repo_name
return link_to(_('Pull request #%s') % pull_request_id,
url('pullrequest_show', repo_name=repo_name,
pull_request_id=pull_request_id))
# action : translated str, callback(extractor), icon
action_map = {
'user_deleted_repo': (_('[deleted] repository'),
None, 'database_delete.png'),
'user_created_repo': (_('[created] repository'),
None, 'database_add.png'),
'user_created_fork': (_('[created] repository as fork'),
None, 'arrow_divide.png'),
'user_forked_repo': (_('[forked] repository'),
get_fork_name, 'arrow_divide.png'),
'user_updated_repo': (_('[updated] repository'),
None, 'database_edit.png'),
'admin_deleted_repo': (_('[delete] repository'),
'admin_created_repo': (_('[created] repository'),
'admin_forked_repo': (_('[forked] repository'),
'admin_updated_repo': (_('[updated] repository'),
'admin_created_user': (_('[created] user'),
get_user_name, 'user_add.png'),
'admin_updated_user': (_('[updated] user'),
get_user_name, 'user_edit.png'),
'admin_created_users_group': (_('[created] users group'),
get_users_group, 'group_add.png'),
'admin_updated_users_group': (_('[updated] users group'),
get_users_group, 'group_edit.png'),
'user_commented_revision': (_('[commented] on revision in repository'),
get_cs_links, 'comment_add.png'),
'user_commented_pull_request': (_('[commented] on pull request for'),
get_pull_request, 'comment_add.png'),
'user_closed_pull_request': (_('[closed] pull request for'),
get_pull_request, 'tick.png'),
'push': (_('[pushed] into'),
get_cs_links, 'script_add.png'),
'push_local': (_('[committed via RhodeCode] into repository'),
get_cs_links, 'script_edit.png'),
'push_remote': (_('[pulled from remote] into repository'),
get_cs_links, 'connect.png'),
'pull': (_('[pulled] from'),
None, 'down_16.png'),
'started_following_repo': (_('[started following] repository'),
None, 'heart_add.png'),
'stopped_following_repo': (_('[stopped following] repository'),
None, 'heart_delete.png'),
action_str = action_map.get(action, action)
if feed:
action = action_str[0].replace('[', '').replace(']', '')
else:
action = action_str[0]\
.replace('[', '<span class="journal_highlight">')\
.replace(']', '</span>')
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
tmpl = """<img src="%s%s" alt="%s"/>"""
ico = action_map.get(action, ['', '', ''])[2]
return literal(tmpl % ((url('/images/icons/')), ico, action))
# returned callbacks we need to call to get
return [lambda: literal(action), action_params_func, action_parser_icon]
#==============================================================================
# PERMS
from rhodecode.lib.auth import HasPermissionAny, HasPermissionAll, \
HasRepoPermissionAny, HasRepoPermissionAll
# GRAVATAR URL
def gravatar_url(email_address, size=30):
from pylons import url ## doh, we need to re-import url to mock it later
if(str2bool(config['app_conf'].get('use_gravatar')) and
config['app_conf'].get('alternative_gravatar_url')):
tmpl = config['app_conf'].get('alternative_gravatar_url', '')
parsed_url = urlparse.urlparse(url.current(qualified=True))
tmpl = tmpl.replace('{email}', email_address)\
.replace('{md5email}', hashlib.md5(email_address.lower()).hexdigest())\
.replace('{md5email}', hashlib.md5(email_address.lower()).hexdigest()) \
.replace('{netloc}', parsed_url.netloc)\
.replace('{scheme}', parsed_url.scheme)\
.replace('{size}', str(size))
return tmpl
if (not str2bool(config['app_conf'].get('use_gravatar')) or
not email_address or email_address == 'anonymous@rhodecode.org'):
f = lambda a, l: min(l, key=lambda x: abs(x - a))
return url("/images/user%s.png" % f(size, [14, 16, 20, 24, 30]))
ssl_enabled = 'https' == request.environ.get('wsgi.url_scheme')
default = 'identicon'
baseurl_nossl = "http://www.gravatar.com/avatar/"
baseurl_ssl = "https://secure.gravatar.com/avatar/"
baseurl = baseurl_ssl if ssl_enabled else baseurl_nossl
if isinstance(email_address, unicode):
#hashlib crashes on unicode items
email_address = safe_str(email_address)
# construct the url
gravatar_url = baseurl + hashlib.md5(email_address.lower()).hexdigest() + "?"
gravatar_url += urllib.urlencode({'d': default, 's': str(size)})
return gravatar_url
# REPO PAGER, PAGER FOR REPOSITORY
class RepoPage(Page):
def __init__(self, collection, page=1, items_per_page=20,
item_count=None, url=None, **kwargs):
"""Create a "RepoPage" instance. special pager for paging
repository
self._url_generator = url
# Safe the kwargs class-wide so they can be used in the pager() method
self.kwargs = kwargs
# Save a reference to the collection
self.original_collection = collection
self.collection = collection
# The self.page is the number of the current page.
# The first page has the number 1!
self.page = int(page) # make it int() if we get it as a string
except (ValueError, TypeError):
self.page = 1
self.items_per_page = items_per_page
# Unless the user tells us how many items the collections has
# we calculate that ourselves.
if item_count is not None:
self.item_count = item_count
self.item_count = len(self.collection)
# Compute the number of the first and last available page
if self.item_count > 0:
self.first_page = 1
self.page_count = int(math.ceil(float(self.item_count) /
self.items_per_page))
self.last_page = self.first_page + self.page_count - 1
# Make sure that the requested page number is the range of
# valid pages
if self.page > self.last_page:
self.page = self.last_page
elif self.page < self.first_page:
self.page = self.first_page
# Note: the number of items on this page can be less than
# items_per_page if the last page is not full
self.first_item = max(0, (self.item_count) - (self.page *
items_per_page))
self.last_item = ((self.item_count - 1) - items_per_page *
(self.page - 1))
self.items = list(self.collection[self.first_item:self.last_item + 1])
# Links to previous and next page
if self.page > self.first_page:
self.previous_page = self.page - 1
self.previous_page = None
if self.page < self.last_page:
self.next_page = self.page + 1
self.next_page = None
# No items available
@@ -66,112 +66,133 @@ TEST_URLS += [
class TestLibs(unittest.TestCase):
def test_uri_filter(self):
from rhodecode.lib.utils2 import uri_filter
for url in TEST_URLS:
self.assertEqual(uri_filter(url[0]), url[1])
def test_credentials_filter(self):
from rhodecode.lib.utils2 import credentials_filter
self.assertEqual(credentials_filter(url[0]), url[2])
def test_str2bool(self):
from rhodecode.lib.utils2 import str2bool
test_cases = [
('t', True),
('true', True),
('y', True),
('yes', True),
('on', True),
('1', True),
('Y', True),
('yeS', True),
('TRUE', True),
('T', True),
('False', False),
('F', False),
('FALSE', False),
('0', False),
('-1', False),
('', False), ]
for case in test_cases:
self.assertEqual(str2bool(case[0]), case[1])
def test_mention_extractor(self):
from rhodecode.lib.utils2 import extract_mentioned_users
sample = (
"@first hi there @marcink here's my email marcin@email.com "
"@lukaszb check @one_more22 it pls @ ttwelve @D[] @one@two@three "
"@MARCIN @maRCiN @2one_more22 @john please see this http://org.pl "
"@marian.user just do it @marco-polo and next extract @marco_polo "
"user.dot hej ! not-needed maril@domain.org"
)
s = sorted([
'first', 'marcink', 'lukaszb', 'one_more22', 'MARCIN', 'maRCiN', 'john',
'marian.user', 'marco-polo', 'marco_polo'
], key=lambda k: k.lower())
self.assertEqual(s, extract_mentioned_users(sample))
def test_age(self):
import calendar
from rhodecode.lib.utils2 import age
n = datetime.datetime.now()
delt = lambda *args, **kwargs: datetime.timedelta(*args, **kwargs)
self.assertEqual(age(n), u'just now')
self.assertEqual(age(n - delt(seconds=1)), u'1 second ago')
self.assertEqual(age(n - delt(seconds=60 * 2)), u'2 minutes ago')
self.assertEqual(age(n - delt(hours=1)), u'1 hour ago')
self.assertEqual(age(n - delt(hours=24)), u'1 day ago')
self.assertEqual(age(n - delt(hours=24 * 5)), u'5 days ago')
self.assertEqual(age(n - delt(hours=24 * (calendar.mdays[n.month-1] + 2))),
u'1 month and 2 days ago')
self.assertEqual(age(n - delt(hours=24 * 400)), u'1 year and 1 month ago')
def test_tag_exctrator(self):
"hello pta[tag] gog [[]] [[] sda ero[or]d [me =>>< sa]"
"[requires] [stale] [see<>=>] [see => http://url.com]"
"[requires => url] [lang => python] [just a tag]"
"[,d] [ => ULR ] [obsolete] [desc]]"
from rhodecode.lib.helpers import desc_stylize
res = desc_stylize(sample)
self.assertTrue('<div class="metatag" tag="tag">tag</div>' in res)
self.assertTrue('<div class="metatag" tag="obsolete">obsolete</div>' in res)
self.assertTrue('<div class="metatag" tag="stale">stale</div>' in res)
self.assertTrue('<div class="metatag" tag="lang">python</div>' in res)
self.assertTrue('<div class="metatag" tag="requires">requires => <a href="/url">url</a></div>' in res)
def test_alternative_gravatar(self):
from rhodecode.lib.helpers import gravatar_url
_md5 = lambda s: hashlib.md5(s).hexdigest()
def fake_conf(**kwargs):
from pylons import config
config['app_conf'] = {}
config['app_conf']['use_gravatar'] = True
config['app_conf'].update(kwargs)
return config
fake = fake_conf(alternative_gravatar_url='http://test.com/{email}')
with mock.patch('pylons.config', fake):
grav = gravatar_url(email_address='test@foo.com', size=24)
assert grav == 'http://test.com/test@foo.com'
class fake_url():
@classmethod
def current(cls, *args, **kwargs):
return 'https://server.com'
with mock.patch('pylons.url', fake_url):
from pylons import url
assert url.current() == 'http://server.com'
fake = fake_conf(alternative_gravatar_url='http://test.com/{md5email}')
em = 'test@foo.com'
grav = gravatar_url(email_address=em, size=24)
assert grav == 'http://test.com/%s' % (_md5(em))
fake = fake_conf(alternative_gravatar_url='http://test.com/{md5email}/{size}')
assert grav == 'http://test.com/%s/%s' % (_md5(em), 24)
fake = fake_conf(alternative_gravatar_url='{scheme}://{netloc}/{md5email}/{size}')
assert grav == 'https://server.com/%s/%s' % (_md5(em), 24)
Status change: