kallithea Changeset - 431689d7f37d

Changeset - 431689d7f37d

Parent rev.

Child rev.

[Not reviewed]

default

0 41 0

Søren Løvborg - 10 years ago 2015-08-31 17:42:57
sorenl@unity3d.com

remove vestiges of Python 2.5 support

We only support Python 2.6 and 2.7; hence we do not need to import
with-statement support from __future__.

27 files changed:

kallithea/bin/kallithea_api.py

kallithea/bin/kallithea_config.py

kallithea/bin/kallithea_gist.py

kallithea/config/routing.py

kallithea/controllers/files.py

kallithea/lib/auth.py

kallithea/lib/indexers/daemon.py

kallithea/lib/paster_commands/cache_keys.py

kallithea/lib/paster_commands/cleanup.py

kallithea/lib/paster_commands/ishell.py

kallithea/lib/paster_commands/make_index.py

kallithea/lib/paster_commands/make_rcextensions.py

kallithea/lib/paster_commands/repo_scan.py

kallithea/lib/paster_commands/update_repoinfo.py

kallithea/lib/pidlock.py

kallithea/lib/profiler.py

kallithea/model/api_key.py

kallithea/model/gist.py

kallithea/model/repo.py

kallithea/model/scm.py

kallithea/tests/api/api_base.py

kallithea/tests/functional/test_admin.py

kallithea/tests/functional/test_login.py

kallithea/tests/models/test_diff_parsers.py

kallithea/tests/other/test_libs.py

kallithea/tests/vcs/test_archives.py

kallithea/tests/vcs/test_branches.py

Changeset was too big and was cut off... Show full diff anyway

0 comments (0 inline, 0 general)

kallithea/bin/kallithea_api.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.bin.kallithea_api
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Api CLI client for Kallithea
 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: Jun 3, 2012
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import sys
 import argparse
 from kallithea.bin.base import json, api_call, RcConf, FORMAT_JSON, FORMAT_PRETTY
 def argparser(argv):
     usage = (
       "kallithea-api [-h] [--format=FORMAT] [--apikey=APIKEY] [--apihost=APIHOST] "
       "[--config=CONFIG] [--save-config] "
       "METHOD <key:val> <key2:val> ...\n"
       "Create config file: kallithea-api --apikey=<key> --apihost=http://your.kallithea.server --save-config"
+    )
     parser = argparse.ArgumentParser(description='Kallithea API cli',
                                      usage=usage)
     ## config
     group = parser.add_argument_group('config')
     group.add_argument('--apikey', help='api access key')
     group.add_argument('--apihost', help='api host')
     group.add_argument('--config', help='config file')
     group.add_argument('--save-config', action='store_true', help='save the given config into a file')
     group = parser.add_argument_group('API')
     group.add_argument('method', metavar='METHOD', nargs='?', type=str, default=None,
             help='API method name to call followed by key:value attributes',
+    )
     group.add_argument('--format', dest='format', type=str,
             help='output format default: `%s` can '
                  'be also `%s`' % (FORMAT_PRETTY, FORMAT_JSON),
             default=FORMAT_PRETTY
+    )
     args, other = parser.parse_known_args()
     return parser, args, other
 def main(argv=None):
     """
     Main execution function for cli
     :param argv:
     """
     if argv is None:
         argv = sys.argv
     conf = None
     parser, args, other = argparser(argv)
     api_credentials_given = (args.apikey and args.apihost)
     if args.save_config:
         if not api_credentials_given:
             raise parser.error('--save-config requires --apikey and --apihost')
         conf = RcConf(config_location=args.config,
                       autocreate=True, config={'apikey': args.apikey,
                                                'apihost': args.apihost})
         sys.exit()
     if not conf:
         conf = RcConf(config_location=args.config, autoload=True)
         if not conf:
             if not api_credentials_given:
                 parser.error('Could not find config file and missing '
                              '--apikey or --apihost in params')
     apikey = args.apikey or conf['apikey']
     apihost = args.apihost or conf['apihost']
     method = args.method
     # if we don't have method here it's an error
     if not method:
         parser.error('Please specify method name')
     try:
         margs = dict(map(lambda s: s.split(':', 1), other))
     except ValueError:
         sys.stderr.write('Error parsing arguments \n')
         sys.exit()
     if args.format == FORMAT_PRETTY:
         print 'Calling method %s => %s' % (method, apihost)
     json_resp = api_call(apikey, apihost, method, **margs)
     error_prefix = ''
     if json_resp['error']:
         error_prefix = 'ERROR:'
         json_data = json_resp['error']
     else:
         json_data = json_resp['result']
     if args.format == FORMAT_JSON:
         print json.dumps(json_data)
     elif args.format == FORMAT_PRETTY:
         print 'Server response \n%s%s' % (
             error_prefix, json.dumps(json_data, indent=4, sort_keys=True)
+        )
     return 0
 if __name__ == '__main__':
     sys.exit(main(sys.argv))

kallithea/bin/kallithea_config.py

➞

Show inline comments

 #!/usr/bin/env python2
 # -*- 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.bin.kallithea_config
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 configuration generator for Kallithea
 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: Jun 18, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import os
 import sys
 import uuid
 import argparse
 from mako.template import Template
 TMPL = 'template.ini.mako'
 here = os.path.dirname(os.path.abspath(__file__))
 def argparser(argv):
     usage = (
       "kallithea-config [-h] [--filename=FILENAME] [--template=TEMPLATE] \n"
       "VARS optional specify extra template variable that will be available in "
       "template. Use comma separated key=val format eg.\n"
       "key1=val1,port=5000,host=127.0.0.1,elements='a\,b\,c'\n"
+    )
     parser = argparse.ArgumentParser(
         description='Kallithea CONFIG generator with variable replacement',
         usage=usage
+    )
     ## config
     group = parser.add_argument_group('CONFIG')
     group.add_argument('--filename', help='Output ini filename.')
     group.add_argument('--template', help='Mako template file to use instead of '
                                           'the default builtin template')
     group.add_argument('--raw', help='Store given mako template as raw without '
                                      'parsing. Use this to create custom template '
                                      'initially', action='store_true')
     group.add_argument('--show-defaults', help='Show all default variables for '
                                                'builtin template', action='store_true')
     args, other = parser.parse_known_args()
     return parser, args, other
 def _escape_split(text, sep):
     """
     Allows for escaping of the separator: e.g. arg='foo\, bar'
     It should be noted that the way bash et. al. do command line parsing, those
     single quotes are required. a shameless ripoff from fabric project.
     """
     escaped_sep = r'\%s' % sep
     if escaped_sep not in text:
         return text.split(sep)
     before, _, after = text.partition(escaped_sep)
     startlist = before.split(sep)  # a regular split is fine here
     unfinished = startlist[-1]
     startlist = startlist[:-1]
     # recurse because there may be more escaped separators
     endlist = _escape_split(after, sep)
     # finish building the escaped value. we use endlist[0] becaue the first
     # part of the string sent in recursion is the rest of the escaped value.
     unfinished += sep + endlist[0]
     return startlist + [unfinished] + endlist[1:]  # put together all the parts
 def _run(argv):
     parser, args, other = argparser(argv)
     if not len(sys.argv) > 1:
         print parser.print_help()
         sys.exit(0)
     # defaults that can be overwritten by arguments
     tmpl_stored_args = {
         'http_server': 'waitress',
         'lang': 'en',
         'database_engine': 'sqlite',
         'host': '127.0.0.1',
         'port': 5000,
         'error_aggregation_service': None,
+    }
     if other:
         # parse arguments, we assume only first is correct
         kwargs = {}
         for el in _escape_split(other[0], ','):
             kv = _escape_split(el, '=')
             if len(kv) == 2:
                 k, v = kv
                 kwargs[k] = v
         # update our template stored args
         tmpl_stored_args.update(kwargs)
     # use default that cannot be replaced
     tmpl_stored_args.update({
         'uuid': lambda: uuid.uuid4().hex,
         'here': os.path.abspath(os.curdir),
     })
     if args.show_defaults:
         for k,v in tmpl_stored_args.iteritems():
             print '%s=%s' % (k, v)
         sys.exit(0)
     try:
         # built in template
         tmpl_file = os.path.join(here, TMPL)
         if args.template:
             tmpl_file = args.template
         with open(tmpl_file, 'rb') as f:
             tmpl_data = f.read().decode('utf-8')
             if args.raw:
                 tmpl = tmpl_data
             else:
                 tmpl = Template(tmpl_data).render(**tmpl_stored_args)
         with open(args.filename, 'wb') as f:
             f.write(tmpl.encode('utf-8'))
         print 'Wrote new config file in %s' % (os.path.abspath(args.filename))
     except Exception:
         from mako import exceptions
         print exceptions.text_error_template().render()
 def main(argv=None):
     """
     Main execution function for cli
     :param argv:
     """
     if argv is None:
         argv = sys.argv
     return _run(argv)
 if __name__ == '__main__':
     sys.exit(main(sys.argv))

kallithea/bin/kallithea_gist.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.bin.kallithea_gist
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Gist CLI client for Kallithea
 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: May 9, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import os
 import sys
 import stat
 import argparse
 import fileinput
 from kallithea.bin.base import json, api_call, RcConf, FORMAT_JSON, FORMAT_PRETTY
 def argparser(argv):
     usage = (
       "kallithea-gist [-h] [--format=FORMAT] [--apikey=APIKEY] [--apihost=APIHOST] "
       "[--config=CONFIG] [--save-config] [GIST OPTIONS] "
       "[filename or stdin use - for terminal stdin ]\n"
       "Create config file: kallithea-gist --apikey=<key> --apihost=http://your.kallithea.server --save-config"
+    )
     parser = argparse.ArgumentParser(description='Kallithea Gist cli',
                                      usage=usage)
     ## config
     group = parser.add_argument_group('config')
     group.add_argument('--apikey', help='api access key')
     group.add_argument('--apihost', help='api host')
     group.add_argument('--config', help='config file path DEFAULT: ~/.config/kallithea')
     group.add_argument('--save-config', action='store_true',
                        help='save the given config into a file')
     group = parser.add_argument_group('GIST')
     group.add_argument('-p', '--private', action='store_true',
                        help='create private Gist')
     group.add_argument('-f', '--filename',
                        help='set uploaded gist filename, '
                             'also defines syntax highlighting')
     group.add_argument('-d', '--description', help='Gist description')
     group.add_argument('-l', '--lifetime', metavar='MINUTES',
                        help='gist lifetime in minutes, -1 (DEFAULT) is forever')
     group.add_argument('--format', dest='format', type=str,
                        help='output format DEFAULT: `%s` can '
                        'be also `%s`' % (FORMAT_PRETTY, FORMAT_JSON),
             default=FORMAT_PRETTY
+    )
     args, other = parser.parse_known_args()
     return parser, args, other
 def _run(argv):
     conf = None
     parser, args, other = argparser(argv)
     api_credentials_given = (args.apikey and args.apihost)
     if args.save_config:
         if not api_credentials_given:
             raise parser.error('--save-config requires --apikey and --apihost')
         conf = RcConf(config_location=args.config,
                       autocreate=True, config={'apikey': args.apikey,
                                                'apihost': args.apihost})
         sys.exit()
     if not conf:
         conf = RcConf(config_location=args.config, autoload=True)
         if not conf:
             if not api_credentials_given:
                 parser.error('Could not find config file and missing '
                              '--apikey or --apihost in params')
     apikey = args.apikey or conf['apikey']
     host = args.apihost or conf['apihost']
     DEFAULT_FILENAME = 'gistfile1.txt'
     if other:
         # skip multifiles for now
         filename = other[0]
         if filename == '-':
             filename = DEFAULT_FILENAME
             gist_content = ''
             for line in fileinput.input('-'):
                 gist_content += line
         else:
             with open(filename, 'rb') as f:
                 gist_content = f.read()
     else:
         filename = DEFAULT_FILENAME
         gist_content = None
         # little bit hacky but cross platform check where the
         # stdin comes from we skip the terminal case it can be handled by '-'
         mode = os.fstat(0).st_mode
         if stat.S_ISFIFO(mode):
             # "stdin is piped"
             gist_content = sys.stdin.read()
         elif stat.S_ISREG(mode):
             # "stdin is redirected"
             gist_content = sys.stdin.read()
         else:
             # "stdin is terminal"
             pass
     # make sure we don't upload binary stuff
     if gist_content and '\0' in gist_content:
         raise Exception('Error: binary files upload is not possible')
     filename = os.path.basename(args.filename or filename)
     if gist_content:
         files = {
             filename: {
                 'content': gist_content,
                 'lexer': None
+            }
+        }
         margs = dict(
             lifetime=args.lifetime,
             description=args.description,
             gist_type='private' if args.private else 'public',
             files=files
+        )
         json_data = api_call(apikey, host, 'create_gist', **margs)['result']
         if args.format == FORMAT_JSON:
             print json.dumps(json_data)
         elif args.format == FORMAT_PRETTY:
             print json_data
             print 'Created %s gist %s' % (json_data['gist']['type'],
                                           json_data['gist']['url'])
     return 0
 def main(argv=None):
     """
     Main execution function for cli
     :param argv:
     """
     if argv is None:
         argv = sys.argv
     try:
         return _run(argv)
     except Exception as e:
         print e
         return 1
 if __name__ == '__main__':
     sys.exit(main(sys.argv))

kallithea/config/routing.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/>.
 """
 Routes configuration
 The more specific and detailed routes should be defined first so they
 may take precedent over the more generic routes. For more information
 refer to the routes manual at http://routes.groovie.org/docs/
 """
 from __future__ import with_statement
 from routes import Mapper
 # prefix for non repository related links needs to be prefixed with `/`
 ADMIN_PREFIX = '/_admin'
 def make_map(config):
     """Create, configure and return the routes Mapper"""
     rmap = Mapper(directory=config['pylons.paths']['controllers'],
                   always_scan=config['debug'])
     rmap.minimization = False
     rmap.explicit = False
     from kallithea.lib.utils import (is_valid_repo, is_valid_repo_group,
                                      get_repo_by_id)
     def check_repo(environ, match_dict):
         """
         check for valid repository for proper 404 handling
         :param environ:
         :param match_dict:
         """
         repo_name = match_dict.get('repo_name')
         if match_dict.get('f_path'):
             #fix for multiple initial slashes that causes errors
             match_dict['f_path'] = match_dict['f_path'].lstrip('/')
         by_id_match = get_repo_by_id(repo_name)
         if by_id_match:
             repo_name = by_id_match
             match_dict['repo_name'] = repo_name
         return is_valid_repo(repo_name, config['base_path'])
     def check_group(environ, match_dict):
         """
         check for valid repository group for proper 404 handling
         :param environ:
         :param match_dict:
         """
         repo_group_name = match_dict.get('group_name')
         return is_valid_repo_group(repo_group_name, config['base_path'])
     def check_group_skip_path(environ, match_dict):
         """
         check for valid repository group for proper 404 handling, but skips
         verification of existing path
         :param environ:
         :param match_dict:
         """
         repo_group_name = match_dict.get('group_name')
         return is_valid_repo_group(repo_group_name, config['base_path'],
                                    skip_path_check=True)
     def check_user_group(environ, match_dict):
         """
         check for valid user group for proper 404 handling
         :param environ:
         :param match_dict:
         """
         return True
     def check_int(environ, match_dict):
         return match_dict.get('id').isdigit()
     # The ErrorController route (handles 404/500 error pages); it should
     # likely stay at the top, ensuring it can always be resolved
     rmap.connect('/error/{action}', controller='error')
     rmap.connect('/error/{action}/{id}', controller='error')
     #==========================================================================
     # CUSTOM ROUTES HERE
     #==========================================================================
     #MAIN PAGE
     rmap.connect('home', '/', controller='home', action='index')
     rmap.connect('about', '/about', controller='home', action='about')
     rmap.connect('repo_switcher_data', '/_repos', controller='home',
                  action='repo_switcher_data')
     rmap.connect('rst_help',
                  "http://docutils.sourceforge.net/docs/user/rst/quickref.html",
                  _static=True)
     rmap.connect('kallithea_project_url', "https://kallithea-scm.org/", _static=True)
     rmap.connect('issues_url', 'https://bitbucket.org/conservancy/kallithea/issues', _static=True)
     #ADMIN REPOSITORY ROUTES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/repos') as m:
         m.connect("repos", "/repos",
                   action="create", conditions=dict(method=["POST"]))
         m.connect("repos", "/repos",
                   action="index", conditions=dict(method=["GET"]))
         m.connect("new_repo", "/create_repository",
                   action="create_repository", conditions=dict(method=["GET"]))
         m.connect("put_repo", "/repos/{repo_name:.*?}",
                   action="update", conditions=dict(method=["PUT"],
                   function=check_repo))
         m.connect("delete_repo", "/repos/{repo_name:.*?}",
                   action="delete", conditions=dict(method=["DELETE"],
                   ))
     #ADMIN REPOSITORY GROUPS ROUTES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/repo_groups') as m:
         m.connect("repos_groups", "/repo_groups",
                   action="create", conditions=dict(method=["POST"]))
         m.connect("repos_groups", "/repo_groups",
                   action="index", conditions=dict(method=["GET"]))
         m.connect("new_repos_group", "/repo_groups/new",
                   action="new", conditions=dict(method=["GET"]))
         m.connect("update_repos_group", "/repo_groups/{group_name:.*?}",
                   action="update", conditions=dict(method=["PUT"],
                                                    function=check_group))
         m.connect("repos_group", "/repo_groups/{group_name:.*?}",
                   action="show", conditions=dict(method=["GET"],
                                                  function=check_group))
         #EXTRAS REPO GROUP ROUTES
         m.connect("edit_repo_group", "/repo_groups/{group_name:.*?}/edit",
                   action="edit",
                   conditions=dict(method=["GET"], function=check_group))
         m.connect("edit_repo_group", "/repo_groups/{group_name:.*?}/edit",
                   action="edit",
                   conditions=dict(method=["PUT"], function=check_group))
         m.connect("edit_repo_group_advanced", "/repo_groups/{group_name:.*?}/edit/advanced",
                   action="edit_repo_group_advanced",
                   conditions=dict(method=["GET"], function=check_group))
         m.connect("edit_repo_group_advanced", "/repo_groups/{group_name:.*?}/edit/advanced",
                   action="edit_repo_group_advanced",
                   conditions=dict(method=["PUT"], function=check_group))
         m.connect("edit_repo_group_perms", "/repo_groups/{group_name:.*?}/edit/permissions",
                   action="edit_repo_group_perms",
                   conditions=dict(method=["GET"], function=check_group))
         m.connect("edit_repo_group_perms", "/repo_groups/{group_name:.*?}/edit/permissions",
                   action="update_perms",
                   conditions=dict(method=["PUT"], function=check_group))
         m.connect("edit_repo_group_perms", "/repo_groups/{group_name:.*?}/edit/permissions",
                   action="delete_perms",
                   conditions=dict(method=["DELETE"], function=check_group))
         m.connect("delete_repo_group", "/repo_groups/{group_name:.*?}",
                   action="delete", conditions=dict(method=["DELETE"],
                                                    function=check_group_skip_path))
     #ADMIN USER ROUTES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/users') as m:
         m.connect("users", "/users",
                   action="create", conditions=dict(method=["POST"]))
         m.connect("users", "/users",
                   action="index", conditions=dict(method=["GET"]))
         m.connect("formatted_users", "/users.{format}",
                   action="index", conditions=dict(method=["GET"]))
         m.connect("new_user", "/users/new",
                   action="new", conditions=dict(method=["GET"]))
         m.connect("update_user", "/users/{id}",
                   action="update", conditions=dict(method=["PUT"]))
         m.connect("delete_user", "/users/{id}",
                   action="delete", conditions=dict(method=["DELETE"]))
         m.connect("edit_user", "/users/{id}/edit",
                   action="edit", conditions=dict(method=["GET"]))
         m.connect("user", "/users/{id}",
                   action="show", conditions=dict(method=["GET"]))
         #EXTRAS USER ROUTES
         m.connect("edit_user_advanced", "/users/{id}/edit/advanced",
                   action="edit_advanced", conditions=dict(method=["GET"]))
         m.connect("edit_user_advanced", "/users/{id}/edit/advanced",
                   action="update_advanced", conditions=dict(method=["PUT"]))
         m.connect("edit_user_api_keys", "/users/{id}/edit/api_keys",
                   action="edit_api_keys", conditions=dict(method=["GET"]))
         m.connect("edit_user_api_keys", "/users/{id}/edit/api_keys",
                   action="add_api_key", conditions=dict(method=["POST"]))
         m.connect("edit_user_api_keys", "/users/{id}/edit/api_keys",
                   action="delete_api_key", conditions=dict(method=["DELETE"]))
         m.connect("edit_user_perms", "/users/{id}/edit/permissions",
                   action="edit_perms", conditions=dict(method=["GET"]))
         m.connect("edit_user_perms", "/users/{id}/edit/permissions",
                   action="update_perms", conditions=dict(method=["PUT"]))
         m.connect("edit_user_emails", "/users/{id}/edit/emails",
                   action="edit_emails", conditions=dict(method=["GET"]))
         m.connect("edit_user_emails", "/users/{id}/edit/emails",
                   action="add_email", conditions=dict(method=["PUT"]))
         m.connect("edit_user_emails", "/users/{id}/edit/emails",
                   action="delete_email", conditions=dict(method=["DELETE"]))
         m.connect("edit_user_ips", "/users/{id}/edit/ips",
                   action="edit_ips", conditions=dict(method=["GET"]))
         m.connect("edit_user_ips", "/users/{id}/edit/ips",
                   action="add_ip", conditions=dict(method=["PUT"]))
         m.connect("edit_user_ips", "/users/{id}/edit/ips",
                   action="delete_ip", conditions=dict(method=["DELETE"]))
     #ADMIN USER GROUPS REST ROUTES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/user_groups') as m:
         m.connect("users_groups", "/user_groups",
                   action="create", conditions=dict(method=["POST"]))
         m.connect("users_groups", "/user_groups",
                   action="index", conditions=dict(method=["GET"]))
         m.connect("new_users_group", "/user_groups/new",
                   action="new", conditions=dict(method=["GET"]))
         m.connect("update_users_group", "/user_groups/{id}",
                   action="update", conditions=dict(method=["PUT"]))
         m.connect("delete_users_group", "/user_groups/{id}",
                   action="delete", conditions=dict(method=["DELETE"]))
         m.connect("edit_users_group", "/user_groups/{id}/edit",
                   action="edit", conditions=dict(method=["GET"]),
                   function=check_user_group)
         m.connect("users_group", "/user_groups/{id}",
                   action="show", conditions=dict(method=["GET"]))
         #EXTRAS USER GROUP ROUTES
         m.connect("edit_user_group_default_perms", "/user_groups/{id}/edit/default_perms",
                   action="edit_default_perms", conditions=dict(method=["GET"]))
         m.connect("edit_user_group_default_perms", "/user_groups/{id}/edit/default_perms",
                   action="update_default_perms", conditions=dict(method=["PUT"]))
         m.connect("edit_user_group_perms", "/user_groups/{id}/edit/perms",
                   action="edit_perms", conditions=dict(method=["GET"]))
         m.connect("edit_user_group_perms", "/user_groups/{id}/edit/perms",
                   action="update_perms", conditions=dict(method=["PUT"]))
         m.connect("edit_user_group_perms", "/user_groups/{id}/edit/perms",
                   action="delete_perms", conditions=dict(method=["DELETE"]))
         m.connect("edit_user_group_advanced", "/user_groups/{id}/edit/advanced",
                   action="edit_advanced", conditions=dict(method=["GET"]))
         m.connect("edit_user_group_members", "/user_groups/{id}/edit/members",
                   action="edit_members", conditions=dict(method=["GET"]))
     #ADMIN PERMISSIONS ROUTES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/permissions') as m:
         m.connect("admin_permissions", "/permissions",
                   action="permission_globals", conditions=dict(method=["POST"]))
         m.connect("admin_permissions", "/permissions",
                   action="permission_globals", conditions=dict(method=["GET"]))
         m.connect("admin_permissions_ips", "/permissions/ips",
                   action="permission_ips", conditions=dict(method=["POST"]))
         m.connect("admin_permissions_ips", "/permissions/ips",
                   action="permission_ips", conditions=dict(method=["GET"]))
         m.connect("admin_permissions_perms", "/permissions/perms",
                   action="permission_perms", conditions=dict(method=["POST"]))
         m.connect("admin_permissions_perms", "/permissions/perms",
                   action="permission_perms", conditions=dict(method=["GET"]))
     #ADMIN DEFAULTS REST ROUTES
     rmap.resource('default', 'defaults',
                   controller='admin/defaults', path_prefix=ADMIN_PREFIX)
     #ADMIN AUTH SETTINGS
     rmap.connect('auth_settings', '%s/auth' % ADMIN_PREFIX,
                  controller='admin/auth_settings', action='auth_settings',
                  conditions=dict(method=["POST"]))
     rmap.connect('auth_home', '%s/auth' % ADMIN_PREFIX,
                  controller='admin/auth_settings')
     #ADMIN SETTINGS ROUTES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/settings') as m:
         m.connect("admin_settings", "/settings",
                   action="settings_vcs", conditions=dict(method=["POST"]))
         m.connect("admin_settings", "/settings",
                   action="settings_vcs", conditions=dict(method=["GET"]))
         m.connect("admin_settings_mapping", "/settings/mapping",
                   action="settings_mapping", conditions=dict(method=["POST"]))
         m.connect("admin_settings_mapping", "/settings/mapping",
                   action="settings_mapping", conditions=dict(method=["GET"]))
         m.connect("admin_settings_global", "/settings/global",
                   action="settings_global", conditions=dict(method=["POST"]))
         m.connect("admin_settings_global", "/settings/global",
                   action="settings_global", conditions=dict(method=["GET"]))
         m.connect("admin_settings_visual", "/settings/visual",
                   action="settings_visual", conditions=dict(method=["POST"]))
         m.connect("admin_settings_visual", "/settings/visual",
                   action="settings_visual", conditions=dict(method=["GET"]))
         m.connect("admin_settings_email", "/settings/email",
                   action="settings_email", conditions=dict(method=["POST"]))
         m.connect("admin_settings_email", "/settings/email",
                   action="settings_email", conditions=dict(method=["GET"]))
         m.connect("admin_settings_hooks", "/settings/hooks",
                   action="settings_hooks", conditions=dict(method=["POST"]))
         m.connect("admin_settings_hooks", "/settings/hooks",
                   action="settings_hooks", conditions=dict(method=["DELETE"]))
         m.connect("admin_settings_hooks", "/settings/hooks",
                   action="settings_hooks", conditions=dict(method=["GET"]))
         m.connect("admin_settings_search", "/settings/search",
                   action="settings_search", conditions=dict(method=["POST"]))
         m.connect("admin_settings_search", "/settings/search",
                   action="settings_search", conditions=dict(method=["GET"]))
         m.connect("admin_settings_system", "/settings/system",
                   action="settings_system", conditions=dict(method=["POST"]))
         m.connect("admin_settings_system", "/settings/system",
                   action="settings_system", conditions=dict(method=["GET"]))
         m.connect("admin_settings_system_update", "/settings/system/updates",
                   action="settings_system_update", conditions=dict(method=["GET"]))
     #ADMIN MY ACCOUNT
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/my_account') as m:
         m.connect("my_account", "/my_account",
                   action="my_account", conditions=dict(method=["GET"]))
         m.connect("my_account", "/my_account",
                   action="my_account", conditions=dict(method=["POST"]))
         m.connect("my_account_password", "/my_account/password",
                   action="my_account_password", conditions=dict(method=["GET"]))
         m.connect("my_account_password", "/my_account/password",
                   action="my_account_password", conditions=dict(method=["POST"]))
         m.connect("my_account_repos", "/my_account/repos",
                   action="my_account_repos", conditions=dict(method=["GET"]))
         m.connect("my_account_watched", "/my_account/watched",
                   action="my_account_watched", conditions=dict(method=["GET"]))
         m.connect("my_account_perms", "/my_account/perms",
                   action="my_account_perms", conditions=dict(method=["GET"]))
         m.connect("my_account_emails", "/my_account/emails",
                   action="my_account_emails", conditions=dict(method=["GET"]))
         m.connect("my_account_emails", "/my_account/emails",
                   action="my_account_emails_add", conditions=dict(method=["POST"]))
         m.connect("my_account_emails", "/my_account/emails",
                   action="my_account_emails_delete", conditions=dict(method=["DELETE"]))
         m.connect("my_account_api_keys", "/my_account/api_keys",
                   action="my_account_api_keys", conditions=dict(method=["GET"]))
         m.connect("my_account_api_keys", "/my_account/api_keys",
                   action="my_account_api_keys_add", conditions=dict(method=["POST"]))
         m.connect("my_account_api_keys", "/my_account/api_keys",
                   action="my_account_api_keys_delete", conditions=dict(method=["DELETE"]))
     #NOTIFICATION REST ROUTES
     with rmap.submapper(path_prefix=ADMIN_PREFIX,
                         controller='admin/notifications') as m:
         m.connect("notifications", "/notifications",
                   action="create", conditions=dict(method=["POST"]))
         m.connect("notifications", "/notifications",
                   action="index", conditions=dict(method=["GET"]))
         m.connect("notifications_mark_all_read", "/notifications/mark_all_read",
                   action="mark_all_read", conditions=dict(method=["GET"]))
         m.connect("formatted_notifications", "/notifications.{format}",
                   action="index", conditions=dict(method=["GET"]))
         m.connect("new_notification", "/notifications/new",
                   action="new", conditions=dict(method=["GET"]))
         m.connect("formatted_new_notification", "/notifications/new.{format}",
                   action="new", conditions=dict(method=["GET"]))
         m.connect("/notifications/{notification_id}",
                   action="update", conditions=dict(method=["PUT"]))
         m.connect("/notifications/{notification_id}",
                   action="delete", conditions=dict(method=["DELETE"]))
         m.connect("edit_notification", "/notifications/{notification_id}/edit",
                   action="edit", conditions=dict(method=["GET"]))
         m.connect("formatted_edit_notification",
                   "/notifications/{notification_id}.{format}/edit",

kallithea/controllers/files.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.controllers.files
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Files controller for Kallithea
 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: Apr 21, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import os
 import logging
 import traceback
 import tempfile
 import shutil
 from pylons import request, response, tmpl_context as c, url
 from pylons.i18n.translation import _
 from pylons.controllers.util import redirect
 from kallithea.lib.utils import jsonify, action_logger
 from kallithea.lib import diffs
 from kallithea.lib import helpers as h
 from kallithea.lib.compat import OrderedDict
 from kallithea.lib.utils2 import convert_line_endings, detect_mode, safe_str,\
     str2bool
 from kallithea.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator
 from kallithea.lib.base import BaseRepoController, render
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.lib.vcs.conf import settings
 from kallithea.lib.vcs.exceptions import RepositoryError, \
     ChangesetDoesNotExistError, EmptyRepositoryError, \
     ImproperArchiveTypeError, VCSError, NodeAlreadyExistsError,\
     NodeDoesNotExistError, ChangesetError, NodeError
 from kallithea.lib.vcs.nodes import FileNode
 from kallithea.model.repo import RepoModel
 from kallithea.model.scm import ScmModel
 from kallithea.model.db import Repository
 from kallithea.controllers.changeset import anchor_url, _ignorews_url,\
     _context_url, get_line_ctx, get_ignore_ws
 from webob.exc import HTTPNotFound
 from kallithea.lib.exceptions import NonRelativePathError
 log = logging.getLogger(__name__)
 class FilesController(BaseRepoController):
     def __before__(self):
         super(FilesController, self).__before__()
         c.cut_off_limit = self.cut_off_limit
     def __get_cs(self, rev, silent_empty=False):
         """
         Safe way to get changeset if error occur it redirects to tip with
         proper message
         :param rev: revision to fetch
         :silent_empty: return None if repository is empty
         """
         try:
             return c.db_repo_scm_instance.get_changeset(rev)
         except EmptyRepositoryError as e:
             if silent_empty:
                 return None
             url_ = url('files_add_home',
                        repo_name=c.repo_name,
                        revision=0, f_path='', anchor='edit')
             add_new = h.link_to(_('Click here to add new file'), url_, class_="alert-link")
             h.flash(h.literal(_('There are no files yet. %s') % add_new),
                     category='warning')
             raise HTTPNotFound()
         except(ChangesetDoesNotExistError, LookupError), e:
             msg = _('Such revision does not exist for this repository')
             h.flash(msg, category='error')
             raise HTTPNotFound()
         except RepositoryError as e:
             h.flash(safe_str(e), category='error')
             raise HTTPNotFound()
     def __get_filenode(self, cs, path):
         """
         Returns file_node or raise HTTP error.
         :param cs: given changeset
         :param path: path to lookup
         """
         try:
             file_node = cs.get_node(path)
             if file_node.is_dir():
                 raise RepositoryError('given path is a directory')
         except(ChangesetDoesNotExistError,), e:
             msg = _('Such revision does not exist for this repository')
             h.flash(msg, category='error')
             raise HTTPNotFound()
         except RepositoryError as e:
             h.flash(safe_str(e), category='error')
             raise HTTPNotFound()
         return file_node
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def index(self, repo_name, revision, f_path, annotate=False):
         # redirect to given revision from form if given
         post_revision = request.POST.get('at_rev', None)
         if post_revision:
             cs = self.__get_cs(post_revision) # FIXME - unused!
         c.revision = revision
         c.changeset = self.__get_cs(revision)
         c.branch = request.GET.get('branch', None)
         c.f_path = f_path
         c.annotate = annotate
         cur_rev = c.changeset.revision
         # prev link
         try:
             prev_rev = c.db_repo_scm_instance.get_changeset(cur_rev).prev(c.branch)
             c.url_prev = url('files_home', repo_name=c.repo_name,
                          revision=prev_rev.raw_id, f_path=f_path)
             if c.branch:
                 c.url_prev += '?branch=%s' % c.branch
         except (ChangesetDoesNotExistError, VCSError):
             c.url_prev = '#'
         # next link
         try:
             next_rev = c.db_repo_scm_instance.get_changeset(cur_rev).next(c.branch)
             c.url_next = url('files_home', repo_name=c.repo_name,
                      revision=next_rev.raw_id, f_path=f_path)
             if c.branch:
                 c.url_next += '?branch=%s' % c.branch
         except (ChangesetDoesNotExistError, VCSError):
             c.url_next = '#'
         # files or dirs
         try:
             c.file = c.changeset.get_node(f_path)
             if c.file.is_file():
                 c.load_full_history = False
                 file_last_cs = c.file.last_changeset
                 c.file_changeset = (c.changeset
                                     if c.changeset.revision < file_last_cs.revision
                                     else file_last_cs)
                 #determine if we're on branch head
                 _branches = c.db_repo_scm_instance.branches
                 c.on_branch_head = revision in _branches.keys() + _branches.values()
                 _hist = []
                 c.file_history = []
                 if c.load_full_history:
                     c.file_history, _hist = self._get_node_history(c.changeset, f_path)
                 c.authors = []
                 for a in set([x.author for x in _hist]):
                     c.authors.append((h.email(a), h.person(a)))
             else:
                 c.authors = c.file_history = []
         except RepositoryError as e:
             h.flash(safe_str(e), category='error')
             raise HTTPNotFound()
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             return render('files/files_ypjax.html')
         # TODO: tags and bookmarks?
         c.revision_options = [(c.changeset.raw_id,
                               _('%s at %s') % (c.changeset.branch, h.short_id(c.changeset.raw_id)))] + \
             [(n, b) for b, n in c.db_repo_scm_instance.branches.items()]
         if c.db_repo_scm_instance.closed_branches:
             prefix = _('(closed)') + ' '
             c.revision_options += [('-', '-')] + \
                 [(n, prefix + b) for b, n in c.db_repo_scm_instance.closed_branches.items()]
         return render('files/files.html')
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     @jsonify
     def history(self, repo_name, revision, f_path):
         changeset = self.__get_cs(revision)
         f_path = f_path
         _file = changeset.get_node(f_path)
         if _file.is_file():
             file_history, _hist = self._get_node_history(changeset, f_path)
             res = []
             for obj in file_history:
                 res.append({
                     'text': obj[1],
                     'children': [{'id': o[0], 'text': o[1]} for o in obj[0]]
                 })
             data = {
                 'more': False,
                 'results': res
+            }
             return data
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def authors(self, repo_name, revision, f_path):
         changeset = self.__get_cs(revision)
         f_path = f_path
         _file = changeset.get_node(f_path)
         if _file.is_file():
             file_history, _hist = self._get_node_history(changeset, f_path)
             c.authors = []
             for a in set([x.author for x in _hist]):
                 c.authors.append((h.email(a), h.person(a)))
             return render('files/files_history_box.html')
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def rawfile(self, repo_name, revision, f_path):
         cs = self.__get_cs(revision)
         file_node = self.__get_filenode(cs, f_path)
         response.content_disposition = 'attachment; filename=%s' % \
             safe_str(f_path.split(Repository.url_sep())[-1])
         response.content_type = file_node.mimetype
         return file_node.content
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def raw(self, repo_name, revision, f_path):
         cs = self.__get_cs(revision)
         file_node = self.__get_filenode(cs, f_path)
         raw_mimetype_mapping = {
             # map original mimetype to a mimetype used for "show as raw"
             # you can also provide a content-disposition to override the
             # default "attachment" disposition.
             # orig_type: (new_type, new_dispo)
             # show images inline:
             'image/x-icon': ('image/x-icon', 'inline'),
             'image/png': ('image/png', 'inline'),
             'image/gif': ('image/gif', 'inline'),
             'image/jpeg': ('image/jpeg', 'inline'),
             'image/svg+xml': ('image/svg+xml', 'inline'),
+        }
         mimetype = file_node.mimetype
         try:
             mimetype, dispo = raw_mimetype_mapping[mimetype]
         except KeyError:
             # we don't know anything special about this, handle it safely
             if file_node.is_binary:
                 # do same as download raw for binary files
                 mimetype, dispo = 'application/octet-stream', 'attachment'
             else:
                 # do not just use the original mimetype, but force text/plain,
                 # otherwise it would serve text/html and that might be unsafe.
                 # Note: underlying vcs library fakes text/plain mimetype if the
                 # mimetype can not be determined and it thinks it is not
                 # binary.This might lead to erroneous text display in some
                 # cases, but helps in other cases, like with text files
                 # without extension.
                 mimetype, dispo = 'text/plain', 'inline'
         if dispo == 'attachment':
             dispo = 'attachment; filename=%s' % \
                         safe_str(f_path.split(os.sep)[-1])
         response.content_disposition = dispo
         response.content_type = mimetype
         return file_node.content
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
     def delete(self, repo_name, revision, f_path):
         repo = c.db_repo
         if repo.enable_locking and repo.locked[0]:
             h.flash(_('This repository has been locked by %s on %s')
                 % (h.person_by_id(repo.locked[0]),
                    h.fmt_date(h.time_to_datetime(repo.locked[1]))),
                 'warning')
             return redirect(h.url('files_home',
                                   repo_name=repo_name, revision='tip'))
         # check if revision is a branch identifier- basically we cannot
         # create multiple heads via file editing
         _branches = repo.scm_instance.branches
         # check if revision is a branch name or branch hash
         if revision not in _branches.keys() + _branches.values():
             h.flash(_('You can only delete files with revision '
                       'being a valid branch '), category='warning')
             return redirect(h.url('files_home',
                                   repo_name=repo_name, revision='tip',
                                   f_path=f_path))
         r_post = request.POST
         c.cs = self.__get_cs(revision)
         c.file = self.__get_filenode(c.cs, f_path)
         c.default_message = _('Deleted file %s via Kallithea') % (f_path)
         c.f_path = f_path
         node_path = f_path
         author = self.authuser.full_contact
         if r_post:
             message = r_post.get('message') or c.default_message
             try:
                 nodes = {
                     node_path: {
                         'content': ''
+                    }
+                }
                 self.scm_model.delete_nodes(
                     user=c.authuser.user_id, repo=c.db_repo,
                     message=message,
                     nodes=nodes,
                     parent_cs=c.cs,
                     author=author,
+                )
                 h.flash(_('Successfully deleted file %s') % f_path,
                         category='success')
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred during commit'), category='error')
             return redirect(url('changeset_home',
                                 repo_name=c.repo_name, revision='tip'))
         return render('files/files_delete.html')
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
     def edit(self, repo_name, revision, f_path):
         repo = c.db_repo
         if repo.enable_locking and repo.locked[0]:
             h.flash(_('This repository has been locked by %s on %s')
                 % (h.person_by_id(repo.locked[0]),
                    h.fmt_date(h.time_to_datetime(repo.locked[1]))),
                 'warning')
             return redirect(h.url('files_home',
                                   repo_name=repo_name, revision='tip'))
         # check if revision is a branch identifier- basically we cannot
         # create multiple heads via file editing
         _branches = repo.scm_instance.branches
         # check if revision is a branch name or branch hash
         if revision not in _branches.keys() + _branches.values():
             h.flash(_('You can only edit files with revision '
                       'being a valid branch '), category='warning')
             return redirect(h.url('files_home',
                                   repo_name=repo_name, revision='tip',
                                   f_path=f_path))
         r_post = request.POST
         c.cs = self.__get_cs(revision)
         c.file = self.__get_filenode(c.cs, f_path)
         if c.file.is_binary:
             return redirect(url('files_home', repo_name=c.repo_name,
                             revision=c.cs.raw_id, f_path=f_path))
         c.default_message = _('Edited file %s via Kallithea') % (f_path)
         c.f_path = f_path
         if r_post:
             old_content = c.file.content
             sl = old_content.splitlines(1)
             first_line = sl[0] if sl else ''
             # modes:  0 - Unix, 1 - Mac, 2 - DOS
             mode = detect_mode(first_line, 0)
             content = convert_line_endings(r_post.get('content', ''), mode)
             message = r_post.get('message') or c.default_message
             author = self.authuser.full_contact
             if content == old_content:
                 h.flash(_('No changes'), category='warning')
                 return redirect(url('changeset_home', repo_name=c.repo_name,
                                     revision='tip'))
             try:
                 self.scm_model.commit_change(repo=c.db_repo_scm_instance,

kallithea/lib/auth.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.auth
 ~~~~~~~~~~~~~~~~~~
 authentication and permission libraries
 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: Apr 4, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import time
 import os
 import logging
 import traceback
 import hashlib
 import itertools
 import collections
 from decorator import decorator
 from pylons import url, request
 from pylons.controllers.util import abort, redirect
 from pylons.i18n.translation import _
 from webhelpers.pylonslib import secure_form
 from sqlalchemy import or_
 from sqlalchemy.orm.exc import ObjectDeletedError
 from sqlalchemy.orm import joinedload
 from kallithea import __platform__, is_windows, is_unix
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 from kallithea.model import meta
 from kallithea.model.meta import Session
 from kallithea.model.user import UserModel
 from kallithea.model.db import User, Repository, Permission, \
     UserToPerm, UserGroupRepoToPerm, UserGroupToPerm, UserGroupMember, \
     RepoGroup, UserGroupRepoGroupToPerm, UserIpMap, UserGroupUserGroupToPerm, \
     UserGroup, UserApiKeys
 from kallithea.lib.utils2 import safe_unicode, aslist
 from kallithea.lib.utils import get_repo_slug, get_repo_group_slug, \
     get_user_group_slug, conditional_cache
 from kallithea.lib.caching_query import FromCache
 log = logging.getLogger(__name__)
 class PasswordGenerator(object):
     """
     This is a simple class for generating password from different sets of
     characters
     usage::
         passwd_gen = PasswordGenerator()
         #print 8-letter password containing only big and small letters
             of alphabet
         passwd_gen.gen_password(8, passwd_gen.ALPHABETS_BIG_SMALL)
     """
     ALPHABETS_NUM = r'''1234567890'''
     ALPHABETS_SMALL = r'''qwertyuiopasdfghjklzxcvbnm'''
     ALPHABETS_BIG = r'''QWERTYUIOPASDFGHJKLZXCVBNM'''
     ALPHABETS_SPECIAL = r'''`-=[]\;',./~!@#$%^&*()_+{}|:"<>?'''
     ALPHABETS_FULL = ALPHABETS_BIG + ALPHABETS_SMALL \
         + ALPHABETS_NUM + ALPHABETS_SPECIAL
     ALPHABETS_ALPHANUM = ALPHABETS_BIG + ALPHABETS_SMALL + ALPHABETS_NUM
     ALPHABETS_BIG_SMALL = ALPHABETS_BIG + ALPHABETS_SMALL
     ALPHABETS_ALPHANUM_BIG = ALPHABETS_BIG + ALPHABETS_NUM
     ALPHABETS_ALPHANUM_SMALL = ALPHABETS_SMALL + ALPHABETS_NUM
     def gen_password(self, length, alphabet=ALPHABETS_FULL):
         assert len(alphabet) <= 256, alphabet
         l = []
         while len(l) < length:
             i = ord(os.urandom(1))
             if i < len(alphabet):
                 l.append(alphabet[i])
         return ''.join(l)
 class KallitheaCrypto(object):
     @classmethod
     def hash_string(cls, str_):
         """
         Cryptographic function used for password hashing based on pybcrypt
         or Python's own OpenSSL wrapper on windows
         :param password: password to hash
         """
         if is_windows:
             return hashlib.sha256(str_).hexdigest()
         elif is_unix:
             import bcrypt
             return bcrypt.hashpw(str_, bcrypt.gensalt(10))
         else:
             raise Exception('Unknown or unsupported platform %s' \
                             % __platform__)
     @classmethod
     def hash_check(cls, password, hashed):
         """
         Checks matching password with it's hashed value, runs different
         implementation based on platform it runs on
         :param password: password
         :param hashed: password in hashed form
         """
         if is_windows:
             return hashlib.sha256(password).hexdigest() == hashed
         elif is_unix:
             import bcrypt
             return bcrypt.hashpw(password, hashed) == hashed
         else:
             raise Exception('Unknown or unsupported platform %s' \
                             % __platform__)
 def get_crypt_password(password):
     return KallitheaCrypto.hash_string(password)
 def check_password(password, hashed):
     return KallitheaCrypto.hash_check(password, hashed)
 def _cached_perms_data(user_id, user_is_admin, user_inherit_default_permissions,
                        explicit, algo):
     RK = 'repositories'
     GK = 'repositories_groups'
     UK = 'user_groups'
     GLOBAL = 'global'
     PERM_WEIGHTS = Permission.PERM_WEIGHTS
     permissions = {RK: {}, GK: {}, UK: {}, GLOBAL: set()}
     def _choose_perm(new_perm, cur_perm):
         new_perm_val = PERM_WEIGHTS[new_perm]
         cur_perm_val = PERM_WEIGHTS[cur_perm]
         if algo == 'higherwin':
             if new_perm_val > cur_perm_val:
                 return new_perm
             return cur_perm
         elif algo == 'lowerwin':
             if new_perm_val < cur_perm_val:
                 return new_perm
             return cur_perm
     #======================================================================
     # fetch default permissions
     #======================================================================
     default_user = User.get_by_username('default', cache=True)
     default_user_id = default_user.user_id
     default_repo_perms = Permission.get_default_perms(default_user_id)
     default_repo_groups_perms = Permission.get_default_group_perms(default_user_id)
     default_user_group_perms = Permission.get_default_user_group_perms(default_user_id)
     if user_is_admin:
         #==================================================================
         # admin users have all rights;
         # based on default permissions, just set everything to admin
         #==================================================================
         permissions[GLOBAL].add('hg.admin')
         permissions[GLOBAL].add('hg.create.write_on_repogroup.true')
         # repositories
         for perm in default_repo_perms:
             r_k = perm.UserRepoToPerm.repository.repo_name
             p = 'repository.admin'
             permissions[RK][r_k] = p
         # repository groups
         for perm in default_repo_groups_perms:
             rg_k = perm.UserRepoGroupToPerm.group.group_name
             p = 'group.admin'
             permissions[GK][rg_k] = p
         # user groups
         for perm in default_user_group_perms:
             u_k = perm.UserUserGroupToPerm.user_group.users_group_name
             p = 'usergroup.admin'
             permissions[UK][u_k] = p
         return permissions
     #==================================================================
     # SET DEFAULTS GLOBAL, REPOS, REPOSITORY GROUPS
     #==================================================================
     # default global permissions taken from the default user
     default_global_perms = UserToPerm.query()\
         .filter(UserToPerm.user_id == default_user_id)\
         .options(joinedload(UserToPerm.permission))
     for perm in default_global_perms:
         permissions[GLOBAL].add(perm.permission.permission_name)
     # defaults for repositories, taken from default user
     for perm in default_repo_perms:
         r_k = perm.UserRepoToPerm.repository.repo_name
         if perm.Repository.private and not (perm.Repository.user_id == user_id):
             # disable defaults for private repos,
             p = 'repository.none'
         elif perm.Repository.user_id == user_id:
             # set admin if owner
             p = 'repository.admin'
         else:
             p = perm.Permission.permission_name
         permissions[RK][r_k] = p
     # defaults for repository groups taken from default user permission
     # on given group
     for perm in default_repo_groups_perms:
         rg_k = perm.UserRepoGroupToPerm.group.group_name
         p = perm.Permission.permission_name
         permissions[GK][rg_k] = p
     # defaults for user groups taken from default user permission
     # on given user group
     for perm in default_user_group_perms:
         u_k = perm.UserUserGroupToPerm.user_group.users_group_name
         p = perm.Permission.permission_name
         permissions[UK][u_k] = p
     #======================================================================
     # !! OVERRIDE GLOBALS !! with user permissions if any found
     #======================================================================
     # those can be configured from groups or users explicitly
     _configurable = set([
         'hg.fork.none', 'hg.fork.repository',
         'hg.create.none', 'hg.create.repository',
         'hg.usergroup.create.false', 'hg.usergroup.create.true'
     ])
     # USER GROUPS comes first
     # user group global permissions
     user_perms_from_users_groups = Session().query(UserGroupToPerm)\
         .options(joinedload(UserGroupToPerm.permission))\
         .join((UserGroupMember, UserGroupToPerm.users_group_id ==
                UserGroupMember.users_group_id))\
         .filter(UserGroupMember.user_id == user_id)\
         .join((UserGroup, UserGroupMember.users_group_id ==
                UserGroup.users_group_id))\
         .filter(UserGroup.users_group_active == True)\
         .order_by(UserGroupToPerm.users_group_id)\
         .all()
     # need to group here by groups since user can be in more than
     # one group
     _grouped = [[x, list(y)] for x, y in
                 itertools.groupby(user_perms_from_users_groups,
                                   lambda x:x.users_group)]
     for gr, perms in _grouped:
         # since user can be in multiple groups iterate over them and
         # select the lowest permissions first (more explicit)
         ##TODO: do this^^
         if not gr.inherit_default_permissions:
             # NEED TO IGNORE all configurable permissions and
             # replace them with explicitly set
             permissions[GLOBAL] = permissions[GLOBAL]\
                                             .difference(_configurable)
         for perm in perms:
             permissions[GLOBAL].add(perm.permission.permission_name)
     # user specific global permissions
     user_perms = Session().query(UserToPerm)\
             .options(joinedload(UserToPerm.permission))\
             .filter(UserToPerm.user_id == user_id).all()
     if not user_inherit_default_permissions:
         # NEED TO IGNORE all configurable permissions and
         # replace them with explicitly set
         permissions[GLOBAL] = permissions[GLOBAL]\
                                         .difference(_configurable)
         for perm in user_perms:
             permissions[GLOBAL].add(perm.permission.permission_name)
     ## END GLOBAL PERMISSIONS
     #======================================================================
     # !! PERMISSIONS FOR REPOSITORIES !!
     #======================================================================
     #======================================================================
     # check if user is part of user groups for this repository and
     # fill in his permission from it. _choose_perm decides of which
     # permission should be selected based on selected method
     #======================================================================
     # user group for repositories permissions
     user_repo_perms_from_users_groups = \
      Session().query(UserGroupRepoToPerm, Permission, Repository,)\
         .join((Repository, UserGroupRepoToPerm.repository_id ==
                Repository.repo_id))\
         .join((Permission, UserGroupRepoToPerm.permission_id ==
                Permission.permission_id))\
         .join((UserGroup, UserGroupRepoToPerm.users_group_id ==
                UserGroup.users_group_id))\
         .filter(UserGroup.users_group_active == True)\
         .join((UserGroupMember, UserGroupRepoToPerm.users_group_id ==
                UserGroupMember.users_group_id))\
         .filter(UserGroupMember.user_id == user_id)\
         .all()
     multiple_counter = collections.defaultdict(int)
     for perm in user_repo_perms_from_users_groups:
         r_k = perm.UserGroupRepoToPerm.repository.repo_name
         multiple_counter[r_k] += 1
         p = perm.Permission.permission_name
         cur_perm = permissions[RK][r_k]
         if perm.Repository.user_id == user_id:
             # set admin if owner
             p = 'repository.admin'
         else:
             if multiple_counter[r_k] > 1:
                 p = _choose_perm(p, cur_perm)
         permissions[RK][r_k] = p
     # user explicit permissions for repositories, overrides any specified
     # by the group permission
     user_repo_perms = Permission.get_default_perms(user_id)
     for perm in user_repo_perms:
         r_k = perm.UserRepoToPerm.repository.repo_name
         cur_perm = permissions[RK][r_k]
         # set admin if owner
         if perm.Repository.user_id == user_id:
             p = 'repository.admin'
         else:
             p = perm.Permission.permission_name
             if not explicit:
                 p = _choose_perm(p, cur_perm)
         permissions[RK][r_k] = p
     #======================================================================
     # !! PERMISSIONS FOR REPOSITORY GROUPS !!
     #======================================================================
     #======================================================================
     # check if user is part of user groups for this repository groups and
     # fill in his permission from it. _choose_perm decides of which
     # permission should be selected based on selected method
     #======================================================================
     # user group for repo groups permissions
     user_repo_group_perms_from_users_groups = \
      Session().query(UserGroupRepoGroupToPerm, Permission, RepoGroup)\
      .join((RepoGroup, UserGroupRepoGroupToPerm.group_id == RepoGroup.group_id))\
      .join((Permission, UserGroupRepoGroupToPerm.permission_id
             == Permission.permission_id))\
      .join((UserGroup, UserGroupRepoGroupToPerm.users_group_id ==
             UserGroup.users_group_id))\
      .filter(UserGroup.users_group_active == True)\
      .join((UserGroupMember, UserGroupRepoGroupToPerm.users_group_id
             == UserGroupMember.users_group_id))\
      .filter(UserGroupMember.user_id == user_id)\
      .all()
     multiple_counter = collections.defaultdict(int)
     for perm in user_repo_group_perms_from_users_groups:
         g_k = perm.UserGroupRepoGroupToPerm.group.group_name
         multiple_counter[g_k] += 1
         p = perm.Permission.permission_name
         cur_perm = permissions[GK][g_k]
         if multiple_counter[g_k] > 1:
             p = _choose_perm(p, cur_perm)
         permissions[GK][g_k] = p
     # user explicit permissions for repository groups
     user_repo_groups_perms = Permission.get_default_group_perms(user_id)
     for perm in user_repo_groups_perms:
         rg_k = perm.UserRepoGroupToPerm.group.group_name
         p = perm.Permission.permission_name
         cur_perm = permissions[GK][rg_k]
         if not explicit:
             p = _choose_perm(p, cur_perm)
         permissions[GK][rg_k] = p
     #======================================================================
     # !! PERMISSIONS FOR USER GROUPS !!
     #======================================================================
     # user group for user group permissions
     user_group_user_groups_perms = \
      Session().query(UserGroupUserGroupToPerm, Permission, UserGroup)\
      .join((UserGroup, UserGroupUserGroupToPerm.target_user_group_id
             == UserGroup.users_group_id))\
      .join((Permission, UserGroupUserGroupToPerm.permission_id
             == Permission.permission_id))\
      .join((UserGroupMember, UserGroupUserGroupToPerm.user_group_id
             == UserGroupMember.users_group_id))\
      .filter(UserGroupMember.user_id == user_id)\
      .join((UserGroup, UserGroupMember.users_group_id ==
             UserGroup.users_group_id), aliased=True, from_joinpoint=True)\
      .filter(UserGroup.users_group_active == True)\
      .all()
     multiple_counter = collections.defaultdict(int)

kallithea/lib/indexers/daemon.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.indexers.daemon
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 A daemon will read from task table and run tasks
 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 26, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import os
 import sys
 import logging
 import traceback
 from shutil import rmtree
 from time import mktime
 from os.path import dirname as dn
 from os.path import join as jn
 # Add location of top level folder to sys.path
 project_path = dn(dn(dn(dn(os.path.realpath(__file__)))))
 sys.path.append(project_path)
 from kallithea.config.conf import INDEX_EXTENSIONS
 from kallithea.model.scm import ScmModel
 from kallithea.model.db import Repository
 from kallithea.lib.utils2 import safe_unicode, safe_str
 from kallithea.lib.indexers import SCHEMA, IDX_NAME, CHGSETS_SCHEMA, \
     CHGSET_IDX_NAME
 from kallithea.lib.vcs.exceptions import ChangesetError, RepositoryError, \
     NodeDoesNotExistError
 from whoosh.index import create_in, open_dir, exists_in
 from whoosh.query import *
 from whoosh.qparser import QueryParser
 log = logging.getLogger('whoosh_indexer')
 class WhooshIndexingDaemon(object):
     """
     Daemon for atomic indexing jobs
     """
     def __init__(self, indexname=IDX_NAME, index_location=None,
                  repo_location=None, sa=None, repo_list=None,
                  repo_update_list=None):
         self.indexname = indexname
         self.index_location = index_location
         if not index_location:
             raise Exception('You have to provide index location')
         self.repo_location = repo_location
         if not repo_location:
             raise Exception('You have to provide repositories location')
         self.repo_paths = ScmModel(sa).repo_scan(self.repo_location)
         #filter repo list
         if repo_list:
             #Fix non-ascii repo names to unicode
             repo_list = map(safe_unicode, repo_list)
             self.filtered_repo_paths = {}
             for repo_name, repo in self.repo_paths.items():
                 if repo_name in repo_list:
                     self.filtered_repo_paths[repo_name] = repo
             self.repo_paths = self.filtered_repo_paths
         #filter update repo list
         self.filtered_repo_update_paths = {}
         if repo_update_list:
             self.filtered_repo_update_paths = {}
             for repo_name, repo in self.repo_paths.items():
                 if repo_name in repo_update_list:
                     self.filtered_repo_update_paths[repo_name] = repo
             self.repo_paths = self.filtered_repo_update_paths
         self.initial = True
         if not os.path.isdir(self.index_location):
             os.makedirs(self.index_location)
             log.info('Cannot run incremental index since it does not '
                      'yet exist running full build')
         elif not exists_in(self.index_location, IDX_NAME):
             log.info('Running full index build as the file content '
                      'index does not exist')
         elif not exists_in(self.index_location, CHGSET_IDX_NAME):
             log.info('Running full index build as the changeset '
                      'index does not exist')
         else:
             self.initial = False
     def _get_index_revision(self, repo):
         db_repo = Repository.get_by_repo_name(repo.name_unicode)
         landing_rev = 'tip'
         if db_repo:
             _rev_type, _rev = db_repo.landing_rev
             landing_rev = _rev
         return landing_rev
     def _get_index_changeset(self, repo, index_rev=None):
         if not index_rev:
             index_rev = self._get_index_revision(repo)
         cs = repo.get_changeset(index_rev)
         return cs
     def get_paths(self, repo):
         """
         recursive walk in root dir and return a set of all path in that dir
         based on repository walk function
         """
         index_paths_ = set()
         try:
             cs = self._get_index_changeset(repo)
             for _topnode, _dirs, files in cs.walk('/'):
                 for f in files:
                     index_paths_.add(jn(safe_str(repo.path), safe_str(f.path)))
         except RepositoryError:
             log.debug(traceback.format_exc())
             pass
         return index_paths_
     def get_node(self, repo, path, index_rev=None):
         """
         gets a filenode based on given full path. It operates on string for
         hg git compatibility.
         :param repo: scm repo instance
         :param path: full path including root location
         :return: FileNode
         """
         # FIXME: paths should be normalized ... or even better: don't include repo.path
         path = safe_str(path)
         repo_path = safe_str(repo.path)
         assert path.startswith(repo_path)
         assert path[len(repo_path)] in (os.path.sep, os.path.altsep)
         node_path = path[len(repo_path) + 1:]
         cs = self._get_index_changeset(repo, index_rev=index_rev)
         node = cs.get_node(node_path)
         return node
     def get_node_mtime(self, node):
         return mktime(node.last_changeset.date.timetuple())
     def add_doc(self, writer, path, repo, repo_name, index_rev=None):
         """
         Adding doc to writer this function itself fetches data from
         the instance of vcs backend
         """
         node = self.get_node(repo, path, index_rev)
         indexed = indexed_w_content = 0
         # we just index the content of chosen files, and skip binary files
         if node.extension in INDEX_EXTENSIONS and not node.is_binary:
             u_content = node.content
             if not isinstance(u_content, unicode):
                 log.warning('  >> %s Could not get this content as unicode '
                             'replacing with empty content' % path)
                 u_content = u''
             else:
                 log.debug('    >> %s [WITH CONTENT]', path)
                 indexed_w_content += 1
         else:
             log.debug('    >> %s', path)
             # just index file name without it's content
             u_content = u''
             indexed += 1
         p = safe_unicode(path)
         writer.add_document(
             fileid=p,
             owner=unicode(repo.contact),
             repository=safe_unicode(repo_name),
             path=p,
             content=u_content,
             modtime=self.get_node_mtime(node),
             extension=node.extension
+        )
         return indexed, indexed_w_content
     def index_changesets(self, writer, repo_name, repo, start_rev=None):
         """
         Add all changeset in the vcs repo starting at start_rev
         to the index writer
         :param writer: the whoosh index writer to add to
         :param repo_name: name of the repository from whence the
           changeset originates including the repository group
         :param repo: the vcs repository instance to index changesets for,
           the presumption is the repo has changesets to index
         :param start_rev=None: the full sha id to start indexing from
           if start_rev is None then index from the first changeset in
           the repo
         """
         if start_rev is None:
             start_rev = repo[0].raw_id
         log.debug('indexing changesets in %s starting at rev: %s',
                   repo_name, start_rev)
         indexed = 0
         cs_iter = repo.get_changesets(start=start_rev)
         total = len(cs_iter)
         for cs in cs_iter:
             log.debug('    >> %s/%s', cs, total)
             writer.add_document(
                 raw_id=unicode(cs.raw_id),
                 owner=unicode(repo.contact),
                 date=cs._timestamp,
                 repository=safe_unicode(repo_name),
                 author=cs.author,
                 message=cs.message,
                 last=cs.last,
                 added=u' '.join([safe_unicode(node.path) for node in cs.added]).lower(),
                 removed=u' '.join([safe_unicode(node.path) for node in cs.removed]).lower(),
                 changed=u' '.join([safe_unicode(node.path) for node in cs.changed]).lower(),
                 parents=u' '.join([cs.raw_id for cs in cs.parents]),
+            )
             indexed += 1
         log.debug('indexed %d changesets for repo %s', indexed, repo_name)
         return indexed
     def index_files(self, file_idx_writer, repo_name, repo):
         """
         Index files for given repo_name
         :param file_idx_writer: the whoosh index writer to add to
         :param repo_name: name of the repository we're indexing
         :param repo: instance of vcs repo
         """
         i_cnt = iwc_cnt = 0
         log.debug('building index for %s @revision:%s', repo.path,
                                                 self._get_index_revision(repo))
         index_rev = self._get_index_revision(repo)
         for idx_path in self.get_paths(repo):
             i, iwc = self.add_doc(file_idx_writer, idx_path, repo, repo_name, index_rev)
             i_cnt += i
             iwc_cnt += iwc
         log.debug('added %s files %s with content for repo %s',
                   i_cnt + iwc_cnt, iwc_cnt, repo.path)
         return i_cnt, iwc_cnt
     def update_changeset_index(self):
         idx = open_dir(self.index_location, indexname=CHGSET_IDX_NAME)
         with idx.searcher() as searcher:
             writer = idx.writer()
             writer_is_dirty = False
             try:
                 indexed_total = 0
                 repo_name = None
                 for repo_name, repo in self.repo_paths.items():
                     # skip indexing if there aren't any revs in the repo
                     num_of_revs = len(repo)
                     if num_of_revs < 1:
                         continue
                     qp = QueryParser('repository', schema=CHGSETS_SCHEMA)
                     q = qp.parse(u"last:t AND %s" % repo_name)
                     results = searcher.search(q)
                     # default to scanning the entire repo
                     last_rev = 0
                     start_id = None
                     if len(results) > 0:
                         # assuming that there is only one result, if not this
                         # may require a full re-index.
                         start_id = results[0]['raw_id']
                         last_rev = repo.get_changeset(revision=start_id).revision
                     # there are new changesets to index or a new repo to index
                     if last_rev == 0 or num_of_revs > last_rev + 1:
                         # delete the docs in the index for the previous
                         # last changeset(s)
                         for hit in results:
                             q = qp.parse(u"last:t AND %s AND raw_id:%s" %
                                             (repo_name, hit['raw_id']))
                             writer.delete_by_query(q)
                         # index from the previous last changeset + all new ones
                         indexed_total += self.index_changesets(writer,
                                                 repo_name, repo, start_id)
                         writer_is_dirty = True
                 log.debug('indexed %s changesets for repo %s',
                           indexed_total, repo_name
+                )
             finally:
                 if writer_is_dirty:
                     log.debug('>> COMMITING CHANGES TO CHANGESET INDEX<<')
                     writer.commit(merge=True)
                     log.debug('>>> FINISHED REBUILDING CHANGESET INDEX <<<')
                 else:
                     log.debug('>> NOTHING TO COMMIT TO CHANGESET INDEX<<')
     def update_file_index(self):
         log.debug((u'STARTING INCREMENTAL INDEXING UPDATE FOR EXTENSIONS %s '
                    'AND REPOS %s') % (INDEX_EXTENSIONS, self.repo_paths.keys()))
         idx = open_dir(self.index_location, indexname=self.indexname)
         # The set of all paths in the index
         indexed_paths = set()
         # The set of all paths we need to re-index
         to_index = set()
         writer = idx.writer()
         writer_is_dirty = False
         try:
             with idx.reader() as reader:
                 # Loop over the stored fields in the index
                 for fields in reader.all_stored_fields():
                     indexed_path = fields['path']
                     indexed_repo_path = fields['repository']
                     indexed_paths.add(indexed_path)
                     if not indexed_repo_path in self.filtered_repo_update_paths:
                         continue
                     repo = self.repo_paths[indexed_repo_path]
                     try:
                         node = self.get_node(repo, indexed_path)
                         # Check if this file was changed since it was indexed
                         indexed_time = fields['modtime']
                         mtime = self.get_node_mtime(node)
                         if mtime > indexed_time:
                             # The file has changed, delete it and add it to
                             # the list of files to reindex
                             log.debug(
                                 'adding to reindex list %s mtime: %s vs %s',
                                     indexed_path, mtime, indexed_time
+                            )
                             writer.delete_by_term('fileid', indexed_path)
                             writer_is_dirty = True
                             to_index.add(indexed_path)
                     except (ChangesetError, NodeDoesNotExistError):
                         # This file was deleted since it was indexed
                         log.debug('removing from index %s', indexed_path)
                         writer.delete_by_term('path', indexed_path)
                         writer_is_dirty = True
             # Loop over the files in the filesystem
             # Assume we have a function that gathers the filenames of the
             # documents to be indexed
             ri_cnt_total = 0  # indexed
             riwc_cnt_total = 0  # indexed with content
             for repo_name, repo in self.repo_paths.items():
                 # skip indexing if there aren't any revisions
                 if len(repo) < 1:
                     continue
                 ri_cnt = 0   # indexed
                 riwc_cnt = 0  # indexed with content
                 for path in self.get_paths(repo):
                     path = safe_unicode(path)
                     if path in to_index or path not in indexed_paths:
                         # This is either a file that's changed, or a new file
                         # that wasn't indexed before. So index it!
                         i, iwc = self.add_doc(writer, path, repo, repo_name)
                         writer_is_dirty = True
                         log.debug('re indexing %s', path)
                         ri_cnt += i
                         ri_cnt_total += 1
                         riwc_cnt += iwc
                         riwc_cnt_total += iwc
                 log.debug('added %s files %s with content for repo %s',
                              ri_cnt + riwc_cnt, riwc_cnt, repo.path
+                )
             log.debug('indexed %s files in total and %s with content',
                         ri_cnt_total, riwc_cnt_total
+            )
         finally:
             if writer_is_dirty:
                 log.debug('>> COMMITING CHANGES TO FILE INDEX <<')
                 writer.commit(merge=True)
                 log.debug('>>> FINISHED REBUILDING FILE INDEX <<<')
             else:
                 log.debug('>> NOTHING TO COMMIT TO FILE INDEX <<')
                 writer.cancel()
     def build_indexes(self):

kallithea/lib/paster_commands/cache_keys.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.paster_commands.cache_keys
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 cleanup-keys paster command for Kallithea
 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: mar 27, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import os
 import sys
 import logging
 from kallithea.model.meta import Session
 from kallithea.lib.utils import BasePasterCommand
 from kallithea.model.db import CacheInvalidation
 # Add location of top level folder to sys.path
 from os.path import dirname as dn
 rc_path = dn(dn(dn(os.path.realpath(__file__))))
 sys.path.append(rc_path)
 log = logging.getLogger(__name__)
 class Command(BasePasterCommand):
     max_args = 1
     min_args = 1
     usage = "CONFIG_FILE"
     group_name = "Kallithea"
     takes_config_file = -1
     parser = BasePasterCommand.standard_parser(verbose=True)
     summary = "Cache keys utils"
     def command(self):
         #get SqlAlchemy session
         self._init_session()
         _caches = CacheInvalidation.query().order_by(CacheInvalidation.cache_key).all()
         if self.options.show:
             for c_obj in _caches:
                 print 'key:%s active:%s' % (c_obj.cache_key, c_obj.cache_active)
         elif self.options.cleanup:
             for c_obj in _caches:
                 Session().delete(c_obj)
                 print 'removing key:%s' % (c_obj.cache_key)
                 Session().commit()
         else:
             print 'nothing done exiting...'
         sys.exit(0)
     def update_parser(self):
         self.parser.add_option(
             '--show',
             action='store_true',
             dest='show',
             help=("show existing cache keys with together with status")
+        )
         self.parser.add_option(
             '--cleanup',
             action="store_true",
             dest="cleanup",
             help="cleanup existing cache keys"
+        )

kallithea/lib/paster_commands/cleanup.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.paster_commands.cleanup
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 cleanup-repos paster command for Kallithea
 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: Jul 14, 2012
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import os
 import sys
 import re
 import shutil
 import logging
 import datetime
 from kallithea.lib.utils import BasePasterCommand, ask_ok, REMOVED_REPO_PAT
 from kallithea.lib.utils2 import safe_str
 from kallithea.model.db import Ui
 # Add location of top level folder to sys.path
 from os.path import dirname as dn
 rc_path = dn(dn(dn(os.path.realpath(__file__))))
 sys.path.append(rc_path)
 log = logging.getLogger(__name__)
 class Command(BasePasterCommand):
     max_args = 1
     min_args = 1
     usage = "CONFIG_FILE"
     group_name = "Kallithea"
     takes_config_file = -1
     parser = BasePasterCommand.standard_parser(verbose=True)
     summary = "Cleanup deleted repos"
     def _parse_older_than(self, val):
         regex = re.compile(r'((?P<days>\d+?)d)?((?P<hours>\d+?)h)?((?P<minutes>\d+?)m)?((?P<seconds>\d+?)s)?')
         parts = regex.match(val)
         if not parts:
             return
         parts = parts.groupdict()
         time_params = {}
         for (name, param) in parts.iteritems():
             if param:
                 time_params[name] = int(param)
         return datetime.timedelta(**time_params)
     def _extract_date(self, name):
         """
         Extract the date part from rm__<date> pattern of removed repos,
         and convert it to datetime object
         :param name:
         """
         date_part = name[4:19]  # 4:19 since we don't parse milisecods
         return datetime.datetime.strptime(date_part, '%Y%m%d_%H%M%S')
     def command(self):
         #get SqlAlchemy session
         self._init_session()
         repos_location = Ui.get_repos_location()
         to_remove = []
         for dn_, dirs, f in os.walk(safe_str(repos_location)):
             alldirs = list(dirs)
             del dirs[:]
             if ('.hg' in alldirs or
                 'objects' in alldirs and ('refs' in alldirs or 'packed-refs' in f)):
                 continue
             for loc in alldirs:
                 if REMOVED_REPO_PAT.match(loc):
                     to_remove.append([os.path.join(dn_, loc),
                                       self._extract_date(loc)])
                 else:
                     dirs.append(loc)
         #filter older than (if present)!
         now = datetime.datetime.now()
         older_than = self.options.older_than
         if older_than:
             to_remove_filtered = []
             older_than_date = self._parse_older_than(older_than)
             for name, date_ in to_remove:
                 repo_age = now - date_
                 if repo_age > older_than_date:
                     to_remove_filtered.append([name, date_])
             to_remove = to_remove_filtered
             print >> sys.stdout, 'removing %s deleted repos older than %s (%s)' \
                 % (len(to_remove), older_than, older_than_date)
         else:
             print >> sys.stdout, 'removing all [%s] deleted repos' \
                 % len(to_remove)
         if self.options.dont_ask or not to_remove:
             # don't ask just remove !
             remove = True
         else:
             remove = ask_ok('the following repositories will be deleted completely:\n%s\n'
                             'are you sure you want to remove them [y/n]?'
                             % ', \n'.join(['%s removed on %s'
                     % (safe_str(x[0]), safe_str(x[1])) for x in to_remove]))
         if remove:
             for path, date_ in to_remove:
                 print >> sys.stdout, 'removing repository %s' % path
                 shutil.rmtree(path)
         else:
             print 'nothing done exiting...'
             sys.exit(0)
     def update_parser(self):
         self.parser.add_option(
             '--older-than',
             action='store',
             dest='older_than',
             help=("only remove repos that have been removed "
                  "at least given time ago. "
                  "The default is to remove all removed repositories. "
                  "Possible suffixes: "
                  "d (days), h (hours), m (minutes), s (seconds). "
                  "For example --older-than=30d deletes repositories "
                  "removed more than 30 days ago.")
+            )
         self.parser.add_option(
             '--dont-ask',
             action="store_true",
             dest="dont_ask",
             help="remove repositories without asking for confirmation."
+        )

kallithea/lib/paster_commands/ishell.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.paster_commands.ishell
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 interactive shell paster command for Kallithea
 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: Apr 4, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import os
 import sys
 import logging
 from kallithea.lib.utils import BasePasterCommand
 # Add location of top level folder to sys.path
 from os.path import dirname as dn
 rc_path = dn(dn(dn(os.path.realpath(__file__))))
 sys.path.append(rc_path)
 log = logging.getLogger(__name__)
 class Command(BasePasterCommand):
     max_args = 1
     min_args = 1
     usage = "CONFIG_FILE"
     group_name = "Kallithea"
     takes_config_file = -1
     parser = BasePasterCommand.standard_parser(verbose=True)
     summary = "Interactive shell"
     def command(self):
         #get SqlAlchemy session
         self._init_session()
         # imports, used in ipython shell
         import os
         import sys
         import time
         import shutil
         import datetime
         from kallithea.model.db import *
         try:
             from IPython import embed
             from IPython.config.loader import Config
             cfg = Config()
             cfg.InteractiveShellEmbed.confirm_exit = False
             embed(config=cfg, banner1="Kallithea IShell.")
         except ImportError:
             print 'ipython installation required for ishell'
             sys.exit(-1)
     def update_parser(self):
         pass

kallithea/lib/paster_commands/make_index.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.paster_commands.make_index
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 make-index paster command for Kallithea
 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: Aug 17, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import os
 import sys
 import logging
 from string import strip
 from kallithea.model.repo import RepoModel
 from kallithea.lib.utils import BasePasterCommand, load_rcextensions
 # Add location of top level folder to sys.path
 from os.path import dirname as dn
 rc_path = dn(dn(dn(os.path.realpath(__file__))))
 sys.path.append(rc_path)
 class Command(BasePasterCommand):
     max_args = 1
     min_args = 1
     usage = "CONFIG_FILE"
     group_name = "Kallithea"
     takes_config_file = -1
     parser = BasePasterCommand.standard_parser(verbose=True)
     summary = "Creates or updates full text search index"
     def command(self):
         logging.config.fileConfig(self.path_to_ini_file)
         #get SqlAlchemy session
         self._init_session()
         from pylons import config
         index_location = config['index_dir']
         load_rcextensions(config['here'])
         repo_location = self.options.repo_location \
             if self.options.repo_location else RepoModel().repos_path
         repo_list = map(strip, self.options.repo_list.split(',')) \
             if self.options.repo_list else None
         repo_update_list = map(strip, self.options.repo_update_list.split(',')) \
             if self.options.repo_update_list else None
         #======================================================================
         # WHOOSH DAEMON
         #======================================================================
         from kallithea.lib.pidlock import LockHeld, DaemonLock
         from kallithea.lib.indexers.daemon import WhooshIndexingDaemon
         try:
             l = DaemonLock(file_=os.path.join(dn(dn(index_location)),
                                               'make_index.lock'))
             WhooshIndexingDaemon(index_location=index_location,
                                  repo_location=repo_location,
                                  repo_list=repo_list,
                                  repo_update_list=repo_update_list)\
                 .run(full_index=self.options.full_index)
             l.release()
         except LockHeld:
             sys.exit(1)
     def update_parser(self):
         self.parser.add_option('--repo-location',
                           action='store',
                           dest='repo_location',
                           help="Specifies repositories location to index OPTIONAL",
+                          )
         self.parser.add_option('--index-only',
                           action='store',
                           dest='repo_list',
                           help="Specifies a comma separated list of repositories "
                                 "to build index on. If not given all repositories "
                                 "are scanned for indexing. OPTIONAL",
+                          )
         self.parser.add_option('--update-only',
                           action='store',
                           dest='repo_update_list',
                           help="Specifies a comma separated list of repositories "
                                 "to re-build index on. OPTIONAL",
+                          )
         self.parser.add_option('-f',
                           action='store_true',
                           dest='full_index',
                           help="Specifies that index should be made full i.e"
                                 " destroy old and build from scratch",
                           default=False)

kallithea/lib/paster_commands/make_rcextensions.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.paster_commands.make_rcextensions
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 make-rcext paster command for Kallithea
 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: Mar 6, 2012
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import os
 import sys
 import pkg_resources
 from kallithea.lib.utils import BasePasterCommand, ask_ok
 # Add location of top level folder to sys.path
 from os.path import dirname as dn
 rc_path = dn(dn(dn(os.path.realpath(__file__))))
 sys.path.append(rc_path)
 class Command(BasePasterCommand):
     max_args = 1
     min_args = 1
     group_name = "Kallithea"
     takes_config_file = -1
     parser = BasePasterCommand.standard_parser(verbose=True)
     summary = "Write template file for extending Kallithea in Python."
     usage = "CONFIG_FILE"
     description = '''\
         A rcextensions directory with a __init__.py file will be created next to
         the ini file. Local customizations in that file will survive upgrades.
         The file contains instructions on how it can be customized.
         '''
     def command(self):
         from pylons import config
         here = config['here']
         content = pkg_resources.resource_string(
             'kallithea', os.path.join('config', 'rcextensions', '__init__.py')
+        )
         ext_file = os.path.join(here, 'rcextensions', '__init__.py')
         if os.path.exists(ext_file):
             msg = ('Extension file already exists, do you want '
                    'to overwrite it ? [y/n]')
             if not ask_ok(msg):
                 print 'Nothing done...'
                 return
         dirname = os.path.dirname(ext_file)
         if not os.path.isdir(dirname):
             os.makedirs(dirname)
         with open(ext_file, 'wb') as f:
             f.write(content)
             print 'Wrote new extensions file to %s' % ext_file
     def update_parser(self):
         pass

kallithea/lib/paster_commands/repo_scan.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.paster_commands.repo_scan
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 repo-scan paster command for Kallithea
 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: Feb 9, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import os
 import sys
 import logging
 from kallithea.model.scm import ScmModel
 from kallithea.lib.utils import BasePasterCommand, repo2db_mapper
 # Add location of top level folder to sys.path
 from os.path import dirname as dn
 rc_path = dn(dn(dn(os.path.realpath(__file__))))
 sys.path.append(rc_path)
 log = logging.getLogger(__name__)
 class Command(BasePasterCommand):
     max_args = 1
     min_args = 1
     usage = "CONFIG_FILE"
     group_name = "Kallithea"
     takes_config_file = -1
     parser = BasePasterCommand.standard_parser(verbose=True)
     summary = "Rescan default location for new repositories"
     def command(self):
         #get SqlAlchemy session
         self._init_session()
         rm_obsolete = self.options.delete_obsolete
         log.info('Now scanning root location for new repos...')
         added, removed = repo2db_mapper(ScmModel().repo_scan(),
                                         remove_obsolete=rm_obsolete)
         added = ', '.join(added) or '-'
         removed = ', '.join(removed) or '-'
         log.info('Scan completed added: %s removed: %s', added, removed)
     def update_parser(self):
         self.parser.add_option(
             '--delete-obsolete',
             action='store_true',
             help="Use this flag do delete repositories that are "
                  "present in Kallithea database but not on the filesystem",
+        )

kallithea/lib/paster_commands/update_repoinfo.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.paster_commands.update_repoinfo
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 update-repoinfo paster command for Kallithea
 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: Jul 14, 2012
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import os
 import sys
 import logging
 import string
 from kallithea.lib.utils import BasePasterCommand
 from kallithea.model.db import Repository
 from kallithea.model.repo import RepoModel
 from kallithea.model.meta import Session
 # Add location of top level folder to sys.path
 from os.path import dirname as dn
 rc_path = dn(dn(dn(os.path.realpath(__file__))))
 sys.path.append(rc_path)
 log = logging.getLogger(__name__)
 class Command(BasePasterCommand):
     max_args = 1
     min_args = 1
     usage = "CONFIG_FILE"
     group_name = "Kallithea"
     takes_config_file = -1
     parser = BasePasterCommand.standard_parser(verbose=True)
     summary = "Updates repositories caches for last changeset"
     def command(self):
         #get SqlAlchemy session
         self._init_session()
         repo_update_list = map(string.strip,
                                self.options.repo_update_list.split(',')) \
                                if self.options.repo_update_list else None
         if repo_update_list:
             repo_list = list(Repository.query()\
                 .filter(Repository.repo_name.in_(repo_update_list)))
         else:
             repo_list = Repository.getAll()
         RepoModel.update_repoinfo(repositories=repo_list)
         Session().commit()
         if self.options.invalidate_cache:
             for r in repo_list:
                 r.set_invalidate()
         print 'Updated cache for %s repositories' % (len(repo_list))
     def update_parser(self):
         self.parser.add_option('--update-only',
                            action='store',
                            dest='repo_update_list',
                            help="Specifies a comma separated list of repositories "
                                 "to update last commit info for. OPTIONAL")
         self.parser.add_option('--invalidate-cache',
                            action='store_true',
                            dest='invalidate_cache',
                            help="Trigger cache invalidation event for repos. "
                                 "OPTIONAL")

kallithea/lib/pidlock.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/>.
 from __future__ import with_statement
 import os
 import errno
 from multiprocessing.util import Finalize
 from kallithea.lib.compat import kill
 class LockHeld(Exception):
     pass
 class DaemonLock(object):
     """daemon locking
     USAGE:
     try:
         l = DaemonLock(file_='/path/tolockfile',desc='test lock')
         main()
         l.release()
     except LockHeld:
         sys.exit(1)
     """
     def __init__(self, file_=None, callbackfn=None,
                  desc='daemon lock', debug=False):
         lock_name = os.path.join(os.path.dirname(__file__), 'running.lock')
         self.pidfile = file_ if file_ else lock_name
         self.callbackfn = callbackfn
         self.desc = desc
         self.debug = debug
         self.held = False
         #run the lock automatically !
         self.lock()
         self._finalize = Finalize(self, DaemonLock._on_finalize,
                                   args=(self, debug), exitpriority=10)
     @staticmethod
     def _on_finalize(lock, debug):
         if lock.held:
             if debug:
                 print 'lock held finalizing and running lock.release()'
             lock.release()
     def lock(self):
         """
         locking function, if lock is present it
         will raise LockHeld exception
         """
         lockname = str(os.getpid())
         if self.debug:
             print 'running lock'
         self.trylock()
         self.makelock(lockname, self.pidfile)
         return True
     def trylock(self):
         running_pid = False
         if self.debug:
             print 'checking for already running process'
         try:
             with open(self.pidfile, 'r') as f:
                 try:
                     running_pid = int(f.readline())
                 except ValueError:
                     running_pid = -1
             if self.debug:
                 print ('lock file present running_pid: %s, '
                        'checking for execution' % (running_pid,))
             # Now we check the PID from lock file matches to the current
             # process PID
             if running_pid:
                 try:
                     kill(running_pid, 0)
                 except OSError as exc:
                     if exc.errno in (errno.ESRCH, errno.EPERM):
                         print ("Lock File is there but"
                                " the program is not running")
                         print "Removing lock file for the: %s" % running_pid
                         self.release()
                     else:
                         raise
                 else:
                     print "You already have an instance of the program running"
                     print "It is running as process %s" % running_pid
                     raise LockHeld()
         except IOError as e:
             if e.errno != 2:
                 raise
     def release(self):
         """releases the pid by removing the pidfile
         """
         if self.debug:
             print 'trying to release the pidlock'
         if self.callbackfn:
             #execute callback function on release
             if self.debug:
                 print 'executing callback function %s' % self.callbackfn
             self.callbackfn()
         try:
             if self.debug:
                 print 'removing pidfile %s' % self.pidfile
             os.remove(self.pidfile)
             self.held = False
         except OSError as e:
             if self.debug:
                 print 'removing pidfile failed %s' % e
             pass
     def makelock(self, lockname, pidfile):
         """
         this function will make an actual lock
         :param lockname: actual pid of file
         :param pidfile: the file to write the pid in
         """
         if self.debug:
             print 'creating a file %s and pid: %s' % (pidfile, lockname)
         dir_, file_ = os.path.split(pidfile)
         if not os.path.isdir(dir_):
             os.makedirs(dir_)
         with open(self.pidfile, 'wb') as f:
             f.write(lockname)
         self.held = True

kallithea/lib/profiler.py

➞

Show inline comments

 from __future__ import with_statement
 import objgraph
 import cProfile
 import pstats
 import cgi
 import pprint
 import threading
 from StringIO import StringIO
 class ProfilingMiddleware(object):
     def __init__(self, app):
         self.lock = threading.Lock()
         self.app = app
     def __call__(self, environ, start_response):
         with self.lock:
             profiler = cProfile.Profile()
             def run_app(*a, **kw):
                 self.response = self.app(environ, start_response)
             profiler.runcall(run_app, environ, start_response)
             profiler.snapshot_stats()
             stats = pstats.Stats(profiler)
             stats.sort_stats('calls') #cumulative
             # Redirect output
             out = StringIO()
             stats.stream = out
             stats.print_stats()
             resp = ''.join(self.response)
             # Lets at least only put this on html-like responses.
             if resp.strip().startswith('<'):
                 ## The profiling info is just appended to the response.
                 ##  Browsers don't mind this.
                 resp += ('<pre style="text-align:left; '
                          'border-top: 4px dashed red; padding: 1em;">')
                 resp += cgi.escape(out.getvalue(), True)
                 ct = objgraph.show_most_common_types()
                 print ct
                 resp += ct if ct else '---'
                 output = StringIO()
                 pprint.pprint(environ, output, depth=3)
                 resp += cgi.escape(output.getvalue(), True)
                 resp += '</pre>'
             return resp

kallithea/model/api_key.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.model.api_key
 ~~~~~~~~~~~~~~~~~~~~~~~
 API key model for Kallithea
 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: Sep 8, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import time
 import logging
 from sqlalchemy import or_
 from kallithea.lib.utils2 import generate_api_key
 from kallithea.model import BaseModel
 from kallithea.model.db import UserApiKeys
 from kallithea.model.meta import Session
 log = logging.getLogger(__name__)
 class ApiKeyModel(BaseModel):
     cls = UserApiKeys
     def create(self, user, description, lifetime=-1):
         """
         :param user: user or user_id
         :param description: description of ApiKey
         :param lifetime: expiration time in seconds
         """
         user = self._get_user(user)
         new_api_key = UserApiKeys()
         new_api_key.api_key = generate_api_key()
         new_api_key.user_id = user.user_id
         new_api_key.description = description
         new_api_key.expires = time.time() + (lifetime * 60) if lifetime != -1 else -1
         Session().add(new_api_key)
         return new_api_key
     def delete(self, api_key, user=None):
         """
         Deletes given api_key, if user is set it also filters the object for
         deletion by given user.
         """
         api_key = UserApiKeys.query().filter(UserApiKeys.api_key == api_key)
         if user is not None:
             user = self._get_user(user)
             api_key = api_key.filter(UserApiKeys.user_id == user.user_id)
         api_key = api_key.scalar()
         Session().delete(api_key)
     def get_api_keys(self, user, show_expired=True):
         user = self._get_user(user)
         user_api_keys = UserApiKeys.query()\
             .filter(UserApiKeys.user_id == user.user_id)
         if not show_expired:
             user_api_keys = user_api_keys\
                 .filter(or_(UserApiKeys.expires == -1,
                             UserApiKeys.expires >= time.time()))
         return user_api_keys

kallithea/model/gist.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.model.gist
 ~~~~~~~~~~~~~~~~~~~~
 gist model for Kallithea
 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: May 9, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import os
 import time
 import logging
 import traceback
 import shutil
 from kallithea.lib.utils2 import safe_unicode, unique_id, safe_int, \
     time_to_datetime, AttributeDict
 from kallithea.lib.compat import json
 from kallithea.model import BaseModel
 from kallithea.model.db import Gist
 from kallithea.model.repo import RepoModel
 from kallithea.model.scm import ScmModel
 log = logging.getLogger(__name__)
 GIST_STORE_LOC = '.rc_gist_store'
 GIST_METADATA_FILE = '.rc_gist_metadata'
 class GistModel(BaseModel):
     cls = Gist
     def _get_gist(self, gist):
         """
         Helper method to get gist by ID, or gist_access_id as a fallback
         :param gist: GistID, gist_access_id, or Gist instance
         """
         return self._get_instance(Gist, gist, callback=Gist.get_by_access_id)
     def __delete_gist(self, gist):
         """
         removes gist from filesystem
         :param gist: gist object
         """
         root_path = RepoModel().repos_path
         rm_path = os.path.join(root_path, GIST_STORE_LOC, gist.gist_access_id)
         log.info("Removing %s", rm_path)
         shutil.rmtree(rm_path)
     def _store_metadata(self, repo, gist_id, gist_access_id, user_id, gist_type,
                         gist_expires):
         """
         store metadata inside the gist, this can be later used for imports
         or gist identification
         """
         metadata = {
             'metadata_version': '1',
             'gist_db_id': gist_id,
             'gist_access_id': gist_access_id,
             'gist_owner_id': user_id,
             'gist_type': gist_type,
             'gist_expires': gist_expires,
             'gist_updated': time.time(),
+        }
         with open(os.path.join(repo.path, '.hg', GIST_METADATA_FILE), 'wb') as f:
             f.write(json.dumps(metadata))
     def get_gist(self, gist):
         return self._get_gist(gist)
     def get_gist_files(self, gist_access_id, revision=None):
         """
         Get files for given gist
         :param gist_access_id:
         """
         repo = Gist.get_by_access_id(gist_access_id)
         cs = repo.scm_instance.get_changeset(revision)
         return cs, [n for n in cs.get_node('/')]
     def create(self, description, owner, gist_mapping,
                gist_type=Gist.GIST_PUBLIC, lifetime=-1):
         """
         :param description: description of the gist
         :param owner: user who created this gist
         :param gist_mapping: mapping {filename:{'content':content},...}
         :param gist_type: type of gist private/public
         :param lifetime: in minutes, -1 == forever
         """
         owner = self._get_user(owner)
         gist_id = safe_unicode(unique_id(20))
         lifetime = safe_int(lifetime, -1)
         gist_expires = time.time() + (lifetime * 60) if lifetime != -1 else -1
         log.debug('set GIST expiration date to: %s',
                   time_to_datetime(gist_expires)
                    if gist_expires != -1 else 'forever')
         #create the Database version
         gist = Gist()
         gist.gist_description = description
         gist.gist_access_id = gist_id
         gist.gist_owner = owner.user_id
         gist.gist_expires = gist_expires
         gist.gist_type = safe_unicode(gist_type)
         self.sa.add(gist)
         self.sa.flush()
         if gist_type == Gist.GIST_PUBLIC:
             # use DB ID for easy to use GIST ID
             gist_id = safe_unicode(gist.gist_id)
             gist.gist_access_id = gist_id
             self.sa.add(gist)
         gist_repo_path = os.path.join(GIST_STORE_LOC, gist_id)
         log.debug('Creating new %s GIST repo in %s', gist_type, gist_repo_path)
         repo = RepoModel()._create_filesystem_repo(
             repo_name=gist_id, repo_type='hg', repo_group=GIST_STORE_LOC)
         processed_mapping = {}
         for filename in gist_mapping:
             if filename != os.path.basename(filename):
                 raise Exception('Filename cannot be inside a directory')
             content = gist_mapping[filename]['content']
             #TODO: expand support for setting explicit lexers
 #             if lexer is None:
 #                 try:
 #                     guess_lexer = pygments.lexers.guess_lexer_for_filename
 #                     lexer = guess_lexer(filename,content)
 #                 except pygments.util.ClassNotFound:
 #                     lexer = 'text'
             processed_mapping[filename] = {'content': content}
         # now create single multifile commit
         message = 'added file'
         message += 's: ' if len(processed_mapping) > 1 else ': '
         message += ', '.join([x for x in processed_mapping])
         #fake Kallithea Repository object
         fake_repo = AttributeDict(dict(
             repo_name=gist_repo_path,
             scm_instance_no_cache=lambda: repo,
         ))
         ScmModel().create_nodes(
             user=owner.user_id, repo=fake_repo,
             message=message,
             nodes=processed_mapping,
             trigger_push_hook=False
+        )
         self._store_metadata(repo, gist.gist_id, gist.gist_access_id,
                              owner.user_id, gist.gist_type, gist.gist_expires)
         return gist
     def delete(self, gist, fs_remove=True):
         gist = self._get_gist(gist)
         try:
             self.sa.delete(gist)
             if fs_remove:
                 self.__delete_gist(gist)
             else:
                 log.debug('skipping removal from filesystem')
         except Exception:
             log.error(traceback.format_exc())
             raise
     def update(self, gist, description, owner, gist_mapping, gist_type,
                lifetime):
         gist = self._get_gist(gist)
         gist_repo = gist.scm_instance
         lifetime = safe_int(lifetime, -1)
         if lifetime == 0:  # preserve old value
             gist_expires = gist.gist_expires
         else:
             gist_expires = time.time() + (lifetime * 60) if lifetime != -1 else -1
         #calculate operation type based on given data
         gist_mapping_op = {}
         for k, v in gist_mapping.items():
             # add, mod, del
             if not v['org_filename'] and v['filename']:
                 op = 'add'
             elif v['org_filename'] and not v['filename']:
                 op = 'del'
             else:
                 op = 'mod'
             v['op'] = op
             gist_mapping_op[k] = v
         gist.gist_description = description
         gist.gist_expires = gist_expires
         gist.owner = owner
         gist.gist_type = gist_type
         self.sa.add(gist)
         self.sa.flush()
         message = 'updated file'
         message += 's: ' if len(gist_mapping) > 1 else ': '
         message += ', '.join([x for x in gist_mapping])
         #fake Kallithea Repository object
         fake_repo = AttributeDict(dict(
             repo_name=gist_repo.path,
             scm_instance_no_cache=lambda: gist_repo,
         ))
         self._store_metadata(gist_repo, gist.gist_id, gist.gist_access_id,
                              owner.user_id, gist.gist_type, gist.gist_expires)
         ScmModel().update_nodes(
             user=owner.user_id,
             repo=fake_repo,
             message=message,
             nodes=gist_mapping_op,
             trigger_push_hook=False
+        )
         return gist

kallithea/model/repo.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.model.repo
 ~~~~~~~~~~~~~~~~~~~~
 Repository model for kallithea
 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: Jun 5, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import os
 import shutil
 import logging
 import traceback
 from datetime import datetime
 from sqlalchemy.orm import subqueryload
 from kallithea.lib.utils import make_ui
 from kallithea.lib.vcs.backends import get_backend
 from kallithea.lib.compat import json
 from kallithea.lib.utils2 import LazyProperty, safe_str, safe_unicode, \
     remove_prefix, obfuscate_url_pw, get_current_authuser
 from kallithea.lib.caching_query import FromCache
 from kallithea.lib.hooks import log_delete_repository
 from kallithea.model import BaseModel
 from kallithea.model.db import Repository, UserRepoToPerm, UserGroupRepoToPerm, \
     UserRepoGroupToPerm, UserGroupRepoGroupToPerm, User, Permission, \
     Statistics, UserGroup, Ui, RepoGroup, RepositoryField
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import HasRepoPermissionAny, HasUserGroupPermissionAny
 from kallithea.lib.exceptions import AttachedForksError
 from kallithea.model.scm import UserGroupList
 log = logging.getLogger(__name__)
 class RepoModel(BaseModel):
     cls = Repository
     URL_SEPARATOR = Repository.url_sep()
     def _get_user_group(self, users_group):
         return self._get_instance(UserGroup, users_group,
                                   callback=UserGroup.get_by_group_name)
     def _get_repo_group(self, repo_group):
         return self._get_instance(RepoGroup, repo_group,
                                   callback=RepoGroup.get_by_group_name)
     def _create_default_perms(self, repository, private):
         # create default permission
         default = 'repository.read'
         def_user = User.get_default_user()
         for p in def_user.user_perms:
             if p.permission.permission_name.startswith('repository.'):
                 default = p.permission.permission_name
                 break
         default_perm = 'repository.none' if private else default
         repo_to_perm = UserRepoToPerm()
         repo_to_perm.permission = Permission.get_by_key(default_perm)
         repo_to_perm.repository = repository
         repo_to_perm.user_id = def_user.user_id
         return repo_to_perm
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
         q = self.sa.query(Ui).filter(Ui.ui_key == '/').one()
         return q.ui_value
     def get(self, repo_id, cache=False):
         repo = self.sa.query(Repository) \
             .filter(Repository.repo_id == repo_id)
         if cache:
             repo = repo.options(FromCache("sql_cache_short",
                                           "get_repo_%s" % repo_id))
         return repo.scalar()
     def get_repo(self, repository):
         return self._get_repo(repository)
     def get_by_repo_name(self, repo_name, cache=False):
         repo = self.sa.query(Repository) \
             .filter(Repository.repo_name == repo_name)
         if cache:
             repo = repo.options(FromCache("sql_cache_short",
                                           "get_repo_%s" % repo_name))
         return repo.scalar()
     def get_all_user_repos(self, user):
         """
         Gets all repositories that user have at least read access
         :param user:
         """
         from kallithea.lib.auth import AuthUser
         user = self._get_user(user)
         repos = AuthUser(user_id=user.user_id).permissions['repositories']
         access_check = lambda r: r[1] in ['repository.read',
                                           'repository.write',
                                           'repository.admin']
         repos = [x[0] for x in filter(access_check, repos.items())]
         return Repository.query().filter(Repository.repo_name.in_(repos))
     def get_users_js(self):
         users = self.sa.query(User).filter(User.active == True).all()
         return json.dumps([
+            {
                 'id': u.user_id,
                 'fname': h.escape(u.name),
                 'lname': h.escape(u.lastname),
                 'nname': u.username,
                 'gravatar_lnk': h.gravatar_url(u.email, size=28),
                 'gravatar_size': 14,
             } for u in users]
+        )
     def get_user_groups_js(self):
         user_groups = self.sa.query(UserGroup) \
             .filter(UserGroup.users_group_active == True) \
             .options(subqueryload(UserGroup.members)) \
             .all()
         user_groups = UserGroupList(user_groups, perm_set=['usergroup.read',
                                                            'usergroup.write',
                                                            'usergroup.admin'])
         return json.dumps([
+            {
                 'id': gr.users_group_id,
                 'grname': gr.users_group_name,
                 'grmembers': len(gr.members),
             } for gr in user_groups]
+        )
     @classmethod
     def _render_datatable(cls, tmpl, *args, **kwargs):
         import kallithea
         from pylons import tmpl_context as c
         from pylons.i18n.translation import _
         _tmpl_lookup = kallithea.CONFIG['pylons.app_globals'].mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         tmpl = template.get_def(tmpl)
         kwargs.update(dict(_=_, h=h, c=c))
         return tmpl.render(*args, **kwargs)
     @classmethod
     def update_repoinfo(cls, repositories=None):
         if not repositories:
             repositories = Repository.getAll()
         for repo in repositories:
             repo.update_changeset_cache()
     def get_repos_as_dict(self, repos_list=None, admin=False, perm_check=True,
                           super_user_actions=False):
         _render = self._render_datatable
         from pylons import tmpl_context as c
         def quick_menu(repo_name):
             return _render('quick_menu', repo_name)
         def repo_lnk(name, rtype, rstate, private, fork_of):
             return _render('repo_name', name, rtype, rstate, private, fork_of,
                            short_name=not admin, admin=False)
         def last_change(last_change):
             return _render("last_change", last_change)
         def rss_lnk(repo_name):
             return _render("rss", repo_name)
         def atom_lnk(repo_name):
             return _render("atom", repo_name)
         def last_rev(repo_name, cs_cache):
             return _render('revision', repo_name, cs_cache.get('revision'),
                            cs_cache.get('raw_id'), cs_cache.get('author'),
                            cs_cache.get('message'))
         def desc(desc):
             return h.urlify_text(desc, truncate=60, stylize=c.visual.stylify_metatags)
         def state(repo_state):
             return _render("repo_state", repo_state)
         def repo_actions(repo_name):
             return _render('repo_actions', repo_name, super_user_actions)
         def owner_actions(user_id, username):
             return _render('user_name', user_id, username)
         repos_data = []
         for repo in repos_list:
             if perm_check:
                 # check permission at this level
                 if not HasRepoPermissionAny(
                         'repository.read', 'repository.write',
                         'repository.admin'
                 )(repo.repo_name, 'get_repos_as_dict check'):
                     continue
             cs_cache = repo.changeset_cache
             row = {
                 "menu": quick_menu(repo.repo_name),
                 "raw_name": repo.repo_name.lower(),
                 "name": repo_lnk(repo.repo_name, repo.repo_type,
                                  repo.repo_state, repo.private, repo.fork),
                 "last_change": last_change(repo.last_db_change),
                 "last_changeset": last_rev(repo.repo_name, cs_cache),
                 "last_rev_raw": cs_cache.get('revision'),
                 "desc": desc(repo.description),
                 "owner": h.person(repo.user),
                 "state": state(repo.repo_state),
                 "rss": rss_lnk(repo.repo_name),
                 "atom": atom_lnk(repo.repo_name),
+            }
             if admin:
                 row.update({
                     "action": repo_actions(repo.repo_name),
                     "owner": owner_actions(repo.user.user_id,
                                            h.person(repo.user))
                 })
             repos_data.append(row)
         return {
             "totalRecords": len(repos_list),
             "startIndex": 0,
             "sort": "name",
             "dir": "asc",
             "records": repos_data
+        }
     def _get_defaults(self, repo_name):
         """
         Gets information about repository, and returns a dict for
         usage in forms
         :param repo_name:
         """
         repo_info = Repository.get_by_repo_name(repo_name)
         if repo_info is None:
             return None
         defaults = repo_info.get_dict()
         group, repo_name, repo_name_full = repo_info.groups_and_repo
         defaults['repo_name'] = repo_name
         defaults['repo_group'] = getattr(group[-1] if group else None,
                                          'group_id', None)
         for strip, k in [(0, 'repo_type'), (1, 'repo_enable_downloads'),
                          (1, 'repo_description'), (1, 'repo_enable_locking'),
                          (1, 'repo_landing_rev'), (0, 'clone_uri'),
                          (1, 'repo_private'), (1, 'repo_enable_statistics')]:
             attr = k
             if strip:
                 attr = remove_prefix(k, 'repo_')
             val = defaults[attr]
             if k == 'repo_landing_rev':
                 val = ':'.join(defaults[attr])
             defaults[k] = val
             if k == 'clone_uri':
                 defaults['clone_uri_hidden'] = repo_info.clone_uri_hidden
         # fill owner
         if repo_info.user:
             defaults.update({'user': repo_info.user.username})
         else:
             replacement_user = User.query().filter(User.admin ==
                                                    True).first().username
             defaults.update({'user': replacement_user})
         # fill repository users
         for p in repo_info.repo_to_perm:
             defaults.update({'u_perm_%s' % p.user.username:
                                  p.permission.permission_name})
         # fill repository groups
         for p in repo_info.users_group_to_perm:
             defaults.update({'g_perm_%s' % p.users_group.users_group_name:
                                  p.permission.permission_name})
         return defaults
     def update(self, repo, **kwargs):
         try:
             cur_repo = self._get_repo(repo)
             org_repo_name = cur_repo.repo_name
             if 'user' in kwargs:
                 cur_repo.user = User.get_by_username(kwargs['user'])
             if 'repo_group' in kwargs:
                 cur_repo.group = RepoGroup.get(kwargs['repo_group'])
             log.debug('Updating repo %s with params:%s', cur_repo, kwargs)
             for k in ['repo_enable_downloads',
                       'repo_description',
                       'repo_enable_locking',
                       'repo_landing_rev',
                       'repo_private',
                       'repo_enable_statistics',
                       ]:
                 if k in kwargs:
                     setattr(cur_repo, remove_prefix(k, 'repo_'), kwargs[k])
             clone_uri = kwargs.get('clone_uri')
             if clone_uri is not None and clone_uri != cur_repo.clone_uri_hidden:
                 cur_repo.clone_uri = clone_uri
             new_name = cur_repo.get_new_name(kwargs['repo_name'])
             cur_repo.repo_name = new_name
             #if private flag is set, reset default permission to NONE
             if kwargs.get('repo_private'):
                 EMPTY_PERM = 'repository.none'
                 RepoModel().grant_user_permission(
                     repo=cur_repo, user='default', perm=EMPTY_PERM
+                )
                 #handle extra fields
             for field in filter(lambda k: k.startswith(RepositoryField.PREFIX),
                                 kwargs):
                 k = RepositoryField.un_prefix_key(field)
                 ex_field = RepositoryField.get_by_key_name(key=k, repo=cur_repo)
                 if ex_field:
                     ex_field.field_value = kwargs[field]
                     self.sa.add(ex_field)
             self.sa.add(cur_repo)
             if org_repo_name != new_name:
                 # rename repository
                 self._rename_filesystem_repo(old=org_repo_name, new=new_name)
             return cur_repo
         except Exception:
             log.error(traceback.format_exc())
             raise
     def _create_repo(self, repo_name, repo_type, description, owner,
                      private=False, clone_uri=None, repo_group=None,
                      landing_rev='rev:tip', fork_of=None,
                      copy_fork_permissions=False, enable_statistics=False,
                      enable_locking=False, enable_downloads=False,
                      copy_group_permissions=False, state=Repository.STATE_PENDING):
         """
         Create repository inside database with PENDING state. This should only be
         executed by create() repo, with exception of importing existing repos.
         """
         from kallithea.model.scm import ScmModel
         owner = self._get_user(owner)
         fork_of = self._get_repo(fork_of)
         repo_group = self._get_repo_group(repo_group)
         try:
             repo_name = safe_unicode(repo_name)
             description = safe_unicode(description)
             # repo name is just a name of repository
             # while repo_name_full is a full qualified name that is combined
             # with name and path of group
             repo_name_full = repo_name
             repo_name = repo_name.split(self.URL_SEPARATOR)[-1]
             new_repo = Repository()
             new_repo.repo_state = state
             new_repo.enable_statistics = False
             new_repo.repo_name = repo_name_full
             new_repo.repo_type = repo_type
             new_repo.user = owner
             new_repo.group = repo_group
             new_repo.description = description or repo_name
             new_repo.private = private
             new_repo.clone_uri = clone_uri
             new_repo.landing_rev = landing_rev
             new_repo.enable_statistics = enable_statistics
             new_repo.enable_locking = enable_locking
             new_repo.enable_downloads = enable_downloads
             if repo_group:
                 new_repo.enable_locking = repo_group.enable_locking
             if fork_of:
                 parent_repo = fork_of

kallithea/model/scm.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.model.scm
 ~~~~~~~~~~~~~~~~~~~
 Scm model for Kallithea
 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: Apr 9, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import os
 import re
 import time
 import traceback
 import logging
 import cStringIO
 import pkg_resources
 from os.path import join as jn
 from sqlalchemy import func
 from pylons.i18n.translation import _
 import kallithea
 from kallithea.lib.vcs import get_backend
 from kallithea.lib.vcs.exceptions import RepositoryError
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 from kallithea.lib.vcs.nodes import FileNode
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea import BACKENDS
 from kallithea.lib import helpers as h
 from kallithea.lib.utils2 import safe_str, safe_unicode, get_server_url,\
     _set_extras
 from kallithea.lib.auth import HasRepoPermissionAny, HasRepoGroupPermissionAny,\
     HasUserGroupPermissionAny, HasPermissionAny, HasPermissionAll
 from kallithea.lib.utils import get_filesystem_repos, make_ui, \
     action_logger
 from kallithea.model import BaseModel
 from kallithea.model.db import Repository, Ui, CacheInvalidation, \
     UserFollowing, UserLog, User, RepoGroup, PullRequest
 from kallithea.lib.hooks import log_push_action
 from kallithea.lib.exceptions import NonRelativePathError, IMCCommitError
 log = logging.getLogger(__name__)
 class UserTemp(object):
     def __init__(self, user_id):
         self.user_id = user_id
     def __repr__(self):
         return "<%s('id:%s')>" % (self.__class__.__name__, self.user_id)
 class RepoTemp(object):
     def __init__(self, repo_id):
         self.repo_id = repo_id
     def __repr__(self):
         return "<%s('id:%s')>" % (self.__class__.__name__, self.repo_id)
 class CachedRepoList(object):
     """
     Cached repo list. Uses super-fast in-memory cache after initialization.
     """
     def __init__(self, db_repo_list, repos_path, order_by=None, perm_set=None):
         self.db_repo_list = db_repo_list
         self.repos_path = repos_path
         self.order_by = order_by
         self.reversed = (order_by or '').startswith('-')
         if not perm_set:
             perm_set = ['repository.read', 'repository.write',
                         'repository.admin']
         self.perm_set = perm_set
     def __len__(self):
         return len(self.db_repo_list)
     def __repr__(self):
         return '<%s (%s)>' % (self.__class__.__name__, self.__len__())
     def __iter__(self):
         # pre-propagated valid_cache_keys to save executing select statements
         # for each repo
         valid_cache_keys = CacheInvalidation.get_valid_cache_keys()
         for dbr in self.db_repo_list:
             scmr = dbr.scm_instance_cached(valid_cache_keys)
             # check permission at this level
             if not HasRepoPermissionAny(
                 *self.perm_set)(dbr.repo_name, 'get repo check'):
                 continue
             try:
                 last_change = scmr.last_change
                 tip = h.get_changeset_safe(scmr, 'tip')
             except Exception:
                 log.error(
                     '%s this repository is present in database but it '
                     'cannot be created as an scm instance, org_exc:%s'
                     % (dbr.repo_name, traceback.format_exc())
+                )
                 continue
             tmp_d = {}
             tmp_d['name'] = dbr.repo_name
             tmp_d['name_sort'] = tmp_d['name'].lower()
             tmp_d['raw_name'] = tmp_d['name'].lower()
             tmp_d['description'] = dbr.description
             tmp_d['description_sort'] = tmp_d['description'].lower()
             tmp_d['last_change'] = last_change
             tmp_d['last_change_sort'] = time.mktime(last_change.timetuple())
             tmp_d['tip'] = tip.raw_id
             tmp_d['tip_sort'] = tip.revision
             tmp_d['rev'] = tip.revision
             tmp_d['contact'] = dbr.user.full_contact
             tmp_d['contact_sort'] = tmp_d['contact']
             tmp_d['owner_sort'] = tmp_d['contact']
             tmp_d['repo_archives'] = list(scmr._get_archives())
             tmp_d['last_msg'] = tip.message
             tmp_d['author'] = tip.author
             tmp_d['dbrepo'] = dbr.get_dict()
             tmp_d['dbrepo_fork'] = dbr.fork.get_dict() if dbr.fork else {}
             yield tmp_d
 class SimpleCachedRepoList(CachedRepoList):
     """
     Lighter version of CachedRepoList without the scm initialisation
     """
     def __iter__(self):
         for dbr in self.db_repo_list:
             # check permission at this level
             if not HasRepoPermissionAny(
                 *self.perm_set)(dbr.repo_name, 'get repo check'):
                 continue
             tmp_d = {
                 'name': dbr.repo_name,
                 'dbrepo': dbr.get_dict(),
                 'dbrepo_fork': dbr.fork.get_dict() if dbr.fork else {}
+            }
             yield tmp_d
 class _PermCheckIterator(object):
     def __init__(self, obj_list, obj_attr, perm_set, perm_checker, extra_kwargs=None):
         """
         Creates iterator from given list of objects, additionally
         checking permission for them from perm_set var
         :param obj_list: list of db objects
         :param obj_attr: attribute of object to pass into perm_checker
         :param perm_set: list of permissions to check
         :param perm_checker: callable to check permissions against
         """
         self.obj_list = obj_list
         self.obj_attr = obj_attr
         self.perm_set = perm_set
         self.perm_checker = perm_checker
         self.extra_kwargs = extra_kwargs or {}
     def __len__(self):
         return len(self.obj_list)
     def __repr__(self):
         return '<%s (%s)>' % (self.__class__.__name__, self.__len__())
     def __iter__(self):
         for db_obj in self.obj_list:
             # check permission at this level
             name = getattr(db_obj, self.obj_attr, None)
             if not self.perm_checker(*self.perm_set)(
                     name, self.__class__.__name__, **self.extra_kwargs):
                 continue
             yield db_obj
 class RepoList(_PermCheckIterator):
     def __init__(self, db_repo_list, perm_set=None, extra_kwargs=None):
         if not perm_set:
             perm_set = ['repository.read', 'repository.write', 'repository.admin']
         super(RepoList, self).__init__(obj_list=db_repo_list,
                     obj_attr='repo_name', perm_set=perm_set,
                     perm_checker=HasRepoPermissionAny,
                     extra_kwargs=extra_kwargs)
 class RepoGroupList(_PermCheckIterator):
     def __init__(self, db_repo_group_list, perm_set=None, extra_kwargs=None):
         if not perm_set:
             perm_set = ['group.read', 'group.write', 'group.admin']
         super(RepoGroupList, self).__init__(obj_list=db_repo_group_list,
                     obj_attr='group_name', perm_set=perm_set,
                     perm_checker=HasRepoGroupPermissionAny,
                     extra_kwargs=extra_kwargs)
 class UserGroupList(_PermCheckIterator):
     def __init__(self, db_user_group_list, perm_set=None, extra_kwargs=None):
         if not perm_set:
             perm_set = ['usergroup.read', 'usergroup.write', 'usergroup.admin']
         super(UserGroupList, self).__init__(obj_list=db_user_group_list,
                     obj_attr='users_group_name', perm_set=perm_set,
                     perm_checker=HasUserGroupPermissionAny,
                     extra_kwargs=extra_kwargs)
 class ScmModel(BaseModel):
     """
     Generic Scm Model
     """
     def __get_repo(self, instance):
         cls = Repository
         if isinstance(instance, cls):
             return instance
         elif isinstance(instance, int) or safe_str(instance).isdigit():
             return cls.get(instance)
         elif isinstance(instance, basestring):
             return cls.get_by_repo_name(instance)
         elif instance is not None:
             raise Exception('given object must be int, basestr or Instance'
                             ' of %s got %s' % (type(cls), type(instance)))
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
         q = self.sa.query(Ui).filter(Ui.ui_key == '/').one()
         return q.ui_value
     def repo_scan(self, repos_path=None):
         """
         Listing of repositories in given path. This path should not be a
         repository itself. Return a dictionary of repository objects
         :param repos_path: path to directory containing repositories
         """
         if repos_path is None:
             repos_path = self.repos_path
         log.info('scanning for repositories in %s', repos_path)
         baseui = make_ui('db')
         repos = {}
         for name, path in get_filesystem_repos(repos_path, recursive=True):
             # name need to be decomposed and put back together using the /
             # since this is internal storage separator for kallithea
             name = Repository.normalize_repo_name(name)
             try:
                 if name in repos:
                     raise RepositoryError('Duplicate repository name %s '
                                           'found in %s' % (name, path))
                 else:
                     klass = get_backend(path[0])
                     if path[0] == 'hg' and path[0] in BACKENDS.keys():
                         repos[name] = klass(safe_str(path[1]), baseui=baseui)
                     if path[0] == 'git' and path[0] in BACKENDS.keys():
                         repos[name] = klass(path[1])
             except OSError:
                 continue
         log.debug('found %s paths with repositories', len(repos))
         return repos
     def get_repos(self, all_repos=None, sort_key=None, simple=False):
         """
         Get all repos from db and for each repo create its
         backend instance and fill that backed with information from database
         :param all_repos: list of repository names as strings
             give specific repositories list, good for filtering
         :param sort_key: initial sorting of repos
         :param simple: use SimpleCachedList - one without the SCM info
         """
         if all_repos is None:
             all_repos = self.sa.query(Repository)\
                         .filter(Repository.group_id == None)\
                         .order_by(func.lower(Repository.repo_name)).all()
         if simple:
             repo_iter = SimpleCachedRepoList(all_repos,
                                              repos_path=self.repos_path,
                                              order_by=sort_key)
         else:
             repo_iter = CachedRepoList(all_repos,
                                        repos_path=self.repos_path,
                                        order_by=sort_key)
         return repo_iter
     def get_repo_groups(self, all_groups=None):
         if all_groups is None:
             all_groups = RepoGroup.query()\
                 .filter(RepoGroup.group_parent_id == None).all()
         return [x for x in RepoGroupList(all_groups)]
     def mark_for_invalidation(self, repo_name, delete=False):
         """
         Mark caches of this repo invalid in the database.
         :param repo_name: the repo for which caches should be marked invalid
         """
         CacheInvalidation.set_invalidate(repo_name, delete=delete)
         repo = Repository.get_by_repo_name(repo_name)
         if repo is not None:
             repo.update_changeset_cache()
     def toggle_following_repo(self, follow_repo_id, user_id):
         f = self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_repo_id == follow_repo_id)\
             .filter(UserFollowing.user_id == user_id).scalar()
         if f is not None:
             try:
                 self.sa.delete(f)
                 action_logger(UserTemp(user_id),
                               'stopped_following_repo',
                               RepoTemp(follow_repo_id))
                 return
             except Exception:
                 log.error(traceback.format_exc())
                 raise
         try:
             f = UserFollowing()
             f.user_id = user_id
             f.follows_repo_id = follow_repo_id
             self.sa.add(f)
             action_logger(UserTemp(user_id),
                           'started_following_repo',
                           RepoTemp(follow_repo_id))
         except Exception:
             log.error(traceback.format_exc())
             raise
     def toggle_following_user(self, follow_user_id, user_id):
         f = self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_user_id == follow_user_id)\
             .filter(UserFollowing.user_id == user_id).scalar()
         if f is not None:
             try:
                 self.sa.delete(f)
                 return
             except Exception:
                 log.error(traceback.format_exc())
                 raise
         try:
             f = UserFollowing()
             f.user_id = user_id
             f.follows_user_id = follow_user_id
             self.sa.add(f)
         except Exception:
             log.error(traceback.format_exc())
             raise
     def is_following_repo(self, repo_name, user_id, cache=False):
         r = self.sa.query(Repository)\
             .filter(Repository.repo_name == repo_name).scalar()
         f = self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_repository == r)\
             .filter(UserFollowing.user_id == user_id).scalar()
         return f is not None
     def is_following_user(self, username, user_id, cache=False):
         u = User.get_by_username(username)
         f = self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_user == u)\

kallithea/tests/api/api_base.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/>.
 """
 tests for api. run with::
     KALLITHEA_WHOOSH_TEST_DISABLE=1 nosetests --with-coverage --cover-package=kallithea.controllers.api.api -x kallithea/tests/api
 """
 from __future__ import with_statement
 import os
 import random
 import mock
 from kallithea.tests import *
 from kallithea.tests.fixture import Fixture
 from kallithea.lib.compat import json
 from kallithea.lib.auth import AuthUser
 from kallithea.model.user import UserModel
 from kallithea.model.user_group import UserGroupModel
 from kallithea.model.repo import RepoModel
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.meta import Session
 from kallithea.model.scm import ScmModel
 from kallithea.model.gist import GistModel
 from kallithea.model.db import Repository, User, Setting
 from kallithea.lib.utils2 import time_to_datetime
 API_URL = '/_admin/api'
 TEST_USER_GROUP = 'test_user_group'
 TEST_REPO_GROUP = 'test_repo_group'
 fixture = Fixture()
 def _build_data(apikey, method, **kw):
     """
     Builds API data with given random ID
     :param random_id:
     """
     random_id = random.randrange(1, 9999)
     return random_id, json.dumps({
         "id": random_id,
         "api_key": apikey,
         "method": method,
         "args": kw
     })
 jsonify = lambda obj: json.loads(json.dumps(obj))
 def crash(*args, **kwargs):
     raise Exception('Total Crash !')
 def api_call(test_obj, params):
     response = test_obj.app.post(API_URL, content_type='application/json',
                                  params=params)
     return response
 ## helpers
 def make_user_group(name=TEST_USER_GROUP):
     gr = fixture.create_user_group(name, cur_user=TEST_USER_ADMIN_LOGIN)
     UserGroupModel().add_user_to_group(user_group=gr,
                                        user=TEST_USER_ADMIN_LOGIN)
     Session().commit()
     return gr
 def make_repo_group(name=TEST_REPO_GROUP):
     gr = fixture.create_repo_group(name, cur_user=TEST_USER_ADMIN_LOGIN)
     Session().commit()
     return gr
 class _BaseTestApi(object):
     REPO = None
     REPO_TYPE = None
     @classmethod
     def setup_class(cls):
         cls.usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         cls.apikey = cls.usr.api_key
         cls.test_user = UserModel().create_or_update(
             username='test-api',
             password='test',
             email='test@example.com',
             firstname='first',
             lastname='last'
+        )
         Session().commit()
         cls.TEST_USER_LOGIN = cls.test_user.username
         cls.apikey_regular = cls.test_user.api_key
     @classmethod
     def teardown_class(cls):
         pass
     def setUp(self):
         self.maxDiff = None
         make_user_group()
         make_repo_group()
     def tearDown(self):
         fixture.destroy_user_group(TEST_USER_GROUP)
         fixture.destroy_gists()
         fixture.destroy_repo_group(TEST_REPO_GROUP)
     def _compare_ok(self, id_, expected, given):
         expected = jsonify({
             'id': id_,
             'error': None,
             'result': expected
         })
         given = json.loads(given)
         self.assertEqual(expected, given)
     def _compare_error(self, id_, expected, given):
         expected = jsonify({
             'id': id_,
             'error': expected,
             'result': None
         })
         given = json.loads(given)
         self.assertEqual(expected, given)
     def test_Optional_object(self):
         from kallithea.controllers.api.api import Optional
         option1 = Optional(None)
         self.assertEqual('<Optional:%s>' % None, repr(option1))
         self.assertEqual(option1(), None)
         self.assertEqual(1, Optional.extract(Optional(1)))
         self.assertEqual('trololo', Optional.extract('trololo'))
     def test_Optional_OAttr(self):
         from kallithea.controllers.api.api import Optional, OAttr
         option1 = Optional(OAttr('apiuser'))
         self.assertEqual('apiuser', Optional.extract(option1))
     def test_OAttr_object(self):
         from kallithea.controllers.api.api import OAttr
         oattr1 = OAttr('apiuser')
         self.assertEqual('<OptionalAttr:apiuser>', repr(oattr1))
         self.assertEqual(oattr1(), oattr1)
     def test_api_wrong_key(self):
         id_, params = _build_data('trololo', 'get_user')
         response = api_call(self, params)
         expected = 'Invalid API key'
         self._compare_error(id_, expected, given=response.body)
     def test_api_missing_non_optional_param(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         response = api_call(self, params)
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_missing_non_optional_param_args_null(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         params = params.replace('"args": {}', '"args": null')
         response = api_call(self, params)
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_missing_non_optional_param_args_bad(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         params = params.replace('"args": {}', '"args": 1')
         response = api_call(self, params)
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_args_is_null(self):
         id_, params = _build_data(self.apikey, 'get_users', )
         params = params.replace('"args": {}', '"args": null')
         response = api_call(self, params)
         self.assertEqual(response.status, '200 OK')
     def test_api_args_is_bad(self):
         id_, params = _build_data(self.apikey, 'get_users', )
         params = params.replace('"args": {}', '"args": 1')
         response = api_call(self, params)
         self.assertEqual(response.status, '200 OK')
     def test_api_args_different_args(self):
         import string
         expected = {
             'ascii_letters': string.ascii_letters,
             'ws': string.whitespace,
             'printables': string.printable
+        }
         id_, params = _build_data(self.apikey, 'test', args=expected)
         response = api_call(self, params)
         self.assertEqual(response.status, '200 OK')
         self._compare_ok(id_, expected, response.body)
     def test_api_get_users(self):
         id_, params = _build_data(self.apikey, 'get_users', )
         response = api_call(self, params)
         ret_all = []
         _users = User.query().filter(User.username != User.DEFAULT_USER) \
             .order_by(User.username).all()
         for usr in _users:
             ret = usr.get_api_data()
             ret_all.append(jsonify(ret))
         expected = ret_all
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user(self):
         id_, params = _build_data(self.apikey, 'get_user',
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data()
         ret['permissions'] = AuthUser(dbuser=usr).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_that_does_not_exist(self):
         id_, params = _build_data(self.apikey, 'get_user',
                                   userid='trololo')
         response = api_call(self, params)
         expected = "user `%s` does not exist" % 'trololo'
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_user_without_giving_userid(self):
         id_, params = _build_data(self.apikey, 'get_user')
         response = api_call(self, params)
         usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data()
         ret['permissions'] = AuthUser(dbuser=usr).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_without_giving_userid_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_user')
         response = api_call(self, params)
         usr = User.get_by_username(self.TEST_USER_LOGIN)
         ret = usr.get_api_data()
         ret['permissions'] = AuthUser(dbuser=usr).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_with_giving_userid_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_user',
                                   userid=self.TEST_USER_LOGIN)
         response = api_call(self, params)
         expected = 'userid is not the same as your user'
         self._compare_error(id_, expected, given=response.body)
     def test_api_pull(self):
         repo_name = 'test_pull'
         r = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         r.clone_uri = os.path.join(TESTS_TMP_PATH, self.REPO)
         Session.add(r)
         Session.commit()
         id_, params = _build_data(self.apikey, 'pull',
                                   repoid=repo_name,)
         response = api_call(self, params)
         expected = {'msg': 'Pulled from `%s`' % repo_name,
                     'repository': repo_name}
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     def test_api_pull_error(self):
         id_, params = _build_data(self.apikey, 'pull',
                                   repoid=self.REPO, )
         response = api_call(self, params)
         expected = 'Unable to pull changes from `%s`' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_rescan_repos(self):
         id_, params = _build_data(self.apikey, 'rescan_repos')
         response = api_call(self, params)
         expected = {'added': [], 'removed': []}
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(ScmModel, 'repo_scan', crash)
     def test_api_rescann_error(self):
         id_, params = _build_data(self.apikey, 'rescan_repos', )
         response = api_call(self, params)
         expected = 'Error occurred during rescan repositories action'
         self._compare_error(id_, expected, given=response.body)
     def test_api_invalidate_cache(self):
         repo = RepoModel().get_by_repo_name(self.REPO)
         repo.scm_instance_cached()  # seed cache
         id_, params = _build_data(self.apikey, 'invalidate_cache',
                                   repoid=self.REPO)
         response = api_call(self, params)
         expected = {
             'msg': "Cache for repository `%s` was invalidated" % (self.REPO,),
             'repository': self.REPO
+        }
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(ScmModel, 'mark_for_invalidation', crash)
     def test_api_invalidate_cache_error(self):
         id_, params = _build_data(self.apikey, 'invalidate_cache',
                                   repoid=self.REPO)
         response = api_call(self, params)
         expected = 'Error occurred during cache invalidation action'
         self._compare_error(id_, expected, given=response.body)
     def test_api_invalidate_cache_regular_user_no_permission(self):
         repo = RepoModel().get_by_repo_name(self.REPO)
         repo.scm_instance_cached() # seed cache
         id_, params = _build_data(self.apikey_regular, 'invalidate_cache',
                                   repoid=self.REPO)
         response = api_call(self, params)
         expected = "repository `%s` does not exist" % (self.REPO,)
         self._compare_error(id_, expected, given=response.body)
     def test_api_lock_repo_lock_aquire(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = {
             'repo': self.REPO, 'locked': True,
             'locked_since': response.json['result']['locked_since'],
             'locked_by': TEST_USER_ADMIN_LOGIN,
             'lock_state_changed': True,
             'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                     % (TEST_USER_ADMIN_LOGIN, self.REPO, True))
+        }
         self._compare_ok(id_, expected, given=response.body)
     def test_api_lock_repo_lock_aquire_by_non_admin(self):
         repo_name = 'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                             cur_user=self.TEST_USER_LOGIN)
         try:
             id_, params = _build_data(self.apikey_regular, 'lock',
                                       repoid=repo_name,
                                       locked=True)
             response = api_call(self, params)
             expected = {
                 'repo': repo_name,
                 'locked': True,
                 'locked_since': response.json['result']['locked_since'],
                 'locked_by': self.TEST_USER_LOGIN,
                 'lock_state_changed': True,
                 'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                         % (self.TEST_USER_LOGIN, repo_name, True))
+            }
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_lock_repo_lock_aquire_non_admin_with_userid(self):
         repo_name = 'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                             cur_user=self.TEST_USER_LOGIN)
         try:
             id_, params = _build_data(self.apikey_regular, 'lock',
                                       userid=TEST_USER_ADMIN_LOGIN,
                                       repoid=repo_name,
                                       locked=True)
             response = api_call(self, params)
             expected = 'userid is not the same as your user'
             self._compare_error(id_, expected, given=response.body)
         finally:

kallithea/tests/functional/test_admin.py

➞

Show inline comments

 from __future__ import with_statement
 import os
 import csv
 import datetime
 from kallithea.tests import *
 from kallithea.model.db import UserLog
 from kallithea.model.meta import Session
 from kallithea.lib.utils2 import safe_unicode
 dn = os.path.dirname
 FIXTURES = os.path.join(dn(dn(os.path.abspath(__file__))), 'fixtures')
 class TestAdminController(TestController):
     @classmethod
     def setup_class(cls):
         UserLog.query().delete()
         Session().commit()
         def strptime(val):
             fmt = '%Y-%m-%d %H:%M:%S'
             if '.' not in val:
                 return datetime.datetime.strptime(val, fmt)
             nofrag, frag = val.split(".")
             date = datetime.datetime.strptime(nofrag, fmt)
             frag = frag[:6]  # truncate to microseconds
             frag += (6 - len(frag)) * '0'  # add 0s
             return date.replace(microsecond=int(frag))
         with open(os.path.join(FIXTURES, 'journal_dump.csv')) as f:
             for row in csv.DictReader(f):
                 ul = UserLog()
                 for k, v in row.iteritems():
                     v = safe_unicode(v)
                     if k == 'action_date':
                         v = strptime(v)
                     if k in ['user_id', 'repository_id']:
                         # nullable due to FK problems
                         v = None
                     setattr(ul, k, v)
                 Session().add(ul)
             Session().commit()
     @classmethod
     def teardown_class(cls):
         UserLog.query().delete()
         Session().commit()
     def test_index(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index'))
         response.mustcontain('Admin Journal')
     def test_filter_all_entries(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',))
         response.mustcontain('2034 Entries')
     def test_filter_journal_filter_exact_match_on_repository(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='repository:xxx'))
         response.mustcontain('3 Entries')
     def test_filter_journal_filter_exact_match_on_repository_CamelCase(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='repository:XxX'))
         response.mustcontain('3 Entries')
     def test_filter_journal_filter_wildcard_on_repository(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='repository:*test*'))
         response.mustcontain('862 Entries')
     def test_filter_journal_filter_prefix_on_repository(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='repository:test*'))
         response.mustcontain('257 Entries')
     def test_filter_journal_filter_prefix_on_repository_CamelCase(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='repository:Test*'))
         response.mustcontain('257 Entries')
     def test_filter_journal_filter_prefix_on_repository_and_user(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='repository:test* AND username:demo'))
         response.mustcontain('130 Entries')
     def test_filter_journal_filter_prefix_on_repository_or_other_repo(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='repository:test* OR repository:xxx'))
         response.mustcontain('260 Entries')  # 257 + 3
     def test_filter_journal_filter_exact_match_on_username(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='username:demo'))
         response.mustcontain('1087 Entries')
     def test_filter_journal_filter_exact_match_on_username_camelCase(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='username:DemO'))
         response.mustcontain('1087 Entries')
     def test_filter_journal_filter_wildcard_on_username(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='username:*test*'))
         response.mustcontain('100 Entries')
     def test_filter_journal_filter_prefix_on_username(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='username:demo*'))
         response.mustcontain('1101 Entries')
     def test_filter_journal_filter_prefix_on_user_or_other_user(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='username:demo OR username:volcan'))
         response.mustcontain('1095 Entries')  # 1087 + 8
     def test_filter_journal_filter_wildcard_on_action(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='action:*pull_request*'))
         response.mustcontain('187 Entries')
     def test_filter_journal_filter_on_date(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='date:20121010'))
         response.mustcontain('47 Entries')
     def test_filter_journal_filter_on_date_2(self):
         self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index',
                                     filter='date:20121020'))
         response.mustcontain('17 Entries')

kallithea/tests/functional/test_login.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 from __future__ import with_statement
 import re
 import time
 import mock
 from kallithea.tests import *
 from kallithea.tests.fixture import Fixture
 from kallithea.lib.utils2 import generate_api_key
 from kallithea.lib.auth import check_password
 from kallithea.lib import helpers as h
 from kallithea.model.api_key import ApiKeyModel
 from kallithea.model import validators
 from kallithea.model.db import User, Notification
 from kallithea.model.meta import Session
 from kallithea.model.user import UserModel
 fixture = Fixture()
 class TestLoginController(TestController):
     def setUp(self):
         self.remove_all_notifications()
         self.assertEqual(Notification.query().all(), [])
     def test_index(self):
         response = self.app.get(url(controller='login', action='index'))
         self.assertEqual(response.status, '200 OK')
         # Test response...
     def test_login_admin_ok(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS})
         self.assertEqual(response.status, '302 Found')
         self.assert_authenticated_user(response, TEST_USER_ADMIN_LOGIN)
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
     def test_login_regular_ok(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS})
         self.assertEqual(response.status, '302 Found')
         self.assert_authenticated_user(response, TEST_USER_REGULAR_LOGIN)
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
     def test_login_ok_came_from(self):
         test_came_from = '/_admin/users'
         response = self.app.post(url(controller='login', action='index',
                                      came_from=test_came_from),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS})
         self.assertEqual(response.status, '302 Found')
         response = response.follow()
         self.assertEqual(response.status, '200 OK')
         response.mustcontain('Users Administration')
     def test_login_do_not_remember(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS,
                                   'remember': False})
         self.assertIn('Set-Cookie', response.headers)
         for cookie in response.headers.getall('Set-Cookie'):
             self.assertFalse(re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE),
                 'Cookie %r has expiration date, but should be a session cookie' % cookie)
     def test_login_remember(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS,
                                   'remember': True})
         self.assertIn('Set-Cookie', response.headers)
         for cookie in response.headers.getall('Set-Cookie'):
             self.assertTrue(re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE),
                 'Cookie %r should have expiration date, but is a session cookie' % cookie)
     def test_logout(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS})
         # Verify that a login session has been established.
         response = self.app.get(url(controller='login', action='index'))
         response = response.follow()
         self.assertIn('authuser', response.session)
         response.click('Log Out')
         # Verify that the login session has been terminated.
         response = self.app.get(url(controller='login', action='index'))
         self.assertNotIn('authuser', response.session)
     @parameterized.expand([
           ('data:text/html,<script>window.alert("xss")</script>',),
           ('mailto:test@example.com',),
           ('file:///etc/passwd',),
           ('ftp://some.ftp.server',),
           ('http://other.domain/bl%C3%A5b%C3%A6rgr%C3%B8d',),
     ])
     def test_login_bad_came_froms(self, url_came_from):
         response = self.app.post(url(controller='login', action='index',
                                      came_from=url_came_from),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS})
         self.assertEqual(response.status, '302 Found')
         self.assertEqual(response._environ['paste.testing_variables']
                          ['tmpl_context'].came_from, '/')
         response = response.follow()
         self.assertEqual(response.status, '200 OK')
     def test_login_short_password(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': 'as'})
         self.assertEqual(response.status, '200 OK')
         response.mustcontain('Enter 3 characters or more')
     def test_login_wrong_username_password(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': 'error',
                                   'password': 'test12'})
         response.mustcontain('Invalid username or password')
     # verify that get arguments are correctly passed along login redirection
     @parameterized.expand([
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå'.encode('utf-8'), 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_redirection_to_login_form_preserves_get_args(self, args, args_encoded):
         with fixture.anon_access(False):
             response = self.app.get(url(controller='summary', action='index',
                                         repo_name=HG_REPO,
                                         **args))
             self.assertEqual(response.status, '302 Found')
             for encoded in args_encoded:
                 self.assertIn(encoded, response.location)
     @parameterized.expand([
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå'.encode('utf-8'), 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_login_form_preserves_get_args(self, args, args_encoded):
         response = self.app.get(url(controller='login', action='index',
                                     came_from = '/_admin/users',
                                     **args))
         for encoded in args_encoded:
             self.assertIn(encoded, response.form.action)
     @parameterized.expand([
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå'.encode('utf-8'), 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_redirection_after_successful_login_preserves_get_args(self, args, args_encoded):
         response = self.app.post(url(controller='login', action='index',
                                      came_from = '/_admin/users',
                                      **args),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS})
         self.assertEqual(response.status, '302 Found')
         for encoded in args_encoded:
             self.assertIn(encoded, response.location)
     @parameterized.expand([
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå'.encode('utf-8'), 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_login_form_after_incorrect_login_preserves_get_args(self, args, args_encoded):
         response = self.app.post(url(controller='login', action='index',
                                      came_from = '/_admin/users',
                                      **args),
                                  {'username': 'error',
                                   'password': 'test12'})
         response.mustcontain('Invalid username or password')
         for encoded in args_encoded:
             self.assertIn(encoded, response.form.action)
     #==========================================================================
     # REGISTRATIONS
     #==========================================================================
     def test_register(self):
         response = self.app.get(url(controller='login', action='register'))
         response.mustcontain('Sign Up')
     def test_register_err_same_username(self):
         uname = TEST_USER_ADMIN_LOGIN
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': uname,
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmail@domain.com',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         msg = validators.ValidUsername()._messages['username_exists']
         msg = h.html_escape(msg % {'username': uname})
         response.mustcontain(msg)
     def test_register_err_same_email(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'test_admin_0',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': TEST_USER_ADMIN_EMAIL,
                                              'firstname': 'test',
                                              'lastname': 'test'})
         msg = validators.UniqSystemEmail()()._messages['email_taken']
         response.mustcontain(msg)
     def test_register_err_same_email_case_sensitive(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'test_admin_1',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': TEST_USER_ADMIN_EMAIL.title(),
                                              'firstname': 'test',
                                              'lastname': 'test'})
         msg = validators.UniqSystemEmail()()._messages['email_taken']
         response.mustcontain(msg)
     def test_register_err_wrong_data(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'xs',
                                              'password': 'test',
                                              'password_confirmation': 'test',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         self.assertEqual(response.status, '200 OK')
         response.mustcontain('An email address must contain a single @')
         response.mustcontain('Enter a value 6 characters long or more')
     def test_register_err_username(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'error user',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         response.mustcontain('An email address must contain a single @')
         response.mustcontain('Username may only contain '
                 'alphanumeric characters underscores, '
                 'periods or dashes and must begin with an '
                 'alphanumeric character')
     def test_register_err_case_sensitive(self):
         usr = TEST_USER_ADMIN_LOGIN.title()
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': usr,
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         response.mustcontain('An email address must contain a single @')
         msg = validators.ValidUsername()._messages['username_exists']
         msg = h.html_escape(msg % {'username': usr})
         response.mustcontain(msg)
     def test_register_special_chars(self):
         response = self.app.post(url(controller='login', action='register'),
                                         {'username': 'xxxaxn',
                                          'password': 'ąćźżąśśśś',
                                          'password_confirmation': 'ąćźżąśśśś',
                                          'email': 'goodmailm@test.plx',
                                          'firstname': 'test',
                                          'lastname': 'test'})
         msg = validators.ValidPassword()._messages['invalid_password']
         response.mustcontain(msg)
     def test_register_password_mismatch(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'xs',
                                              'password': '123qwe',
                                              'password_confirmation': 'qwe123',
                                              'email': 'goodmailm@test.plxa',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         msg = validators.ValidPasswordsMatch('password', 'password_confirmation')._messages['password_mismatch']
         response.mustcontain(msg)
     def test_register_ok(self):
         username = 'test_regular4'
         password = 'qweqwe'
         email = 'username@test.com'
         name = 'testname'
         lastname = 'testlastname'
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': username,
                                              'password': password,
                                              'password_confirmation': password,
                                              'email': email,
                                              'firstname': name,
                                              'lastname': lastname,
                                              'admin': True})  # This should be overriden
         self.assertEqual(response.status, '302 Found')
         self.checkSessionFlash(response, 'You have successfully registered into Kallithea')
         ret = Session().query(User).filter(User.username == 'test_regular4').one()
         self.assertEqual(ret.username, username)
         self.assertEqual(check_password(password, ret.password), True)
         self.assertEqual(ret.email, email)
         self.assertEqual(ret.name, name)
         self.assertEqual(ret.lastname, lastname)
         self.assertNotEqual(ret.api_key, None)
         self.assertEqual(ret.admin, False)
     #==========================================================================
     # PASSWORD RESET
     #==========================================================================
     def test_forgot_password_wrong_mail(self):
         bad_email = 'username%wrongmail.org'
         response = self.app.post(
                         url(controller='login', action='password_reset'),
                             {'email': bad_email, }
+        )
         response.mustcontain('An email address must contain a single @')
     def test_forgot_password(self):
         response = self.app.get(url(controller='login',
                                     action='password_reset'))
         self.assertEqual(response.status, '200 OK')
         username = 'test_password_reset_1'
         password = 'qweqwe'
         email = 'username@python-works.com'
         name = 'passwd'
         lastname = 'reset'
         timestamp = int(time.time())
         new = User()
         new.username = username
         new.password = password
         new.email = email
         new.name = name
         new.lastname = lastname
         new.api_key = generate_api_key()
         Session().add(new)
         Session().commit()
         response = self.app.post(url(controller='login',
                                      action='password_reset'),
                                  {'email': email, })
         self.checkSessionFlash(response, 'A password reset confirmation code has been sent')
         response = response.follow()
         # BAD TOKEN
         token = "bad"
         response = self.app.post(url(controller='login',
                                      action='password_reset_confirmation'),
                                  {'email': email,
                                   'timestamp': timestamp,
                                   'password': "p@ssw0rd",
                                   'password_confirm': "p@ssw0rd",
                                   'token': token,
                                  })

kallithea/tests/models/test_diff_parsers.py

➞

Show inline comments

 from __future__ import with_statement
 from kallithea.tests import *
 from kallithea.lib.diffs import DiffProcessor, NEW_FILENODE, DEL_FILENODE, \
     MOD_FILENODE, RENAMED_FILENODE, CHMOD_FILENODE, BIN_FILENODE, COPIED_FILENODE
 from kallithea.tests.fixture import Fixture
 fixture = Fixture()
 DIFF_FIXTURES = {
     'hg_diff_add_single_binary_file.diff': [
         ('US Warszawa.jpg', 'A',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {NEW_FILENODE: 'new file 100755',
                   BIN_FILENODE: 'binary diff not shown'}}),
     ],
     'hg_diff_mod_single_binary_file.diff': [
         ('US Warszawa.jpg', 'M',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {MOD_FILENODE: 'modified file',
                   BIN_FILENODE: 'binary diff not shown'}}),
     ],
     'hg_diff_mod_single_file_and_rename_and_chmod.diff': [
         ('README', 'R',
          {'added': 3,
           'deleted': 0,
           'binary': False,
           'ops': {RENAMED_FILENODE: 'file renamed from README.rst to README',
                   CHMOD_FILENODE: 'modified file chmod 100755 => 100644'}}),
     ],
     'hg_diff_mod_file_and_rename.diff': [
         ('README.rst', 'R',
          {'added': 3,
           'deleted': 0,
           'binary': False,
           'ops': {RENAMED_FILENODE: 'file renamed from README to README.rst'}}),
     ],
     'hg_diff_del_single_binary_file.diff': [
         ('US Warszawa.jpg', 'D',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {DEL_FILENODE: 'deleted file',
                   BIN_FILENODE: 'binary diff not shown'}}),
     ],
     'hg_diff_chmod_and_mod_single_binary_file.diff': [
         ('gravatar.png', 'M',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {CHMOD_FILENODE: 'modified file chmod 100644 => 100755',
                   BIN_FILENODE: 'binary diff not shown'}}),
     ],
     'hg_diff_chmod.diff': [
         ('file', 'M',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {CHMOD_FILENODE: 'modified file chmod 100755 => 100644'}}),
     ],
     'hg_diff_rename_file.diff': [
         ('file_renamed', 'R',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {RENAMED_FILENODE: 'file renamed from file to file_renamed'}}),
     ],
     'hg_diff_rename_and_chmod_file.diff': [
         ('README', 'R',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {CHMOD_FILENODE: 'modified file chmod 100644 => 100755',
                   RENAMED_FILENODE: 'file renamed from README.rst to README'}}),
     ],
     'hg_diff_binary_and_normal.diff': [
         ('img/baseline-10px.png', 'A',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {NEW_FILENODE: 'new file 100644',
                   BIN_FILENODE: 'binary diff not shown'}}),
         ('img/baseline-20px.png', 'D',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {DEL_FILENODE: 'deleted file',
                   BIN_FILENODE: 'binary diff not shown'}}),
         ('index.html', 'M',
          {'added': 3,
           'deleted': 2,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('js/global.js', 'D',
          {'added': 0,
           'deleted': 75,
           'binary': False,
           'ops': {DEL_FILENODE: 'deleted file'}}),
         ('js/jquery/hashgrid.js', 'A',
          {'added': 340,
           'deleted': 0,
           'binary': False,
           'ops': {NEW_FILENODE: 'new file 100755'}}),
         ('less/docs.less', 'M',
          {'added': 34,
           'deleted': 0,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('less/scaffolding.less', 'M',
          {'added': 1,
           'deleted': 3,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('readme.markdown', 'M',
          {'added': 1,
           'deleted': 10,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
     ],
     'git_diff_chmod.diff': [
         ('work-horus.xls', 'M',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {CHMOD_FILENODE: 'modified file chmod 100644 => 100755'}})
     ],
     'git_diff_rename_file.diff': [
         ('file.xls', 'R',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {RENAMED_FILENODE: 'file renamed from work-horus.xls to file.xls'}})
     ],
     'git_diff_mod_single_binary_file.diff': [
         ('US Warszawa.jpg', 'M',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {MOD_FILENODE: 'modified file',
                   BIN_FILENODE: 'binary diff not shown'}})
     ],
     'git_diff_binary_and_normal.diff': [
         ('img/baseline-10px.png', 'A',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {NEW_FILENODE: 'new file 100644',
                   BIN_FILENODE: 'binary diff not shown'}}),
         ('img/baseline-20px.png', 'D',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {DEL_FILENODE: 'deleted file',
                   BIN_FILENODE: 'binary diff not shown'}}),
         ('index.html', 'M',
          {'added': 3,
           'deleted': 2,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('js/global.js', 'D',
          {'added': 0,
           'deleted': 75,
           'binary': False,
           'ops': {DEL_FILENODE: 'deleted file'}}),
         ('js/jquery/hashgrid.js', 'A',
          {'added': 340,
           'deleted': 0,
           'binary': False,
           'ops': {NEW_FILENODE: 'new file 100755'}}),
         ('less/docs.less', 'M',
          {'added': 34,
           'deleted': 0,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('less/scaffolding.less', 'M',
          {'added': 1,
           'deleted': 3,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('readme.markdown', 'M',
          {'added': 1,
           'deleted': 10,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
     ],
     'diff_with_diff_data.diff': [
         ('vcs/backends/base.py', 'M',
          {'added': 18,
           'deleted': 2,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('vcs/backends/git/repository.py', 'M',
          {'added': 46,
           'deleted': 15,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('vcs/backends/hg.py', 'M',
          {'added': 22,
           'deleted': 3,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('vcs/tests/test_git.py', 'M',
          {'added': 5,
           'deleted': 5,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
         ('vcs/tests/test_repository.py', 'M',
          {'added': 174,
           'deleted': 2,
           'binary': False,
           'ops': {MOD_FILENODE: 'modified file'}}),
     ],
     'git_diff_modify_binary_file.diff': [
         ('file.name', 'M',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {MOD_FILENODE: 'modified file',
                   BIN_FILENODE: 'binary diff not shown'}})
     ],
     'hg_diff_copy_file.diff': [
         ('file2', 'M',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {COPIED_FILENODE: 'file copied from file1 to file2'}}),
     ],
     'hg_diff_copy_and_modify_file.diff': [
         ('file3', 'M',
          {'added': 1,
           'deleted': 0,
           'binary': False,
           'ops': {COPIED_FILENODE: 'file copied from file2 to file3',
                   MOD_FILENODE: 'modified file'}}),
     ],
     'hg_diff_copy_and_chmod_file.diff': [
         ('file4', 'M',
          {'added': 0,
           'deleted': 0,
           'binary': True,
           'ops': {COPIED_FILENODE: 'file copied from file3 to file4',
                   CHMOD_FILENODE: 'modified file chmod 100644 => 100755'}}),
     ],
     'hg_diff_copy_chmod_and_edit_file.diff': [
         ('file5', 'M',
          {'added': 2,
           'deleted': 1,
           'binary': False,
           'ops': {COPIED_FILENODE: 'file copied from file4 to file5',
                   CHMOD_FILENODE: 'modified file chmod 100755 => 100644',
                   MOD_FILENODE: 'modified file'}}),
     ],
     'hg_diff_rename_space_cr.diff': [
         ('oh yes', 'R',
          {'added': 3,
           'deleted': 2,
           'binary': False,
           'ops': {RENAMED_FILENODE: 'file renamed from oh no to oh yes'}}),
     ],
+}
 class DiffLibTest(BaseTestCase):
     @parameterized.expand([(x,) for x in DIFF_FIXTURES])
     def test_diff(self, diff_fixture):
         diff = fixture.load_resource(diff_fixture, strip=False)
         diff_proc = DiffProcessor(diff)
         diff_proc_d = diff_proc.prepare()
         data = [(x['filename'], x['operation'], x['stats']) for x in diff_proc_d]
         expected_data = DIFF_FIXTURES[diff_fixture]
         self.assertListEqual(expected_data, data)
     def test_diff_markup(self):
         diff = fixture.load_resource('markuptest.diff', strip=False)
         diff_proc = DiffProcessor(diff)
         diff_proc_d = diff_proc.prepare()
         chunks = diff_proc_d[0]['chunks']
         self.assertFalse(chunks[0])
         #from pprint import pprint; pprint(chunks[1])
         l = ['\n']
         for d in chunks[1]:
             l.append('%(action)-7s %(new_lineno)3s %(old_lineno)3s %(line)r\n' % d)
         s = ''.join(l)
         print s
         self.assertEqual(s, r'''
 context ... ... u'@@ -51,5 +51,12 @@\n'
 unmod    51  51 u'<u>\t</u>begin();\n'
 unmod    52  52 u'<u>\t</u>\n'
 add      53     u'<u>\t</u>int foo;<u class="cr"></u>\n'
 add      54     u'<u>\t</u>int bar; <u class="cr"></u>\n'
 add      55     u'<u>\t</u>int baz;<u>\t</u><u class="cr"></u>\n'
 add      56     u'<u>\t</u>int space; <i></i>'
 add      57     u'<u>\t</u>int tab;<u>\t</u>\n'
 add      58     u'<u>\t</u>\n'
 unmod    59  53 u' <i></i>'
 del          54 u'<u>\t</u><del>#define MAX_STEPS (48)</del>\n'
 add      60     u'<u>\t</u><ins><u class="cr"></u></ins>\n'
 add      61     u'<u>\t</u>#define MAX_STEPS (64)<u class="cr"></u>\n'
 unmod    62  55 u'\n'
 ''')

kallithea/tests/other/test_libs.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.tests.other.test_libs
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Package for testing various lib/helper functions in kallithea
 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: Jun 9, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import with_statement
 import datetime
 import hashlib
 import mock
 from kallithea.tests import *
 from kallithea.lib.utils2 import AttributeDict
 from kallithea.model.db import Repository
 proto = 'http'
 TEST_URLS = [
     ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://username@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://username:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],
      '%s://127.0.0.1:8080' % proto),
     ('%s://domain.org' % proto, ['%s://' % proto, 'domain.org'],
      '%s://domain.org' % proto),
     ('%s://user:pass@domain.org:8080' % proto, ['%s://' % proto, 'domain.org',
                                                 '8080'],
      '%s://domain.org:8080' % proto),
+]
 proto = 'https'
 TEST_URLS += [
     ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://username@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://username:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],
      '%s://127.0.0.1:8080' % proto),
     ('%s://domain.org' % proto, ['%s://' % proto, 'domain.org'],
      '%s://domain.org' % proto),
     ('%s://user:pass@domain.org:8080' % proto, ['%s://' % proto, 'domain.org',
                                                 '8080'],
      '%s://domain.org:8080' % proto),
+]
 class TestLibs(BaseTestCase):
     @parameterized.expand(TEST_URLS)
     def test_uri_filter(self, test_url, expected, expected_creds):
         from kallithea.lib.utils2 import uri_filter
         self.assertEqual(uri_filter(test_url), expected)
     @parameterized.expand(TEST_URLS)
     def test_credentials_filter(self, test_url, expected, expected_creds):
         from kallithea.lib.utils2 import credentials_filter
         self.assertEqual(credentials_filter(test_url), expected_creds)
     @parameterized.expand([('t', True),
                            ('true', True),
                            ('y', True),
                            ('yes', True),
                            ('on', True),
                            ('1', True),
                            ('Y', True),
                            ('yeS', True),
                            ('Y', True),
                            ('TRUE', True),
                            ('T', True),
                            ('False', False),
                            ('F', False),
                            ('FALSE', False),
                            ('0', False),
                            ('-1', False),
                            ('', False)
     ])
     def test_str2bool(self, str_bool, expected):
         from kallithea.lib.utils2 import str2bool
         self.assertEqual(str2bool(str_bool), expected)
     def test_mention_extractor(self):
         from kallithea.lib.utils2 import extract_mentioned_users
         sample = (
             "@first hi there @world here's my email username@email.com "
             "@lukaszb check @one_more22 it pls @ ttwelve @D[] @one@two@three "
             "@UPPER    @cAmEL @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([
             '2one_more22', 'first', 'lukaszb', 'one', 'one_more22', 'UPPER', 'cAmEL', 'john',
             'marian.user', 'marco-polo', 'marco_polo', 'world'], key=lambda k: k.lower())
         self.assertEqual(s, extract_mentioned_users(sample))
     @parameterized.expand([
         (dict(), u'just now'),
         (dict(seconds= -1), u'1 second ago'),
         (dict(seconds= -60 * 2), u'2 minutes ago'),
         (dict(hours= -1), u'1 hour ago'),
         (dict(hours= -24), u'1 day ago'),
         (dict(hours= -24 * 5), u'5 days ago'),
         (dict(months= -1), u'1 month ago'),
         (dict(months= -1, days= -2), u'1 month and 2 days ago'),
         (dict(months= -1, days= -20), u'1 month and 19 days ago'),
         (dict(years= -1, months= -1), u'1 year and 1 month ago'),
         (dict(years= -1, months= -10), u'1 year and 10 months ago'),
         (dict(years= -2, months= -4), u'2 years and 4 months ago'),
         (dict(years= -2, months= -11), u'2 years and 11 months ago'),
         (dict(years= -3, months= -2), u'3 years and 2 months ago'),
     ])
     def test_age(self, age_args, expected):
         from kallithea.lib.utils2 import age
         from dateutil import relativedelta
         n = datetime.datetime(year=2012, month=5, day=17)
         delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
         self.assertEqual(age(n + delt(**age_args), now=n), expected)
     @parameterized.expand([
         (dict(), u'just now'),
         (dict(seconds= -1), u'1 second ago'),
         (dict(seconds= -60 * 2), u'2 minutes ago'),
         (dict(hours= -1), u'1 hour ago'),
         (dict(hours= -24), u'1 day ago'),
         (dict(hours= -24 * 5), u'5 days ago'),
         (dict(months= -1), u'1 month ago'),
         (dict(months= -1, days= -2), u'1 month ago'),
         (dict(months= -1, days= -20), u'1 month ago'),
         (dict(years= -1, months= -1), u'13 months ago'),
         (dict(years= -1, months= -10), u'22 months ago'),
         (dict(years= -2, months= -4), u'2 years ago'),
         (dict(years= -2, months= -11), u'3 years ago'),
         (dict(years= -3, months= -2), u'3 years ago'),
         (dict(years= -4, months= -8), u'5 years ago'),
     ])
     def test_age_short(self, age_args, expected):
         from kallithea.lib.utils2 import age
         from dateutil import relativedelta
         n = datetime.datetime(year=2012, month=5, day=17)
         delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
         self.assertEqual(age(n + delt(**age_args), show_short_version=True, now=n), expected)
     @parameterized.expand([
         (dict(), u'just now'),
         (dict(seconds=1), u'in 1 second'),
         (dict(seconds=60 * 2), u'in 2 minutes'),
         (dict(hours=1), u'in 1 hour'),
         (dict(hours=24), u'in 1 day'),
         (dict(hours=24 * 5), u'in 5 days'),
         (dict(months=1), u'in 1 month'),
         (dict(months=1, days=1), u'in 1 month and 1 day'),
         (dict(years=1, months=1), u'in 1 year and 1 month')
     ])
     def test_age_in_future(self, age_args, expected):
         from kallithea.lib.utils2 import age
         from dateutil import relativedelta
         n = datetime.datetime(year=2012, month=5, day=17)
         delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
         self.assertEqual(age(n + delt(**age_args), now=n), expected)
     def test_tag_exctrator(self):
         sample = (
             "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 kallithea.lib.helpers import urlify_text
         res = urlify_text(sample, stylize=True)
         self.assertIn('<div class="metatag" tag="tag">tag</div>', res)
         self.assertIn('<div class="metatag" tag="obsolete">obsolete</div>', res)
         self.assertIn('<div class="metatag" tag="stale">stale</div>', res)
         self.assertIn('<div class="metatag" tag="lang">python</div>', res)
         self.assertIn('<div class="metatag" tag="requires">requires =&gt; <a href="/url">url</a></div>', res)
         self.assertIn('<div class="metatag" tag="tag">tag</div>', res)
     def test_alternative_gravatar(self):
         from kallithea.lib.helpers import gravatar_url
         _md5 = lambda s: hashlib.md5(s).hexdigest()
         #mock pylons.url
         class fake_url(object):
             @classmethod
             def current(cls, *args, **kwargs):
                 return 'https://server.com'
         #mock pylons.tmpl_context
         def fake_tmpl_context(_url):
             _c = AttributeDict()
             _c.visual = AttributeDict()
             _c.visual.use_gravatar = True
             _c.visual.gravatar_url = _url
             return _c
         with mock.patch('pylons.url', fake_url):
             fake = fake_tmpl_context(_url='http://test.com/{email}')
             with mock.patch('pylons.tmpl_context', fake):
                     from pylons import url
                     assert url.current() == 'https://server.com'
                     grav = gravatar_url(email_address='test@foo.com', size=24)
                     assert grav == 'http://test.com/test@foo.com'
             fake = fake_tmpl_context(_url='http://test.com/{email}')
             with mock.patch('pylons.tmpl_context', fake):
                 grav = gravatar_url(email_address='test@foo.com', size=24)
                 assert grav == 'http://test.com/test@foo.com'
             fake = fake_tmpl_context(_url='http://test.com/{md5email}')
             with mock.patch('pylons.tmpl_context', fake):
                 em = 'test@foo.com'
                 grav = gravatar_url(email_address=em, size=24)
                 assert grav == 'http://test.com/%s' % (_md5(em))
             fake = fake_tmpl_context(_url='http://test.com/{md5email}/{size}')
             with mock.patch('pylons.tmpl_context', fake):
                 em = 'test@foo.com'
                 grav = gravatar_url(email_address=em, size=24)
                 assert grav == 'http://test.com/%s/%s' % (_md5(em), 24)
             fake = fake_tmpl_context(_url='{scheme}://{netloc}/{md5email}/{size}')
             with mock.patch('pylons.tmpl_context', fake):
                 em = 'test@foo.com'
                 grav = gravatar_url(email_address=em, size=24)
                 assert grav == 'https://server.com/%s/%s' % (_md5(em), 24)
     @parameterized.expand([
         (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '', 'http://vps1:8000/group/repo1'),
         (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/group/repo1'),
         (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '/prefix', 'http://vps1:8000/prefix/group/repo1'),
         (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/prefix', 'http://user@vps1:8000/prefix/group/repo1'),
         (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '/prefix', 'http://username@vps1:8000/prefix/group/repo1'),
         (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/prefix/', 'http://user@vps1:8000/prefix/group/repo1'),
         (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '/prefix/', 'http://username@vps1:8000/prefix/group/repo1'),
         ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {}, '', 'http://vps1:8000/_23'),
         ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/_23'),
         ('http://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/_23'),
         ('http://{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://vps1:8000/_23'),
         ('https://{user}@proxy1.server.com/{repo}', 'group/repo1', {'user': 'username'}, '', 'https://username@proxy1.server.com/group/repo1'),
         ('https://{user}@proxy1.server.com/{repo}', 'group/repo1', {}, '', 'https://proxy1.server.com/group/repo1'),
         ('https://proxy1.server.com/{user}/{repo}', 'group/repo1', {'user': 'username'}, '', 'https://proxy1.server.com/username/group/repo1'),
     ])
     def test_clone_url_generator(self, tmpl, repo_name, overrides, prefix, expected):
         from kallithea.lib.utils2 import get_clone_url
         clone_url = get_clone_url(uri_tmpl=tmpl, qualified_home_url='http://vps1:8000'+prefix,
                                   repo_name=repo_name, repo_id=23, **overrides)
         self.assertEqual(clone_url, expected)
     def _quick_url(self, text, tmpl="""<a class="revision-link" href="%s">%s</a>""", url_=None):
         """
         Changes `some text url[foo]` => `some text <a href="/">foo</a>
         :param text:
         """
         import re
         # quickly change expected url[] into a link
         URL_PAT = re.compile(r'(?:url\[)(.+?)(?:\])')
         def url_func(match_obj):
             _url = match_obj.groups()[0]
             return tmpl % (url_ or '/some-url', _url)
         return URL_PAT.sub(url_func, text)
     @parameterized.expand([
       ("",
        ""),
       ("git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68",
        "git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68"),
       ("from rev 000000000000",
        "from rev url[000000000000]"),
       ("from rev 000000000000123123 also rev 000000000000",
        "from rev url[000000000000123123] also rev url[000000000000]"),
       ("this should-000 00",
        "this should-000 00"),
       ("longtextffffffffff rev 123123123123",
        "longtextffffffffff rev url[123123123123]"),
       ("rev ffffffffffffffffffffffffffffffffffffffffffffffffff",
        "rev ffffffffffffffffffffffffffffffffffffffffffffffffff"),
       ("ffffffffffff some text traalaa",
        "url[ffffffffffff] some text traalaa"),
        ("""Multi line
        123123123123
        some text 123123123123
        sometimes !
        """,
        """Multi line
        url[123123123123]
        some text url[123123123123]
        sometimes !
        """)
     ])
     def test_urlify_changesets(self, sample, expected):
         def fake_url(self, *args, **kwargs):
             return '/some-url'
         expected = self._quick_url(expected)
         with mock.patch('pylons.url', fake_url):
             from kallithea.lib.helpers import urlify_changesets
             self.assertEqual(urlify_changesets(sample, 'repo_name'), expected)
     @parameterized.expand([
       ("",
        "",
        ""),
       ("https://svn.apache.org/repos",
        "url[https://svn.apache.org/repos]",
        "https://svn.apache.org/repos"),
       ("http://svn.apache.org/repos",
        "url[http://svn.apache.org/repos]",
        "http://svn.apache.org/repos"),
       ("from rev a also rev http://google.com",
        "from rev a also rev url[http://google.com]",
        "http://google.com"),
        ("""Multi line
        https://foo.bar.com
        some text lalala""",
        """Multi line
        url[https://foo.bar.com]
        some text lalala""",
        "https://foo.bar.com")
     ])
     def test_urlify_test(self, sample, expected, url_):
         from kallithea.lib.helpers import urlify_text
         expected = self._quick_url(expected,
                                    tmpl="""<a href="%s">%s</a>""", url_=url_)
         self.assertEqual(urlify_text(sample), expected)
     @parameterized.expand([
       ("", None),
       ("/_2", '2'),
       ("_2", '2'),
       ("/_2/", '2'),
       ("_2/", '2'),
       ("/_21", '21'),
       ("_21", '21'),
       ("/_21/", '21'),
       ("_21/", '21'),
       ("/_21/foobar", '21'),
       ("_21/121", '21'),
       ("/_21/_12", '21'),
       ("_21/prefix/foo", '21'),
     ])
     def test_get_repo_by_id(self, test, expected):
         from kallithea.lib.utils import _extract_id_from_repo_name
         _test = _extract_id_from_repo_name(test)
         self.assertEqual(_test, expected, msg='url:%s, got:`%s` expected: `%s`'
                                               % (test, _test, expected))

kallithea/tests/vcs/test_archives.py

➞

Show inline comments

 from __future__ import with_statement
 import os
 import tarfile
 import zipfile
 import datetime
 import tempfile
 import StringIO
 from kallithea.tests.vcs.base import _BackendTestMixin
 from kallithea.tests.vcs.conf import SCM_TESTS
 from kallithea.lib.vcs.exceptions import VCSError
 from kallithea.lib.vcs.nodes import FileNode
 from kallithea.lib.vcs.utils.compat import unittest
 class ArchivesTestCaseMixin(_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('%d/file_%d.txt' % (x, x),
                         content='Foobar %d' % x),
                 ],
+            }
     def test_archive_zip(self):
         path = tempfile.mkstemp()[1]
         with open(path, 'wb') as f:
             self.tip.fill_archive(stream=f, kind='zip', prefix='repo')
         out = zipfile.ZipFile(path)
         for x in xrange(5):
             node_path = '%d/file_%d.txt' % (x, x)
             decompressed = StringIO.StringIO()
             decompressed.write(out.read('repo/' + node_path))
             self.assertEqual(
                 decompressed.getvalue(),
                 self.tip.get_node(node_path).content)
     def test_archive_tgz(self):
         path = tempfile.mkstemp()[1]
         with open(path, 'wb') as f:
             self.tip.fill_archive(stream=f, kind='tgz', prefix='repo')
         outdir = tempfile.mkdtemp()
         outfile = tarfile.open(path, 'r|gz')
         outfile.extractall(outdir)
         for x in xrange(5):
             node_path = '%d/file_%d.txt' % (x, x)
             self.assertEqual(
                 open(os.path.join(outdir, 'repo/' + node_path)).read(),
                 self.tip.get_node(node_path).content)
     def test_archive_tbz2(self):
         path = tempfile.mkstemp()[1]
         with open(path, 'w+b') as f:
             self.tip.fill_archive(stream=f, kind='tbz2', prefix='repo')
         outdir = tempfile.mkdtemp()
         outfile = tarfile.open(path, 'r|bz2')
         outfile.extractall(outdir)
         for x in xrange(5):
             node_path = '%d/file_%d.txt' % (x, x)
             self.assertEqual(
                 open(os.path.join(outdir, 'repo/' + node_path)).read(),
                 self.tip.get_node(node_path).content)
     def test_archive_default_stream(self):
         tmppath = tempfile.mkstemp()[1]
         with open(tmppath, 'w') as stream:
             self.tip.fill_archive(stream=stream)
         mystream = StringIO.StringIO()
         self.tip.fill_archive(stream=mystream)
         mystream.seek(0)
         with open(tmppath, 'r') as f:
             self.assertEqual(f.read(), mystream.read())
     def test_archive_wrong_kind(self):
         with self.assertRaises(VCSError):
             self.tip.fill_archive(kind='wrong kind')
     def test_archive_empty_prefix(self):
         with self.assertRaises(VCSError):
             self.tip.fill_archive(prefix='')
     def test_archive_prefix_with_leading_slash(self):
         with self.assertRaises(VCSError):
             self.tip.fill_archive(prefix='/any')
 # For each backend create test case class
 for alias in SCM_TESTS:
     attrs = {
         'backend_alias': alias,
+    }
     cls_name = ''.join(('%s archive test' % alias).title().split())
     bases = (ArchivesTestCaseMixin, unittest.TestCase)
     globals()[cls_name] = type(cls_name, bases, attrs)
 if __name__ == '__main__':
     unittest.main()

kallithea/tests/vcs/test_branches.py

➞

Show inline comments

 from __future__ import with_statement
 import datetime
 from kallithea.lib import vcs
 from kallithea.lib.vcs.utils.compat import unittest
 from kallithea.lib.vcs.nodes import FileNode
 from kallithea.tests.vcs.base import _BackendTestMixin
 from kallithea.tests.vcs.conf import SCM_TESTS
 class BranchesTestCaseMixin(_BackendTestMixin):
     @classmethod
     def _get_commits(cls):
         commits = [
+            {
                 'message': 'Initial commit',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('foobar', content='Foobar'),
                     FileNode('foobar2', content='Foobar II'),
                     FileNode('foo/bar/baz', content='baz here!'),
                 ],
             },
+            {
                 'message': 'Changes...',
                 'author': 'Jane Doe <jane.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 21),
                 'added': [
                     FileNode('some/new.txt', content='news...'),
                 ],
                 'changed': [
                     FileNode('foobar', 'Foobar I'),
                 ],
                 'removed': [],
             },
+        ]
         return commits
     def test_simple(self):
         tip = self.repo.get_changeset()
         self.assertEqual(tip.date, datetime.datetime(2010, 1, 1, 21))
     def test_new_branch(self):
         # This check must not be removed to ensure the 'branches' LazyProperty
         # gets hit *before* the new 'foobar' branch got created:
         self.assertFalse('foobar' in self.repo.branches)
         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',
+        )
         self.assertTrue('foobar' in self.repo.branches)
         self.assertEqual(foobar_tip.branch, 'foobar')
     def test_new_head(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],
+        )
         self.assertEqual(newest_tip.branch,
             self.backend_class.DEFAULT_BRANCH_NAME)
     def test_branch_with_slash_in_name(self):
         self.imc.add(vcs.nodes.FileNode('extrafile', content='Some data\n'))
         self.imc.commit(u'Branch with a slash!', author=u'joe',
             branch='issue/123')
         self.assertTrue('issue/123' in self.repo.branches)
     def test_branch_with_slash_in_name_and_similar_without(self):
         self.imc.add(vcs.nodes.FileNode('extrafile', content='Some data\n'))
         self.imc.commit(u'Branch with a slash!', author=u'joe',
             branch='issue/123')
         self.imc.add(vcs.nodes.FileNode('extrafile II', content='Some data\n'))
         self.imc.commit(u'Branch without a slash...', author=u'joe',
             branch='123')
         self.assertIn('issue/123', self.repo.branches)
         self.assertIn('123', self.repo.branches)
 # For each backend create test case class
 for alias in SCM_TESTS:
     attrs = {
         'backend_alias': alias,
+    }
     cls_name = ''.join(('%s branches test' % alias).title().split())
     bases = (BranchesTestCaseMixin, unittest.TestCase)
     globals()[cls_name] = type(cls_name, bases, attrs)
 if __name__ == '__main__':
     unittest.main()

Changeset was too big and was cut off... Show full diff anyway

0 comments (0 inline, 0 general)