kallithea Changeset - 388843a3a3c0

Changeset - 388843a3a3c0

Parent rev.

Child rev.

[Not reviewed]

beta

0 5 0

Marcin Kuzminski - 13 years ago 2012-06-26 21:57:31
marcin@python-works.com

Updated create_or_update method to not change API key when password is not updated

5 files changed with 29 insertions and 38 deletions:

rhodecode/lib/db_manage.py

rhodecode/model/user.py

rhodecode/tests/functional/test_admin_notifications.py

rhodecode/tests/functional/test_forks.py

rhodecode/tests/test_models.py

0 comments (0 inline, 0 general)

rhodecode/lib/db_manage.py

➞

Show inline comments

@@ @@ -372,166 +372,166 @@ class DbManage(object): @@
                 .filter(UserRepoGroupToPerm.group == g)\
                 .filter(UserRepoGroupToPerm.user == def_usr)\
                 .scalar()
             if default is None:
                 log.debug('missing default permission for group %s adding' % g)
                 ReposGroupModel()._create_default_perms(g)
     def config_prompt(self, test_repo_path='', retries=3, defaults={}):
         _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 path_ok is False:
             retries -= 1
             return self.config_prompt(test_repo_path, retries)
         return 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
         hgsettings1 = RhodeCodeSetting('realm', 'RhodeCode authentication')
         hgsettings2 = RhodeCodeSetting('title', 'RhodeCode')
         hgsettings3 = RhodeCodeSetting('ga_code', '')
         self.sa.add(web1)
         self.sa.add(web2)
         self.sa.add(web3)
         self.sa.add(web4)
         self.sa.add(paths)
         self.sa.add(hgsettings1)
         self.sa.add(hgsettings2)
         self.sa.add(hgsettings3)
         self.create_ldap_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,
-                                     name='RhodeCode', lastname='Admin',
+                                     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',
-                              name='Anonymous', lastname='User')
+                              firstname='Anonymous', lastname='User')
     def create_permissions(self):
         # module.(access|create|change|delete)_[name]
         # module.(none|read|write|admin)
         perms = [
          ('repository.none', 'Repository no access'),
          ('repository.read', 'Repository read access'),
          ('repository.write', 'Repository write access'),
          ('repository.admin', 'Repository admin access'),
          ('group.none', 'Repositories Group no access'),
          ('group.read', 'Repositories Group read access'),
          ('group.write', 'Repositories Group write access'),
          ('group.admin', 'Repositories Group admin access'),
          ('hg.admin', 'Hg Administrator'),
          ('hg.create.repository', 'Repository create'),
          ('hg.create.none', 'Repository creation disabled'),
          ('hg.register.none', 'Register disabled'),
          ('hg.register.manual_activate', 'Register new user with RhodeCode '
                                          'without manual activation'),
          ('hg.register.auto_activate', 'Register new user with RhodeCode '
                                         'without auto activation'),
+        ]
         for p in perms:
             if not Permission.get_by_key(p[0]):
                 new_perm = Permission()
                 new_perm.permission_name = p[0]
                 new_perm.permission_longname = p[1]
                 self.sa.add(new_perm)
     def populate_default_permissions(self):
         log.info('creating default user permissions')
         default_user = self.sa.query(User)\
         .filter(User.username == 'default').scalar()
         reg_perm = UserToPerm()
         reg_perm.user = default_user
         reg_perm.permission = self.sa.query(Permission)\
         .filter(Permission.permission_name == 'hg.register.manual_activate')\
         .scalar()
         create_repo_perm = UserToPerm()
         create_repo_perm.user = default_user
         create_repo_perm.permission = self.sa.query(Permission)\
         .filter(Permission.permission_name == 'hg.create.repository')\
         .scalar()
         default_repo_perm = UserToPerm()
         default_repo_perm.user = default_user
         default_repo_perm.permission = self.sa.query(Permission)\
         .filter(Permission.permission_name == 'repository.read')\
         .scalar()
         self.sa.add(reg_perm)
         self.sa.add(create_repo_perm)
         self.sa.add(default_repo_perm)

rhodecode/model/user.py

➞

Show inline comments

@@ @@ -8,228 +8,232 @@ @@
     :created_on: Apr 9, 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 pylons import url
 from pylons.i18n.translation import _
 from sqlalchemy.exc import DatabaseError
 from sqlalchemy.orm import joinedload
 from rhodecode.lib.utils2 import safe_unicode, generate_api_key
 from rhodecode.lib.caching_query import FromCache
 from rhodecode.model import BaseModel
 from rhodecode.model.db import User, UserRepoToPerm, Repository, Permission, \
     UserToPerm, UsersGroupRepoToPerm, UsersGroupToPerm, UsersGroupMember, \
     Notification, RepoGroup, UserRepoGroupToPerm, UsersGroupRepoGroupToPerm, \
     UserEmailMap
 from rhodecode.lib.exceptions import DefaultUserException, \
     UserOwnsReposException
 log = logging.getLogger(__name__)
 PERM_WEIGHTS = {
     'repository.none': 0,
     'repository.read': 1,
     'repository.write': 3,
     'repository.admin': 4,
     'group.none': 0,
     'group.read': 1,
     'group.write': 3,
     'group.admin': 4,
+}
 class UserModel(BaseModel):
     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 get_by_username(self, username, cache=False, case_insensitive=False):
         if case_insensitive:
             user = self.sa.query(User).filter(User.username.ilike(username))
         else:
             user = self.sa.query(User)\
                 .filter(User.username == username)
         if cache:
             user = user.options(FromCache("sql_cache_short",
                                           "get_user_%s" % username))
         return user.scalar()
     def get_by_api_key(self, api_key, cache=False):
         return User.get_by_api_key(api_key, cache)
     def create(self, form_data):
         from rhodecode.lib.auth import get_crypt_password
         try:
             new_user = User()
             for k, v in form_data.items():
                 if k == 'password':
                     v = get_crypt_password(v)
                 setattr(new_user, k, v)
             new_user.api_key = generate_api_key(form_data['username'])
             self.sa.add(new_user)
             return new_user
         except:
             log.error(traceback.format_exc())
             raise
     def create_or_update(self, username, password, email, name, lastname,
                          active=True, admin=False, ldap_dn=None):
     def create_or_update(self, username, password, email, firstname='',
                          lastname='', active=True, admin=False, ldap_dn=None):
         """
         Creates a new instance if not found, or updates current one
         :param username:
         :param password:
         :param email:
         :param active:
-        :param name:
+        :param firstname:
         :param lastname:
         :param active:
         :param admin:
         :param ldap_dn:
         """
         from rhodecode.lib.auth import get_crypt_password
         log.debug('Checking for %s account in RhodeCode database' % username)
         user = User.get_by_username(username, case_insensitive=True)
         if user is None:
             log.debug('creating new user %s' % username)
             new_user = User()
             edit = False
         else:
             log.debug('updating user %s' % username)
             new_user = user
             edit = True
         try:
             new_user.username = username
             new_user.admin = admin
             new_user.password = get_crypt_password(password)
             new_user.api_key = generate_api_key(username)
             # set password only if creating an user or password is changed
             if edit is False or user.password != password:
                 new_user.password = get_crypt_password(password)
                 new_user.api_key = generate_api_key(username)
             new_user.email = email
             new_user.active = active
             new_user.ldap_dn = safe_unicode(ldap_dn) if ldap_dn else None
-            new_user.name = name
+            new_user.name = firstname
             new_user.lastname = lastname
             self.sa.add(new_user)
             return new_user
         except (DatabaseError,):
             log.error(traceback.format_exc())
             raise
     def create_for_container_auth(self, username, attrs):
         """
         Creates the given user if it's not already in the database
         :param username:
         :param attrs:
         """
         if self.get_by_username(username, case_insensitive=True) is None:
             # autogenerate email for container account without one
             generate_email = lambda usr: '%s@container_auth.account' % usr
             try:
                 new_user = User()
                 new_user.username = username
                 new_user.password = None
                 new_user.api_key = generate_api_key(username)
                 new_user.email = attrs['email']
                 new_user.active = attrs.get('active', True)
                 new_user.name = attrs['name'] or generate_email(username)
                 new_user.lastname = attrs['lastname']
                 self.sa.add(new_user)
                 return new_user
             except (DatabaseError,):
                 log.error(traceback.format_exc())
                 self.sa.rollback()
                 raise
         log.debug('User %s already exists. Skipping creation of account'
                   ' for container auth.', username)
         return None
     def create_ldap(self, username, password, user_dn, attrs):
         """
         Checks if user is in database, if not creates this user marked
         as ldap user
         :param username:
         :param password:
         :param user_dn:
         :param attrs:
         """
         from rhodecode.lib.auth import get_crypt_password
         log.debug('Checking for such ldap account in RhodeCode database')
         if self.get_by_username(username, case_insensitive=True) is None:
             # autogenerate email for ldap account without one
             generate_email = lambda usr: '%s@ldap.account' % usr
             try:
                 new_user = User()
                 username = username.lower()
                 # add ldap account always lowercase
                 new_user.username = username
                 new_user.password = get_crypt_password(password)
                 new_user.api_key = generate_api_key(username)
                 new_user.email = attrs['email'] or generate_email(username)
                 new_user.active = attrs.get('active', True)
                 new_user.ldap_dn = safe_unicode(user_dn)
                 new_user.name = attrs['name']
                 new_user.lastname = attrs['lastname']
                 self.sa.add(new_user)
                 return new_user
             except (DatabaseError,):
                 log.error(traceback.format_exc())
                 self.sa.rollback()
                 raise
         log.debug('this %s user exists skipping creation of ldap account',
                   username)
         return None
     def create_registration(self, form_data):
         from rhodecode.model.notification import NotificationModel
         try:
             form_data['admin'] = False
             new_user = self.create(form_data)
             self.sa.add(new_user)
             self.sa.flush()
             # notification to admins
             subject = _('new user registration')
             body = ('New user registration\n'
                     '---------------------\n'
                     '- Username: %s\n'
                     '- Full Name: %s\n'
                     '- Email: %s\n')

rhodecode/tests/functional/test_admin_notifications.py

➞

Show inline comments

 from rhodecode.tests import *
 from rhodecode.model.db import Notification, User, UserNotification
 from rhodecode.model.user import UserModel
 from rhodecode.model.notification import NotificationModel
 from rhodecode.model.meta import Session
 class TestNotificationsController(TestController):
     def tearDown(self):
         for n in Notification.query().all():
             inst = Notification.get(n.notification_id)
             Session.delete(inst)
         Session.commit()
     def test_index(self):
         self.log_user()
         u1 = UserModel().create_or_update(username='u1', password='qweqwe',
                                                email='u1@rhodecode.org',
                                                name='u1', lastname='u1').user_id
                                                firstname='u1', lastname='u1')
         u1 = u1.user_id
         response = self.app.get(url('notifications'))
         self.assertTrue('''<div class="table">No notifications here yet</div>'''
                         in response.body)
         cur_user = self._get_logged_user()
         NotificationModel().create(created_by=u1, subject=u'test_notification_1',
                                    body=u'notification_1',
                                    recipients=[cur_user])
         Session.commit()
         response = self.app.get(url('notifications'))
         self.assertTrue(u'test_notification_1' in response.body)
 #    def test_index_as_xml(self):
 #        response = self.app.get(url('formatted_notifications', format='xml'))
+#
 #    def test_create(self):
 #        response = self.app.post(url('notifications'))
+#
 #    def test_new(self):
 #        response = self.app.get(url('new_notification'))
+#
 #    def test_new_as_xml(self):
 #        response = self.app.get(url('formatted_new_notification', format='xml'))
+#
 #    def test_update(self):
 #        response = self.app.put(url('notification', notification_id=1))
+#
 #    def test_update_browser_fakeout(self):
 #        response = self.app.post(url('notification', notification_id=1), params=dict(_method='put'))
     def test_delete(self):
         self.log_user()
         cur_user = self._get_logged_user()
         u1 = UserModel().create_or_update(username='u1', password='qweqwe',
                                                email='u1@rhodecode.org',
-                                               name='u1', lastname='u1')
+                                               firstname='u1', lastname='u1')
         u2 = UserModel().create_or_update(username='u2', password='qweqwe',
                                                email='u2@rhodecode.org',
-                                               name='u2', lastname='u2')
+                                               firstname='u2', lastname='u2')
         # make notifications
         notification = NotificationModel().create(created_by=cur_user,
                                                   subject=u'test',
                                                   body=u'hi there',
                                                   recipients=[cur_user, u1, u2])
         Session.commit()
         u1 = User.get(u1.user_id)
         u2 = User.get(u2.user_id)
         # check DB
         get_notif = lambda un:[x.notification for x in un]
+        get_notif = lambda un: [x.notification for x in un]
         self.assertEqual(get_notif(cur_user.notifications), [notification])
         self.assertEqual(get_notif(u1.notifications), [notification])
         self.assertEqual(get_notif(u2.notifications), [notification])
         cur_usr_id = cur_user.user_id
         response = self.app.delete(url('notification',
                                        notification_id=
                                        notification.notification_id))
         cur_user = User.get(cur_usr_id)
         self.assertEqual(cur_user.notifications, [])
 #    def test_delete_browser_fakeout(self):
 #        response = self.app.post(url('notification', notification_id=1), params=dict(_method='delete'))
     def test_show(self):
         self.log_user()
         cur_user = self._get_logged_user()
         u1 = UserModel().create_or_update(username='u1', password='qweqwe',
                                                email='u1@rhodecode.org',
-                                               name='u1', lastname='u1')
+                                               firstname='u1', lastname='u1')
         u2 = UserModel().create_or_update(username='u2', password='qweqwe',
                                                email='u2@rhodecode.org',
-                                               name='u2', lastname='u2')
+                                               firstname='u2', lastname='u2')
         notification = NotificationModel().create(created_by=cur_user,
                                                   subject=u'test',
                                                   body=u'hi there',
                                                   recipients=[cur_user, u1, u2])
         response = self.app.get(url('notification',
                                     notification_id=notification.notification_id))
 #    def test_show_as_xml(self):
 #        response = self.app.get(url('formatted_notification', notification_id=1, format='xml'))
+#
 #    def test_edit(self):
 #        response = self.app.get(url('edit_notification', notification_id=1))
+#
 #    def test_edit_as_xml(self):
 #        response = self.app.get(url('formatted_edit_notification', notification_id=1, format='xml'))

rhodecode/tests/functional/test_forks.py

➞

Show inline comments

 from rhodecode.tests import *
 from rhodecode.model.db import Repository
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.user import UserModel
 class TestForksController(TestController):
     def setUp(self):
         self.username = u'forkuser'
         self.password = u'qweqwe'
         self.u1 = UserModel().create_or_update(
             username=self.username, password=self.password,
-            email=u'fork_king@rhodecode.org', name=u'u1', lastname=u'u1'
+            email=u'fork_king@rhodecode.org', firstname=u'u1', lastname=u'u1'
+        )
         self.Session.commit()
     def tearDown(self):
         self.Session.delete(self.u1)
         self.Session.commit()
     def test_index(self):
         self.log_user()
         repo_name = HG_REPO
         response = self.app.get(url(controller='forks', action='forks',
                                     repo_name=repo_name))
         self.assertTrue("""There are no forks yet""" in response.body)
     def test_index_with_fork_hg(self):
         self.log_user()
         # create a fork
         fork_name = HG_FORK
         description = 'fork of vcs test'
         repo_name = HG_REPO
         org_repo = Repository.get_by_repo_name(repo_name)
         response = self.app.post(url(controller='forks',
                                      action='fork_create',
                                     repo_name=repo_name),
                                     {'repo_name': fork_name,
                                      'repo_group': '',
                                      'fork_parent_id': org_repo.repo_id,
                                      'repo_type': 'hg',
                                      'description': description,
                                      'private': 'False',
                                      'landing_rev': 'tip'})
         response = self.app.get(url(controller='forks', action='forks',
                                     repo_name=repo_name))
         response.mustcontain(
             """<a href="/%s/summary">%s</a>""" % (fork_name, fork_name)
+        )
         #remove this fork
         response = self.app.delete(url('repo', repo_name=fork_name))
     def test_index_with_fork_git(self):
         self.log_user()
         # create a fork
         fork_name = GIT_FORK
         description = 'fork of vcs test'
         repo_name = GIT_REPO
         org_repo = Repository.get_by_repo_name(repo_name)
         response = self.app.post(url(controller='forks',
                                      action='fork_create',
                                     repo_name=repo_name),
                                     {'repo_name': fork_name,
                                      'repo_group': '',
                                      'fork_parent_id': org_repo.repo_id,
                                      'repo_type': 'git',
                                      'description': description,
                                      'private': 'False',
                                      'landing_rev': 'tip'})
         response = self.app.get(url(controller='forks', action='forks',
                                     repo_name=repo_name))
         response.mustcontain(
             """<a href="/%s/summary">%s</a>""" % (fork_name, fork_name)
+        )
         #remove this fork
         response = self.app.delete(url('repo', repo_name=fork_name))
     def test_z_fork_create(self):
         self.log_user()
         fork_name = HG_FORK
         description = 'fork of vcs test'
         repo_name = HG_REPO
         org_repo = Repository.get_by_repo_name(repo_name)
         response = self.app.post(url(controller='forks', action='fork_create',
                                     repo_name=repo_name),
                                     {'repo_name':fork_name,
                                      'repo_group':'',
                                      'fork_parent_id':org_repo.repo_id,
                                      'repo_type':'hg',
                                      'description':description,
                                      'private':'False',
                                      'landing_rev': 'tip'})
         #test if we have a message that fork is ok
         self.checkSessionFlash(response,
                 'forked %s repository as %s' % (repo_name, fork_name))
         #test if the fork was created in the database
         fork_repo = self.Session.query(Repository)\
             .filter(Repository.repo_name == fork_name).one()

rhodecode/tests/test_models.py

➞

Show inline comments

@@ @@ -92,491 +92,491 @@ class TestReposGroups(unittest.TestCase) @@
         self.assertEqual(RepoGroup.get(sg1.group_id), None)
         self.assertFalse(self.__check_path('deteteme'))
         sg1 = _make_group('deleteme', parent_id=self.g1.group_id)
         self.__delete_group(sg1.group_id)
         self.assertEqual(RepoGroup.get(sg1.group_id), None)
         self.assertFalse(self.__check_path('test1', 'deteteme'))
     def test_rename_single_group(self):
         sg1 = _make_group('initial')
         new_sg1 = self.__update_group(sg1.group_id, 'after')
         self.assertTrue(self.__check_path('after'))
         self.assertEqual(RepoGroup.get_by_group_name('initial'), None)
     def test_update_group_parent(self):
         sg1 = _make_group('initial', parent_id=self.g1.group_id)
         new_sg1 = self.__update_group(sg1.group_id, 'after', parent_id=self.g1.group_id)
         self.assertTrue(self.__check_path('test1', 'after'))
         self.assertEqual(RepoGroup.get_by_group_name('test1/initial'), None)
         new_sg1 = self.__update_group(sg1.group_id, 'after', parent_id=self.g3.group_id)
         self.assertTrue(self.__check_path('test3', 'after'))
         self.assertEqual(RepoGroup.get_by_group_name('test3/initial'), None)
         new_sg1 = self.__update_group(sg1.group_id, 'hello')
         self.assertTrue(self.__check_path('hello'))
         self.assertEqual(RepoGroup.get_by_group_name('hello'), new_sg1)
     def test_subgrouping_with_repo(self):
         g1 = _make_group('g1')
         g2 = _make_group('g2')
         # create new repo
         form_data = dict(repo_name='john',
                          repo_name_full='john',
                          fork_name=None,
                          description=None,
                          repo_group=None,
                          private=False,
                          repo_type='hg',
                          clone_uri=None,
                          landing_rev='tip')
         cur_user = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         r = RepoModel().create(form_data, cur_user)
         self.assertEqual(r.repo_name, 'john')
         # put repo into group
         form_data = form_data
         form_data['repo_group'] = g1.group_id
         form_data['perms_new'] = []
         form_data['perms_updates'] = []
         RepoModel().update(r.repo_name, form_data)
         self.assertEqual(r.repo_name, 'g1/john')
         self.__update_group(g1.group_id, 'g1', parent_id=g2.group_id)
         self.assertTrue(self.__check_path('g2', 'g1'))
         # test repo
         self.assertEqual(r.repo_name, RepoGroup.url_sep().join(['g2', 'g1', r.just_name]))
     def test_move_to_root(self):
         g1 = _make_group('t11')
         Session.commit()
         g2 = _make_group('t22', parent_id=g1.group_id)
         Session.commit()
         self.assertEqual(g2.full_path, 't11/t22')
         self.assertTrue(self.__check_path('t11', 't22'))
         g2 = self.__update_group(g2.group_id, 'g22', parent_id=None)
         Session.commit()
         self.assertEqual(g2.group_name, 'g22')
         # we moved out group from t1 to '' so it's full path should be 'g2'
         self.assertEqual(g2.full_path, 'g22')
         self.assertFalse(self.__check_path('t11', 't22'))
         self.assertTrue(self.__check_path('g22'))
 class TestUser(unittest.TestCase):
     def __init__(self, methodName='runTest'):
         Session.remove()
         super(TestUser, self).__init__(methodName=methodName)
     def test_create_and_remove(self):
         usr = UserModel().create_or_update(username=u'test_user',
                                            password=u'qweqwe',
                                      email=u'u232@rhodecode.org',
-                                     name=u'u1', lastname=u'u1')
+                                     firstname=u'u1', lastname=u'u1')
         Session.commit()
         self.assertEqual(User.get_by_username(u'test_user'), usr)
         # make users group
         users_group = UsersGroupModel().create('some_example_group')
         Session.commit()
         UsersGroupModel().add_user_to_group(users_group, usr)
         Session.commit()
         self.assertEqual(UsersGroup.get(users_group.users_group_id), users_group)
         self.assertEqual(UsersGroupMember.query().count(), 1)
         UserModel().delete(usr.user_id)
         Session.commit()
         self.assertEqual(UsersGroupMember.query().all(), [])
     def test_additonal_email_as_main(self):
         usr = UserModel().create_or_update(username=u'test_user',
                                            password=u'qweqwe',
                                      email=u'main_email@rhodecode.org',
-                                     name=u'u1', lastname=u'u1')
+                                     firstname=u'u1', lastname=u'u1')
         Session.commit()
         def do():
             m = UserEmailMap()
             m.email = u'main_email@rhodecode.org'
             m.user = usr
             Session.add(m)
             Session.commit()
         self.assertRaises(AttributeError, do)
         UserModel().delete(usr.user_id)
         Session.commit()
     def test_extra_email_map(self):
         usr = UserModel().create_or_update(username=u'test_user',
                                            password=u'qweqwe',
                                      email=u'main_email@rhodecode.org',
-                                     name=u'u1', lastname=u'u1')
+                                     firstname=u'u1', lastname=u'u1')
         Session.commit()
         m = UserEmailMap()
         m.email = u'main_email2@rhodecode.org'
         m.user = usr
         Session.add(m)
         Session.commit()
         u = User.get_by_email(email='main_email@rhodecode.org')
         self.assertEqual(usr.user_id, u.user_id)
         self.assertEqual(usr.username, u.username)
         u = User.get_by_email(email='main_email2@rhodecode.org')
         self.assertEqual(usr.user_id, u.user_id)
         self.assertEqual(usr.username, u.username)
         u = User.get_by_email(email='main_email3@rhodecode.org')
         self.assertEqual(None, u)
         UserModel().delete(usr.user_id)
         Session.commit()
 class TestNotifications(unittest.TestCase):
     def __init__(self, methodName='runTest'):
         Session.remove()
         self.u1 = UserModel().create_or_update(username=u'u1',
                                         password=u'qweqwe',
                                         email=u'u1@rhodecode.org',
-                                        name=u'u1', lastname=u'u1')
+                                        firstname=u'u1', lastname=u'u1')
         Session.commit()
         self.u1 = self.u1.user_id
         self.u2 = UserModel().create_or_update(username=u'u2',
                                         password=u'qweqwe',
                                         email=u'u2@rhodecode.org',
                                         name=u'u2', lastname=u'u3')
         Session.commit()
         self.u2 = self.u2.user_id
         self.u3 = UserModel().create_or_update(username=u'u3',
                                         password=u'qweqwe',
                                         email=u'u3@rhodecode.org',
                                         name=u'u3', lastname=u'u3')
         Session.commit()
         self.u3 = self.u3.user_id
         super(TestNotifications, self).__init__(methodName=methodName)
     def _clean_notifications(self):
         for n in Notification.query().all():
             Session.delete(n)
         Session.commit()
         self.assertEqual(Notification.query().all(), [])
     def tearDown(self):
         self._clean_notifications()
     def test_create_notification(self):
         self.assertEqual([], Notification.query().all())
         self.assertEqual([], UserNotification.query().all())
         usrs = [self.u1, self.u2]
         notification = NotificationModel().create(created_by=self.u1,
                                            subject=u'subj', body=u'hi there',
                                            recipients=usrs)
         Session.commit()
         u1 = User.get(self.u1)
         u2 = User.get(self.u2)
         u3 = User.get(self.u3)
         notifications = Notification.query().all()
         self.assertEqual(len(notifications), 1)
         unotification = UserNotification.query()\
             .filter(UserNotification.notification == notification).all()
         self.assertEqual(notifications[0].recipients, [u1, u2])
         self.assertEqual(notification.notification_id,
                          notifications[0].notification_id)
         self.assertEqual(len(unotification), len(usrs))
         self.assertEqual([x.user.user_id for x in unotification], usrs)
     def test_user_notifications(self):
         self.assertEqual([], Notification.query().all())
         self.assertEqual([], UserNotification.query().all())
         notification1 = NotificationModel().create(created_by=self.u1,
                                             subject=u'subj', body=u'hi there1',
                                             recipients=[self.u3])
         Session.commit()
         notification2 = NotificationModel().create(created_by=self.u1,
                                             subject=u'subj', body=u'hi there2',
                                             recipients=[self.u3])
         Session.commit()
         u3 = Session.query(User).get(self.u3)
         self.assertEqual(sorted([x.notification for x in u3.notifications]),
                          sorted([notification2, notification1]))
     def test_delete_notifications(self):
         self.assertEqual([], Notification.query().all())
         self.assertEqual([], UserNotification.query().all())
         notification = NotificationModel().create(created_by=self.u1,
                                            subject=u'title', body=u'hi there3',
                                     recipients=[self.u3, self.u1, self.u2])
         Session.commit()
         notifications = Notification.query().all()
         self.assertTrue(notification in notifications)
         Notification.delete(notification.notification_id)
         Session.commit()
         notifications = Notification.query().all()
         self.assertFalse(notification in notifications)
         un = UserNotification.query().filter(UserNotification.notification
                                              == notification).all()
         self.assertEqual(un, [])
     def test_delete_association(self):
         self.assertEqual([], Notification.query().all())
         self.assertEqual([], UserNotification.query().all())
         notification = NotificationModel().create(created_by=self.u1,
                                            subject=u'title', body=u'hi there3',
                                     recipients=[self.u3, self.u1, self.u2])
         Session.commit()
         unotification = UserNotification.query()\
                             .filter(UserNotification.notification ==
                                     notification)\
                             .filter(UserNotification.user_id == self.u3)\
                             .scalar()
         self.assertEqual(unotification.user_id, self.u3)
         NotificationModel().delete(self.u3,
                                    notification.notification_id)
         Session.commit()
         u3notification = UserNotification.query()\
                             .filter(UserNotification.notification ==
                                     notification)\
                             .filter(UserNotification.user_id == self.u3)\
                             .scalar()
         self.assertEqual(u3notification, None)
         # notification object is still there
         self.assertEqual(Notification.query().all(), [notification])
         #u1 and u2 still have assignments
         u1notification = UserNotification.query()\
                             .filter(UserNotification.notification ==
                                     notification)\
                             .filter(UserNotification.user_id == self.u1)\
                             .scalar()
         self.assertNotEqual(u1notification, None)
         u2notification = UserNotification.query()\
                             .filter(UserNotification.notification ==
                                     notification)\
                             .filter(UserNotification.user_id == self.u2)\
                             .scalar()
         self.assertNotEqual(u2notification, None)
     def test_notification_counter(self):
         self._clean_notifications()
         self.assertEqual([], Notification.query().all())
         self.assertEqual([], UserNotification.query().all())
         NotificationModel().create(created_by=self.u1,
                             subject=u'title', body=u'hi there_delete',
                             recipients=[self.u3, self.u1])
         Session.commit()
         self.assertEqual(NotificationModel()
                          .get_unread_cnt_for_user(self.u1), 1)
         self.assertEqual(NotificationModel()
                          .get_unread_cnt_for_user(self.u2), 0)
         self.assertEqual(NotificationModel()
                          .get_unread_cnt_for_user(self.u3), 1)
         notification = NotificationModel().create(created_by=self.u1,
                                            subject=u'title', body=u'hi there3',
                                     recipients=[self.u3, self.u1, self.u2])
         Session.commit()
         self.assertEqual(NotificationModel()
                          .get_unread_cnt_for_user(self.u1), 2)
         self.assertEqual(NotificationModel()
                          .get_unread_cnt_for_user(self.u2), 1)
         self.assertEqual(NotificationModel()
                          .get_unread_cnt_for_user(self.u3), 2)
 class TestUsers(unittest.TestCase):
     def __init__(self, methodName='runTest'):
         super(TestUsers, self).__init__(methodName=methodName)
     def setUp(self):
         self.u1 = UserModel().create_or_update(username=u'u1',
                                         password=u'qweqwe',
                                         email=u'u1@rhodecode.org',
-                                        name=u'u1', lastname=u'u1')
+                                        firstname=u'u1', lastname=u'u1')
     def tearDown(self):
         perm = Permission.query().all()
         for p in perm:
             UserModel().revoke_perm(self.u1, p)
         UserModel().delete(self.u1)
         Session.commit()
     def test_add_perm(self):
         perm = Permission.query().all()[0]
         UserModel().grant_perm(self.u1, perm)
         Session.commit()
         self.assertEqual(UserModel().has_perm(self.u1, perm), True)
     def test_has_perm(self):
         perm = Permission.query().all()
         for p in perm:
             has_p = UserModel().has_perm(self.u1, p)
             self.assertEqual(False, has_p)
     def test_revoke_perm(self):
         perm = Permission.query().all()[0]
         UserModel().grant_perm(self.u1, perm)
         Session.commit()
         self.assertEqual(UserModel().has_perm(self.u1, perm), True)
         #revoke
         UserModel().revoke_perm(self.u1, perm)
         Session.commit()
         self.assertEqual(UserModel().has_perm(self.u1, perm), False)
 class TestPermissions(unittest.TestCase):
     def __init__(self, methodName='runTest'):
         super(TestPermissions, self).__init__(methodName=methodName)
     def setUp(self):
         self.u1 = UserModel().create_or_update(
             username=u'u1', password=u'qweqwe',
-            email=u'u1@rhodecode.org', name=u'u1', lastname=u'u1'
+            email=u'u1@rhodecode.org', firstname=u'u1', lastname=u'u1'
+        )
         self.u2 = UserModel().create_or_update(
             username=u'u2', password=u'qweqwe',
-            email=u'u2@rhodecode.org', name=u'u2', lastname=u'u2'
+            email=u'u2@rhodecode.org', firstname=u'u2', lastname=u'u2'
+        )
         self.anon = User.get_by_username('default')
         self.a1 = UserModel().create_or_update(
             username=u'a1', password=u'qweqwe',
-            email=u'a1@rhodecode.org', name=u'a1', lastname=u'a1', admin=True
+            email=u'a1@rhodecode.org', firstname=u'a1', lastname=u'a1', admin=True
+        )
         Session.commit()
     def tearDown(self):
         if hasattr(self, 'test_repo'):
             RepoModel().delete(repo=self.test_repo)
         UserModel().delete(self.u1)
         UserModel().delete(self.u2)
         UserModel().delete(self.a1)
         if hasattr(self, 'g1'):
             ReposGroupModel().delete(self.g1.group_id)
         if hasattr(self, 'g2'):
             ReposGroupModel().delete(self.g2.group_id)
         if hasattr(self, 'ug1'):
             UsersGroupModel().delete(self.ug1, force=True)
         Session.commit()
     def test_default_perms_set(self):
         u1_auth = AuthUser(user_id=self.u1.user_id)
         perms = {
             'repositories_groups': {},
             'global': set([u'hg.create.repository', u'repository.read',
                            u'hg.register.manual_activate']),
             'repositories': {u'vcs_test_hg': u'repository.read'}
+        }
         self.assertEqual(u1_auth.permissions['repositories'][HG_REPO],
                          perms['repositories'][HG_REPO])
         new_perm = 'repository.write'
         RepoModel().grant_user_permission(repo=HG_REPO, user=self.u1, perm=new_perm)
         Session.commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         self.assertEqual(u1_auth.permissions['repositories'][HG_REPO], new_perm)
     def test_default_admin_perms_set(self):
         a1_auth = AuthUser(user_id=self.a1.user_id)
         perms = {
             'repositories_groups': {},
             'global': set([u'hg.admin']),
             'repositories': {u'vcs_test_hg': u'repository.admin'}
+        }
         self.assertEqual(a1_auth.permissions['repositories'][HG_REPO],
                          perms['repositories'][HG_REPO])
         new_perm = 'repository.write'
         RepoModel().grant_user_permission(repo=HG_REPO, user=self.a1, perm=new_perm)
         Session.commit()
         # cannot really downgrade admins permissions !? they still get's set as
         # admin !
         u1_auth = AuthUser(user_id=self.a1.user_id)
         self.assertEqual(u1_auth.permissions['repositories'][HG_REPO],
                          perms['repositories'][HG_REPO])
     def test_default_group_perms(self):
         self.g1 = _make_group('test1', skip_if_exists=True)
         self.g2 = _make_group('test2', skip_if_exists=True)
         u1_auth = AuthUser(user_id=self.u1.user_id)
         perms = {
             'repositories_groups': {u'test1': 'group.read', u'test2': 'group.read'},
             'global': set([u'hg.create.repository', u'repository.read', u'hg.register.manual_activate']),
             'repositories': {u'vcs_test_hg': u'repository.read'}
+        }
         self.assertEqual(u1_auth.permissions['repositories'][HG_REPO],
                          perms['repositories'][HG_REPO])
         self.assertEqual(u1_auth.permissions['repositories_groups'],
                          perms['repositories_groups'])
     def test_default_admin_group_perms(self):
         self.g1 = _make_group('test1', skip_if_exists=True)
         self.g2 = _make_group('test2', skip_if_exists=True)
         a1_auth = AuthUser(user_id=self.a1.user_id)
         perms = {
             'repositories_groups': {u'test1': 'group.admin', u'test2': 'group.admin'},
             'global': set(['hg.admin']),
             'repositories': {u'vcs_test_hg': 'repository.admin'}
+        }
         self.assertEqual(a1_auth.permissions['repositories'][HG_REPO],
                          perms['repositories'][HG_REPO])
         self.assertEqual(a1_auth.permissions['repositories_groups'],
                          perms['repositories_groups'])
     def test_propagated_permission_from_users_group(self):
         # make group
         self.ug1 = UsersGroupModel().create('G1')
         # add user to group
         UsersGroupModel().add_user_to_group(self.ug1, self.u1)
         # set permission to lower
         new_perm = 'repository.none'
         RepoModel().grant_user_permission(repo=HG_REPO, user=self.u1, perm=new_perm)
         Session.commit()
         u1_auth = AuthUser(user_id=self.u1.user_id)
         self.assertEqual(u1_auth.permissions['repositories'][HG_REPO],
                          new_perm)

0 comments (0 inline, 0 general)