kallithea Changeset - 7cde75eac0fe

Changeset - 7cde75eac0fe

Parent rev.

Child rev.

[Not reviewed]

beta

0 3 0

Marcin Kuzminski - 13 years ago 2013-03-07 20:06:45
marcin@python-works.com

get_locks API function draft

3 files changed with 74 insertions and 6 deletions:

rhodecode/controllers/api/api.py

rhodecode/model/db.py

rhodecode/tests/api/api_base.py

0 comments (0 inline, 0 general)

rhodecode/controllers/api/api.py

➞

Show inline comments

@@ @@ -12,57 +12,58 @@ @@
 """
 # 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; version 2
 # of the License or (at your opinion) any later version of the license.
+#
 # 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, write to the Free Software
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
 # MA  02110-1301, USA.
 import traceback
 import logging
 from rhodecode.controllers.api import JSONRPCController, JSONRPCError
 from rhodecode.lib.auth import PasswordGenerator, AuthUser, \
     HasPermissionAllDecorator, HasPermissionAnyDecorator, \
     HasPermissionAnyApi, HasRepoPermissionAnyApi
 from rhodecode.lib.utils import map_groups, repo2db_mapper
 from rhodecode.lib.utils2 import str2bool, time_to_datetime
+from rhodecode.lib.utils2 import str2bool, time_to_datetime, safe_int
 from rhodecode.lib import helpers as h
 from rhodecode.model.meta import Session
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.user import UserModel
 from rhodecode.model.users_group import UserGroupModel
 from rhodecode.model.permission import PermissionModel
 from rhodecode.model.db import Repository, RhodeCodeSetting, UserIpMap
 from rhodecode.lib.compat import json
 log = logging.getLogger(__name__)
 class OptionalAttr(object):
     """
     Special Optional Option that defines other attribute
     """
     def __init__(self, attr_name):
         self.attr_name = attr_name
     def __repr__(self):
         return '<OptionalAttr:%s>' % self.attr_name
     def __call__(self):
         return self
 #alias
 OAttr = OptionalAttr
 class Optional(object):
     """
     Defines an optional parameter::
@@ @@ -253,66 +254,104 @@ class ApiController(JSONRPCController): @@
                 raise JSONRPCError(
                     'userid is not the same as your user'
+                )
         else:
             raise JSONRPCError('repository `%s` does not exist' % (repoid))
         if isinstance(userid, Optional):
             userid = apiuser.user_id
         user = get_user_or_error(userid)
         if isinstance(locked, Optional):
             lockobj = Repository.getlock(repo)
             if lockobj[0] is None:
                 return ('Repo `%s` not locked. Locked=`False`.'
                         % (repo.repo_name))
             else:
                 userid, time_ = lockobj
                 user = get_user_or_error(userid)
                 return ('Repo `%s` locked by `%s`. Locked=`True`. '
                         'Locked since: `%s`'
                     % (repo.repo_name, user.username,
-                       h.fmt_date(time_to_datetime(time_))))
+                       json.dumps(time_to_datetime(time_))))
         else:
             locked = str2bool(locked)
             try:
                 if locked:
                     Repository.lock(repo, user.user_id)
                 else:
                     Repository.unlock(repo)
                 return ('User `%s` set lock state for repo `%s` to `%s`'
                         % (user.username, repo.repo_name, locked))
             except Exception:
                 log.error(traceback.format_exc())
                 raise JSONRPCError(
                     'Error occurred locking repository `%s`' % repo.repo_name
+                )
     def get_locks(self, apiuser, userid=Optional(OAttr('apiuser'))):
         """
         Get all locks for given userid, if
         this command is runned by non-admin account userid is set to user
         who is calling this method, thus returning locks for himself
         :param apiuser:
         :param userid:
         """
         if HasPermissionAnyApi('hg.admin')(user=apiuser):
             pass
         else:
             #make sure normal user does not pass someone else userid,
             #he is not allowed to do that
             if not isinstance(userid, Optional) and userid != apiuser.user_id:
                 raise JSONRPCError(
                     'userid is not the same as your user'
+                )
         ret = []
         if isinstance(userid, Optional):
             user = None
         else:
             user = get_user_or_error(userid)
         #show all locks
         for r in Repository.getAll():
             userid, time_ = r.locked
             if time_:
                 _api_data = r.get_api_data()
                 # if we use userfilter just show the locks for this user
                 if user:
                     if safe_int(userid) == user.user_id:
                         ret.append(_api_data)
                 else:
                     ret.append(_api_data)
         return ret
     @HasPermissionAllDecorator('hg.admin')
     def show_ip(self, apiuser, userid):
         """
         Shows IP address as seen from RhodeCode server, together with all
         defined IP addresses for given user
         :param apiuser:
         :param userid:
         """
         user = get_user_or_error(userid)
         ips = UserIpMap.query().filter(UserIpMap.user == user).all()
         return dict(
             ip_addr_server=self.ip_addr,
             user_ips=ips
+        )
     def get_user(self, apiuser, userid=Optional(OAttr('apiuser'))):
         """"
         Get a user by username, or userid, if userid is given
         :param apiuser:
         :param userid:
         """
         if HasPermissionAnyApi('hg.admin')(user=apiuser) is False:

rhodecode/model/db.py

➞

Show inline comments

@@ @@ -26,49 +26,49 @@ @@
 import os
 import logging
 import datetime
 import traceback
 import hashlib
 import time
 from collections import defaultdict
 from sqlalchemy import *
 from sqlalchemy.ext.hybrid import hybrid_property
 from sqlalchemy.orm import relationship, joinedload, class_mapper, validates
 from sqlalchemy.exc import DatabaseError
 from beaker.cache import cache_region, region_invalidate
 from webob.exc import HTTPNotFound
 from pylons.i18n.translation import lazy_ugettext as _
 from rhodecode.lib.vcs import get_backend
 from rhodecode.lib.vcs.utils.helpers import get_scm
 from rhodecode.lib.vcs.exceptions import VCSError
 from rhodecode.lib.vcs.utils.lazy import LazyProperty
 from rhodecode.lib.vcs.backends.base import EmptyChangeset
 from rhodecode.lib.utils2 import str2bool, safe_str, get_changeset_safe, \
     safe_unicode, remove_suffix, remove_prefix
+    safe_unicode, remove_suffix, remove_prefix, time_to_datetime
 from rhodecode.lib.compat import json
 from rhodecode.lib.caching_query import FromCache
 from rhodecode.model.meta import Base, Session
 URL_SEP = '/'
 log = logging.getLogger(__name__)
 #==============================================================================
 # BASE CLASSES
 #==============================================================================
 _hash_key = lambda k: hashlib.md5(safe_str(k)).hexdigest()
 class BaseModel(object):
     """
     Base Model for all classess
     """
     @classmethod
     def _get_keys(cls):
         """return column names for this model """
         return class_mapper(cls).c.keys()
@@ @@ -951,49 +951,53 @@ class Repository(Base, BaseModel): @@
         from rhodecode.lib.utils import is_valid_repo
         return is_valid_repo(repo_name, cls.base_path())
     def get_api_data(self):
         """
         Common function for generating repo api data
         """
         repo = self
         data = dict(
             repo_id=repo.repo_id,
             repo_name=repo.repo_name,
             repo_type=repo.repo_type,
             clone_uri=repo.clone_uri,
             private=repo.private,
             created_on=repo.created_on,
             description=repo.description,
             landing_rev=repo.landing_rev,
             owner=repo.user.username,
             fork_of=repo.fork.repo_name if repo.fork else None,
             enable_statistics=repo.enable_statistics,
             enable_locking=repo.enable_locking,
             enable_downloads=repo.enable_downloads,
             last_changeset=repo.changeset_cache
             last_changeset=repo.changeset_cache,
             locked_by=User.get(self.locked[0]).get_api_data() \
                 if self.locked[0] else None,
             locked_date=time_to_datetime(self.locked[1]) \
                 if self.locked[1] else None
+        )
         rc_config = RhodeCodeSetting.get_app_settings()
         repository_fields = str2bool(rc_config.get('rhodecode_repository_fields'))
         if repository_fields:
             for f in self.extra_fields:
                 data[f.field_key_prefixed] = f.field_value
         return data
     @classmethod
     def lock(cls, repo, user_id):
         repo.locked = [user_id, time.time()]
         Session().add(repo)
         Session().commit()
     @classmethod
     def unlock(cls, repo):
         repo.locked = None
         Session().add(repo)
         Session().commit()
     @classmethod
     def getlock(cls, repo):
         return repo.locked

rhodecode/tests/api/api_base.py

➞

Show inline comments

@@ @@ -350,70 +350,95 @@ class BaseTestApi(object): @@
         response = api_call(self, params)
         expected = 'repository `%s` does not exist' % (self.REPO)
         self._compare_error(id_, expected, given=response.body)
     def test_api_lock_repo_lock_release(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   repoid=self.REPO,
                                   locked=False)
         response = api_call(self, params)
         expected = ('User `%s` set lock state for repo `%s` to `%s`'
                    % (TEST_USER_ADMIN_LOGIN, self.REPO, False))
         self._compare_ok(id_, expected, given=response.body)
     def test_api_lock_repo_lock_aquire_optional_userid(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = ('User `%s` set lock state for repo `%s` to `%s`'
                    % (TEST_USER_ADMIN_LOGIN, self.REPO, True))
         self._compare_ok(id_, expected, given=response.body)
     def test_api_lock_repo_lock_optional_locked(self):
         from rhodecode.lib import helpers
         from rhodecode.lib.utils2 import  time_to_datetime
-        _locked_since = helpers.fmt_date(time_to_datetime(Repository\
+        _locked_since = json.dumps(time_to_datetime(Repository\
                                     .get_by_repo_name(self.REPO).locked[1]))
         id_, params = _build_data(self.apikey, 'lock',
                                   repoid=self.REPO)
         response = api_call(self, params)
         expected = ('Repo `%s` locked by `%s`. Locked=`True`. Locked since: `%s`'
                    % (self.REPO, TEST_USER_ADMIN_LOGIN, _locked_since))
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(Repository, 'lock', crash)
     def test_api_lock_error(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = 'Error occurred locking repository `%s`' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_locks_regular_user(self):
         id_, params = _build_data(self.apikey_regular, 'get_locks')
         response = api_call(self, params)
         expected = []
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_locks_with_userid_regular_user(self):
         id_, params = _build_data(self.apikey_regular, 'get_locks',
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         expected = 'userid is not the same as your user'
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_locks(self):
         id_, params = _build_data(self.apikey, 'get_locks')
         response = api_call(self, params)
         expected = []
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_locks_with_userid(self):
         id_, params = _build_data(self.apikey, 'get_locks',
                                   userid=TEST_USER_REGULAR_LOGIN)
         response = api_call(self, params)
         expected = []
         self._compare_ok(id_, expected, given=response.body)
     def test_api_create_existing_user(self):
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=TEST_USER_ADMIN_LOGIN,
                                   email='test@foo.com',
                                   password='trololo')
         response = api_call(self, params)
         expected = "user `%s` already exist" % TEST_USER_ADMIN_LOGIN
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_user_with_existing_email(self):
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=TEST_USER_ADMIN_LOGIN + 'new',
                                   email=TEST_USER_REGULAR_EMAIL,
                                   password='trololo')
         response = api_call(self, params)
         expected = "email `%s` already exist" % TEST_USER_REGULAR_EMAIL
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_user(self):
         username = 'test_new_api_user'
         email = username + "@foo.com"

0 comments (0 inline, 0 general)