gr = fixture.create_user_group(name, cur_user=TEST_USER_ADMIN_LOGIN)

    UserGroupModel().add_user_to_group(user_group=gr,

                                       user=TEST_USER_ADMIN_LOGIN)

    Session().commit()

    return gr

def make_repo_group(name=TEST_REPO_GROUP):

    gr = fixture.create_repo_group(name, cur_user=TEST_USER_ADMIN_LOGIN)

    Session().commit()

    return gr

class BaseTestApi(object):

    REPO = None

    REPO_TYPE = None

    @classmethod

    def setup_class(cls):

        cls.usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)

        cls.apikey = cls.usr.api_key

        cls.test_user = UserModel().create_or_update(

            username='test-api',

            password='test',

            firstname='first',

            lastname='last'

        )

        Session().commit()

        cls.TEST_USER_LOGIN = cls.test_user.username

        cls.apikey_regular = cls.test_user.api_key

    @classmethod

    def teardown_class(cls):

        pass

    def setUp(self):

        self.maxDiff = None

        make_user_group()

        make_repo_group()

    def tearDown(self):

        fixture.destroy_user_group(TEST_USER_GROUP)

        fixture.destroy_gists()

        fixture.destroy_repo_group(TEST_REPO_GROUP)

    def _compare_ok(self, id_, expected, given):

        expected = jsonify({

            'id': id_,

        )

        try:

            expected = ret

            self._compare_ok(id_, expected, given=response.body)

        finally:

            fixture.destroy_user(usr.user_id)

    @mock.patch.object(UserModel, 'create_or_update', crash)

    def test_api_create_user_when_exception_happened(self):

        username = 'test_new_api_user'

        email = username + "@foo.com"

        id_, params = _build_data(self.apikey, 'create_user',

                                  username=username,

                                  email=email,

                                  password='trololo')

        response = api_call(self, params)

        expected = 'failed to create user `%s`' % username

        self._compare_error(id_, expected, given=response.body)

    def test_api_delete_user(self):

        usr = UserModel().create_or_update(username=u'test_user',

                                           password=u'qweqwe',

                                           firstname=u'u1', lastname=u'u1')

        Session().commit()

        username = usr.username

        email = usr.email

        usr_id = usr.user_id

        ## DELETE THIS USER NOW

        id_, params = _build_data(self.apikey, 'delete_user',

                                  userid=username, )

        response = api_call(self, params)

        ret = {'msg': 'deleted user ID:%s %s' % (usr_id, username),

               'user': None}

        expected = ret

        self._compare_ok(id_, expected, given=response.body)

    @mock.patch.object(UserModel, 'delete', crash)

    def test_api_delete_user_when_exception_happened(self):

        usr = UserModel().create_or_update(username=u'test_user',

                                           password=u'qweqwe',

                                           firstname=u'u1', lastname=u'u1')

        Session().commit()

        username = usr.username

        id_, params = _build_data(self.apikey, 'delete_user',

                                  userid=username, )

        response = api_call(self, params)

        ret = 'failed to delete user ID:%s %s' % (usr.user_id,

                                                  usr.username)

        expected = ret

        self._compare_error(id_, expected, given=response.body)

    @parameterized.expand([('firstname', 'new_username'),

                           ('lastname', 'new_username'),

                           ('email', 'new_username'),

                           ('admin', True),

                           ('admin', False),

                           ('extern_type', 'ldap'),

                           ('extern_type', None),

                           ('extern_name', 'test'),

                           ('extern_name', None),

                           ('active', False),

                           ('active', True),

                           ('password', 'newpass')

        # fix the repo name if passed as repo_name_full

        if defs['repo_name']:

            defs['repo_name'] = defs['repo_name'].split('/')[-1]

        return defs

    def _get_group_create_params(self, **custom):

        defs = dict(

            group_name=None,

            group_description='DESC',

            group_parent_id=None,

            perms_updates=[],

            perms_new=[],

            enable_locking=False,

            recursive=False

        )

        defs.update(custom)

        return defs

    def _get_user_create_params(self, name, **custom):

        defs = dict(

            username=name,

            password='qweqwe',

            firstname='TestUser',

            lastname='Test',

            active=True,

            admin=False,

            extern_type='rhodecode',

            extern_name=None

        )

        defs.update(custom)

        return defs

    def _get_user_group_create_params(self, name, **custom):

        defs = dict(

            users_group_name=name,

            user_group_description='DESC',

            users_group_active=True,

            user_group_data={},

        )

        defs.update(custom)

        return defs

    def create_repo(self, name, **kwargs):

        if 'skip_if_exists' in kwargs:

from rhodecode.tests import *

from rhodecode.model.db import Notification, User

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

                                          firstname='u1', lastname='u1')

        u1 = u1.user_id

        response = self.app.get(url('notifications'))

        response.mustcontain('<div class="table">No notifications here yet</div>')

        cur_user = self._get_logged_user()

        notif = 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'))

        response.mustcontain('id="notification_%s"' % notif.notification_id)

    def test_delete(self):

        self.log_user()

        cur_user = self._get_logged_user()

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

                                               firstname='u1', lastname='u1')

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

                                               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]

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

        self.assertEqual(response.body, 'ok')

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

                                          firstname='u1', lastname='u1')

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

                                          firstname='u2', lastname='u2')

        subject = u'test'

        notif_body = u'hi there'

        notification = NotificationModel().create(created_by=cur_user,

                                                  subject=subject,

                                                  body=notif_body,

                                                  recipients=[cur_user, u1, u2])

        response = self.app.get(url('notification',

                                    notification_id=notification.notification_id))

        response.mustcontain(subject)

        response.mustcontain(notif_body)

        self.assertRaises(NoResultFound, get_user), 'found user in database'

    def test_new(self):

        self.log_user()

        response = self.app.get(url('new_user'))

    @parameterized.expand(

        [('firstname', {'firstname': 'new_username'}),

         ('lastname', {'lastname': 'new_username'}),

         ('admin', {'admin': True}),

         ('admin', {'admin': False}),

         ('extern_type', {'extern_type': 'ldap'}),

         ('extern_type', {'extern_type': None}),

         ('extern_name', {'extern_name': 'test'}),

         ('extern_name', {'extern_name': None}),

         ('active', {'active': False}),

         ('active', {'active': True}),

         ('email', {'email': 'some@email.com'}),

        # ('new_password', {'new_password': 'foobar123',

        #                   'password_confirmation': 'foobar123'})

        ])

    def test_update(self, name, attrs):

        self.log_user()

        usr = fixture.create_user(self.test_user_1, password='qweqwe',

                                  extern_type='rhodecode',

                                  extern_name=self.test_user_1,

                                  skip_if_exists=True)

        Session().commit()

        params = usr.get_api_data()

        params.update({'password_confirmation': ''})

        params.update({'new_password': ''})

        params.update(attrs)

        if name == 'email':

            params['emails'] = [attrs['email']]

        if name == 'extern_type':

            #cannot update this via form, expected value is original one

            params['extern_type'] = "rhodecode"

        if name == 'extern_name':

            #cannot update this via form, expected value is original one

            params['extern_name'] = self.test_user_1

            # special case since this user is not

                                          # logged in yet his data is not filled

                                          # so we use creation data

        response = self.app.put(url('user', id=usr.user_id), params)

        self.checkSessionFlash(response, 'User updated successfully')

        updated_user = User.get_by_username(self.test_user_1)

from rhodecode.tests import *

class _BaseTest(TestController):

    """

    Write all tests here

    """

    REPO = None

    REPO_TYPE = None

    NEW_REPO = None

    REPO_FORK = None

    @classmethod

    def setup_class(cls):

        pass

    @classmethod

    def teardown_class(cls):

        pass

    def setUp(self):

        self.username = u'forkuser'

        self.password = u'qweqwe'

        self.u1 = fixture.create_user(self.username, password=self.password,

        Session().commit()

    def tearDown(self):

        Session().delete(self.u1)

        Session().commit()

    def test_index(self):

        self.log_user()

        repo_name = self.REPO

        response = self.app.get(url(controller='forks', action='forks',

                                    repo_name=repo_name))

        response.mustcontain("""There are no forks yet""")

    def test_no_permissions_to_fork(self):

        usr = self.log_user(TEST_USER_REGULAR_LOGIN,

                            TEST_USER_REGULAR_PASS)['user_id']

        user_model = UserModel()

        user_model.revoke_perm(usr, 'hg.fork.repository')

        user_model.grant_perm(usr, 'hg.fork.none')

        u = UserModel().get(usr)

        u.inherit_default_permissions = False

        Session().commit()

        # try create a fork

        response = self.app.post(url(controller='login', action='index'),

                                 {'username': 'test_regular',

                                  'password': 'test12'})

        self.assertEqual(response.status, '302 Found')

        self.assertEqual(response.session['rhodecode_user'].get('username'),

                         'test_regular')

        response = response.follow()

        response.mustcontain('/%s' % HG_REPO)

    def test_login_ok_came_from(self):

        test_came_from = '/_admin/users'

        response = self.app.post(url(controller='login', action='index',

                                     came_from=test_came_from),

                                 {'username': 'test_admin',

                                  'password': 'test12'})

        self.assertEqual(response.status, '302 Found')

        response = response.follow()

        self.assertEqual(response.status, '200 OK')

        response.mustcontain('Users administration')

    @parameterized.expand([

          ('data:text/html,<script>window.alert("xss")</script>',),

          ('file:///etc/passwd',),

          ('ftp://some.ftp.server',),

          ('http://other.domain',),

    ])

    def test_login_bad_came_froms(self, url_came_from):

        response = self.app.post(url(controller='login', action='index',

                                     came_from=url_came_from),

                                 {'username': 'test_admin',

                                  'password': 'test12'})

        self.assertEqual(response.status, '302 Found')

        self.assertEqual(response._environ['paste.testing_variables']

                         ['tmpl_context'].came_from, '/')

        response = response.follow()

        self.assertEqual(response.status, '200 OK')

    def test_login_short_password(self):

        response = self.app.post(url(controller='login', action='index'),

                                 {'username': 'test_admin',

                                  'password': 'as'})

        self.assertEqual(response.status, '200 OK')

        response.mustcontain('Enter 3 characters or more')

        response = self.app.post(url('my_account_emails'),

                                 {'del_email_id': email_id, '_method': 'delete'})

        self.checkSessionFlash(response, 'Removed email from user')

        response = self.app.get(url('my_account_emails'))

        response.mustcontain('No additional emails specified')

    @parameterized.expand(

        [('firstname', {'firstname': 'new_username'}),

         ('lastname', {'lastname': 'new_username'}),

         ('admin', {'admin': True}),

         ('admin', {'admin': False}),

         ('extern_type', {'extern_type': 'ldap'}),

         ('extern_type', {'extern_type': None}),

         #('extern_name', {'extern_name': 'test'}),

         #('extern_name', {'extern_name': None}),

         ('active', {'active': False}),

         ('active', {'active': True}),

         ('email', {'email': 'some@email.com'}),

        # ('new_password', {'new_password': 'foobar123',

        #                   'password_confirmation': 'foobar123'})

        ])

    def test_my_account_update(self, name, attrs):

        usr = fixture.create_user(self.test_user_1, password='qweqwe',

                                  extern_type='rhodecode',

                                  extern_name=self.test_user_1,

                                  skip_if_exists=True)

        params = usr.get_api_data()  # current user data

        user_id = usr.user_id

        self.log_user(username=self.test_user_1, password='qweqwe')

        params.update({'password_confirmation': ''})

        params.update({'new_password': ''})

        params.update({'extern_type': 'rhodecode'})

        params.update({'extern_name': self.test_user_1})

        params.update(attrs)

        response = self.app.post(url('my_account'), params)

        self.checkSessionFlash(response,

                               'Your account was updated successfully')

        updated_user = User.get_by_username(self.test_user_1)

        updated_params = updated_user.get_api_data()

        updated_params.update({'password_confirmation': ''})

        updated_params.update({'new_password': ''})

        params['last_login'] = updated_params['last_login']

    structure

     [g0] - group `g0` with 3 subgroups

     |

     |__[g0_1] group g0_1 with 2 groups 0 repos

     |  |

     |  |__[g0_1_1] group g0_1_1 with 1 group 2 repos

     |  |   |__<g0/g0_1/g0_1_1/g0_1_1_r1>

     |  |   |__<g0/g0_1/g0_1_1/g0_1_1_r2>

     |  |__<g0/g0_1/g0_1_r1>

     |

     |__[g0_2] 2 repos

     |  |

     |  |__<g0/g0_2/g0_2_r1>

     |  |__<g0/g0_2/g0_2_r2>

     |

     |__[g0_3] 1 repo

        |

        |_<g0/g0_3/g0_3_r1>

        |_<g0/g0_3/g0_3_r2_private>

    """

    test_u1 = UserModel().create_or_update(

        username=u'test_u1', password=u'qweqwe',

    )

    g0 = fixture.create_repo_group('g0')

    g0_1 = fixture.create_repo_group('g0_1', group_parent_id=g0)

    g0_1_1 = fixture.create_repo_group('g0_1_1', group_parent_id=g0_1)

    g0_1_1_r1 = fixture.create_repo('g0/g0_1/g0_1_1/g0_1_1_r1', repo_group=g0_1_1)

    g0_1_1_r2 = fixture.create_repo('g0/g0_1/g0_1_1/g0_1_1_r2', repo_group=g0_1_1)

    g0_1_r1 = fixture.create_repo('g0/g0_1/g0_1_r1', repo_group=g0_1)

    g0_2 = fixture.create_repo_group('g0_2', group_parent_id=g0)

    g0_2_r1 = fixture.create_repo('g0/g0_2/g0_2_r1', repo_group=g0_2)

    g0_2_r2 = fixture.create_repo('g0/g0_2/g0_2_r2', repo_group=g0_2)

    g0_3 = fixture.create_repo_group('g0_3', group_parent_id=g0)

    g0_3_r1 = fixture.create_repo('g0/g0_3/g0_3_r1', repo_group=g0_3)

    g0_3_r2_private = fixture.create_repo('g0/g0_3/g0_3_r1_private',

                                          repo_group=g0_3, repo_private=True)

    return test_u1

def expected_count(group_name, objects=False):

    repo_group = RepoGroup.get_by_group_name(group_name=group_name)

    objs = repo_group.recursive_groups_and_repos()

    if objects:

        return objs

    return len(objs)

from rhodecode.tests import *

from rhodecode.model.db import User, Notification, UserNotification

from rhodecode.model.user import UserModel

from rhodecode.model.meta import Session

from rhodecode.model.notification import NotificationModel

class TestNotifications(BaseTestCase):

    def __init__(self, methodName='runTest'):

        Session.remove()

        self.u1 = UserModel().create_or_update(username=u'u1',

                                        password=u'qweqwe',

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

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

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

from rhodecode.model.meta import Session

from rhodecode.model.user_group import UserGroupModel

from rhodecode.lib.auth import AuthUser

from rhodecode.model.permission import PermissionModel

fixture = Fixture()

class TestPermissions(BaseTestCase):

    def __init__(self, methodName='runTest'):

        super(TestPermissions, self).__init__(methodName=methodName)

    @classmethod

    def setUpClass(cls):

        #recreate default user to get a clean start

        PermissionModel().create_default_permissions(user=User.DEFAULT_USER,

                                                     force=True)

        Session().commit()

    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.u3 = UserModel().create_or_update(

            username=u'u3', password=u'qweqwe',

        )

        self.anon = User.get_default_user()

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

        UserModel().delete(self.a1)

        if hasattr(self, 'g1'):

            RepoGroupModel().delete(self.g1.group_id)

        if hasattr(self, 'g2'):

            RepoGroupModel().delete(self.g2.group_id)

        if hasattr(self, 'ug1'):

            UserGroupModel().delete(self.ug1, force=True)

        Session().commit()

    def test_default_perms_set(self):

        u1_auth = AuthUser(user_id=self.u1.user_id)

        perms = {

from rhodecode.tests import *

from rhodecode.model.db import User, UserGroup, UserGroupMember, UserEmailMap,\

    Permission

from rhodecode.model.user import UserModel

from rhodecode.model.meta import Session

from rhodecode.model.user_group import UserGroupModel

from rhodecode.tests.fixture import Fixture

fixture = Fixture()

class TestUser(BaseTestCase):

    def __init__(self, methodName='runTest'):

        Session.remove()

        super(TestUser, self).__init__(methodName=methodName)

    def tearDown(self):

        Session.remove()

    def test_create_and_remove(self):

        usr = UserModel().create_or_update(username=u'test_user',

                                           password=u'qweqwe',

                                           firstname=u'u1', lastname=u'u1')

        Session().commit()

        self.assertEqual(User.get_by_username(u'test_user'), usr)

        # make user group

        user_group = fixture.create_user_group('some_example_group')

        Session().commit()

        UserGroupModel().add_user_to_group(user_group, usr)

        Session().commit()

        self.assertEqual(UserGroup.get(user_group.users_group_id), user_group)

        self.assertEqual(UserGroupMember.query().count(), 1)

        UserModel().delete(usr.user_id)

        Session().commit()

        self.assertEqual(UserGroupMember.query().all(), [])

    def test_additonal_email_as_main(self):

        usr = UserModel().create_or_update(username=u'test_user',

                                           password=u'qweqwe',

                                     firstname=u'u1', lastname=u'u1')

        Session().commit()

        def do():

            m = UserEmailMap()

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

                                     firstname=u'u1', lastname=u'u1')

        Session().commit()

        m = UserEmailMap()

        m.user = usr

        Session().add(m)

        Session().commit()

        self.assertEqual(usr.user_id, u.user_id)

        self.assertEqual(usr.username, u.username)

        self.assertEqual(usr.user_id, u.user_id)

        self.assertEqual(usr.username, u.username)

        self.assertEqual(None, u)

        UserModel().delete(usr.user_id)

        Session().commit()

class TestUsers(BaseTestCase):

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

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

        Session.remove()

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

#!/bin/sh

psql -U postgres -h localhost -c 'drop database if exists rhodecode;'

psql -U postgres -h localhost -c 'create database rhodecode;'

paster setup-rhodecode rc.ini --force-yes --user=marcink --password=qweqwe --email=marcin@python-blog.com --repos=/home/marcink/repos --no-public-access

API_KEY=`psql -R " " -A -U postgres -h localhost -c "select api_key from users where admin=TRUE" -d rhodecode | awk '{print $2}'`

echo "run those after running server"

paster serve rc.ini --pid-file=rc.pid --daemon

sleep 3

rhodecode-api --apikey=$API_KEY --apihost=http://127.0.0.1:5001 create_user_group group_name:demo12

rhodecode-api --apikey=$API_KEY --apihost=http://127.0.0.1:5001 add_user_to_user_group usergroupid:demo12 userid:demo1

rhodecode-api --apikey=$API_KEY --apihost=http://127.0.0.1:5001 add_user_to_user_group usergroupid:demo12 userid:demo2

echo "killing server"

kill `cat rc.pid`

rm rc.pid