@@ -691,185 +691,177 @@ def create_test_env(repos_test_path, con
tar.close()
cur_dir = dn(dn(abspath(__file__)))
tar = tarfile.open(jn(cur_dir, 'tests', 'fixtures', "vcs_test_git.tar.gz"))
tar.extractall(jn(TESTS_TMP_PATH, GIT_REPO))
#LOAD VCS test stuff
from kallithea.tests.vcs import setup_package
setup_package()
#==============================================================================
# PASTER COMMANDS
class BasePasterCommand(Command):
"""
Abstract Base Class for paster commands.
The celery commands are somewhat aggressive about loading
celery.conf, and since our module sets the `CELERY_LOADER`
environment variable to our loader, we have to bootstrap a bit and
make sure we've had a chance to load the pylons config off of the
command line, otherwise everything fails.
min_args = 1
min_args_error = "Please provide a paster config file as an argument."
takes_config_file = 1
requires_config_file = True
def notify_msg(self, msg, log=False):
"""Make a notification to user, additionally if logger is passed
it logs this action using given logger
:param msg: message that will be printed to user
:param log: logging instance, to use to additionally log this message
if log and isinstance(log, logging):
log(msg)
def run(self, args):
Overrides Command.run
Checks for a config file argument and loads it.
if len(args) < self.min_args:
raise BadCommand(
self.min_args_error % {'min_args': self.min_args,
'actual_args': len(args)})
# Decrement because we're going to lob off the first argument.
# @@ This is hacky
self.min_args -= 1
self.bootstrap_config(args[0])
self.update_parser()
return super(BasePasterCommand, self).run(args[1:])
def update_parser(self):
Abstract method. Allows for the class's parser to be updated
before the superclass's `run` method is called. Necessary to
allow options/arguments to be passed through to the underlying
celery command.
raise NotImplementedError("Abstract Method.")
def bootstrap_config(self, conf):
Loads the pylons configuration.
from pylons import config as pylonsconfig
self.path_to_ini_file = os.path.realpath(conf)
conf = paste.deploy.appconfig('config:' + self.path_to_ini_file)
pylonsconfig.init_app(conf.global_conf, conf.local_conf)
def _init_session(self):
Inits SqlAlchemy Session
logging.config.fileConfig(self.path_to_ini_file)
from pylons import config
from kallithea.model import init_model
from kallithea.lib.utils2 import engine_from_config
#get to remove repos !!
add_cache(config)
engine = engine_from_config(config, 'sqlalchemy.db1.')
init_model(engine)
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 properly work.
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)
ver = (stdout.split(' ')[-1] or '').strip() or '0.0.0'
if len(ver.split('.')) > 3:
#StrictVersion needs to be only 3 element type
ver = '.'.join(ver.split('.')[:3])
try:
_ver = StrictVersion(ver)
except ValueError:
_ver = StrictVersion('0.0.0')
stderr = traceback.format_exc()
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')
req_ver = '1.7.4'
to_old_git = False
if _ver < StrictVersion(req_ver):
to_old_git = True
if 'git' in BACKENDS:
log.debug('Git executable: "%s" version detected: %s'
% (settings.GIT_EXECUTABLE_PATH, stdout))
if stderr:
log.warning('Unable to detect git version, org error was: %r' % stderr)
elif to_old_git:
log.warning('Kallithea detected git version %s, which is too old '
'for the system to function properly. Make sure '
'its version is at least %s' % (ver, req_ver))
return _ver
log.debug('Git executable: "%s" version %s detected: %s'
% (settings.GIT_EXECUTABLE_PATH, ver, stdout))
log.warning('Error detecting git version: %r' % stderr)
elif ver < req_ver:
'for the system to function properly. '
'Please upgrade to version %s or later.' % (ver, req_ver))
return ver
@decorator.decorator
def jsonify(func, *args, **kwargs):
"""Action decorator that formats output for JSON
Given a function that will return content, this decorator will turn
the result into JSON, with a content-type of 'application/json' and
output it.
from pylons.decorators.util import get_pylons
from kallithea.lib.compat import json
pylons = get_pylons(args)
pylons.response.headers['Content-Type'] = 'application/json; charset=utf-8'
data = func(*args, **kwargs)
if isinstance(data, (list, tuple)):
msg = "JSON responses with Array envelopes are susceptible to " \
"cross-site data leak attacks, see " \
"http://wiki.pylonshq.com/display/pylonsfaq/Warnings"
warnings.warn(msg, Warning, 2)
log.warning(msg)
log.debug("Returning JSON wrapped action output")
return json.dumps(data, encoding='utf-8')
def conditional_cache(region, prefix, condition, func):
Conditional caching function use like::
def _c(arg):
#heavy computation function
return data
# denpending from condition the compute is wrapped in cache or not
compute = conditional_cache('short_term', 'cache_desc', codnition=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
Status change: