"""Pylons application test package

This package assumes the Pylons environment is already loaded, such as

when this script is imported from the `nosetests --with-pylons=test.ini`

command.

This module initializes the application via ``websetup`` (`paster

setup-app`) and provides the base testing objects.

nosetests -x - fail on first error

nosetests rhodecode.tests.functional.test_admin_settings:TestSettingsController.test_my_account

nosetests --pdb --pdb-failures

nosetests --with-coverage --cover-package=rhodecode.model.validators rhodecode.tests.test_validators

optional FLAGS:

    RC_WHOOSH_TEST_DISABLE=1 - skip whoosh index building and tests

    RC_NO_TMP_PATH=1 - disable new temp path for tests, used mostly for test_vcs_operations

"""

import os

import time

import logging

import datetime

import hashlib

import tempfile

from os.path import join as jn

from unittest import TestCase

from tempfile import _RandomNameSequence

from paste.deploy import loadapp

from paste.script.appinstall import SetupCommand

import pylons

import pylons.test

from pylons import config, url

from pylons.i18n.translation import _get_translator

from routes.util import URLGenerator

from webtest import TestApp

from nose.plugins.skip import SkipTest

from rhodecode import is_windows

from rhodecode.model.meta import Session

from rhodecode.model.db import User

from rhodecode.tests.nose_parametrized import parameterized

from rhodecode.lib.utils2 import safe_unicode, safe_str

os.environ['TZ'] = 'UTC'

if not is_windows:

    time.tzset()

log = logging.getLogger(__name__)

__all__ = [

    'parameterized', 'environ', 'url', 'get_new_dir', 'TestController',

    'TESTS_TMP_PATH', 'HG_REPO', 'GIT_REPO', 'NEW_HG_REPO', 'NEW_GIT_REPO',

    'HG_FORK', 'GIT_FORK', 'TEST_USER_ADMIN_LOGIN', 'TEST_USER_ADMIN_PASS',

    'TEST_USER_REGULAR_LOGIN', 'TEST_USER_REGULAR_PASS',

    'TEST_USER_REGULAR_EMAIL', 'TEST_USER_REGULAR2_LOGIN',

    'TEST_USER_REGULAR2_PASS', 'TEST_USER_REGULAR2_EMAIL', 'TEST_HG_REPO',

    'TEST_HG_REPO_CLONE', 'TEST_HG_REPO_PULL', 'TEST_GIT_REPO',

    'TEST_GIT_REPO_CLONE', 'TEST_GIT_REPO_PULL', 'HG_REMOTE_REPO',

    'GIT_REMOTE_REPO', 'SCM_TESTS',

]

# Invoke websetup with the current config file

# SetupCommand('setup-app').run([config_file])

environ = {}

#SOME GLOBALS FOR TESTS

TESTS_TMP_PATH = jn('/', 'tmp', 'rc_test_%s' % _RandomNameSequence().next())

TEST_USER_ADMIN_LOGIN = 'test_admin'

TEST_USER_ADMIN_PASS = 'test12'

TEST_USER_ADMIN_EMAIL = 'test_admin@mail.com'

TEST_USER_REGULAR_LOGIN = 'test_regular'

TEST_USER_REGULAR_PASS = 'test12'

TEST_USER_REGULAR_EMAIL = 'test_regular@mail.com'

TEST_USER_REGULAR2_LOGIN = 'test_regular2'

TEST_USER_REGULAR2_PASS = 'test12'

TEST_USER_REGULAR2_EMAIL = 'test_regular2@mail.com'

HG_REPO = 'vcs_test_hg'

GIT_REPO = 'vcs_test_git'

NEW_HG_REPO = 'vcs_test_hg_new'

NEW_GIT_REPO = 'vcs_test_git_new'

HG_FORK = 'vcs_test_hg_fork'

GIT_FORK = 'vcs_test_git_fork'

## VCS

SCM_TESTS = ['hg', 'git']

uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))

GIT_REMOTE_REPO = 'git://github.com/codeinn/vcs.git'

TEST_GIT_REPO = jn(TESTS_TMP_PATH, GIT_REPO)

TEST_GIT_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)

TEST_GIT_REPO_PULL = jn(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)

HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'

TEST_HG_REPO = jn(TESTS_TMP_PATH, HG_REPO)

TEST_HG_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)

TEST_HG_REPO_PULL = jn(TESTS_TMP_PATH, 'vcshgpull%s' % uniq_suffix)

TEST_DIR = tempfile.gettempdir()

TEST_REPO_PREFIX = 'vcs-test'

# cached repos if any !

# comment out to get some other repos from bb or github

GIT_REMOTE_REPO = jn(TESTS_TMP_PATH, GIT_REPO)

HG_REMOTE_REPO = jn(TESTS_TMP_PATH, HG_REPO)

#skip ldap tests if LDAP lib is not installed

ldap_lib_installed = False

try:

    import ldap

    ldap_lib_installed = True

except ImportError:

    # means that python-ldap is not installed

    pass

def get_new_dir(title):

    """

    Returns always new directory path.

    """

    from rhodecode.tests.vcs.utils import get_normalized_path

    name = TEST_REPO_PREFIX

    if title:

        name = '-'.join((name, title))

    hex = hashlib.sha1(str(time.time())).hexdigest()

    name = '-'.join((name, hex))

    path = os.path.join(TEST_DIR, name)

    return get_normalized_path(path)

    def __init__(self, *args, **kwargs):

        self.index_location = config['app_conf']['index_dir']

    def log_user(self, username=TEST_USER_ADMIN_LOGIN,

                 password=TEST_USER_ADMIN_PASS):

        self._logged_username = username

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

                                 {'username': username,

                                  'password': password})

        if 'invalid user name' in response.body:

            self.fail('could not login using %s %s' % (username, password))

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

        ses = response.session['rhodecode_user']

        self.assertEqual(ses.get('username'), username)

        response = response.follow()

        self.assertEqual(ses.get('is_authenticated'), True)

        return response.session['rhodecode_user']

    def _get_logged_user(self):

        return User.get_by_username(self._logged_username)

    def checkSessionFlash(self, response, msg):

        self.assertTrue('flash' in response.session,

                        msg='Response session:%r have no flash'

                        % response.session)

        if not msg in response.session['flash'][0][1]:

            msg = u'msg `%s` not found in session flash: got `%s` instead' % (

                      msg, response.session['flash'][0][1])

            self.fail(safe_str(msg))

from rhodecode.tests import *

from rhodecode.tests.fixture import Fixture

from rhodecode.model.repos_group import ReposGroupModel

from rhodecode.model.repo import RepoModel

from rhodecode.model.db import RepoGroup, Repository, User

from rhodecode.model.user import UserModel

from rhodecode.lib.auth import AuthUser

from rhodecode.model.meta import Session

fixture = Fixture()

def _destroy_project_tree(test_u1_id):

    Session.remove()

    repos_group = RepoGroup.get_by_group_name(group_name='g0')

    for el in reversed(repos_group.recursive_groups_and_repos()):

        if isinstance(el, Repository):

            RepoModel().delete(el)

        elif isinstance(el, RepoGroup):

            ReposGroupModel().delete(el, force_delete=True)

    u = User.get(test_u1_id)

    Session().delete(u)

    Session().commit()

def _create_project_tree():

    """

    Creates a tree of groups and repositories to test permissions

    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>

     |

from __future__ import with_statement

import os

from rhodecode.tests import *

from rhodecode.lib.diffs import DiffProcessor, NEW_FILENODE, DEL_FILENODE, \

    MOD_FILENODE, RENAMED_FILENODE, CHMOD_FILENODE, BIN_FILENODE

dn = os.path.dirname

FIXTURES = os.path.join(dn(dn(os.path.abspath(__file__))), 'fixtures')

DIFF_FIXTURES = {

    'hg_diff_add_single_binary_file.diff': [

        ('US Warszawa.jpg', 'A',

         {'added': 0,

          'deleted': 0,

          'binary': True,

          'ops': {NEW_FILENODE: 'new file 100755',

                  BIN_FILENODE: 'binary diff not shown'}}),

    ],

    'hg_diff_mod_single_binary_file.diff': [

        ('US Warszawa.jpg', 'M',

         {'added': 0,

          'deleted': 0,

          'binary': True,

          'ops': {MOD_FILENODE: 'modified file',

                  BIN_FILENODE: 'binary diff not shown'}}),

    ],

    'hg_diff_mod_single_file_and_rename_and_chmod.diff': [

        ('README', 'M',

         {'added': 3,

          'deleted': 0,

          'binary': False,

          'ops': {MOD_FILENODE: 'modified file',

                  RENAMED_FILENODE: 'file renamed from README.rst to README',

                  CHMOD_FILENODE: 'modified file chmod 100755 => 100644'}}),

    ],

    'hg_diff_rename_and_chmod_file.diff': [

        ('README', 'M',

         {'added': 3,

          'deleted': 0,

          'binary': False,

          'ops': {MOD_FILENODE: 'modified file',

                  BIN_FILENODE: 'binary diff not shown'}}),

    ],

    'hg_diff_del_single_binary_file.diff': [

        ('US Warszawa.jpg', 'D',

         {'added': 0,

          'deleted': 0,

          'binary': True,

          'ops': {DEL_FILENODE: 'deleted file',

          'binary': False,

          'ops': {DEL_FILENODE: 'deleted file'}}),

    ],

    'diff_with_diff_data.diff': [

        ('vcs/backends/base.py', 'M',

         {'added': 18,

          'deleted': 2,

          'binary': False,

          'ops': {MOD_FILENODE: 'modified file'}}),

        ('vcs/backends/git/repository.py', 'M',

         {'added': 46,

          'deleted': 15,

          'binary': False,

          'ops': {MOD_FILENODE: 'modified file'}}),

        ('vcs/backends/hg.py', 'M',

         {'added': 22,

          'deleted': 3,

          'binary': False,

          'ops': {MOD_FILENODE: 'modified file'}}),

        ('vcs/tests/test_git.py', 'M',

         {'added': 5,

          'deleted': 5,

          'binary': False,

          'ops': {MOD_FILENODE: 'modified file'}}),

        ('vcs/tests/test_repository.py', 'M',

         {'added': 174,

          'deleted': 2,

          'binary': False,

          'ops': {MOD_FILENODE: 'modified file'}}),

    ],

#     'large_diff.diff': [

#         ('.hgignore', 'A', {'deleted': 0, 'binary': False, 'added': 3, 'ops': {1: 'new file 100644'}}),

#         ('MANIFEST.in', 'A', {'deleted': 0, 'binary': False, 'added': 3, 'ops': {1: 'new file 100644'}}),

#         ('README.txt', 'A', {'deleted': 0, 'binary': False, 'added': 19, 'ops': {1: 'new file 100644'}}),

#         ('development.ini', 'A', {'deleted': 0, 'binary': False, 'added': 116, 'ops': {1: 'new file 100644'}}),

#         ('docs/index.txt', 'A', {'deleted': 0, 'binary': False, 'added': 19, 'ops': {1: 'new file 100644'}}),

#         ('ez_setup.py', 'A', {'deleted': 0, 'binary': False, 'added': 276, 'ops': {1: 'new file 100644'}}),

#         ('hgapp.py', 'A', {'deleted': 0, 'binary': False, 'added': 26, 'ops': {1: 'new file 100644'}}),

#         ('hgwebdir.config', 'A', {'deleted': 0, 'binary': False, 'added': 21, 'ops': {1: 'new file 100644'}}),

#         ('pylons_app.egg-info/PKG-INFO', 'A', {'deleted': 0, 'binary': False, 'added': 10, 'ops': {1: 'new file 100644'}}),

#         ('pylons_app.egg-info/SOURCES.txt', 'A', {'deleted': 0, 'binary': False, 'added': 33, 'ops': {1: 'new file 100644'}}),

#         ('pylons_app.egg-info/dependency_links.txt', 'A', {'deleted': 0, 'binary': False, 'added': 1, 'ops': {1: 'new file 100644'}}),

#         #TODO:

#     ],

}

    @parameterized.expand([(x,) for x in DIFF_FIXTURES])

    def test_diff(self, diff_fixture):

        with open(os.path.join(FIXTURES, diff_fixture)) as f:

            diff = f.read()

        diff_proc = DiffProcessor(diff)

        diff_proc_d = diff_proc.prepare()

        data = [(x['filename'], x['operation'], x['stats']) for x in diff_proc_d]

        expected_data = DIFF_FIXTURES[diff_fixture]

        self.assertListEqual(expected_data, data)

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

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

        Session.remove()

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

                                        password=u'qweqwe',

                                        email=u'u1@rhodecode.org',

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

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

                                        email=u'u3@rhodecode.org',

                                        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)

        Session().commit()

        u1 = User.get(self.u1)

        u2 = User.get(self.u2)

        u3 = User.get(self.u3)

from rhodecode.tests import *

from rhodecode.tests.fixture import Fixture

from rhodecode.model.repos_group import ReposGroupModel

from rhodecode.model.repo import RepoModel

from rhodecode.model.db import RepoGroup, User, UserGroupRepoGroupToPerm,\

    Permission, UserToPerm

from rhodecode.model.user import UserModel

from rhodecode.model.meta import Session

from rhodecode.model.users_group import UserGroupModel

from rhodecode.lib.auth import AuthUser

from rhodecode.model.permission import PermissionModel

fixture = Fixture()

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

        )

        self.u2 = UserModel().create_or_update(

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

            email=u'u2@rhodecode.org', firstname=u'u2', lastname=u'u2'

        )

        self.u3 = UserModel().create_or_update(

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

            email=u'u3@rhodecode.org', firstname=u'u3', lastname=u'u3'

        )

        self.anon = User.get_default_user()

        self.a1 = UserModel().create_or_update(

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

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

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

            UserGroupModel().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'}

from rhodecode.tests import *

from rhodecode.model.meta import Session

from rhodecode.tests.fixture import Fixture

from rhodecode.model.repo import RepoModel

from rhodecode.model.db import Repository

from rhodecode.lib.exceptions import AttachedForksError

fixture = Fixture()

    def setUp(self):

        pass

    def tearDown(self):

        Session.remove()

    def test_remove_repo(self):

        repo = fixture.create_repo(name='test-repo-1')

        Session().commit()

        RepoModel().delete(repo=repo)

        Session().commit()

        self.assertEqual(None, Repository.get_by_repo_name(repo_name='test-repo-1'))

    def test_remove_repo_repo_raises_exc_when_attached_forks(self):

        repo = fixture.create_repo(name='test-repo-1')

        Session().commit()

        fixture.create_fork(repo.repo_name, 'test-repo-fork-1')

        Session().commit()

        self.assertRaises(AttachedForksError, lambda: RepoModel().delete(repo=repo))

    def test_remove_repo_delete_forks(self):

        repo = fixture.create_repo(name='test-repo-1')

        Session().commit()

        fork = fixture.create_fork(repo.repo_name, 'test-repo-fork-1')

        Session().commit()

        #fork of fork

        fixture.create_fork(fork.repo_name, 'test-repo-fork-fork-1')

        Session().commit()

        RepoModel().delete(repo=repo, forks='delete')

        Session().commit()

        self.assertEqual(None, Repository.get_by_repo_name(repo_name='test-repo-1'))

        self.assertEqual(None, Repository.get_by_repo_name(repo_name='test-repo-fork-1'))

        self.assertEqual(None, Repository.get_by_repo_name(repo_name='test-repo-fork-fork-1'))

    def test_remove_repo_detach_forks(self):

        repo = fixture.create_repo(name='test-repo-1')

        Session().commit()

        fork = fixture.create_fork(repo.repo_name, 'test-repo-fork-1')

import os

from sqlalchemy.exc import IntegrityError

from rhodecode.tests import *

from rhodecode.tests.fixture import Fixture

from rhodecode.model.repos_group import ReposGroupModel

from rhodecode.model.repo import RepoModel

from rhodecode.model.db import RepoGroup

from rhodecode.model.meta import Session

fixture = Fixture()

def _update_group(id_, group_name, desc='desc', parent_id=None):

    form_data = fixture._get_group_create_params(group_name=group_name,

                                                 group_desc=desc,

                                                 group_parent_id=parent_id)

    gr = ReposGroupModel().update(id_, form_data)

    return gr

def _update_repo(name, **kwargs):

    form_data = fixture._get_repo_create_params(**kwargs)

    if not 'repo_name' in kwargs:

        form_data['repo_name'] = name

    if not 'perms_new' in kwargs:

        form_data['perms_new'] = []

    if not 'perms_updates' in kwargs:

        form_data['perms_updates'] = []

    r = RepoModel().update(name, **form_data)

    return r

    def setUp(self):

        self.g1 = fixture.create_group('test1', skip_if_exists=True)

        self.g2 = fixture.create_group('test2', skip_if_exists=True)

        self.g3 = fixture.create_group('test3', skip_if_exists=True)

    def tearDown(self):

        Session.remove()

    def __check_path(self, *path):

        """

        Checks the path for existance !

        """

        path = [TESTS_TMP_PATH] + list(path)

        path = os.path.join(*path)

        return os.path.isdir(path)

    def _check_folders(self):

        print os.listdir(TESTS_TMP_PATH)

    def __delete_group(self, id_):

        ReposGroupModel().delete(id_)

    def test_create_group(self):

        g = fixture.create_group('newGroup')

        Session().commit()

        self.assertEqual(g.full_path, 'newGroup')

        self.assertTrue(self.__check_path('newGroup'))

    def test_create_same_name_group(self):

        self.assertRaises(IntegrityError, lambda: fixture.create_group('newGroup'))

        Session().rollback()

    def test_same_subgroup(self):

        sg1 = fixture.create_group('sub1', group_parent_id=self.g1.group_id)

        self.assertEqual(sg1.parent_group, self.g1)

        self.assertEqual(sg1.full_path, 'test1/sub1')

        self.assertTrue(self.__check_path('test1', 'sub1'))

        ssg1 = fixture.create_group('subsub1', group_parent_id=sg1.group_id)

        self.assertEqual(ssg1.parent_group, sg1)

        self.assertEqual(ssg1.full_path, 'test1/sub1/subsub1')

        self.assertTrue(self.__check_path('test1', 'sub1', 'subsub1'))

    def test_remove_group(self):

        sg1 = fixture.create_group('deleteme')

        self.__delete_group(sg1.group_id)

import functools

from rhodecode.tests import *

from rhodecode.model.repos_group import ReposGroupModel

from rhodecode.model.db import RepoGroup, Repository, User

from rhodecode.model.meta import Session

from nose.tools import with_setup

from rhodecode.tests.models.common import _create_project_tree, check_tree_perms, \

    _get_perms, _check_expected_count, expected_count, _destroy_project_tree

test_u1_id = None

_get_repo_perms = None

_get_group_perms = None

def permissions_setup_func(group_name='g0', perm='group.read', recursive=True):

    """

    Resets all permissions to perm attribute

    """

    repos_group = RepoGroup.get_by_group_name(group_name=group_name)

    if not repos_group:

        raise Exception('Cannot get group %s' % group_name)

    perms_updates = [[test_u1_id, perm, 'user']]

    ReposGroupModel()._update_permissions(repos_group,

                                          perms_updates=perms_updates,

                                          recursive=recursive,

                                          check_perms=False)

    Session().commit()

def setup_module():

    global test_u1_id, _get_repo_perms, _get_group_perms

    test_u1 = _create_project_tree()

    Session().commit()

    test_u1_id = test_u1.user_id

    _get_repo_perms = functools.partial(_get_perms, key='repositories',

                                        test_u1_id=test_u1_id)

    _get_group_perms = functools.partial(_get_perms, key='repositories_groups',

                                         test_u1_id=test_u1_id)

def teardown_module():

    _destroy_project_tree(test_u1_id)

@with_setup(permissions_setup_func)

def test_user_permissions_on_group_without_recursive_mode():

    # set permission to g0 non-recursive mode

    recursive = False

    group = 'g0'

    permissions_setup_func(group, 'group.write', recursive=recursive)

    items = [x for x in _get_repo_perms(group, recursive)]

    expected = 0

    assert len(items) == expected, ' %s != %s' % (len(items), expected)

    for name, perm in items:

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.users_group import UserGroupModel

from rhodecode.tests.fixture import Fixture

fixture = Fixture()

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

                                     email=u'u232@rhodecode.org',

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

        Session().commit()

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

        # make user group

        users_group = fixture.create_user_group('some_example_group')

        Session().commit()

        UserGroupModel().add_user_to_group(users_group, usr)

        Session().commit()

        self.assertEqual(UserGroup.get(users_group.users_group_id), users_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',

                                     email=u'main_email@rhodecode.org',

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

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

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

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