kallithea Changeset - 3d1fcf67f299

Changeset - 3d1fcf67f299

Parent rev.

Child rev.

[Not reviewed]

default

0 14 0

Søren Løvborg - 9 years ago 2016-08-03 17:01:57
sorenl@unity3d.com

model: drop BaseModel.cls

It no longer serves a purpose.

14 files changed with 1 insertions and 24 deletions:

kallithea/model/__init__.py

kallithea/model/api_key.py

kallithea/model/changeset_status.py

kallithea/model/comment.py

kallithea/model/gist.py

kallithea/model/notification.py

kallithea/model/permission.py

kallithea/model/pull_request.py

kallithea/model/repo.py

kallithea/model/repo_group.py

kallithea/model/repo_permission.py

kallithea/model/user.py

kallithea/model/user_group.py

kallithea/tests/fixture.py

0 comments (0 inline, 0 general)

kallithea/model/__init__.py

➞

Show inline comments

@@ @@ -27,98 +27,96 @@ Original author and date, and relevant c @@
 :example:
     .. code-block:: python
        from paste.deploy import appconfig
        from pylons import config
        from sqlalchemy import engine_from_config
        from kallithea.config.environment import load_environment
        conf = appconfig('config:development.ini', relative_to = './../../')
        load_environment(conf.global_conf, conf.local_conf)
        engine = engine_from_config(config, 'sqlalchemy.')
        init_model(engine)
        # RUN YOUR CODE HERE
 """
 import logging
 from kallithea.model import meta
 from kallithea.lib.utils2 import safe_str, obfuscate_url_pw
 log = logging.getLogger(__name__)
 def init_model(engine):
     """
     Initializes db session, bind the engine with the metadata,
     Call this before using any of the tables or classes in the model,
     preferably once in application start
     :param engine: engine to bind to
     """
     engine_str = obfuscate_url_pw(str(engine.url))
     log.info("initializing db for %s", engine_str)
     meta.Base.metadata.bind = engine
 class BaseModel(object):
     """
     Base Model for all Kallithea models, it adds sql alchemy session
     into instance of model
     :param sa: If passed it reuses this session instead of creating a new one
     """
     cls = None  # override in child class
     def __init__(self, sa=None):
         if sa is not None:
             self.sa = sa
         else:
             self.sa = meta.Session()
     def _get_instance(self, cls, instance, callback=None):
         """
         Gets instance of given cls using some simple lookup mechanism.
         :param cls: class to fetch
         :param instance: int or Instance
         :param callback: callback to call if all lookups failed
         """
         if isinstance(instance, cls):
             return instance
         elif isinstance(instance, (int, long)) or safe_str(instance).isdigit():
             return cls.get(instance)
         else:
             if instance is not None:
                 if callback is None:
                     raise Exception(
                         'given object must be int, long or Instance of %s '
                         'got %s, no callback provided' % (cls, type(instance))
+                    )
                 else:
                     return callback(instance)
     def _get_user(self, user):
         """
         Helper method to get user by ID, or username fallback
         :param user: UserID, username, or User instance
         """
         from kallithea.model.db import User
         return self._get_instance(User, user,
                                   callback=User.get_by_username)
     def _get_repo(self, repository):
         """
         Helper method to get repository by ID, or repository name
         :param repository: RepoID, repository name or Repository Instance
         """
         from kallithea.model.db import Repository
         return self._get_instance(Repository, repository,
                                   callback=Repository.get_by_repo_name)

kallithea/model/api_key.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.api_key
 ~~~~~~~~~~~~~~~~~~~~~~~
 API key model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Sep 8, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import time
 import logging
 from sqlalchemy import or_
 from kallithea.lib.utils2 import generate_api_key
 from kallithea.model import BaseModel
 from kallithea.model.db import UserApiKeys
 from kallithea.model.meta import Session
 log = logging.getLogger(__name__)
 class ApiKeyModel(BaseModel):
     cls = UserApiKeys
     def create(self, user, description, lifetime=-1):
         """
         :param user: user or user_id
         :param description: description of ApiKey
         :param lifetime: expiration time in seconds
         """
         user = self._get_user(user)
         new_api_key = UserApiKeys()
         new_api_key.api_key = generate_api_key()
         new_api_key.user_id = user.user_id
         new_api_key.description = description
         new_api_key.expires = time.time() + (lifetime * 60) if lifetime != -1 else -1
         Session().add(new_api_key)
         return new_api_key
     def delete(self, api_key, user=None):
         """
         Deletes given api_key, if user is set it also filters the object for
         deletion by given user.
         """
         api_key = UserApiKeys.query().filter(UserApiKeys.api_key == api_key)
         if user is not None:
             user = self._get_user(user)
             api_key = api_key.filter(UserApiKeys.user_id == user.user_id)
         api_key = api_key.scalar()
         Session().delete(api_key)
     def get_api_keys(self, user, show_expired=True):
         user = self._get_user(user)
         user_api_keys = UserApiKeys.query() \
             .filter(UserApiKeys.user_id == user.user_id)
         if not show_expired:
             user_api_keys = user_api_keys \
                 .filter(or_(UserApiKeys.expires == -1,
                             UserApiKeys.expires >= time.time()))
         return user_api_keys

kallithea/model/changeset_status.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.changeset_status
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Changeset status controller
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 30, 2012
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 from sqlalchemy.orm import joinedload
 from kallithea.model import BaseModel
 from kallithea.model.db import ChangesetStatus, PullRequest
 from kallithea.lib.exceptions import StatusChangeOnClosedPullRequestError
 log = logging.getLogger(__name__)
 class ChangesetStatusModel(BaseModel):
     cls = ChangesetStatus
     def _get_status_query(self, repo, revision, pull_request,
                           with_revisions=False):
         repo = self._get_repo(repo)
         q = ChangesetStatus.query() \
             .filter(ChangesetStatus.repo == repo)
         if not with_revisions:
             # only report the latest vote across all users! TODO: be smarter!
             q = q.filter(ChangesetStatus.version == 0)
         if revision:
             q = q.filter(ChangesetStatus.revision == revision)
         elif pull_request:
             pull_request = self._get_instance(PullRequest, pull_request)
             q = q.filter(ChangesetStatus.pull_request == pull_request)
         else:
             raise Exception('Please specify revision or pull_request')
         q = q.order_by(ChangesetStatus.version.asc())
         return q
     def _calculate_status(self, statuses):
         """
         Given a list of statuses, calculate the resulting status, according to
         the policy: approve if consensus, reject when at least one reject.
         """
         if not statuses:
             return ChangesetStatus.STATUS_UNDER_REVIEW
         if all(st and st.status == ChangesetStatus.STATUS_APPROVED for st in statuses):
             return ChangesetStatus.STATUS_APPROVED
         if any(st and st.status == ChangesetStatus.STATUS_REJECTED for st in statuses):
             return ChangesetStatus.STATUS_REJECTED
         return ChangesetStatus.STATUS_UNDER_REVIEW
     def calculate_pull_request_result(self, pull_request):
         """
         Return a tuple (reviewers, pending reviewers, pull request status)
         Only approve and reject counts as valid votes.
         """
         # collect latest votes from all voters
         cs_statuses = dict()
         for st in reversed(self.get_statuses(pull_request.org_repo,
                                              pull_request=pull_request,
                                              with_revisions=True)):

kallithea/model/comment.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.comment
 ~~~~~~~~~~~~~~~~~~~~~~~
 comments model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Nov 11, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 from pylons.i18n.translation import _
 from collections import defaultdict
 from kallithea.lib.utils2 import extract_mentioned_users, safe_unicode
 from kallithea.lib import helpers as h
 from kallithea.model import BaseModel
 from kallithea.model.db import ChangesetComment, User, \
     Notification, PullRequest
 from kallithea.model.notification import NotificationModel
 from kallithea.model.meta import Session
 log = logging.getLogger(__name__)
 class ChangesetCommentsModel(BaseModel):
     cls = ChangesetComment
     def _get_notification_data(self, repo, comment, user, comment_text,
                                line_no=None, revision=None, pull_request=None,
                                status_change=None, closing_pr=False):
         """
         :returns: tuple (subj,body,recipients,notification_type,email_kwargs)
         """
         # make notification
         body = comment_text  # text of the comment
         line = ''
         if line_no:
             line = _('on line %s') % line_no
         #changeset
         if revision:
             notification_type = Notification.TYPE_CHANGESET_COMMENT
             cs = repo.scm_instance.get_changeset(revision)
             desc = cs.short_id
             threading = ['%s-rev-%s@%s' % (repo.repo_name, revision, h.canonical_hostname())]
             if line_no: # TODO: url to file _and_ line number
                 threading.append('%s-rev-%s-line-%s@%s' % (repo.repo_name, revision, line_no,
                                                            h.canonical_hostname()))
             comment_url = h.canonical_url('changeset_home',
                 repo_name=repo.repo_name,
                 revision=revision,
                 anchor='comment-%s' % comment.comment_id)
             subj = safe_unicode(
                 h.link_to('Re changeset: %(desc)s %(line)s' % \
                           {'desc': desc, 'line': line},
                           comment_url)
+            )
             # get the current participants of this changeset
             recipients = ChangesetComment.get_users(revision=revision)
             # add changeset author if it's known locally
             cs_author = User.get_from_cs_author(cs.author)
             if not cs_author:
                 #use repo owner if we cannot extract the author correctly
                 # FIXME: just use committer name even if not a user
                 cs_author = repo.user
             recipients += [cs_author]
             email_kwargs = {
                 'status_change': status_change,
                 'cs_comment_user': user.full_name_and_username,
                 'cs_target_repo': h.canonical_url('summary_home', repo_name=repo.repo_name),
                 'cs_comment_url': comment_url,
                 'raw_id': revision,
                 'message': cs.message,
                 'message_short': h.shorter(cs.message, 50, firstline=True),

kallithea/model/gist.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.gist
 ~~~~~~~~~~~~~~~~~~~~
 gist model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: May 9, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import time
 import logging
 import traceback
 import shutil
 from kallithea.lib.utils2 import safe_unicode, unique_id, safe_int, \
     time_to_datetime, AttributeDict
 from kallithea.lib.compat import json
 from kallithea.model import BaseModel
 from kallithea.model.db import Gist
 from kallithea.model.repo import RepoModel
 from kallithea.model.scm import ScmModel
 log = logging.getLogger(__name__)
 GIST_STORE_LOC = '.rc_gist_store'
 GIST_METADATA_FILE = '.rc_gist_metadata'
 class GistModel(BaseModel):
     cls = Gist
     def _get_gist(self, gist):
         """
         Helper method to get gist by ID, or gist_access_id as a fallback
         :param gist: GistID, gist_access_id, or Gist instance
         """
         return self._get_instance(Gist, gist, callback=Gist.get_by_access_id)
     def __delete_gist(self, gist):
         """
         removes gist from filesystem
         :param gist: gist object
         """
         root_path = RepoModel().repos_path
         rm_path = os.path.join(root_path, GIST_STORE_LOC, gist.gist_access_id)
         log.info("Removing %s", rm_path)
         shutil.rmtree(rm_path)
     def _store_metadata(self, repo, gist_id, gist_access_id, user_id, gist_type,
                         gist_expires):
         """
         store metadata inside the gist, this can be later used for imports
         or gist identification
         """
         metadata = {
             'metadata_version': '1',
             'gist_db_id': gist_id,
             'gist_access_id': gist_access_id,
             'gist_owner_id': user_id,
             'gist_type': gist_type,
             'gist_expires': gist_expires,
             'gist_updated': time.time(),
+        }
         with open(os.path.join(repo.path, '.hg', GIST_METADATA_FILE), 'wb') as f:
             f.write(json.dumps(metadata))
     def get_gist(self, gist):
         return self._get_gist(gist)
     def get_gist_files(self, gist_access_id, revision=None):
         """
         Get files for given gist
         :param gist_access_id:
         """
         repo = Gist.get_by_access_id(gist_access_id)

kallithea/model/notification.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.notification
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Model for notifications
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Nov 20, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 from pylons import tmpl_context as c
 from pylons.i18n.translation import _
 from sqlalchemy.orm import joinedload, subqueryload
 import kallithea
 from kallithea.lib import helpers as h
 from kallithea.lib.utils2 import safe_unicode
 from kallithea.model import BaseModel
 from kallithea.model.db import Notification, User, UserNotification
 from kallithea.model.meta import Session
 log = logging.getLogger(__name__)
 class NotificationModel(BaseModel):
     cls = Notification
     def create(self, created_by, subject, body, recipients=None,
                type_=Notification.TYPE_MESSAGE, with_email=True,
                email_kwargs=None, repo_name=None):
         """
         Creates notification of given type
         :param created_by: int, str or User instance. User who created this
             notification
         :param subject:
         :param body:
         :param recipients: list of int, str or User objects, when None
             is given send to all admins
         :param type_: type of notification
         :param with_email: send email with this notification
         :param email_kwargs: additional dict to pass as args to email template
         """
         from kallithea.lib.celerylib import tasks, run_task
         email_kwargs = email_kwargs or {}
         if recipients and not getattr(recipients, '__iter__', False):
             raise Exception('recipients must be a list or iterable')
         created_by_obj = self._get_user(created_by)
         recipients_objs = []
         if recipients:
             for u in recipients:
                 obj = self._get_user(u)
                 if obj is not None:
                     recipients_objs.append(obj)
                 else:
                     # TODO: inform user that requested operation couldn't be completed
                     log.error('cannot email unknown user %r', u)
             recipients_objs = set(recipients_objs)
             log.debug('sending notifications %s to %s',
                 type_, recipients_objs
+            )
         elif recipients is None:
             # empty recipients means to all admins
             recipients_objs = User.query().filter(User.admin == True).all()
             log.debug('sending notifications %s to admins: %s',
                 type_, recipients_objs
+            )
         #else: silently skip notification mails?
         # TODO: inform user who are notified
         notif = Notification.create(
             created_by=created_by_obj, subject=subject,

kallithea/model/permission.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.permission
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 permissions model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Aug 20, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 from sqlalchemy.exc import DatabaseError
 from kallithea.model import BaseModel
 from kallithea.model.db import User, Permission, UserToPerm, UserRepoToPerm, \
     UserRepoGroupToPerm, UserUserGroupToPerm
 from kallithea.lib.utils2 import str2bool
 log = logging.getLogger(__name__)
 class PermissionModel(BaseModel):
     """
     Permissions model for Kallithea
     """
     cls = Permission
     def create_permissions(self):
         """
         Create permissions for whole system
         """
         for p in Permission.PERMS:
             if not Permission.get_by_key(p[0]):
                 new_perm = Permission()
                 new_perm.permission_name = p[0]
                 self.sa.add(new_perm)
     def create_default_permissions(self, user, force=False):
         """
         Create missing default permissions for user. If force is set, the default
         permissions for the user are reset, otherwise only missing permissions are
         created.
         :param user:
         """
         user = self._get_user(user)
         def _make_perm(perm):
             new_perm = UserToPerm()
             new_perm.user = user
             new_perm.permission = Permission.get_by_key(perm)
             return new_perm
         def _get_group(perm_name):
             return '.'.join(perm_name.split('.')[:1])
         perms = UserToPerm.query().filter(UserToPerm.user == user).all()
         defined_perms_groups = map(_get_group,
                                 (x.permission.permission_name for x in perms))
         log.debug('GOT ALREADY DEFINED:%s', perms)
         DEFAULT_PERMS = Permission.DEFAULT_USER_PERMISSIONS
         if force:
             for perm in perms:
                 self.sa.delete(perm)
             self.sa.commit()
             defined_perms_groups = []
         # For every default permission that needs to be created, we check if
         # its group is already defined. If it's not, we create default permission.
         for perm_name in DEFAULT_PERMS:
             gr = _get_group(perm_name)
             if gr not in defined_perms_groups:
                 log.debug('GR:%s not found, creating permission %s',
                           gr, perm_name)
                 new_perm = _make_perm(perm_name)

kallithea/model/pull_request.py

➞

Show inline comments

@@ @@ -2,98 +2,96 @@ @@
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.pull_request
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 pull request model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jun 6, 2012
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import datetime
 from pylons.i18n.translation import _
 from sqlalchemy.orm import joinedload
 from kallithea.model.meta import Session
 from kallithea.lib import helpers as h
 from kallithea.lib.exceptions import UserInvalidException
 from kallithea.model import BaseModel
 from kallithea.model.db import PullRequest, PullRequestReviewers, Notification, \
     ChangesetStatus, User
 from kallithea.model.notification import NotificationModel
 from kallithea.lib.utils2 import extract_mentioned_users, safe_unicode
 log = logging.getLogger(__name__)
 class PullRequestModel(BaseModel):
     cls = PullRequest
     def get_pullrequest_cnt_for_user(self, user):
         return PullRequest.query() \
                                 .join(PullRequestReviewers) \
                                 .filter(PullRequestReviewers.user_id == user) \
                                 .filter(PullRequest.status != PullRequest.STATUS_CLOSED) \
                                 .count()
     def get_all(self, repo_name, from_=False, closed=False):
         """Get all PRs for repo.
         Default is all PRs to the repo, PRs from the repo if from_.
         Closed PRs are only included if closed is true."""
         repo = self._get_repo(repo_name)
         q = PullRequest.query()
         if from_:
             q = q.filter(PullRequest.org_repo == repo)
         else:
             q = q.filter(PullRequest.other_repo == repo)
         if not closed:
             q = q.filter(PullRequest.status != PullRequest.STATUS_CLOSED)
         return q.order_by(PullRequest.created_on.desc()).all()
     def _get_valid_reviewers(self, seq):
         """ Generate User objects from a sequence of user IDs, usernames or
         User objects. Raises UserInvalidException if the DEFAULT user is
         specified, or if a given ID or username does not match any user.
         """
         for user_spec in seq:
             user = self._get_user(user_spec)
             if user is None or user.username == User.DEFAULT_USER:
                 raise UserInvalidException(user_spec)
             yield user
     def create(self, created_by, org_repo, org_ref, other_repo, other_ref,
                revisions, reviewers, title, description=None):
         from kallithea.model.changeset_status import ChangesetStatusModel
         created_by_user = self._get_user(created_by)
         org_repo = self._get_repo(org_repo)
         other_repo = self._get_repo(other_repo)
         new = PullRequest()
         new.org_repo = org_repo
         new.org_ref = org_ref
         new.other_repo = other_repo
         new.other_ref = other_ref
         new.revisions = revisions
         new.title = title
         new.description = description

kallithea/model/repo.py

➞

Show inline comments

@@ @@ -11,97 +11,96 @@ @@
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.repo
 ~~~~~~~~~~~~~~~~~~~~
 Repository model for kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jun 5, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import shutil
 import logging
 import traceback
 from datetime import datetime
 from sqlalchemy.orm import subqueryload
 from kallithea.lib.utils import make_ui
 from kallithea.lib.vcs.backends import get_backend
 from kallithea.lib.compat import json
 from kallithea.lib.utils2 import LazyProperty, safe_str, safe_unicode, \
     remove_prefix, obfuscate_url_pw, get_current_authuser
 from kallithea.lib.caching_query import FromCache
 from kallithea.lib.hooks import log_delete_repository
 from kallithea.model import BaseModel
 from kallithea.model.db import Repository, UserRepoToPerm, UserGroupRepoToPerm, \
     UserRepoGroupToPerm, UserGroupRepoGroupToPerm, User, Permission, \
     Statistics, UserGroup, Ui, RepoGroup, RepositoryField
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import HasRepoPermissionAny, HasUserGroupPermissionAny
 from kallithea.lib.exceptions import AttachedForksError
 from kallithea.model.scm import UserGroupList
 log = logging.getLogger(__name__)
 class RepoModel(BaseModel):
     cls = Repository
     URL_SEPARATOR = Repository.url_sep()
     def _get_user_group(self, users_group):
         return self._get_instance(UserGroup, users_group,
                                   callback=UserGroup.get_by_group_name)
     def _get_repo_group(self, repo_group):
         return self._get_instance(RepoGroup, repo_group,
                                   callback=RepoGroup.get_by_group_name)
     def _create_default_perms(self, repository, private):
         # create default permission
         default = 'repository.read'
         def_user = User.get_default_user()
         for p in def_user.user_perms:
             if p.permission.permission_name.startswith('repository.'):
                 default = p.permission.permission_name
                 break
         default_perm = 'repository.none' if private else default
         repo_to_perm = UserRepoToPerm()
         repo_to_perm.permission = Permission.get_by_key(default_perm)
         repo_to_perm.repository = repository
         repo_to_perm.user_id = def_user.user_id
         return repo_to_perm
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
         q = self.sa.query(Ui).filter(Ui.ui_key == '/').one()
         return q.ui_value
     def get(self, repo_id, cache=False):
         repo = self.sa.query(Repository) \
             .filter(Repository.repo_id == repo_id)
         if cache:
             repo = repo.options(FromCache("sql_cache_short",
                                           "get_repo_%s" % repo_id))
         return repo.scalar()
     def get_repo(self, repository):

kallithea/model/repo_group.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.repo_group
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 repo group model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jan 25, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import logging
 import traceback
 import shutil
 import datetime
 from kallithea.lib.utils2 import LazyProperty
 from kallithea.model import BaseModel
 from kallithea.model.db import RepoGroup, Ui, UserRepoGroupToPerm, \
     User, Permission, UserGroupRepoGroupToPerm, UserGroup, Repository
 log = logging.getLogger(__name__)
 class RepoGroupModel(BaseModel):
     cls = RepoGroup
     def _get_user_group(self, users_group):
         return self._get_instance(UserGroup, users_group,
                                   callback=UserGroup.get_by_group_name)
     def _get_repo_group(self, repo_group):
         return self._get_instance(RepoGroup, repo_group,
                                   callback=RepoGroup.get_by_group_name)
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
         q = Ui.get_by_key('paths', '/')
         return q.ui_value
     def _create_default_perms(self, new_group):
         # create default permission
         default_perm = 'group.read'
         def_user = User.get_default_user()
         for p in def_user.user_perms:
             if p.permission.permission_name.startswith('group.'):
                 default_perm = p.permission.permission_name
                 break
         repo_group_to_perm = UserRepoGroupToPerm()
         repo_group_to_perm.permission = Permission.get_by_key(default_perm)
         repo_group_to_perm.group = new_group
         repo_group_to_perm.user_id = def_user.user_id
         return repo_group_to_perm
     def _create_group(self, group_name):
         """
         makes repository group on filesystem
         :param repo_name:
         :param parent_id:
         """
         create_path = os.path.join(self.repos_path, group_name)
         log.debug('creating new group in %s', create_path)
         if os.path.isdir(create_path):
             raise Exception('That directory already exists !')
         os.makedirs(create_path)

kallithea/model/repo_permission.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.repo_permission
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 repository permission model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Oct 1, 2011
 :author: nvinot, marcink
 """
 import logging
 from kallithea.model import BaseModel
 from kallithea.model.db import UserRepoToPerm, UserGroupRepoToPerm, \
     Permission
 log = logging.getLogger(__name__)
 class RepositoryPermissionModel(BaseModel):
     cls = UserRepoToPerm
     def get_user_permission(self, repository, user):
         repository = self._get_repo(repository)
         user = self._get_user(user)
         return UserRepoToPerm.query() \
                 .filter(UserRepoToPerm.user == user) \
                 .filter(UserRepoToPerm.repository == repository) \
                 .scalar()
     def update_user_permission(self, repository, user, permission):
         permission = Permission.get_by_key(permission)
         current = self.get_user_permission(repository, user)
         if current:
             if not current.permission is permission:
                 current.permission = permission
         else:
             p = UserRepoToPerm()
             p.user = user
             p.repository = repository
             p.permission = permission
             self.sa.add(p)
     def delete_user_permission(self, repository, user):
         current = self.get_user_permission(repository, user)
         if current:
             self.sa.delete(current)
     def get_users_group_permission(self, repository, users_group):
         return UserGroupRepoToPerm.query() \
                 .filter(UserGroupRepoToPerm.users_group == users_group) \
                 .filter(UserGroupRepoToPerm.repository == repository) \
                 .scalar()
     def update_users_group_permission(self, repository, users_group,
                                       permission):
         permission = Permission.get_by_key(permission)
         current = self.get_users_group_permission(repository, users_group)
         if current:
             if not current.permission is permission:
                 current.permission = permission
         else:
             p = UserGroupRepoToPerm()
             p.users_group = users_group
             p.repository = repository
             p.permission = permission
             self.sa.add(p)
     def delete_users_group_permission(self, repository, users_group):

kallithea/model/user.py

➞

Show inline comments

@@ @@ -9,98 +9,96 @@ @@
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.user
 ~~~~~~~~~~~~~~~~~~~~
 users model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 9, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import hashlib
 import hmac
 import logging
 import time
 import traceback
 from pylons import config
 from pylons.i18n.translation import _
 from sqlalchemy.exc import DatabaseError
 from kallithea import EXTERN_TYPE_INTERNAL
 from kallithea.lib.utils2 import safe_str, generate_api_key, get_current_authuser
 from kallithea.lib.caching_query import FromCache
 from kallithea.model import BaseModel
 from kallithea.model.db import User, UserToPerm, Notification, \
     UserEmailMap, UserIpMap
 from kallithea.lib.exceptions import DefaultUserException, \
     UserOwnsReposException
 from kallithea.model.meta import Session
 log = logging.getLogger(__name__)
 class UserModel(BaseModel):
     password_reset_token_lifetime = 86400 # 24 hours
     cls = User
     def get(self, user_id, cache=False):
         user = self.sa.query(User)
         if cache:
             user = user.options(FromCache("sql_cache_short",
                                           "get_user_%s" % user_id))
         return user.get(user_id)
     def get_user(self, user):
         return self._get_user(user)
     def create(self, form_data, cur_user=None):
         if not cur_user:
             cur_user = getattr(get_current_authuser(), 'username', None)
         from kallithea.lib.hooks import log_create_user, \
             check_allowed_create_user
         _fd = form_data
         user_data = {
             'username': _fd['username'],
             'password': _fd['password'],
             'email': _fd['email'],
             'firstname': _fd['firstname'],
             'lastname': _fd['lastname'],
             'active': _fd['active'],
             'admin': False
+        }
         # raises UserCreationError if it's not allowed
         check_allowed_create_user(user_data, cur_user)
         from kallithea.lib.auth import get_crypt_password
         new_user = User()
         for k, v in form_data.items():
             if k == 'password':
                 v = get_crypt_password(v)
             if k == 'firstname':
                 k = 'name'
             setattr(new_user, k, v)
         new_user.api_key = generate_api_key()
         self.sa.add(new_user)
         log_create_user(new_user.get_dict(), cur_user)
         return new_user
     def create_or_update(self, username, password, email, firstname=u'',
                          lastname=u'', active=True, admin=False,
                          extern_type=None, extern_name=None, cur_user=None):
         """

kallithea/model/user_group.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.user_group
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 user group model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Oct 1, 2011
 :author: nvinot, marcink
 """
 import logging
 import traceback
 from kallithea.model import BaseModel
 from kallithea.model.db import UserGroupMember, UserGroup, \
     UserGroupRepoToPerm, Permission, UserGroupToPerm, User, UserUserGroupToPerm, \
     UserGroupUserGroupToPerm
 from kallithea.lib.exceptions import UserGroupsAssignedException, \
     RepoGroupAssignmentError
 log = logging.getLogger(__name__)
 class UserGroupModel(BaseModel):
     cls = UserGroup
     def _get_user_group(self, user_group):
         return self._get_instance(UserGroup, user_group,
                                   callback=UserGroup.get_by_group_name)
     def _create_default_perms(self, user_group):
         # create default permission
         default_perm = 'usergroup.read'
         def_user = User.get_default_user()
         for p in def_user.user_perms:
             if p.permission.permission_name.startswith('usergroup.'):
                 default_perm = p.permission.permission_name
                 break
         user_group_to_perm = UserUserGroupToPerm()
         user_group_to_perm.permission = Permission.get_by_key(default_perm)
         user_group_to_perm.user_group = user_group
         user_group_to_perm.user_id = def_user.user_id
         return user_group_to_perm
     def _update_permissions(self, user_group, perms_new=None,
                             perms_updates=None):
         from kallithea.lib.auth import HasUserGroupPermissionAny
         if not perms_new:
             perms_new = []
         if not perms_updates:
             perms_updates = []
         # update permissions
         for member, perm, member_type in perms_updates:
             if member_type == 'user':
                 # this updates existing one
                 self.grant_user_permission(
                     user_group=user_group, user=member, perm=perm
+                )
             else:
                 #check if we have permissions to alter this usergroup
                 if HasUserGroupPermissionAny('usergroup.read', 'usergroup.write',
                                              'usergroup.admin')(member):
                     self.grant_user_group_permission(
                         target_user_group=user_group, user_group=member, perm=perm
+                    )
         # set new permissions
         for member, perm, member_type in perms_new:
             if member_type == 'user':
                 self.grant_user_permission(
                     user_group=user_group, user=member, perm=perm
+                )

kallithea/tests/fixture.py

➞

Show inline comments

@@ @@ -192,97 +192,97 @@ class Fixture(object): @@
         gr = RepoGroup.get_by_group_name(gr.group_name)
         return gr
     def destroy_repo_group(self, repogroupid):
         RepoGroupModel().delete(repogroupid)
         Session().commit()
     def create_user(self, name, **kwargs):
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             user = User.get_by_username(name)
             if user:
                 return user
         form_data = self._get_user_create_params(name, **kwargs)
         user = UserModel().create(form_data)
         Session().commit()
         user = User.get_by_username(user.username)
         return user
     def destroy_user(self, userid):
         UserModel().delete(userid)
         Session().commit()
     def create_user_group(self, name, **kwargs):
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             gr = UserGroup.get_by_group_name(group_name=name)
             if gr:
                 return gr
         form_data = self._get_user_group_create_params(name, **kwargs)
         owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
         user_group = UserGroupModel().create(
             name=form_data['users_group_name'],
             description=form_data['user_group_description'],
             owner=owner, active=form_data['users_group_active'],
             group_data=form_data['user_group_data'])
         Session().commit()
         user_group = UserGroup.get_by_group_name(user_group.users_group_name)
         return user_group
     def destroy_user_group(self, usergroupid):
         UserGroupModel().delete(user_group=usergroupid, force=True)
         Session().commit()
     def create_gist(self, **kwargs):
         form_data = {
             'description': u'new-gist',
             'owner': TEST_USER_ADMIN_LOGIN,
-            'gist_type': GistModel.cls.GIST_PUBLIC,
+            'gist_type': Gist.GIST_PUBLIC,
             'lifetime': -1,
             'gist_mapping': {'filename1.txt':{'content':'hello world'},}
+        }
         form_data.update(kwargs)
         gist = GistModel().create(
             description=form_data['description'],owner=form_data['owner'],
             gist_mapping=form_data['gist_mapping'], gist_type=form_data['gist_type'],
             lifetime=form_data['lifetime']
+        )
         Session().commit()
         return gist
     def destroy_gists(self, gistid=None):
         for g in Gist.get_all():
             if gistid:
                 if gistid == g.gist_access_id:
                     GistModel().delete(g)
             else:
                 GistModel().delete(g)
         Session().commit()
     def load_resource(self, resource_name, strip=True):
         with open(os.path.join(FIXTURES, resource_name), 'rb') as f:
             source = f.read()
             if strip:
                 source = source.strip()
         return source
     def commit_change(self, repo, filename, content, message, vcs_type, parent=None, newfile=False):
         repo = Repository.get_by_repo_name(repo)
         _cs = parent
         if not parent:
             _cs = EmptyChangeset(alias=vcs_type)
         if newfile:
             nodes = {
                 filename: {
                     'content': content
+                }
+            }
             cs = ScmModel().create_nodes(
                 user=TEST_USER_ADMIN_LOGIN, repo=repo,
                 message=message,
                 nodes=nodes,
                 parent_cs=_cs,
                 author=TEST_USER_ADMIN_LOGIN,

0 comments (0 inline, 0 general)