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,

                                     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',

    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)\

    '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

        """

        Creates a new instance if not found, or updates current one

        :param username:

        :param password:

        :param email:

        :param active:

        :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()

        else:

            log.debug('updating user %s' % username)

            new_user = user

        try:

            new_user.username = username

            new_user.admin = admin

            new_user.email = email

            new_user.active = active

            new_user.ldap_dn = safe_unicode(ldap_dn) if ldap_dn else None

            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.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',

        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',

        u2 = UserModel().create_or_update(username='u2', password='qweqwe',

                                               email='u2@rhodecode.org',

        # 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

        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_show(self):

        self.log_user()

        cur_user = self._get_logged_user()

        u1 = UserModel().create_or_update(username='u1', password='qweqwe',

                                               email='u1@rhodecode.org',

        u2 = UserModel().create_or_update(username='u2', password='qweqwe',

                                               email='u2@rhodecode.org',

        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))

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,

        )

        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

                         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',

        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',

        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',

        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',

        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])

                            .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',

    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',

        )

        self.u2 = UserModel().create_or_update(

            username=u'u2', password=u'qweqwe',

        )

        self.anon = User.get_by_username('default')

        self.a1 = UserModel().create_or_update(

            username=u'a1', password=u'qweqwe',

        )

        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()