Changeset - 388843a3a3c0
[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:
0 comments (0 inline, 0 general)
rhodecode/lib/db_manage.py
Show inline comments
 
@@ -456,34 +456,34 @@ class DbManage(object):
 
        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'),
rhodecode/model/user.py
Show inline comments
 
@@ -92,60 +92,64 @@ class UserModel(BaseModel):
 
            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:
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'))
 
@@ -49,71 +50,57 @@ class TestNotificationsController(TestCo
 
#    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
 
@@ -3,25 +3,25 @@ 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))
rhodecode/tests/test_models.py
Show inline comments
 
@@ -176,65 +176,65 @@ class TestReposGroups(unittest.TestCase)
 
        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)
 

	
 
@@ -246,25 +246,25 @@ class TestUser(unittest.TestCase):
 

	
 
        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',
 
@@ -424,25 +424,25 @@ class TestNotifications(unittest.TestCas
 
                         .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)
 
@@ -465,34 +465,34 @@ class TestUsers(unittest.TestCase):
 
        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'):
0 comments (0 inline, 0 general)