kallithea Changeset - b84c83b651de

Changeset - b84c83b651de

Parent rev.

Child rev.

[Not reviewed]

beta

0 9 0

Marcin Kuzminski - 12 years ago 2013-05-22 03:15:44
marcin@python-works.com

replace equality comparision to None

9 files changed with 11 insertions and 11 deletions:

rhodecode/bin/ldap_sync.py

rhodecode/controllers/changeset.py

rhodecode/lib/db_manage.py

rhodecode/lib/middleware/simplegit.py

rhodecode/lib/rcmail/response.py

rhodecode/lib/vcs/subprocessio.py

rhodecode/tests/functional/test_admin_settings.py

rhodecode/tests/other/test_validators.py

rhodecode/tests/scripts/test_concurency.py

0 comments (0 inline, 0 general)

rhodecode/bin/ldap_sync.py

➞

Show inline comments

 # 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/>.
 import ldap
 import urllib2
 import uuid
 try:
     from rhodecode.lib.compat import json
 except ImportError:
     try:
         import simplejson as json
     except ImportError:
         import json
 from ConfigParser import ConfigParser
 config = ConfigParser()
 config.read('ldap_sync.conf')
 class InvalidResponseIDError(Exception):
     """ Request and response don't have the same UUID. """
 class RhodecodeResponseError(Exception):
     """ Response has an error, something went wrong with request execution. """
 class UserAlreadyInGroupError(Exception):
     """ User is already a member of the target group. """
 class UserNotInGroupError(Exception):
     """ User is not a member of the target group. """
 class RhodecodeAPI():
     def __init__(self, url, key):
         self.url = url
         self.key = key
     def get_api_data(self, uid, method, args):
         """Prepare dict for API post."""
         return {
             "id": uid,
             "api_key": self.key,
             "method": method,
             "args": args
+        }
     def rhodecode_api_post(self, method, args):
         """Send a generic API post to Rhodecode.
         This will generate the UUID for validation check after the
         response is returned. Handle errors and get the result back.
         """
         uid = str(uuid.uuid1())
         data = self.get_api_data(uid, method, args)
         data = json.dumps(data)
         headers = {'content-type': 'text/plain'}
         req = urllib2.Request(self.url, data, headers)
         response = urllib2.urlopen(req)
         response = json.load(response)
         if uid != response["id"]:
             raise InvalidResponseIDError("UUID does not match.")
-        if response["error"] != None:
+        if response["error"] is not None:
             raise RhodecodeResponseError(response["error"])
         return response["result"]
     def create_group(self, name, active=True):
         """Create the Rhodecode user group."""
         args = {
             "group_name": name,
             "active": str(active)
+        }
         self.rhodecode_api_post("create_users_group", args)
     def add_membership(self, group, username):
         """Add specific user to a group."""
         args = {
             "usersgroupid": group,
             "userid": username
+        }
         result = self.rhodecode_api_post("add_user_to_users_group", args)
         if not result["success"]:
             raise UserAlreadyInGroupError("User %s already in group %s." %
                                           (username, group))
     def remove_membership(self, group, username):
         """Remove specific user from a group."""
         args = {
             "usersgroupid": group,
             "userid": username
+        }
         result = self.rhodecode_api_post("remove_user_from_users_group", args)
         if not result["success"]:
             raise UserNotInGroupError("User %s not in group %s." %
                                       (username, group))
     def get_group_members(self, name):
         """Get the list of member usernames from a user group."""
         args = {"usersgroupid": name}
         members = self.rhodecode_api_post("get_users_group", args)['members']
         member_list = []
         for member in members:
             member_list.append(member["username"])
         return member_list
     def get_group(self, name):
         """Return group info."""
         args = {"usersgroupid": name}
         return self.rhodecode_api_post("get_users_group", args)
     def get_user(self, username):
         """Return user info."""
         args = {"userid": username}
         return self.rhodecode_api_post("get_user", args)
 class LdapClient():
     def __init__(self, uri, user, key, base_dn):
         self.client = ldap.initialize(uri, trace_level=0)
         self.client.set_option(ldap.OPT_REFERRALS, 0)
         self.client.simple_bind(user, key)
         self.base_dn = base_dn
     def __del__(self):
         self.client.unbind()
     def get_groups(self):
         """Get all the groups in form of dict {group_name: group_info,...}."""
         searchFilter = "objectClass=groupOfUniqueNames"
         result = self.client.search_s(self.base_dn, ldap.SCOPE_SUBTREE,
                                       searchFilter)
         groups = {}
         for group in result:
             groups[group[1]['cn'][0]] = group[1]
         return groups
     def get_group_users(self, groups, group):
         """Returns all the users belonging to a single group.
         Based on the list of groups and memberships, returns all the
         users belonging to a single group, searching recursively.
         """
         users = []
         for member in groups[group]["uniqueMember"]:
             member = self.parse_member_string(member)
             if member[0] == "uid":
                 users.append(member[1])
             elif member[0] == "cn":
                 users += self.get_group_users(groups, member[1])
         return users
     def parse_member_string(self, member):
         """Parses the member string and returns a touple of type and name.
         Unique member can be either user or group. Users will have 'uid' as
         prefix while groups will have 'cn'.
         """
         member = member.split(",")[0]
         return member.split('=')
 class LdapSync(object):
     def __init__(self):
         self.ldap_client = LdapClient(config.get("default", "ldap_uri"),
                                       config.get("default", "ldap_user"),
                                       config.get("default", "ldap_key"),
                                       config.get("default", "base_dn"))
         self.rhodocode_api = RhodecodeAPI(config.get("default", "api_url"),
                                           config.get("default", "api_key"))
     def update_groups_from_ldap(self):
         """Add all the groups from LDAP to Rhodecode."""
         added = existing = 0
         groups = self.ldap_client.get_groups()
         for group in groups:
             try:
                 self.rhodecode_api.create_group(group)
                 added += 1
             except Exception:
                 existing += 1
         return added, existing
     def update_memberships_from_ldap(self, group):
         """Update memberships in rhodecode based on the LDAP groups."""
         groups = self.ldap_client.get_groups()
         group_users = self.ldap_client.get_group_users(groups, group)
         # Delete memberships first from each group which are not part
         # of the group any more.
         rhodecode_members = self.rhodecode_api.get_group_members(group)
         for rhodecode_member in rhodecode_members:
             if rhodecode_member not in group_users:
                 try:
                     self.rhodocode_api.remove_membership(group,
                                                          rhodecode_member)
                 except UserNotInGroupError:
                     pass
         # Add memberships.
         for member in group_users:
             try:
                 self.rhodecode_api.add_membership(group, member)
             except UserAlreadyInGroupError:
                 # TODO: handle somehow maybe..
                 pass
 if __name__ == '__main__':
     sync = LdapSync()
     print sync.update_groups_from_ldap()
     for gr in sync.ldap_client.get_groups():
         # TODO: exception when user does not exist during add membership...
         # How should we handle this.. Either sync users as well at this step,
         # or just ignore those who don't exist. If we want the second case,
         # we need to find a way to recognize the right exception (we always get
         # RhodecodeResponseError with no error code so maybe by return msg (?)
         sync.update_memberships_from_ldap(gr)

rhodecode/controllers/changeset.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.controllers.changeset
     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
     changeset controller for pylons showoing changes beetween
     revisions
     :created_on: Apr 25, 2010
     :author: marcink
     :copyright: (C) 2010-2012 Marcin Kuzminski <marcin@python-works.com>
     :license: GPLv3, see COPYING for more details.
 """
 # 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/>.
 import logging
 import traceback
 from collections import defaultdict
 from webob.exc import HTTPForbidden, HTTPBadRequest, HTTPNotFound
 from pylons import tmpl_context as c, url, request, response
 from pylons.i18n.translation import _
 from pylons.controllers.util import redirect
 from rhodecode.lib.utils import jsonify
 from rhodecode.lib.vcs.exceptions import RepositoryError, \
     ChangesetDoesNotExistError
 import rhodecode.lib.helpers as h
 from rhodecode.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator,\
     NotAnonymous
 from rhodecode.lib.base import BaseRepoController, render
 from rhodecode.lib.utils import action_logger
 from rhodecode.lib.compat import OrderedDict
 from rhodecode.lib import diffs
 from rhodecode.model.db import ChangesetComment, ChangesetStatus
 from rhodecode.model.comment import ChangesetCommentsModel
 from rhodecode.model.changeset_status import ChangesetStatusModel
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.lib.diffs import LimitedDiffContainer
 from rhodecode.lib.exceptions import StatusChangeOnClosedPullRequestError
 from rhodecode.lib.vcs.backends.base import EmptyChangeset
 from rhodecode.lib.utils2 import safe_unicode
 log = logging.getLogger(__name__)
 def _update_with_GET(params, GET):
     for k in ['diff1', 'diff2', 'diff']:
         params[k] += GET.getall(k)
 def anchor_url(revision, path, GET):
     fid = h.FID(revision, path)
     return h.url.current(anchor=fid, **dict(GET))
 def get_ignore_ws(fid, GET):
     ig_ws_global = GET.get('ignorews')
     ig_ws = filter(lambda k: k.startswith('WS'), GET.getall(fid))
     if ig_ws:
         try:
             return int(ig_ws[0].split(':')[-1])
         except Exception:
             pass
     return ig_ws_global
 def _ignorews_url(GET, fileid=None):
     fileid = str(fileid) if fileid else None
     params = defaultdict(list)
     _update_with_GET(params, GET)
     lbl = _('Show white space')
     ig_ws = get_ignore_ws(fileid, GET)
     ln_ctx = get_line_ctx(fileid, GET)
     # global option
     if fileid is None:
         if ig_ws is None:
             params['ignorews'] += [1]
             lbl = _('Ignore white space')
         ctx_key = 'context'
         ctx_val = ln_ctx
     # per file options
     else:
         if ig_ws is None:
             params[fileid] += ['WS:1']
             lbl = _('Ignore white space')
         ctx_key = fileid
         ctx_val = 'C:%s' % ln_ctx
     # if we have passed in ln_ctx pass it along to our params
     if ln_ctx:
         params[ctx_key] += [ctx_val]
     params['anchor'] = fileid
     img = h.image(h.url('/images/icons/text_strikethrough.png'), lbl, class_='icon')
     return h.link_to(img, h.url.current(**params), title=lbl, class_='tooltip')
 def get_line_ctx(fid, GET):
     ln_ctx_global = GET.get('context')
     if fid:
         ln_ctx = filter(lambda k: k.startswith('C'), GET.getall(fid))
     else:
         _ln_ctx = filter(lambda k: k.startswith('C'), GET)
         ln_ctx = GET.get(_ln_ctx[0]) if _ln_ctx  else ln_ctx_global
         if ln_ctx:
             ln_ctx = [ln_ctx]
     if ln_ctx:
         retval = ln_ctx[0].split(':')[-1]
     else:
         retval = ln_ctx_global
     try:
         return int(retval)
     except Exception:
         return 3
 def _context_url(GET, fileid=None):
     """
     Generates url for context lines
     :param fileid:
     """
     fileid = str(fileid) if fileid else None
     ig_ws = get_ignore_ws(fileid, GET)
     ln_ctx = (get_line_ctx(fileid, GET) or 3) * 2
     params = defaultdict(list)
     _update_with_GET(params, GET)
     # global option
     if fileid is None:
         if ln_ctx > 0:
             params['context'] += [ln_ctx]
         if ig_ws:
             ig_ws_key = 'ignorews'
             ig_ws_val = 1
     # per file option
     else:
         params[fileid] += ['C:%s' % ln_ctx]
         ig_ws_key = fileid
         ig_ws_val = 'WS:%s' % 1
     if ig_ws:
         params[ig_ws_key] += [ig_ws_val]
     lbl = _('%s line context') % ln_ctx
     params['anchor'] = fileid
     img = h.image(h.url('/images/icons/table_add.png'), lbl, class_='icon')
     return h.link_to(img, h.url.current(**params), title=lbl, class_='tooltip')
 class ChangesetController(BaseRepoController):
     def __before__(self):
         super(ChangesetController, self).__before__()
         c.affected_files_cut_off = 60
         repo_model = RepoModel()
         c.users_array = repo_model.get_users_js()
         c.users_groups_array = repo_model.get_users_groups_js()
     def _index(self, revision, method):
         c.anchor_url = anchor_url
         c.ignorews_url = _ignorews_url
         c.context_url = _context_url
         c.fulldiff = fulldiff = request.GET.get('fulldiff')
         #get ranges of revisions if preset
         rev_range = revision.split('...')[:2]
         enable_comments = True
         try:
             if len(rev_range) == 2:
                 enable_comments = False
                 rev_start = rev_range[0]
                 rev_end = rev_range[1]
                 rev_ranges = c.rhodecode_repo.get_changesets(start=rev_start,
                                                              end=rev_end)
             else:
                 rev_ranges = [c.rhodecode_repo.get_changeset(revision)]
             c.cs_ranges = list(rev_ranges)
             if not c.cs_ranges:
                 raise RepositoryError('Changeset range returned empty result')
         except (RepositoryError, ChangesetDoesNotExistError, Exception), e:
             log.error(traceback.format_exc())
             h.flash(str(e), category='error')
             raise HTTPNotFound()
         c.changes = OrderedDict()
         c.lines_added = 0  # count of lines added
         c.lines_deleted = 0  # count of lines removes
         c.changeset_statuses = ChangesetStatus.STATUSES
         c.comments = []
         c.statuses = []
         c.inline_comments = []
         c.inline_cnt = 0
         # Iterate over ranges (default changeset view is always one changeset)
         for changeset in c.cs_ranges:
             inlines = []
             if method == 'show':
                 c.statuses.extend([ChangesetStatusModel().get_status(
                             c.rhodecode_db_repo.repo_id, changeset.raw_id)])
                 c.comments.extend(ChangesetCommentsModel()\
                                   .get_comments(c.rhodecode_db_repo.repo_id,
                                                 revision=changeset.raw_id))
                 #comments from PR
                 st = ChangesetStatusModel().get_statuses(
                             c.rhodecode_db_repo.repo_id, changeset.raw_id,
                             with_revisions=True)
                 # from associated statuses, check the pull requests, and
                 # show comments from them
                 prs = set([x.pull_request for x in
-                           filter(lambda x: x.pull_request != None, st)])
+                           filter(lambda x: x.pull_request is not None, st)])
                 for pr in prs:
                     c.comments.extend(pr.comments)
                 inlines = ChangesetCommentsModel()\
                             .get_inline_comments(c.rhodecode_db_repo.repo_id,
                                                  revision=changeset.raw_id)
                 c.inline_comments.extend(inlines)
             c.changes[changeset.raw_id] = []
             cs2 = changeset.raw_id
             cs1 = changeset.parents[0].raw_id if changeset.parents else EmptyChangeset()
             context_lcl = get_line_ctx('', request.GET)
             ign_whitespace_lcl = ign_whitespace_lcl = get_ignore_ws('', request.GET)
             _diff = c.rhodecode_repo.get_diff(cs1, cs2,
                 ignore_whitespace=ign_whitespace_lcl, context=context_lcl)
             diff_limit = self.cut_off_limit if not fulldiff else None
             diff_processor = diffs.DiffProcessor(_diff,
                                                  vcs=c.rhodecode_repo.alias,
                                                  format='gitdiff',
                                                  diff_limit=diff_limit)
             cs_changes = OrderedDict()
             if method == 'show':
                 _parsed = diff_processor.prepare()
                 c.limited_diff = False
                 if isinstance(_parsed, LimitedDiffContainer):
                     c.limited_diff = True
                 for f in _parsed:
                     st = f['stats']
                     c.lines_added += st['added']
                     c.lines_deleted += st['deleted']
                     fid = h.FID(changeset.raw_id, f['filename'])
                     diff = diff_processor.as_html(enable_comments=enable_comments,
                                                   parsed_lines=[f])
                     cs_changes[fid] = [cs1, cs2, f['operation'], f['filename'],
                                        diff, st]
             else:
                 # downloads/raw we only need RAW diff nothing else
                 diff = diff_processor.as_raw()
                 cs_changes[''] = [None, None, None, None, diff, None]
             c.changes[changeset.raw_id] = cs_changes
         #sort comments by how they were generated
         c.comments = sorted(c.comments, key=lambda x: x.comment_id)
         # count inline comments
         for __, lines in c.inline_comments:
             for comments in lines.values():
                 c.inline_cnt += len(comments)
         if len(c.cs_ranges) == 1:
             c.changeset = c.cs_ranges[0]
             c.parent_tmpl = ''.join(['# Parent  %s\n' % x.raw_id
                                      for x in c.changeset.parents])
         if method == 'download':
             response.content_type = 'text/plain'
             response.content_disposition = 'attachment; filename=%s.diff' \
                                             % revision[:12]
             return diff
         elif method == 'patch':
             response.content_type = 'text/plain'
             c.diff = safe_unicode(diff)
             return render('changeset/patch_changeset.html')
         elif method == 'raw':
             response.content_type = 'text/plain'
             return diff
         elif method == 'show':
             if len(c.cs_ranges) == 1:
                 return render('changeset/changeset.html')
             else:
                 return render('changeset/changeset_range.html')
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def index(self, revision, method='show'):
         return self._index(revision, method=method)
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def changeset_raw(self, revision):
         return self._index(revision, method='raw')
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def changeset_patch(self, revision):
         return self._index(revision, method='patch')
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def changeset_download(self, revision):
         return self._index(revision, method='download')
     @LoginRequired()
     @NotAnonymous()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     @jsonify
     def comment(self, repo_name, revision):
         status = request.POST.get('changeset_status')
         change_status = request.POST.get('change_changeset_status')
         text = request.POST.get('text')
         if status and change_status:
             text = text or (_('Status change -> %s')
                             % ChangesetStatus.get_status_lbl(status))
         c.co = comm = ChangesetCommentsModel().create(
             text=text,
             repo=c.rhodecode_db_repo.repo_id,
             user=c.rhodecode_user.user_id,
             revision=revision,
             f_path=request.POST.get('f_path'),
             line_no=request.POST.get('line'),
             status_change=(ChangesetStatus.get_status_lbl(status)
                            if status and change_status else None)
+        )
         # get status if set !
         if status and change_status:
             # if latest status was from pull request and it's closed
             # disallow changing status !
             # dont_allow_on_closed_pull_request = True !
             try:
                 ChangesetStatusModel().set_status(
                     c.rhodecode_db_repo.repo_id,
                     status,
                     c.rhodecode_user.user_id,
                     comm,
                     revision=revision,
                     dont_allow_on_closed_pull_request=True
+                )
             except StatusChangeOnClosedPullRequestError:
                 log.error(traceback.format_exc())
                 msg = _('Changing status on a changeset associated with '
                         'a closed pull request is not allowed')
                 h.flash(msg, category='warning')
                 return redirect(h.url('changeset_home', repo_name=repo_name,
                                       revision=revision))
         action_logger(self.rhodecode_user,
                       'user_commented_revision:%s' % revision,
                       c.rhodecode_db_repo, self.ip_addr, self.sa)
         Session().commit()
         if not request.environ.get('HTTP_X_PARTIAL_XHR'):
             return redirect(h.url('changeset_home', repo_name=repo_name,
                                   revision=revision))
         #only ajax below
         data = {
            'target_id': h.safeid(h.safe_unicode(request.POST.get('f_path'))),
+        }
         if comm:
             data.update(comm.get_dict())
             data.update({'rendered_text':
                          render('changeset/changeset_comment_block.html')})
         return data
     @LoginRequired()
     @NotAnonymous()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     def preview_comment(self):
         if not request.environ.get('HTTP_X_PARTIAL_XHR'):
             raise HTTPBadRequest()
         text = request.POST.get('text')
         if text:
             return h.rst_w_mentions(text)
         return ''
     @LoginRequired()
     @NotAnonymous()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     @jsonify
     def delete_comment(self, repo_name, comment_id):
         co = ChangesetComment.get(comment_id)
         owner = co.author.user_id == c.rhodecode_user.user_id
         if h.HasPermissionAny('hg.admin', 'repository.admin')() or owner:
             ChangesetCommentsModel().delete(comment=co)
             Session().commit()
             return True
         else:
             raise HTTPForbidden()
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')
     @jsonify
     def changeset_info(self, repo_name, revision):
         if request.is_xhr:
             try:
                 return c.rhodecode_repo.get_changeset(revision)
             except ChangesetDoesNotExistError, e:
                 return EmptyChangeset(message=str(e))
         else:
             raise HTTPBadRequest()

rhodecode/lib/db_manage.py

➞

Show inline comments

@@ @@ -133,593 +133,593 @@ class DbManage(object): @@
         repository_path = jn(dn(dn(dn(os.path.realpath(__file__)))),
                              'rhodecode/lib/dbmigrate')
         db_uri = self.dburi
         try:
             curr_version = api.db_version(db_uri, repository_path)
             msg = ('Found current database under version'
                  ' control with version %s' % curr_version)
         except (RuntimeError, DatabaseNotControlledError):
             curr_version = 1
             msg = ('Current database is not under version control. Setting'
                    ' as version %s' % curr_version)
             api.version_control(db_uri, repository_path, curr_version)
         notify(msg)
         if curr_version == __dbversion__:
             sys.exit('This database is already at the newest version')
         # clear cache keys
         log.info("Clearing cache keys now...")
         CacheInvalidation.clear_cache()
         #======================================================================
         # UPGRADE STEPS
         #======================================================================
         class UpgradeSteps(object):
             """
             Those steps follow schema versions so for example schema
             for example schema with seq 002 == step_2 and so on.
             """
             def __init__(self, klass):
                 self.klass = klass
             def step_0(self):
                 # step 0 is the schema upgrade, and than follow proper upgrades
                 notify('attempting to do database upgrade from '
                        'version %s to version %s' %(curr_version, __dbversion__))
                 api.upgrade(db_uri, repository_path, __dbversion__)
                 notify('Schema upgrade completed')
             def step_1(self):
                 pass
             def step_2(self):
                 notify('Patching repo paths for newer version of RhodeCode')
                 self.klass.fix_repo_paths()
                 notify('Patching default user of RhodeCode')
                 self.klass.fix_default_user()
                 log.info('Changing ui settings')
                 self.klass.create_ui_settings()
             def step_3(self):
                 notify('Adding additional settings into RhodeCode db')
                 self.klass.fix_settings()
                 notify('Adding ldap defaults')
                 self.klass.create_ldap_options(skip_existing=True)
             def step_4(self):
                 notify('create permissions and fix groups')
                 self.klass.create_permissions()
                 self.klass.fixup_groups()
             def step_5(self):
                 pass
             def step_6(self):
                 notify('re-checking permissions')
                 self.klass.create_permissions()
                 notify('installing new UI options')
                 sett4 = RhodeCodeSetting('show_public_icon', True)
                 Session().add(sett4)
                 sett5 = RhodeCodeSetting('show_private_icon', True)
                 Session().add(sett5)
                 sett6 = RhodeCodeSetting('stylify_metatags', False)
                 Session().add(sett6)
                 notify('fixing old PULL hook')
                 _pull = RhodeCodeUi.get_by_key('preoutgoing.pull_logger')
                 if _pull:
                     _pull.ui_key = RhodeCodeUi.HOOK_PULL
                     Session().add(_pull)
                 notify('fixing old PUSH hook')
                 _push = RhodeCodeUi.get_by_key('pretxnchangegroup.push_logger')
                 if _push:
                     _push.ui_key = RhodeCodeUi.HOOK_PUSH
                     Session().add(_push)
                 notify('installing new pre-push hook')
                 hooks4 = RhodeCodeUi()
                 hooks4.ui_section = 'hooks'
                 hooks4.ui_key = RhodeCodeUi.HOOK_PRE_PUSH
                 hooks4.ui_value = 'python:rhodecode.lib.hooks.pre_push'
                 Session().add(hooks4)
                 notify('installing new pre-pull hook')
                 hooks6 = RhodeCodeUi()
                 hooks6.ui_section = 'hooks'
                 hooks6.ui_key = RhodeCodeUi.HOOK_PRE_PULL
                 hooks6.ui_value = 'python:rhodecode.lib.hooks.pre_pull'
                 Session().add(hooks6)
                 notify('installing hgsubversion option')
                 # enable hgsubversion disabled by default
                 hgsubversion = RhodeCodeUi()
                 hgsubversion.ui_section = 'extensions'
                 hgsubversion.ui_key = 'hgsubversion'
                 hgsubversion.ui_value = ''
                 hgsubversion.ui_active = False
                 Session().add(hgsubversion)
                 notify('installing hg git option')
                 # enable hggit disabled by default
                 hggit = RhodeCodeUi()
                 hggit.ui_section = 'extensions'
                 hggit.ui_key = 'hggit'
                 hggit.ui_value = ''
                 hggit.ui_active = False
                 Session().add(hggit)
                 notify('re-check default permissions')
                 default_user = User.get_by_username(User.DEFAULT_USER)
                 perm = Permission.get_by_key('hg.fork.repository')
                 reg_perm = UserToPerm()
                 reg_perm.user = default_user
                 reg_perm.permission = perm
                 Session().add(reg_perm)
             def step_7(self):
                 perm_fixes = self.klass.reset_permissions(User.DEFAULT_USER)
                 Session().commit()
                 if perm_fixes:
                     notify('There was an inconsistent state of permissions '
                            'detected for default user. Permissions are now '
                            'reset to the default value for default user. '
                            'Please validate and check default permissions '
                            'in admin panel')
             def step_8(self):
                 self.klass.create_permissions()
                 self.klass.populate_default_permissions()
                 self.klass.create_default_options(skip_existing=True)
                 Session().commit()
             def step_9(self):
                 perm_fixes = self.klass.reset_permissions(User.DEFAULT_USER)
                 Session().commit()
                 if perm_fixes:
                     notify('There was an inconsistent state of permissions '
                            'detected for default user. Permissions are now '
                            'reset to the default value for default user. '
                            'Please validate and check default permissions '
                            'in admin panel')
             def step_10(self):
                 pass
             def step_11(self):
                 self.klass.update_repo_info()
             def step_12(self):
                 self.klass.create_permissions()
                 Session().commit()
                 self.klass.populate_default_permissions()
                 Session().commit()
                 #fix all usergroups
                 ug_model = UserGroupModel()
                 for ug in UserGroup.get_all():
                     perm_obj = ug_model._create_default_perms(ug)
                     Session().add(perm_obj)
                 Session().commit()
         upgrade_steps = [0] + range(curr_version + 1, __dbversion__ + 1)
         # CALL THE PROPER ORDER OF STEPS TO PERFORM FULL UPGRADE
         _step = None
         for step in upgrade_steps:
             notify('performing upgrade step %s' % step)
             getattr(UpgradeSteps(self), 'step_%s' % step)()
             self.sa.commit()
             _step = step
         notify('upgrade to version %s successful' % _step)
     def fix_repo_paths(self):
         """
         Fixes a old rhodecode version path into new one without a '*'
         """
         paths = self.sa.query(RhodeCodeUi)\
                 .filter(RhodeCodeUi.ui_key == '/')\
                 .scalar()
         paths.ui_value = paths.ui_value.replace('*', '')
         try:
             self.sa.add(paths)
             self.sa.commit()
         except Exception:
             self.sa.rollback()
             raise
     def fix_default_user(self):
         """
         Fixes a old default user with some 'nicer' default values,
         used mostly for anonymous access
         """
         def_user = self.sa.query(User)\
                 .filter(User.username == 'default')\
                 .one()
         def_user.name = 'Anonymous'
         def_user.lastname = 'User'
         def_user.email = 'anonymous@rhodecode.org'
         try:
             self.sa.add(def_user)
             self.sa.commit()
         except Exception:
             self.sa.rollback()
             raise
     def fix_settings(self):
         """
         Fixes rhodecode settings adds ga_code key for google analytics
         """
         hgsettings3 = RhodeCodeSetting('ga_code', '')
         try:
             self.sa.add(hgsettings3)
             self.sa.commit()
         except Exception:
             self.sa.rollback()
             raise
     def admin_prompt(self, second=False):
         if not self.tests:
             import getpass
             # defaults
             defaults = self.cli_args
             username = defaults.get('username')
             password = defaults.get('password')
             email = defaults.get('email')
             def get_password():
                 password = getpass.getpass('Specify admin password '
                                            '(min 6 chars):')
                 confirm = getpass.getpass('Confirm password:')
                 if password != confirm:
                     log.error('passwords mismatch')
                     return False
                 if len(password) < 6:
                     log.error('password is to short use at least 6 characters')
                     return False
                 return password
             if username is None:
                 username = raw_input('Specify admin username:')
             if password is None:
                 password = get_password()
                 if not password:
                     #second try
                     password = get_password()
                     if not password:
                         sys.exit()
             if email is None:
                 email = raw_input('Specify admin email:')
             self.create_user(username, password, email, True)
         else:
             log.info('creating admin and regular test users')
             from rhodecode.tests import TEST_USER_ADMIN_LOGIN, \
             TEST_USER_ADMIN_PASS, TEST_USER_ADMIN_EMAIL, \
             TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS, \
             TEST_USER_REGULAR_EMAIL, TEST_USER_REGULAR2_LOGIN, \
             TEST_USER_REGULAR2_PASS, TEST_USER_REGULAR2_EMAIL
             self.create_user(TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS,
                              TEST_USER_ADMIN_EMAIL, True)
             self.create_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS,
                              TEST_USER_REGULAR_EMAIL, False)
             self.create_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS,
                              TEST_USER_REGULAR2_EMAIL, False)
     def create_ui_settings(self):
         """
         Creates ui settings, fills out hooks
         and disables dotencode
         """
         #HOOKS
         hooks1_key = RhodeCodeUi.HOOK_UPDATE
         hooks1_ = self.sa.query(RhodeCodeUi)\
             .filter(RhodeCodeUi.ui_key == hooks1_key).scalar()
         hooks1 = RhodeCodeUi() if hooks1_ is None else hooks1_
         hooks1.ui_section = 'hooks'
         hooks1.ui_key = hooks1_key
         hooks1.ui_value = 'hg update >&2'
         hooks1.ui_active = False
         self.sa.add(hooks1)
         hooks2_key = RhodeCodeUi.HOOK_REPO_SIZE
         hooks2_ = self.sa.query(RhodeCodeUi)\
             .filter(RhodeCodeUi.ui_key == hooks2_key).scalar()
         hooks2 = RhodeCodeUi() if hooks2_ is None else hooks2_
         hooks2.ui_section = 'hooks'
         hooks2.ui_key = hooks2_key
         hooks2.ui_value = 'python:rhodecode.lib.hooks.repo_size'
         self.sa.add(hooks2)
         hooks3 = RhodeCodeUi()
         hooks3.ui_section = 'hooks'
         hooks3.ui_key = RhodeCodeUi.HOOK_PUSH
         hooks3.ui_value = 'python:rhodecode.lib.hooks.log_push_action'
         self.sa.add(hooks3)
         hooks4 = RhodeCodeUi()
         hooks4.ui_section = 'hooks'
         hooks4.ui_key = RhodeCodeUi.HOOK_PRE_PUSH
         hooks4.ui_value = 'python:rhodecode.lib.hooks.pre_push'
         self.sa.add(hooks4)
         hooks5 = RhodeCodeUi()
         hooks5.ui_section = 'hooks'
         hooks5.ui_key = RhodeCodeUi.HOOK_PULL
         hooks5.ui_value = 'python:rhodecode.lib.hooks.log_pull_action'
         self.sa.add(hooks5)
         hooks6 = RhodeCodeUi()
         hooks6.ui_section = 'hooks'
         hooks6.ui_key = RhodeCodeUi.HOOK_PRE_PULL
         hooks6.ui_value = 'python:rhodecode.lib.hooks.pre_pull'
         self.sa.add(hooks6)
         # enable largefiles
         largefiles = RhodeCodeUi()
         largefiles.ui_section = 'extensions'
         largefiles.ui_key = 'largefiles'
         largefiles.ui_value = ''
         self.sa.add(largefiles)
         # enable hgsubversion disabled by default
         hgsubversion = RhodeCodeUi()
         hgsubversion.ui_section = 'extensions'
         hgsubversion.ui_key = 'hgsubversion'
         hgsubversion.ui_value = ''
         hgsubversion.ui_active = False
         self.sa.add(hgsubversion)
         # enable hggit disabled by default
         hggit = RhodeCodeUi()
         hggit.ui_section = 'extensions'
         hggit.ui_key = 'hggit'
         hggit.ui_value = ''
         hggit.ui_active = False
         self.sa.add(hggit)
     def create_ldap_options(self, skip_existing=False):
         """Creates ldap settings"""
         for k, v in [('ldap_active', 'false'), ('ldap_host', ''),
                     ('ldap_port', '389'), ('ldap_tls_kind', 'PLAIN'),
                     ('ldap_tls_reqcert', ''), ('ldap_dn_user', ''),
                     ('ldap_dn_pass', ''), ('ldap_base_dn', ''),
                     ('ldap_filter', ''), ('ldap_search_scope', ''),
                     ('ldap_attr_login', ''), ('ldap_attr_firstname', ''),
                     ('ldap_attr_lastname', ''), ('ldap_attr_email', '')]:
-            if skip_existing and RhodeCodeSetting.get_by_name(k) != None:
+            if skip_existing and RhodeCodeSetting.get_by_name(k) is not None:
                 log.debug('Skipping option %s' % k)
                 continue
             setting = RhodeCodeSetting(k, v)
             self.sa.add(setting)
     def create_default_options(self, skip_existing=False):
         """Creates default settings"""
         for k, v in [
             ('default_repo_enable_locking',  False),
             ('default_repo_enable_downloads', False),
             ('default_repo_enable_statistics', False),
             ('default_repo_private', False),
             ('default_repo_type', 'hg')]:
-            if skip_existing and RhodeCodeSetting.get_by_name(k) != None:
+            if skip_existing and RhodeCodeSetting.get_by_name(k) is not None:
                 log.debug('Skipping option %s' % k)
                 continue
             setting = RhodeCodeSetting(k, v)
             self.sa.add(setting)
     def fixup_groups(self):
         def_usr = User.get_default_user()
         for g in RepoGroup.query().all():
             g.group_name = g.get_new_name(g.name)
             self.sa.add(g)
             # get default perm
             default = UserRepoGroupToPerm.query()\
                 .filter(UserRepoGroupToPerm.group == g)\
                 .filter(UserRepoGroupToPerm.user == def_usr)\
                 .scalar()
             if default is None:
                 log.debug('missing default permission for group %s adding' % g)
                 perm_obj = ReposGroupModel()._create_default_perms(g)
                 self.sa.add(perm_obj)
     def reset_permissions(self, username):
         """
         Resets permissions to default state, usefull when old systems had
         bad permissions, we must clean them up
         :param username:
         """
         default_user = User.get_by_username(username)
         if not default_user:
             return
         u2p = UserToPerm.query()\
             .filter(UserToPerm.user == default_user).all()
         fixed = False
         if len(u2p) != len(Permission.DEFAULT_USER_PERMISSIONS):
             for p in u2p:
                 Session().delete(p)
             fixed = True
             self.populate_default_permissions()
         return fixed
     def update_repo_info(self):
         RepoModel.update_repoinfo()
     def config_prompt(self, test_repo_path='', retries=3):
         defaults = self.cli_args
         _path = defaults.get('repos_location')
         if retries == 3:
             log.info('Setting up repositories config')
         if _path is not None:
             path = _path
         elif not self.tests and not test_repo_path:
             path = raw_input(
                  'Enter a valid absolute path to store repositories. '
                  'All repositories in that path will be added automatically:'
+            )
         else:
             path = test_repo_path
         path_ok = True
         # check proper dir
         if not os.path.isdir(path):
             path_ok = False
             log.error('Given path %s is not a valid directory' % path)
         elif not os.path.isabs(path):
             path_ok = False
             log.error('Given path %s is not an absolute path' % path)
         # check write access
         elif not os.access(path, os.W_OK) and path_ok:
             path_ok = False
             log.error('No write permission to given path %s' % path)
         if retries == 0:
             sys.exit('max retries reached')
         if not path_ok:
             retries -= 1
             return self.config_prompt(test_repo_path, retries)
         real_path = os.path.normpath(os.path.realpath(path))
         if real_path != os.path.normpath(path):
             if not ask_ok(('Path looks like a symlink, Rhodecode will store '
                            'given path as %s ? [y/n]') % (real_path)):
                 log.error('Canceled by user')
                 sys.exit(-1)
         return real_path
     def create_settings(self, path):
         self.create_ui_settings()
         #HG UI OPTIONS
         web1 = RhodeCodeUi()
         web1.ui_section = 'web'
         web1.ui_key = 'push_ssl'
         web1.ui_value = 'false'
         web2 = RhodeCodeUi()
         web2.ui_section = 'web'
         web2.ui_key = 'allow_archive'
         web2.ui_value = 'gz zip bz2'
         web3 = RhodeCodeUi()
         web3.ui_section = 'web'
         web3.ui_key = 'allow_push'
         web3.ui_value = '*'
         web4 = RhodeCodeUi()
         web4.ui_section = 'web'
         web4.ui_key = 'baseurl'
         web4.ui_value = '/'
         paths = RhodeCodeUi()
         paths.ui_section = 'paths'
         paths.ui_key = '/'
         paths.ui_value = path
         phases = RhodeCodeUi()
         phases.ui_section = 'phases'
         phases.ui_key = 'publish'
         phases.ui_value = False
         sett1 = RhodeCodeSetting('realm', 'RhodeCode authentication')
         sett2 = RhodeCodeSetting('title', 'RhodeCode')
         sett3 = RhodeCodeSetting('ga_code', '')
         sett4 = RhodeCodeSetting('show_public_icon', True)
         sett5 = RhodeCodeSetting('show_private_icon', True)
         sett6 = RhodeCodeSetting('stylify_metatags', False)
         self.sa.add(web1)
         self.sa.add(web2)
         self.sa.add(web3)
         self.sa.add(web4)
         self.sa.add(paths)
         self.sa.add(sett1)
         self.sa.add(sett2)
         self.sa.add(sett3)
         self.sa.add(sett4)
         self.sa.add(sett5)
         self.sa.add(sett6)
         self.create_ldap_options()
         self.create_default_options()
         log.info('created ui config')
     def create_user(self, username, password, email='', admin=False):
         log.info('creating user %s' % username)
         UserModel().create_or_update(username, password, email,
                                      firstname='RhodeCode', lastname='Admin',
                                      active=True, admin=admin)
     def create_default_user(self):
         log.info('creating default user')
         # create default user for handling default permissions.
         UserModel().create_or_update(username='default',
                               password=str(uuid.uuid1())[:8],
                               email='anonymous@rhodecode.org',
                               firstname='Anonymous', lastname='User')
     def create_permissions(self):
         """
         Creates all permissions defined in the system
         """
         # module.(access|create|change|delete)_[name]
         # module.(none|read|write|admin)
         log.info('creating permissions')
         PermissionModel(self.sa).create_permissions()
     def populate_default_permissions(self):
         """
         Populate default permissions. It will create only the default
         permissions that are missing, and not alter already defined ones
         """
         log.info('creating default user permissions')
         PermissionModel(self.sa).create_default_permissions(user=User.DEFAULT_USER)
     @staticmethod
     def check_waitress():
         """
         Function executed at the end of setup
         """
         if not __py_version__ >= (2, 6):
             notify('Python2.5 detected, please switch '
                    'egg:waitress#main -> egg:Paste#http '
                    'in your .ini file')

rhodecode/lib/middleware/simplegit.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.lib.middleware.simplegit
     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
     SimpleGit middleware for handling git protocol request (push/clone etc.)
     It's implemented with basic auth function
     :created_on: Apr 28, 2010
     :author: marcink
     :copyright: (C) 2010-2012 Marcin Kuzminski <marcin@python-works.com>
     :license: GPLv3, see COPYING for more details.
 """
 # 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/>.
 import os
 import re
 import logging
 import traceback
 from dulwich import server as dulserver
 from dulwich.web import LimitedInputFilter, GunzipFilter
 from rhodecode.lib.exceptions import HTTPLockedRC
 from rhodecode.lib.hooks import pre_pull
 class SimpleGitUploadPackHandler(dulserver.UploadPackHandler):
     def handle(self):
         write = lambda x: self.proto.write_sideband(1, x)
         graph_walker = dulserver.ProtocolGraphWalker(self,
                                                      self.repo.object_store,
                                                      self.repo.get_peeled)
         objects_iter = self.repo.fetch_objects(
           graph_walker.determine_wants, graph_walker, self.progress,
           get_tagged=self.get_tagged)
         # Did the process short-circuit (e.g. in a stateless RPC call)? Note
         # that the client still expects a 0-object pack in most cases.
         if objects_iter is None:
             return
         self.progress("counting objects: %d, done.\n" % len(objects_iter))
         dulserver.write_pack_objects(dulserver.ProtocolFile(None, write),
                                      objects_iter)
         messages = []
         messages.append('thank you for using rhodecode')
         for msg in messages:
             self.progress(msg + "\n")
         # we are done
         self.proto.write("0000")
 dulserver.DEFAULT_HANDLERS = {
   #git-ls-remote, git-clone, git-fetch and git-pull
   'git-upload-pack': SimpleGitUploadPackHandler,
   #git-push
   'git-receive-pack': dulserver.ReceivePackHandler,
+}
 # not used for now until dulwich get's fixed
 #from dulwich.repo import Repo
 #from dulwich.web import make_wsgi_chain
 from paste.httpheaders import REMOTE_USER, AUTH_TYPE
 from webob.exc import HTTPNotFound, HTTPForbidden, HTTPInternalServerError, \
     HTTPBadRequest, HTTPNotAcceptable
 from rhodecode.lib.utils2 import safe_str, fix_PATH, get_server_url,\
     _set_extras
 from rhodecode.lib.base import BaseVCSController
 from rhodecode.lib.auth import get_container_username
 from rhodecode.lib.utils import is_valid_repo, make_ui
 from rhodecode.lib.compat import json
 from rhodecode.model.db import User, RhodeCodeUi
 log = logging.getLogger(__name__)
 GIT_PROTO_PAT = re.compile(r'^/(.+)/(info/refs|git-upload-pack|git-receive-pack)')
 def is_git(environ):
     path_info = environ['PATH_INFO']
     isgit_path = GIT_PROTO_PAT.match(path_info)
     log.debug('pathinfo: %s detected as GIT %s' % (
-        path_info, isgit_path != None)
+        path_info, isgit_path is not None)
+    )
     return isgit_path
 class SimpleGit(BaseVCSController):
     def _handle_request(self, environ, start_response):
         if not is_git(environ):
             return self.application(environ, start_response)
         if not self._check_ssl(environ, start_response):
             return HTTPNotAcceptable('SSL REQUIRED !')(environ, start_response)
         ip_addr = self._get_ip_addr(environ)
         username = None
         self._git_first_op = False
         # skip passing error to error controller
         environ['pylons.status_code_redirect'] = True
         #======================================================================
         # EXTRACT REPOSITORY NAME FROM ENV
         #======================================================================
         try:
             repo_name = self.__get_repository(environ)
             log.debug('Extracted repo name is %s' % repo_name)
         except Exception:
             return HTTPInternalServerError()(environ, start_response)
         # quick check if that dir exists...
         if not is_valid_repo(repo_name, self.basepath, 'git'):
             return HTTPNotFound()(environ, start_response)
         #======================================================================
         # GET ACTION PULL or PUSH
         #======================================================================
         action = self.__get_action(environ)
         #======================================================================
         # CHECK ANONYMOUS PERMISSION
         #======================================================================
         if action in ['pull', 'push']:
             anonymous_user = self.__get_user('default')
             username = anonymous_user.username
             anonymous_perm = self._check_permission(action, anonymous_user,
                                                     repo_name, ip_addr)
             if not anonymous_perm or not anonymous_user.active:
                 if not anonymous_perm:
                     log.debug('Not enough credentials to access this '
                               'repository as anonymous user')
                 if not anonymous_user.active:
                     log.debug('Anonymous access is disabled, running '
                               'authentication')
                 #==============================================================
                 # DEFAULT PERM FAILED OR ANONYMOUS ACCESS IS DISABLED SO WE
                 # NEED TO AUTHENTICATE AND ASK FOR AUTH USER PERMISSIONS
                 #==============================================================
                 # Attempting to retrieve username from the container
                 username = get_container_username(environ, self.config)
                 # If not authenticated by the container, running basic auth
                 if not username:
                     self.authenticate.realm = \
                         safe_str(self.config['rhodecode_realm'])
                     result = self.authenticate(environ)
                     if isinstance(result, str):
                         AUTH_TYPE.update(environ, 'basic')
                         REMOTE_USER.update(environ, result)
                         username = result
                     else:
                         return result.wsgi_application(environ, start_response)
                 #==============================================================
                 # CHECK PERMISSIONS FOR THIS REQUEST USING GIVEN USERNAME
                 #==============================================================
                 try:
                     user = self.__get_user(username)
                     if user is None or not user.active:
                         return HTTPForbidden()(environ, start_response)
                     username = user.username
                 except Exception:
                     log.error(traceback.format_exc())
                     return HTTPInternalServerError()(environ, start_response)
                 #check permissions for this repository
                 perm = self._check_permission(action, user, repo_name, ip_addr)
                 if not perm:
                     return HTTPForbidden()(environ, start_response)
         # extras are injected into UI object and later available
         # in hooks executed by rhodecode
         from rhodecode import CONFIG
         server_url = get_server_url(environ)
         extras = {
             'ip': ip_addr,
             'username': username,
             'action': action,
             'repository': repo_name,
             'scm': 'git',
             'config': CONFIG['__file__'],
             'server_url': server_url,
             'make_lock': None,
             'locked_by': [None, None]
+        }
         #===================================================================
         # GIT REQUEST HANDLING
         #===================================================================
         str_repo_name = safe_str(repo_name)
         repo_path = os.path.join(safe_str(self.basepath),str_repo_name)
         log.debug('Repository path is %s' % repo_path)
         # CHECK LOCKING only if it's not ANONYMOUS USER
         if username != User.DEFAULT_USER:
             log.debug('Checking locking on repository')
             (make_lock,
              locked,
              locked_by) = self._check_locking_state(
                             environ=environ, action=action,
                             repo=repo_name, user_id=user.user_id
+                       )
             # store the make_lock for later evaluation in hooks
             extras.update({'make_lock': make_lock,
                            'locked_by': locked_by})
         fix_PATH()
         log.debug('HOOKS extras is %s' % extras)
         baseui = make_ui('db')
         self.__inject_extras(repo_path, baseui, extras)
         try:
             self._handle_githooks(repo_name, action, baseui, environ)
             log.info('%s action on GIT repo "%s" by "%s" from %s' %
                      (action, str_repo_name, safe_str(username), ip_addr))
             app = self.__make_app(repo_name, repo_path, extras)
             return app(environ, start_response)
         except HTTPLockedRC, e:
             _code = CONFIG.get('lock_ret_code')
             log.debug('Repository LOCKED ret code %s!' % (_code))
             return e(environ, start_response)
         except Exception:
             log.error(traceback.format_exc())
             return HTTPInternalServerError()(environ, start_response)
         finally:
             # invalidate cache on push
             if action == 'push':
                 self._invalidate_cache(repo_name)
     def __make_app(self, repo_name, repo_path, extras):
         """
         Make an wsgi application using dulserver
         :param repo_name: name of the repository
         :param repo_path: full path to the repository
         """
         from rhodecode.lib.middleware.pygrack import make_wsgi_app
         app = make_wsgi_app(
             repo_root=safe_str(self.basepath),
             repo_name=repo_name,
             extras=extras,
+        )
         app = GunzipFilter(LimitedInputFilter(app))
         return app
     def __get_repository(self, environ):
         """
         Get's repository name out of PATH_INFO header
         :param environ: environ where PATH_INFO is stored
         """
         try:
             environ['PATH_INFO'] = self._get_by_id(environ['PATH_INFO'])
             repo_name = GIT_PROTO_PAT.match(environ['PATH_INFO']).group(1)
         except Exception:
             log.error(traceback.format_exc())
             raise
         return repo_name
     def __get_user(self, username):
         return User.get_by_username(username)
     def __get_action(self, environ):
         """
         Maps git request commands into a pull or push command.
         :param environ:
         """
         service = environ['QUERY_STRING'].split('=')
         if len(service) > 1:
             service_cmd = service[1]
             mapping = {
                 'git-receive-pack': 'push',
                 'git-upload-pack': 'pull',
+            }
             op = mapping[service_cmd]
             self._git_stored_op = op
             return op
         else:
             # try to fallback to stored variable as we don't know if the last
             # operation is pull/push
             op = getattr(self, '_git_stored_op', 'pull')
         return op
     def _handle_githooks(self, repo_name, action, baseui, environ):
         """
         Handles pull action, push is handled by post-receive hook
         """
         from rhodecode.lib.hooks import log_pull_action
         service = environ['QUERY_STRING'].split('=')
         if len(service) < 2:
             return
         from rhodecode.model.db import Repository
         _repo = Repository.get_by_repo_name(repo_name)
         _repo = _repo.scm_instance
         _hooks = dict(baseui.configitems('hooks')) or {}
         if action == 'pull':
             # stupid git, emulate pre-pull hook !
             pre_pull(ui=baseui, repo=_repo._repo)
         if action == 'pull' and _hooks.get(RhodeCodeUi.HOOK_PULL):
             log_pull_action(ui=baseui, repo=_repo._repo)
     def __inject_extras(self, repo_path, baseui, extras={}):
         """
         Injects some extra params into baseui instance
         :param baseui: baseui instance
         :param extras: dict with extra params to put into baseui
         """
         _set_extras(extras)

rhodecode/lib/rcmail/response.py

➞

Show inline comments

 # The code in this module is entirely lifted from the Lamson project
 # (http://lamsonproject.org/).  Its copyright is:
 # Copyright (c) 2008, Zed A. Shaw
 # All rights reserved.
 # It is provided under this license:
 # Redistribution and use in source and binary forms, with or without
 # modification, are permitted provided that the following conditions are met:
 # * Redistributions of source code must retain the above copyright notice, this
 #   list of conditions and the following disclaimer.
 # * Redistributions in binary form must reproduce the above copyright notice,
 #   this list of conditions and the following disclaimer in the documentation
 #   and/or other materials provided with the distribution.
 # * Neither the name of the Zed A. Shaw nor the names of its contributors may
 #   be used to endorse or promote products derived from this software without
 #   specific prior written permission.
 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
 # FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
 # COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
 # INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 # (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 # SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 # HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
 # STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 # POSSIBILITY OF SUCH DAMAGE.
 import os
 import mimetypes
 import string
 from email import encoders
 from email.charset import Charset
 from email.utils import parseaddr
 from email.mime.base import MIMEBase
 ADDRESS_HEADERS_WHITELIST = ['From', 'To', 'Delivered-To', 'Cc']
 DEFAULT_ENCODING = "utf-8"
 VALUE_IS_EMAIL_ADDRESS = lambda v: '@' in v
 def normalize_header(header):
     return string.capwords(header.lower(), '-')
 class EncodingError(Exception):
     """Thrown when there is an encoding error."""
     pass
 class MailBase(object):
     """MailBase is used as the basis of lamson.mail and contains the basics of
     encoding an email.  You actually can do all your email processing with this
     class, but it's more raw.
     """
     def __init__(self, items=()):
         self.headers = dict(items)
         self.parts = []
         self.body = None
         self.content_encoding = {'Content-Type': (None, {}),
                                  'Content-Disposition': (None, {}),
                                  'Content-Transfer-Encoding': (None, {})}
     def __getitem__(self, key):
         return self.headers.get(normalize_header(key), None)
     def __len__(self):
         return len(self.headers)
     def __iter__(self):
         return iter(self.headers)
     def __contains__(self, key):
         return normalize_header(key) in self.headers
     def __setitem__(self, key, value):
         self.headers[normalize_header(key)] = value
     def __delitem__(self, key):
         del self.headers[normalize_header(key)]
     def __nonzero__(self):
-        return self.body != None or len(self.headers) > 0 or len(self.parts) > 0
+        return self.body is not None or len(self.headers) > 0 or len(self.parts) > 0
     def keys(self):
         """Returns the sorted keys."""
         return sorted(self.headers.keys())
     def attach_file(self, filename, data, ctype, disposition):
         """
         A file attachment is a raw attachment with a disposition that
         indicates the file name.
         """
         assert filename, "You can't attach a file without a filename."
         ctype = ctype.lower()
         part = MailBase()
         part.body = data
         part.content_encoding['Content-Type'] = (ctype, {'name': filename})
         part.content_encoding['Content-Disposition'] = (disposition,
                                                         {'filename': filename})
         self.parts.append(part)
     def attach_text(self, data, ctype):
         """
         This attaches a simpler text encoded part, which doesn't have a
         filename.
         """
         ctype = ctype.lower()
         part = MailBase()
         part.body = data
         part.content_encoding['Content-Type'] = (ctype, {})
         self.parts.append(part)
     def walk(self):
         for p in self.parts:
             yield p
             for x in p.walk():
                 yield x
 class MailResponse(object):
     """
     You are given MailResponse objects from the lamson.view methods, and
     whenever you want to generate an email to send to someone.  It has the
     same basic functionality as MailRequest, but it is designed to be written
     to, rather than read from (although you can do both).
     You can easily set a Body or Html during creation or after by passing it
     as __init__ parameters, or by setting those attributes.
     You can initially set the From, To, and Subject, but they are headers so
     use the dict notation to change them: msg['From'] = 'joe@test.com'.
     The message is not fully crafted until right when you convert it with
     MailResponse.to_message.  This lets you change it and work with it, then
     send it out when it's ready.
     """
     def __init__(self, To=None, From=None, Subject=None, Body=None, Html=None,
                  separator="; "):
         self.Body = Body
         self.Html = Html
         self.base = MailBase([('To', To), ('From', From), ('Subject', Subject)])
         self.multipart = self.Body and self.Html
         self.attachments = []
         self.separator = separator
     def __contains__(self, key):
         return self.base.__contains__(key)
     def __getitem__(self, key):
         return self.base.__getitem__(key)
     def __setitem__(self, key, val):
         return self.base.__setitem__(key, val)
     def __delitem__(self, name):
         del self.base[name]
     def attach(self, filename=None, content_type=None, data=None,
                disposition=None):
         """
         Simplifies attaching files from disk or data as files.  To attach
         simple text simple give data and a content_type.  To attach a file,
         give the data/content_type/filename/disposition combination.
         For convenience, if you don't give data and only a filename, then it
         will read that file's contents when you call to_message() later.  If
         you give data and filename then it will assume you've filled data
         with what the file's contents are and filename is just the name to
         use.
         """
         assert filename or data, ("You must give a filename or some data to "
                                   "attach.")
         assert data or os.path.exists(filename), ("File doesn't exist, and no "
                                                   "data given.")
         self.multipart = True
         if filename and not content_type:
             content_type, encoding = mimetypes.guess_type(filename)
         assert content_type, ("No content type given, and couldn't guess "
                               "from the filename: %r" % filename)
         self.attachments.append({'filename': filename,
                                  'content_type': content_type,
                                  'data': data,
                                  'disposition': disposition,})
     def attach_part(self, part):
         """
         Attaches a raw MailBase part from a MailRequest (or anywhere)
         so that you can copy it over.
         """
         self.multipart = True
         self.attachments.append({'filename': None,
                                  'content_type': None,
                                  'data': None,
                                  'disposition': None,
                                  'part': part,
                                  })
     def attach_all_parts(self, mail_request):
         """
         Used for copying the attachment parts of a mail.MailRequest
         object for mailing lists that need to maintain attachments.
         """
         for part in mail_request.all_parts():
             self.attach_part(part)
         self.base.content_encoding = mail_request.base.content_encoding.copy()
     def clear(self):
         """
         Clears out the attachments so you can redo them.  Use this to keep the
         headers for a series of different messages with different attachments.
         """
         del self.attachments[:]
         del self.base.parts[:]
         self.multipart = False
     def update(self, message):
         """
         Used to easily set a bunch of heading from another dict
         like object.
         """
         for k in message.keys():
             self.base[k] = message[k]
     def __str__(self):
         """
         Converts to a string.
         """
         return self.to_message().as_string()
     def _encode_attachment(self, filename=None, content_type=None, data=None,
                            disposition=None, part=None):
         """
         Used internally to take the attachments mentioned in self.attachments
         and do the actual encoding in a lazy way when you call to_message.
         """
         if part:
             self.base.parts.append(part)
         elif filename:
             if not data:
                 data = open(filename).read()
             self.base.attach_file(filename, data, content_type,
                                   disposition or 'attachment')
         else:
             self.base.attach_text(data, content_type)
         ctype = self.base.content_encoding['Content-Type'][0]
         if ctype and not ctype.startswith('multipart'):
             self.base.content_encoding['Content-Type'] = ('multipart/mixed', {})
     def to_message(self):
         """
         Figures out all the required steps to finally craft the
         message you need and return it.  The resulting message
         is also available as a self.base attribute.
         What is returned is a Python email API message you can
         use with those APIs.  The self.base attribute is the raw
         lamson.encoding.MailBase.
         """
         del self.base.parts[:]
         if self.Body and self.Html:
             self.multipart = True
             self.base.content_encoding['Content-Type'] = (
                 'multipart/alternative', {})
         if self.multipart:
             self.base.body = None
             if self.Body:
                 self.base.attach_text(self.Body, 'text/plain')
             if self.Html:
                 self.base.attach_text(self.Html, 'text/html')
             for args in self.attachments:
                 self._encode_attachment(**args)
         elif self.Body:
             self.base.body = self.Body
             self.base.content_encoding['Content-Type'] = ('text/plain', {})
         elif self.Html:
             self.base.body = self.Html
             self.base.content_encoding['Content-Type'] = ('text/html', {})
         return to_message(self.base, separator=self.separator)
     def all_parts(self):
         """
         Returns all the encoded parts.  Only useful for debugging
         or inspecting after calling to_message().
         """
         return self.base.parts
     def keys(self):
         return self.base.keys()
 def to_message(mail, separator="; "):
     """
     Given a MailBase message, this will construct a MIMEPart
     that is canonicalized for use with the Python email API.
     """
     ctype, params = mail.content_encoding['Content-Type']
     if not ctype:
         if mail.parts:
             ctype = 'multipart/mixed'
         else:
             ctype = 'text/plain'
     else:
         if mail.parts:
             assert ctype.startswith(("multipart", "message")), \
                    "Content type should be multipart or message, not %r" % ctype
     # adjust the content type according to what it should be now
     mail.content_encoding['Content-Type'] = (ctype, params)
     try:
         out = MIMEPart(ctype, **params)
     except TypeError, exc:  # pragma: no cover
         raise EncodingError("Content-Type malformed, not allowed: %r; "
                             "%r (Python ERROR: %s" %
                             (ctype, params, exc.message))
     for k in mail.keys():
         if k in ADDRESS_HEADERS_WHITELIST:
             out[k.encode('ascii')] = header_to_mime_encoding(
                                          mail[k],
                                          not_email=False,
                                          separator=separator
+                                     )
         else:
             out[k.encode('ascii')] = header_to_mime_encoding(
                                          mail[k],
                                          not_email=True
+                                    )
     out.extract_payload(mail)
     # go through the children
     for part in mail.parts:
         out.attach(to_message(part))
     return out
 class MIMEPart(MIMEBase):
     """
     A reimplementation of nearly everything in email.mime to be more useful
     for actually attaching things.  Rather than one class for every type of
     thing you'd encode, there's just this one, and it figures out how to
     encode what you ask it.
     """
     def __init__(self, type, **params):
         self.maintype, self.subtype = type.split('/')
         MIMEBase.__init__(self, self.maintype, self.subtype, **params)
     def add_text(self, content):
         # this is text, so encode it in canonical form
         try:
             encoded = content.encode('ascii')
             charset = 'ascii'
         except UnicodeError:
             encoded = content.encode('utf-8')
             charset = 'utf-8'
         self.set_payload(encoded, charset=charset)
     def extract_payload(self, mail):
-        if mail.body == None:
+        if mail.body is None:
             return  # only None, '' is still ok
         ctype, ctype_params = mail.content_encoding['Content-Type']
         cdisp, cdisp_params = mail.content_encoding['Content-Disposition']
         assert ctype, ("Extract payload requires that mail.content_encoding "
                        "have a valid Content-Type.")
         if ctype.startswith("text/"):
             self.add_text(mail.body)
         else:
             if cdisp:
                 # replicate the content-disposition settings
                 self.add_header('Content-Disposition', cdisp, **cdisp_params)
             self.set_payload(mail.body)
             encoders.encode_base64(self)
     def __repr__(self):
         return "<MIMEPart '%s/%s': %r, %r, multipart=%r>" % (
             self.subtype,
             self.maintype,
             self['Content-Type'],
             self['Content-Disposition'],
             self.is_multipart())
 def header_to_mime_encoding(value, not_email=False, separator=", "):
     if not value:
         return ""
     encoder = Charset(DEFAULT_ENCODING)
     if type(value) == list:
         return separator.join(properly_encode_header(
             v, encoder, not_email) for v in value)
     else:
         return properly_encode_header(value, encoder, not_email)
 def properly_encode_header(value, encoder, not_email):
     """
     The only thing special (weird) about this function is that it tries
     to do a fast check to see if the header value has an email address in
     it.  Since random headers could have an email address, and email addresses
     have weird special formatting rules, we have to check for it.
     Normally this works fine, but in Librelist, we need to "obfuscate" email
     addresses by changing the '@' to '-AT-'.  This is where
     VALUE_IS_EMAIL_ADDRESS exists.  It's a simple lambda returning True/False
     to check if a header value has an email address.  If you need to make this
     check different, then change this.
     """
     try:
         return value.encode("ascii")
     except UnicodeEncodeError:
         if not not_email and VALUE_IS_EMAIL_ADDRESS(value):
             # this could have an email address, make sure we don't screw it up
             name, address = parseaddr(value)
             return '"%s" <%s>' % (
                 encoder.header_encode(name.encode("utf-8")), address)
         return encoder.header_encode(value.encode("utf-8"))

rhodecode/lib/vcs/subprocessio.py

➞

Show inline comments

 """
 Module provides a class allowing to wrap communication over subprocess.Popen
 input, output, error streams into a meaningfull, non-blocking, concurrent
 stream processor exposing the output data as an iterator fitting to be a
 return value passed by a WSGI applicaiton to a WSGI server per PEP 3333.
 Copyright (c) 2011  Daniel Dotsenko <dotsa@hotmail.com>
 This file is part of git_http_backend.py Project.
 git_http_backend.py Project is free software: you can redistribute it and/or
 modify it under the terms of the GNU Lesser General Public License as
 published by the Free Software Foundation, either version 2.1 of the License,
 or (at your option) any later version.
 git_http_backend.py Project 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 Lesser General Public License for more details.
 You should have received a copy of the GNU Lesser General Public License
 along with git_http_backend.py Project.
 If not, see <http://www.gnu.org/licenses/>.
 """
 import os
 import subprocess
 from rhodecode.lib.vcs.utils.compat import deque, Event, Thread, _bytes, _bytearray
 class StreamFeeder(Thread):
     """
     Normal writing into pipe-like is blocking once the buffer is filled.
     This thread allows a thread to seep data from a file-like into a pipe
     without blocking the main thread.
     We close inpipe once the end of the source stream is reached.
     """
     def __init__(self, source):
         super(StreamFeeder, self).__init__()
         self.daemon = True
         filelike = False
         self.bytes = _bytes()
         if type(source) in (type(''), _bytes, _bytearray):  # string-like
             self.bytes = _bytes(source)
         else:  # can be either file pointer or file-like
             if type(source) in (int, long):  # file pointer it is
                 ## converting file descriptor (int) stdin into file-like
                 try:
                     source = os.fdopen(source, 'rb', 16384)
                 except Exception:
                     pass
             # let's see if source is file-like by now
             try:
                 filelike = source.read
             except Exception:
                 pass
         if not filelike and not self.bytes:
             raise TypeError("StreamFeeder's source object must be a readable "
                             "file-like, a file descriptor, or a string-like.")
         self.source = source
         self.readiface, self.writeiface = os.pipe()
     def run(self):
         t = self.writeiface
         if self.bytes:
             os.write(t, self.bytes)
         else:
             s = self.source
             b = s.read(4096)
             while b:
                 os.write(t, b)
                 b = s.read(4096)
         os.close(t)
     @property
     def output(self):
         return self.readiface
 class InputStreamChunker(Thread):
     def __init__(self, source, target, buffer_size, chunk_size):
         super(InputStreamChunker, self).__init__()
         self.daemon = True  # die die die.
         self.source = source
         self.target = target
         self.chunk_count_max = int(buffer_size / chunk_size) + 1
         self.chunk_size = chunk_size
         self.data_added = Event()
         self.data_added.clear()
         self.keep_reading = Event()
         self.keep_reading.set()
         self.EOF = Event()
         self.EOF.clear()
         self.go = Event()
         self.go.set()
     def stop(self):
         self.go.clear()
         self.EOF.set()
         try:
             # this is not proper, but is done to force the reader thread let
             # go of the input because, if successful, .close() will send EOF
             # down the pipe.
             self.source.close()
         except:
             pass
     def run(self):
         s = self.source
         t = self.target
         cs = self.chunk_size
         ccm = self.chunk_count_max
         kr = self.keep_reading
         da = self.data_added
         go = self.go
         try:
             b = s.read(cs)
         except ValueError:
             b = ''
         while b and go.is_set():
             if len(t) > ccm:
                 kr.clear()
                 kr.wait(2)
 #                # this only works on 2.7.x and up
 #                if not kr.wait(10):
 #                    raise Exception("Timed out while waiting for input to be read.")
                 # instead we'll use this
                 if len(t) > ccm + 3:
                     raise IOError("Timed out while waiting for input from subprocess.")
             t.append(b)
             da.set()
             b = s.read(cs)
         self.EOF.set()
         da.set()  # for cases when done but there was no input.
 class BufferedGenerator():
     """
     Class behaves as a non-blocking, buffered pipe reader.
     Reads chunks of data (through a thread)
     from a blocking pipe, and attaches these to an array (Deque) of chunks.
     Reading is halted in the thread when max chunks is internally buffered.
     The .next() may operate in blocking or non-blocking fashion by yielding
     '' if no data is ready
     to be sent or by not returning until there is some data to send
     When we get EOF from underlying source pipe we raise the marker to raise
     StopIteration after the last chunk of data is yielded.
     """
     def __init__(self, source, buffer_size=65536, chunk_size=4096,
                  starting_values=[], bottomless=False):
         if bottomless:
             maxlen = int(buffer_size / chunk_size)
         else:
             maxlen = None
         self.data = deque(starting_values, maxlen)
         self.worker = InputStreamChunker(source, self.data, buffer_size,
                                          chunk_size)
         if starting_values:
             self.worker.data_added.set()
         self.worker.start()
     ####################
     # Generator's methods
     ####################
     def __iter__(self):
         return self
     def next(self):
         while not len(self.data) and not self.worker.EOF.is_set():
             self.worker.data_added.clear()
             self.worker.data_added.wait(0.2)
         if len(self.data):
             self.worker.keep_reading.set()
             return _bytes(self.data.popleft())
         elif self.worker.EOF.is_set():
             raise StopIteration
     def throw(self, type, value=None, traceback=None):
         if not self.worker.EOF.is_set():
             raise type(value)
     def start(self):
         self.worker.start()
     def stop(self):
         self.worker.stop()
     def close(self):
         try:
             self.worker.stop()
             self.throw(GeneratorExit)
         except (GeneratorExit, StopIteration):
             pass
     def __del__(self):
         self.close()
     ####################
     # Threaded reader's infrastructure.
     ####################
     @property
     def input(self):
         return self.worker.w
     @property
     def data_added_event(self):
         return self.worker.data_added
     @property
     def data_added(self):
         return self.worker.data_added.is_set()
     @property
     def reading_paused(self):
         return not self.worker.keep_reading.is_set()
     @property
     def done_reading_event(self):
         """
         Done_reding does not mean that the iterator's buffer is empty.
         Iterator might have done reading from underlying source, but the read
         chunks might still be available for serving through .next() method.
         @return An Event class instance.
         """
         return self.worker.EOF
     @property
     def done_reading(self):
         """
         Done_reding does not mean that the iterator's buffer is empty.
         Iterator might have done reading from underlying source, but the read
         chunks might still be available for serving through .next() method.
         @return An Bool value.
         """
         return self.worker.EOF.is_set()
     @property
     def length(self):
         """
         returns int.
         This is the lenght of the que of chunks, not the length of
         the combined contents in those chunks.
         __len__() cannot be meaningfully implemented because this
         reader is just flying throuh a bottomless pit content and
         can only know the lenght of what it already saw.
         If __len__() on WSGI server per PEP 3333 returns a value,
         the responce's length will be set to that. In order not to
         confuse WSGI PEP3333 servers, we will not implement __len__
         at all.
         """
         return len(self.data)
     def prepend(self, x):
         self.data.appendleft(x)
     def append(self, x):
         self.data.append(x)
     def extend(self, o):
         self.data.extend(o)
     def __getitem__(self, i):
         return self.data[i]
 class SubprocessIOChunker(object):
     """
     Processor class wrapping handling of subprocess IO.
     In a way, this is a "communicate()" replacement with a twist.
     - We are multithreaded. Writing in and reading out, err are all sep threads.
     - We support concurrent (in and out) stream processing.
     - The output is not a stream. It's a queue of read string (bytes, not unicode)
       chunks. The object behaves as an iterable. You can "for chunk in obj:" us.
     - We are non-blocking in more respects than communicate()
       (reading from subprocess out pauses when internal buffer is full, but
        does not block the parent calling code. On the flip side, reading from
        slow-yielding subprocess may block the iteration until data shows up. This
        does not block the parallel inpipe reading occurring parallel thread.)
     The purpose of the object is to allow us to wrap subprocess interactions into
     and interable that can be passed to a WSGI server as the application's return
     value. Because of stream-processing-ability, WSGI does not have to read ALL
     of the subprocess's output and buffer it, before handing it to WSGI server for
     HTTP response. Instead, the class initializer reads just a bit of the stream
     to figure out if error ocurred or likely to occur and if not, just hands the
     further iteration over subprocess output to the server for completion of HTTP
     response.
     The real or perceived subprocess error is trapped and raised as one of
     EnvironmentError family of exceptions
     Example usage:
     #    try:
     #        answer = SubprocessIOChunker(
     #            cmd,
     #            input,
     #            buffer_size = 65536,
     #            chunk_size = 4096
     #            )
     #    except (EnvironmentError) as e:
     #        print str(e)
     #        raise e
+    #
     #    return answer
     """
     def __init__(self, cmd, inputstream=None, buffer_size=65536,
                  chunk_size=4096, starting_values=[], **kwargs):
         """
         Initializes SubprocessIOChunker
         :param cmd: A Subprocess.Popen style "cmd". Can be string or array of strings
         :param inputstream: (Default: None) A file-like, string, or file pointer.
         :param buffer_size: (Default: 65536) A size of total buffer per stream in bytes.
         :param chunk_size: (Default: 4096) A max size of a chunk. Actual chunk may be smaller.
         :param starting_values: (Default: []) An array of strings to put in front of output que.
         """
         if inputstream:
             input_streamer = StreamFeeder(inputstream)
             input_streamer.start()
             inputstream = input_streamer.output
         _shell = kwargs.get('shell', True)
         if isinstance(cmd, (list, tuple)):
             cmd = ' '.join(cmd)
         kwargs['shell'] = _shell
         _p = subprocess.Popen(cmd,
             bufsize=-1,
             stdin=inputstream,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             **kwargs
+        )
         bg_out = BufferedGenerator(_p.stdout, buffer_size, chunk_size, starting_values)
         bg_err = BufferedGenerator(_p.stderr, 16000, 1, bottomless=True)
         while not bg_out.done_reading and not bg_out.reading_paused and not bg_err.length:
             # doing this until we reach either end of file, or end of buffer.
             bg_out.data_added_event.wait(1)
             bg_out.data_added_event.clear()
         # at this point it's still ambiguous if we are done reading or just full buffer.
         # Either way, if error (returned by ended process, or implied based on
         # presence of stuff in stderr output) we error out.
         # Else, we are happy.
         _returncode = _p.poll()
-        if _returncode or (_returncode == None and bg_err.length):
+        if _returncode or (_returncode is None and bg_err.length):
             try:
                 _p.terminate()
             except:
                 pass
             bg_out.stop()
             bg_err.stop()
             err = '%s' % ''.join(bg_err)
             if err:
                 raise EnvironmentError("Subprocess exited due to an error:\n" + err)
             raise EnvironmentError("Subprocess exited with non 0 ret code:%s" % _returncode)
         self.process = _p
         self.output = bg_out
         self.error = bg_err
     def __iter__(self):
         return self
     def next(self):
         if self.process.poll():
             err = '%s' % ''.join(self.error)
             raise EnvironmentError("Subprocess exited due to an error:\n" + err)
         return self.output.next()
     def throw(self, type, value=None, traceback=None):
         if self.output.length or not self.output.done_reading:
             raise type(value)
     def close(self):
         try:
             self.process.terminate()
         except:
             pass
         try:
             self.output.close()
         except:
             pass
         try:
             self.error.close()
         except:
             pass
     def __del__(self):
         self.close()

rhodecode/tests/functional/test_admin_settings.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 from rhodecode.lib.auth import get_crypt_password, check_password
 from rhodecode.model.db import User, RhodeCodeSetting, Repository
 from rhodecode.tests import *
 from rhodecode.lib import helpers as h
 from rhodecode.model.user import UserModel
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.meta import Session
 class TestAdminSettingsController(TestController):
     def test_index(self):
         response = self.app.get(url('admin_settings'))
         # Test response...
     def test_index_as_xml(self):
         response = self.app.get(url('formatted_admin_settings', format='xml'))
     def test_create(self):
         response = self.app.post(url('admin_settings'))
     def test_new(self):
         response = self.app.get(url('admin_new_setting'))
     def test_new_as_xml(self):
         response = self.app.get(url('formatted_admin_new_setting', format='xml'))
     def test_update(self):
         response = self.app.put(url('admin_setting', setting_id=1))
     def test_update_browser_fakeout(self):
         response = self.app.post(url('admin_setting', setting_id=1), params=dict(_method='put'))
     def test_delete(self):
         response = self.app.delete(url('admin_setting', setting_id=1))
     def test_delete_browser_fakeout(self):
         response = self.app.post(url('admin_setting', setting_id=1), params=dict(_method='delete'))
     def test_show(self):
         response = self.app.get(url('admin_setting', setting_id=1))
     def test_show_as_xml(self):
         response = self.app.get(url('formatted_admin_setting', setting_id=1, format='xml'))
     def test_edit(self):
         response = self.app.get(url('admin_edit_setting', setting_id=1))
     def test_edit_as_xml(self):
         response = self.app.get(url('formatted_admin_edit_setting',
                                     setting_id=1, format='xml'))
     def test_ga_code_active(self):
         self.log_user()
         old_title = 'RhodeCode'
         old_realm = 'RhodeCode authentication'
         new_ga_code = 'ga-test-123456789'
         response = self.app.post(url('admin_setting', setting_id='global'),
                                      params=dict(
                                                  _method='put',
                                                  rhodecode_title=old_title,
                                                  rhodecode_realm=old_realm,
                                                  rhodecode_ga_code=new_ga_code
                                                  ))
         self.checkSessionFlash(response, 'Updated application settings')
         self.assertEqual(RhodeCodeSetting
                          .get_app_settings()['rhodecode_ga_code'], new_ga_code)
         response = response.follow()
         response.mustcontain("""_gaq.push(['_setAccount', '%s']);""" % new_ga_code)
     def test_ga_code_inactive(self):
         self.log_user()
         old_title = 'RhodeCode'
         old_realm = 'RhodeCode authentication'
         new_ga_code = ''
         response = self.app.post(url('admin_setting', setting_id='global'),
                                      params=dict(
                                                  _method='put',
                                                  rhodecode_title=old_title,
                                                  rhodecode_realm=old_realm,
                                                  rhodecode_ga_code=new_ga_code
                                                  ))
         self.checkSessionFlash(response, 'Updated application settings')
         self.assertEqual(RhodeCodeSetting
                         .get_app_settings()['rhodecode_ga_code'], new_ga_code)
         response = response.follow()
         response.mustcontain(no=["_gaq.push(['_setAccount', '%s']);" % new_ga_code])
     def test_title_change(self):
         self.log_user()
         old_title = 'RhodeCode'
         new_title = old_title + '_changed'
         old_realm = 'RhodeCode authentication'
         for new_title in ['Changed', 'Żółwik', old_title]:
             response = self.app.post(url('admin_setting', setting_id='global'),
                                          params=dict(
                                                      _method='put',
                                                      rhodecode_title=new_title,
                                                      rhodecode_realm=old_realm,
                                                      rhodecode_ga_code=''
                                                      ))
             self.checkSessionFlash(response, 'Updated application settings')
             self.assertEqual(RhodeCodeSetting
                              .get_app_settings()['rhodecode_title'],
                              new_title.decode('utf-8'))
             response = response.follow()
             response.mustcontain("""<h1><a href="/">%s</a></h1>""" % new_title)
     def test_my_account(self):
         self.log_user()
         response = self.app.get(url('admin_settings_my_account'))
         response.mustcontain('value="test_admin')
     @parameterized.expand([('firstname', 'new_username'),
                            ('lastname', 'new_username'),
                            ('admin', True),
                            ('admin', False),
                            ('ldap_dn', 'test'),
                            ('ldap_dn', None),
                            ('active', False),
                            ('active', True),
                            ('email', 'some@email.com'),
                            ])
     def test_my_account_update(self, name, expected):
         uname = 'testme'
         usr = UserModel().create_or_update(username=uname, password='qweqwe',
                                            email='testme@rhodecod.org')
         Session().commit()
         params = usr.get_api_data()
         user_id = usr.user_id
         self.log_user(username=uname, password='qweqwe')
         params.update({name: expected})
         params.update({'password_confirmation': ''})
         params.update({'new_password': ''})
         try:
             response = self.app.put(url('admin_settings_my_account_update',
                                         id=user_id), params)
             self.checkSessionFlash(response,
                                    'Your account was updated successfully')
             updated_user = User.get_by_username(uname)
             updated_params = updated_user.get_api_data()
             updated_params.update({'password_confirmation': ''})
             updated_params.update({'new_password': ''})
             params['last_login'] = updated_params['last_login']
             if name == 'email':
                 params['emails'] = [expected]
             if name == 'ldap_dn':
                 #cannot update this via form
                 params['ldap_dn'] = None
             if name == 'active':
                 #my account cannot deactivate account
                 params['active'] = True
             if name == 'admin':
                 #my account cannot make you an admin !
                 params['admin'] = False
             self.assertEqual(params, updated_params)
         finally:
             UserModel().delete('testme')
     def test_my_account_update_err_email_exists(self):
         self.log_user()
         new_email = 'test_regular@mail.com'  # already exisitn email
         response = self.app.put(url('admin_settings_my_account_update'),
                                 params=dict(
                                     username='test_admin',
                                     new_password='test12',
                                     password_confirmation='test122',
                                     firstname='NewName',
                                     lastname='NewLastname',
                                     email=new_email,)
+                                )
         response.mustcontain('This e-mail address is already taken')
     def test_my_account_update_err(self):
         self.log_user('test_regular2', 'test12')
         new_email = 'newmail.pl'
         response = self.app.post(url('admin_settings_my_account_update'),
                                  params=dict(
                                             _method='put',
                                             username='test_admin',
                                             new_password='test12',
                                             password_confirmation='test122',
                                             firstname='NewName',
                                             lastname='NewLastname',
                                             email=new_email,)
+                                 )
         response.mustcontain('An email address must contain a single @')
         from rhodecode.model import validators
         msg = validators.ValidUsername(edit=False,
                                     old_data={})._messages['username_exists']
         msg = h.html_escape(msg % {'username': 'test_admin'})
         response.mustcontain(u"%s" % msg)
     def test_set_repo_fork_has_no_self_id(self):
         self.log_user()
         repo = Repository.get_by_repo_name(HG_REPO)
         response = self.app.get(url('edit_repo', repo_name=HG_REPO))
         opt = """<option value="%s">vcs_test_git</option>""" % repo.repo_id
         response.mustcontain(no=[opt])
     def test_set_fork_of_repo(self):
         self.log_user()
         repo = Repository.get_by_repo_name(HG_REPO)
         repo2 = Repository.get_by_repo_name(GIT_REPO)
         response = self.app.put(url('repo_as_fork', repo_name=HG_REPO),
                                  params=dict(
                                     id_fork_of=repo2.repo_id
                                  ))
         repo = Repository.get_by_repo_name(HG_REPO)
         repo2 = Repository.get_by_repo_name(GIT_REPO)
         self.checkSessionFlash(response,
         'Marked repo %s as fork of %s' % (repo.repo_name, repo2.repo_name))
         assert repo.fork == repo2
         response = response.follow()
         # check if given repo is selected
         opt = """<option value="%s" selected="selected">%s</option>""" % (
                                                 repo2.repo_id, repo2.repo_name)
         response.mustcontain(opt)
         # clean session flash
         #response = self.app.get(url('edit_repo', repo_name=HG_REPO))
         ## mark it as None
         response = self.app.put(url('repo_as_fork', repo_name=HG_REPO),
                                  params=dict(
                                     id_fork_of=None
                                  ))
         repo = Repository.get_by_repo_name(HG_REPO)
         repo2 = Repository.get_by_repo_name(GIT_REPO)
         self.checkSessionFlash(response,
         'Marked repo %s as fork of %s' % (repo.repo_name, "Nothing"))
-        assert repo.fork == None
+        assert repo.fork is None
     def test_set_fork_of_same_repo(self):
         self.log_user()
         repo = Repository.get_by_repo_name(HG_REPO)
         response = self.app.put(url('repo_as_fork', repo_name=HG_REPO),
                                  params=dict(
                                     id_fork_of=repo.repo_id
                                  ))
         self.checkSessionFlash(response,
                                'An error occurred during this operation')

rhodecode/tests/other/test_validators.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 import formencode
 from rhodecode.tests import *
 from rhodecode.model import validators as v
 from rhodecode.model.users_group import UserGroupModel
 from rhodecode.model.meta import Session
 from rhodecode.model.repos_group import ReposGroupModel
 from rhodecode.model.db import ChangesetStatus, Repository
 from rhodecode.model.changeset_status import ChangesetStatusModel
 from rhodecode.tests.fixture import Fixture
 fixture = Fixture()
 class TestReposGroups(BaseTestCase):
     def setUp(self):
         pass
     def tearDown(self):
         Session.remove()
     def test_Message_extractor(self):
         validator = v.ValidUsername()
         self.assertRaises(formencode.Invalid, validator.to_python, 'default')
         class StateObj(object):
             pass
         self.assertRaises(formencode.Invalid,
                           validator.to_python, 'default', StateObj)
     def test_ValidUsername(self):
         validator = v.ValidUsername()
         self.assertRaises(formencode.Invalid, validator.to_python, 'default')
         self.assertRaises(formencode.Invalid, validator.to_python, 'new_user')
         self.assertRaises(formencode.Invalid, validator.to_python, '.,')
         self.assertRaises(formencode.Invalid, validator.to_python,
                           TEST_USER_ADMIN_LOGIN)
         self.assertEqual('test', validator.to_python('test'))
         validator = v.ValidUsername(edit=True, old_data={'user_id': 1})
     def test_ValidRepoUser(self):
         validator = v.ValidRepoUser()
         self.assertRaises(formencode.Invalid, validator.to_python, 'nouser')
         self.assertEqual(TEST_USER_ADMIN_LOGIN,
                          validator.to_python(TEST_USER_ADMIN_LOGIN))
     def test_ValidUserGroup(self):
         validator = v.ValidUserGroup()
         self.assertRaises(formencode.Invalid, validator.to_python, 'default')
         self.assertRaises(formencode.Invalid, validator.to_python, '.,')
         gr = fixture.create_user_group('test')
         gr2 = fixture.create_user_group('tes2')
         Session().commit()
         self.assertRaises(formencode.Invalid, validator.to_python, 'test')
-        assert gr.users_group_id != None
+        assert gr.users_group_id is not None
         validator = v.ValidUserGroup(edit=True,
                                     old_data={'users_group_id':
                                               gr2.users_group_id})
         self.assertRaises(formencode.Invalid, validator.to_python, 'test')
         self.assertRaises(formencode.Invalid, validator.to_python, 'TesT')
         self.assertRaises(formencode.Invalid, validator.to_python, 'TEST')
         UserGroupModel().delete(gr)
         UserGroupModel().delete(gr2)
         Session().commit()
     def test_ValidReposGroup(self):
         validator = v.ValidReposGroup()
         model = ReposGroupModel()
         self.assertRaises(formencode.Invalid, validator.to_python,
                           {'group_name': HG_REPO, })
         gr = model.create(group_name='test_gr', group_description='desc',
                           parent=None,
                           just_db=True,
                           owner=TEST_USER_ADMIN_LOGIN)
         self.assertRaises(formencode.Invalid,
                           validator.to_python, {'group_name': gr.group_name, })
         validator = v.ValidReposGroup(edit=True,
                                       old_data={'group_id':  gr.group_id})
         self.assertRaises(formencode.Invalid,
                           validator.to_python, {
                                         'group_name': gr.group_name + 'n',
                                         'group_parent_id': gr.group_id
                                         })
         model.delete(gr)
     def test_ValidPassword(self):
         validator = v.ValidPassword()
         self.assertEqual('lol', validator.to_python('lol'))
         self.assertEqual(None, validator.to_python(None))
         self.assertRaises(formencode.Invalid, validator.to_python, 'ąćżź')
     def test_ValidPasswordsMatch(self):
         validator = v.ValidPasswordsMatch()
         self.assertRaises(formencode.Invalid,
                     validator.to_python, {'password': 'pass',
                                           'password_confirmation': 'pass2'})
         self.assertRaises(formencode.Invalid,
                     validator.to_python, {'new_password': 'pass',
                                           'password_confirmation': 'pass2'})
         self.assertEqual({'new_password': 'pass',
                           'password_confirmation': 'pass'},
                     validator.to_python({'new_password': 'pass',
                                          'password_confirmation': 'pass'}))
         self.assertEqual({'password': 'pass',
                           'password_confirmation': 'pass'},
                     validator.to_python({'password': 'pass',
                                          'password_confirmation': 'pass'}))
     def test_ValidAuth(self):
         validator = v.ValidAuth()
         valid_creds = {
             'username': TEST_USER_REGULAR2_LOGIN,
             'password': TEST_USER_REGULAR2_PASS,
+        }
         invalid_creds = {
             'username': 'err',
             'password': 'err',
+        }
         self.assertEqual(valid_creds, validator.to_python(valid_creds))
         self.assertRaises(formencode.Invalid,
                           validator.to_python, invalid_creds)
     def test_ValidAuthToken(self):
         validator = v.ValidAuthToken()
         # this is untestable without a threadlocal
 #        self.assertRaises(formencode.Invalid,
 #                          validator.to_python, 'BadToken')
         validator
     def test_ValidRepoName(self):
         validator = v.ValidRepoName()
         self.assertRaises(formencode.Invalid,
                           validator.to_python, {'repo_name': ''})
         self.assertRaises(formencode.Invalid,
                           validator.to_python, {'repo_name': HG_REPO})
         gr = ReposGroupModel().create(group_name='group_test',
                                       group_description='desc',
                                       parent=None,
                                       owner=TEST_USER_ADMIN_LOGIN)
         self.assertRaises(formencode.Invalid,
                           validator.to_python, {'repo_name': gr.group_name})
         #TODO: write an error case for that ie. create a repo withinh a group
 #        self.assertRaises(formencode.Invalid,
 #                          validator.to_python, {'repo_name': 'some',
 #                                                'repo_group': gr.group_id})
     def test_ValidForkName(self):
         # this uses ValidRepoName validator
         assert True
     @parameterized.expand([
         ('test', 'test'), ('lolz!', 'lolz'), ('  aavv', 'aavv'),
         ('ala ma kota', 'ala-ma-kota'), ('@nooo', 'nooo'),
         ('$!haha lolz !', 'haha-lolz'), ('$$$$$', ''), ('{}OK!', 'OK'),
         ('/]re po', 're-po')])
     def test_SlugifyName(self, name, expected):
         validator = v.SlugifyName()
         self.assertEqual(expected, validator.to_python(name))
     def test_ValidCloneUri(self):
             #TODO: write this one
             pass
     def test_ValidForkType(self):
             validator = v.ValidForkType(old_data={'repo_type': 'hg'})
             self.assertEqual('hg', validator.to_python('hg'))
             self.assertRaises(formencode.Invalid, validator.to_python, 'git')
     def test_ValidPerms(self):
             #TODO: write this one
             pass
     def test_ValidSettings(self):
         validator = v.ValidSettings()
         self.assertEqual({'pass': 'pass'},
                          validator.to_python(value={'user': 'test',
                                                     'pass': 'pass'}))
         self.assertEqual({'user2': 'test', 'pass': 'pass'},
                          validator.to_python(value={'user2': 'test',
                                                     'pass': 'pass'}))
     def test_ValidPath(self):
             validator = v.ValidPath()
             self.assertEqual(TESTS_TMP_PATH,
                              validator.to_python(TESTS_TMP_PATH))
             self.assertRaises(formencode.Invalid, validator.to_python,
                               '/no_such_dir')
     def test_UniqSystemEmail(self):
         validator = v.UniqSystemEmail(old_data={})
         self.assertEqual('mail@python.org',
                          validator.to_python('MaiL@Python.org'))
         email = TEST_USER_REGULAR2_EMAIL
         self.assertRaises(formencode.Invalid, validator.to_python, email)
     def test_ValidSystemEmail(self):
         validator = v.ValidSystemEmail()
         email = TEST_USER_REGULAR2_EMAIL
         self.assertEqual(email, validator.to_python(email))
         self.assertRaises(formencode.Invalid, validator.to_python, 'err')
     def test_LdapLibValidator(self):
         if ldap_lib_installed:
             validator = v.LdapLibValidator()
             self.assertEqual("DN", validator.to_python('DN'))
         else:
             validator = v.LdapLibValidator()
             self.assertRaises(v.LdapImportError, validator.to_python, 'err')
     def test_AttrLoginValidator(self):
         validator = v.AttrLoginValidator()
         self.assertEqual('DN_attr', validator.to_python('DN_attr'))
     def test_NotReviewedRevisions(self):
         repo_id = Repository.get_by_repo_name(HG_REPO).repo_id
         validator = v.NotReviewedRevisions(repo_id)
         rev = '0' * 40
         # add status for a rev, that should throw an error because it is already
         # reviewed
         new_status = ChangesetStatus()
         new_status.author = ChangesetStatusModel()._get_user(TEST_USER_ADMIN_LOGIN)
         new_status.repo = ChangesetStatusModel()._get_repo(HG_REPO)
         new_status.status = ChangesetStatus.STATUS_APPROVED
         new_status.comment = None
         new_status.revision = rev
         Session().add(new_status)
         Session().commit()
         try:
             self.assertRaises(formencode.Invalid, validator.to_python, [rev])
         finally:
             Session().delete(new_status)
             Session().commit()

rhodecode/tests/scripts/test_concurency.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     rhodecode.tests.test_hg_operations
     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
     Test suite for making push/pull operations
     :created_on: Dec 30, 2010
     :author: marcink
     :copyright: (C) 2010-2012 Marcin Kuzminski <marcin@python-works.com>
     :license: GPLv3, see COPYING for more details.
 """
 # 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/>.
 import os
 import sys
 import shutil
 import logging
 from os.path import join as jn
 from os.path import dirname as dn
 from tempfile import _RandomNameSequence
 from subprocess import Popen, PIPE
 from paste.deploy import appconfig
 from pylons import config
 from sqlalchemy import engine_from_config
 from rhodecode.lib.utils import add_cache
 from rhodecode.model import init_model
 from rhodecode.model import meta
 from rhodecode.model.db import User, Repository
 from rhodecode.lib.auth import get_crypt_password
 from rhodecode.tests import TESTS_TMP_PATH, NEW_HG_REPO, HG_REPO
 from rhodecode.config.environment import load_environment
 rel_path = dn(dn(dn(dn(os.path.abspath(__file__)))))
 conf = appconfig('config:rc.ini', relative_to=rel_path)
 load_environment(conf.global_conf, conf.local_conf)
 add_cache(conf)
 USER = 'test_admin'
 PASS = 'test12'
 HOST = 'rc.local'
 METHOD = 'pull'
 DEBUG = True
 log = logging.getLogger(__name__)
 class Command(object):
     def __init__(self, cwd):
         self.cwd = cwd
     def execute(self, cmd, *args):
         """Runs command on the system with given ``args``.
         """
         command = cmd + ' ' + ' '.join(args)
         log.debug('Executing %s' % command)
         if DEBUG:
             print command
         p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd)
         stdout, stderr = p.communicate()
         if DEBUG:
             print stdout, stderr
         return stdout, stderr
 def get_session():
     engine = engine_from_config(conf, 'sqlalchemy.db1.')
     init_model(engine)
     sa = meta.Session
     return sa
 def create_test_user(force=True):
     print 'creating test user'
     sa = get_session()
     user = sa.query(User).filter(User.username == USER).scalar()
     if force and user is not None:
         print 'removing current user'
         for repo in sa.query(Repository).filter(Repository.user == user).all():
             sa.delete(repo)
         sa.delete(user)
         sa.commit()
     if user is None or force:
         print 'creating new one'
         new_usr = User()
         new_usr.username = USER
         new_usr.password = get_crypt_password(PASS)
         new_usr.email = 'mail@mail.com'
         new_usr.name = 'test'
         new_usr.lastname = 'lasttestname'
         new_usr.active = True
         new_usr.admin = True
         sa.add(new_usr)
         sa.commit()
     print 'done'
 def create_test_repo(force=True):
     print 'creating test repo'
     from rhodecode.model.repo import RepoModel
     sa = get_session()
     user = sa.query(User).filter(User.username == USER).scalar()
     if user is None:
         raise Exception('user not found')
     repo = sa.query(Repository).filter(Repository.repo_name == HG_REPO).scalar()
     if repo is None:
         print 'repo not found creating'
         form_data = {'repo_name': HG_REPO,
                      'repo_type': 'hg',
                      'private':False,
                      'clone_uri': '' }
         rm = RepoModel(sa)
         rm.base_path = '/home/hg'
         rm.create(form_data, user)
     print 'done'
 def set_anonymous_access(enable=True):
     sa = get_session()
     user = sa.query(User).filter(User.username == 'default').one()
     user.active = enable
     sa.add(user)
     sa.commit()
 def get_anonymous_access():
     sa = get_session()
     return sa.query(User).filter(User.username == 'default').one().active
 #==============================================================================
 # TESTS
 #==============================================================================
 def test_clone_with_credentials(no_errors=False, repo=HG_REPO, method=METHOD,
                                 seq=None, backend='hg'):
     cwd = path = jn(TESTS_TMP_PATH, repo)
-    if seq == None:
+    if seq is None:
         seq = _RandomNameSequence().next()
     try:
         shutil.rmtree(path, ignore_errors=True)
         os.makedirs(path)
         #print 'made dirs %s' % jn(path)
     except OSError:
         raise
     clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s' % \
                   {'user': USER,
                    'pass': PASS,
                    'host': HOST,
                    'cloned_repo': repo, }
     dest = path + seq
     if method == 'pull':
         stdout, stderr = Command(cwd).execute(backend, method, '--cwd', dest, clone_url)
     else:
         stdout, stderr = Command(cwd).execute(backend, method, clone_url, dest)
         print stdout,'sdasdsadsa'
         if not no_errors:
             if backend == 'hg':
                 assert """adding file changes""" in stdout, 'no messages about cloning'
                 assert """abort""" not in stderr , 'got error from clone'
             elif backend == 'git':
                 assert """Cloning into""" in stdout, 'no messages about cloning'
 if __name__ == '__main__':
     try:
         create_test_user(force=False)
         seq = None
         import time
         try:
             METHOD = sys.argv[3]
         except Exception:
             pass
         try:
             backend = sys.argv[4]
         except Exception:
             backend = 'hg'
         if METHOD == 'pull':
             seq = _RandomNameSequence().next()
             test_clone_with_credentials(repo=sys.argv[1], method='clone',
                                         seq=seq, backend=backend)
         s = time.time()
         for i in range(1, int(sys.argv[2]) + 1):
             print 'take', i
             test_clone_with_credentials(repo=sys.argv[1], method=METHOD,
                                         seq=seq, backend=backend)
         print 'time taken %.3f' % (time.time() - s)
     except Exception, e:
         sys.exit('stop on %s' % e)

0 comments (0 inline, 0 general)