from webhelpers.text import collapse, remove_formatting, strip_tags

from rhodecode.lib.vcs import get_backend

from rhodecode.lib.vcs.backends.base import BaseChangeset

from rhodecode.lib.vcs.utils.lazy import LazyProperty

from rhodecode.lib.vcs.utils.helpers import get_scm

from rhodecode.lib.vcs.exceptions import VCSError

from rhodecode.lib.caching_query import FromCache

from rhodecode.model import meta

from rhodecode.model.db import Repository, User, RhodeCodeUi, \

from rhodecode.model.meta import Session

from rhodecode.model.repos_group import ReposGroupModel

from rhodecode.lib.utils2 import safe_str, safe_unicode

from rhodecode.lib.vcs.utils.fakemod import create_module

log = logging.getLogger(__name__)

REMOVED_REPO_PAT = re.compile(r'rm__\d{8}_\d{6}_\d{6}__.*')

def recursive_replace(str_, replace=' '):

    """

    :param user: user that made this action, can be a unique username string or

        object containing user_id attribute

    :param action: action to log, should be on of predefined unique actions for

        easy translations

    :param repo: string name of repository or object containing repo_id,

        that action was made on

    :param ipaddr: optional ip address from what the action was made

    :param sa: optional sqlalchemy session

    """

    if not sa:

    try:

        if hasattr(user, 'user_id'):

            user_obj = user

        elif isinstance(user, basestring):

            user_obj = User.get_by_username(user)

        else:

            raise Exception('You have to provide user object or username')

        if hasattr(repo, 'repo_id'):

            repo_obj = Repository.get(repo.repo_id)

            repo_name = repo_obj.repo_name

        if not os.path.isfile(path):

            log.debug('hgrc file is not present at %s skipping...' % path)

            return False

        log.debug('reading hgrc from %s' % path)

        cfg = config.config()

        cfg.read(path)

        for section in ui_sections:

            for k, v in cfg.items(section):

                log.debug('settings ui from file[%s]%s:%s' % (section, k, v))

                baseui.setconfig(section, k, v)

    elif read_from == 'db':

        ret = sa.query(RhodeCodeUi)\

            .options(FromCache("sql_cache_short", "get_hg_ui_settings"))\

            .all()

        hg_ui = ret

        for ui_ in hg_ui:

            if ui_.ui_active:

                log.debug('settings ui from db[%s]%s:%s', ui_.ui_section,

                          ui_.ui_key, ui_.ui_value)

                baseui.setconfig(ui_.ui_section, ui_.ui_key, ui_.ui_value)

        meta.Session.remove()

    def get_file_size(self, path):

        return 0

def map_groups(path):

    """

    Given a full path to a repository, create all nested groups that this

    repo is inside. This function creates parent-child relationships between

    groups and creates default perms for all new groups.

    :param paths: full path to repository

    """

    groups = path.split(Repository.url_sep())

    parent = None

    group = None

    # last element is repo in nested groups structure

    groups = groups[:-1]

    rgm = ReposGroupModel(sa)

    for lvl, group_name in enumerate(groups):

        group_name = '/'.join(groups[:lvl] + [group_name])

        group = RepoGroup.get_by_group_name(group_name)

        desc = '%s group' % group_name

def repo2db_mapper(initial_repo_list, remove_obsolete=False):

    """

    maps all repos given in initial_repo_list, non existing repositories

    are created, if remove_obsolete is True it also check for db entries

    that are not in initial_repo_list and removes them.

    :param initial_repo_list: list of repositories found by scanning methods

    :param remove_obsolete: check for obsolete entries in database

    """

    from rhodecode.model.repo import RepoModel

    rm = RepoModel()

    user = sa.query(User).filter(User.admin == True).first()

    if user is None:

        raise Exception('Missing administrative account !')

    added = []

    for name, repo in initial_repo_list.items():

        group = map_groups(name)

            added.append(name)

    sa.commit()

    removed = []

    if remove_obsolete:

        # remove from database those repositories that are not in the filesystem

        for repo in sa.query(Repository).all():

            if repo.repo_name not in initial_repo_list.keys():

                log.debug("Removing non existing repository found in db %s" %

                          repo.repo_name)

                removed.append(repo.repo_name)

                sa.delete(repo)

                sa.commit()

    if not os.path.isdir(repos_test_path):

        log.debug('Creating testdir %s' % repos_test_path)

        os.makedirs(repos_test_path)

    dbmanage = DbManage(log_sql=True, dbconf=dbconf, root=config['here'],

                        tests=True)

    dbmanage.create_tables(override=True)

    dbmanage.create_settings(dbmanage.config_prompt(repos_test_path))

    dbmanage.create_default_user()

    dbmanage.admin_prompt()

    dbmanage.create_permissions()

    dbmanage.populate_default_permissions()

    # PART TWO make test repo

    log.debug('making test vcs repositories')

    idx_path = config['app_conf']['index_dir']

    data_path = config['app_conf']['cache_dir']

    #clean index and data

    if idx_path and os.path.exists(idx_path):

        log.debug('remove %s' % idx_path)

        shutil.rmtree(idx_path)

    if data_path and os.path.exists(data_path):

        except:

            log.error(traceback.format_exc())

            raise

    def create_repo(self, repo_name, repo_type, description, owner,

                    private=False, clone_uri=None, repos_group=None,

                    landing_rev='tip', just_db=False, fork_of=None,

                    copy_fork_permissions=False):

        from rhodecode.model.scm import ScmModel

        owner = self._get_user(owner)

        fork_of = self._get_repo(fork_of)

        try:

            # repo name is just a name of repository

            # while repo_name_full is a full qualified name that is combined

            # with name and path of group

            repo_name = repo_name.split(self.URL_SEPARATOR)[-1]

            new_repo = Repository()

            new_repo.enable_statistics = False

            new_repo.repo_name = repo_name_full

            new_repo.repo_type = repo_type

            new_repo.user = owner

            new_repo.description = description or repo_name

            new_repo.private = private

            new_repo.clone_uri = clone_uri

            new_repo.landing_rev = landing_rev

            if fork_of:

                parent_repo = fork_of

                new_repo.fork = parent_repo

            self.sa.add(new_repo)

            def _create_default_perms():

                                              perm.permission)

                    for perm in group_perms:

                        UsersGroupRepoToPerm.create(perm.users_group, new_repo,

                                                    perm.permission)

                else:

                    _create_default_perms()

            else:

                _create_default_perms()

            if not just_db:

                self.__create_repo(repo_name, repo_type,

                                   clone_uri)

                log_create_repository(new_repo.get_dict(),

                                      created_by=owner.username)

            # now automatically start following this repository as owner

            ScmModel(self.sa).toggle_following_repo(new_repo.repo_id,

                                                    owner.user_id)

            return new_repo

        except:

            log.error(traceback.format_exc())

            raise

    def create(self, form_data, cur_user, just_db=False, fork=None):

        repo_name = form_data['repo_name_full']

        repo_type = form_data['repo_type']

        description = form_data['description']

        owner = cur_user

        private = form_data['private']

        clone_uri = form_data.get('clone_uri')

        repos_group = form_data['repo_group']

        :param repo_name:

        """

        try:

            obj = self.sa.query(Statistics)\

                    .filter(Statistics.repository ==

                            self.get_by_repo_name(repo_name))\

                    .one()

            self.sa.delete(obj)

        except:

            log.error(traceback.format_exc())

            raise

        """

        makes repository on filesystem. It's group aware means it'll create

        a repository within a group, and alter the paths accordingly of

        group location

        :param repo_name:

        :param alias:

        :param parent_id:

        :param clone_uri:

        """

        from rhodecode.lib.utils import is_valid_repo, is_valid_repos_group

        else:

            new_parent_path = ''

        # we need to make it str for mercurial

        repo_path = os.path.join(*map(lambda x: safe_str(x),

                                [self.repos_path, new_parent_path, repo_name]))

        # check if this path is not a repository

        if is_valid_repo(repo_path, self.repos_path):

            raise Exception('This path %s is a valid repository' % repo_path)

        # check if this path is a group

        Deletes a group from a filesystem

        :param group: instance of group from database

        """

        paths = group.full_path.split(RepoGroup.url_sep())

        paths = os.sep.join(paths)

        rm_path = os.path.join(self.repos_path, paths)

        if os.path.isdir(rm_path):

            # delete only if that path really exists

            os.rmdir(rm_path)

        try:

            new_repos_group = RepoGroup()

            new_repos_group.group_description = group_description

            new_repos_group.parent_group = self.__get_repos_group(parent)

            new_repos_group.group_name = new_repos_group.get_new_name(group_name)

            self.sa.add(new_repos_group)

            self._create_default_perms(new_repos_group)

            if not just_db:

                # we need to flush here, in order to check if database won't

                # throw any exceptions, create filesystem dirs at the very end

import random

import mock

from rhodecode.tests import *

from rhodecode.lib.compat import json

from rhodecode.lib.auth import AuthUser

from rhodecode.model.user import UserModel

from rhodecode.model.users_group import UsersGroupModel

from rhodecode.model.repo import RepoModel

API_URL = '/_admin/api'

def _build_data(apikey, method, **kw):

    """

    Builds API data with given random ID

    :param random_id:

    :type random_id:

    """

    random_id = random.randrange(1, 9999)

                                 params=params)

        usr = UserModel().get_by_username(username)

        ret = dict(

            msg='created new user `%s`' % username,

            user=jsonify(usr.get_api_data())

        )

        expected = ret

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

        UserModel().delete(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 = self.app.post(API_URL, content_type='application/json',

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

                                           email=u'u232@rhodecode.org',

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

        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 = self.app.post(API_URL, content_type='application/json',

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

                                           email=u'u232@rhodecode.org',

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

        username = usr.username

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

                                  userid=username,)

        response = self.app.post(API_URL, content_type='application/json',

                                 params=params)

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

                                             usr.username)

        expected = ret

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

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

                                 params=params)

        ret = 'failed to update user `%s`' % usr.user_id

        expected = ret

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

    def test_api_get_repo(self):

        new_group = 'some_new_group'

        make_users_group(new_group)

        RepoModel().grant_users_group_permission(repo=self.REPO,

                                                 group_name=new_group,

                                                 perm='repository.read')

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

                                  repoid=self.REPO)

        response = self.app.post(API_URL, content_type='application/json',

                                 params=params)

        repo = RepoModel().get_by_repo_name(self.REPO)

        ret = repo.get_api_data()

        members = []

        for user in repo.repo_to_perm:

            perm = user.permission.permission_name

            user = user.user

        id_, params = _build_data(self.apikey, 'get_users_groups',)

        response = self.app.post(API_URL, content_type='application/json',

                                 params=params)

        expected = []

        for gr_name in [TEST_USERS_GROUP, 'test_users_group2']:

            users_group = UsersGroupModel().get_group(gr_name)

            ret = users_group.get_api_data()

            expected.append(ret)

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

        UsersGroupModel().delete(users_group='test_users_group2')

    def test_api_create_users_group(self):

        group_name = 'some_new_group'

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

                                  group_name=group_name)

        response = self.app.post(API_URL, content_type='application/json',

                                 params=params)

        ret = {

            'msg': 'created new users group `%s`' % group_name,

            'users_group': jsonify(UsersGroupModel()\

                                   .get_by_name(group_name)\

        group_name = 'exception_happens'

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

                                  group_name=group_name)

        response = self.app.post(API_URL, content_type='application/json',

                                 params=params)

        expected = 'failed to create group `%s`' % group_name

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

    def test_api_add_user_to_users_group(self):

        gr_name = 'test_group'

        UsersGroupModel().create(gr_name)

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

                                  usersgroupid=gr_name,

                                  userid=TEST_USER_ADMIN_LOGIN)

        response = self.app.post(API_URL, content_type='application/json',

                                 params=params)

        expected = {

                    'msg': 'added member `%s` to users group `%s`' % (

                                TEST_USER_ADMIN_LOGIN, gr_name

                            ),

                    'success': True}

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

        UsersGroupModel().delete(users_group=gr_name)

    def test_api_add_user_to_users_group_that_doesnt_exist(self):

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

                                  usersgroupid='false-group',

                                  userid=TEST_USER_ADMIN_LOGIN)

        response = self.app.post(API_URL, content_type='application/json',

                                 params=params)

        expected = 'users group `%s` does not exist' % 'false-group'

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

    @mock.patch.object(UsersGroupModel, 'add_user_to_group', crash)

    def test_api_add_user_to_users_group_exception_occurred(self):

        gr_name = 'test_group'

        UsersGroupModel().create(gr_name)

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

                                  usersgroupid=gr_name,

                                  userid=TEST_USER_ADMIN_LOGIN)

        response = self.app.post(API_URL, content_type='application/json',

                                 params=params)

        expected = 'failed to add member to users group `%s`' % gr_name

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

        UsersGroupModel().delete(users_group=gr_name)

    def test_api_remove_user_from_users_group(self):

        gr_name = 'test_group_3'

        gr = UsersGroupModel().create(gr_name)

        UsersGroupModel().add_user_to_group(gr, user=TEST_USER_ADMIN_LOGIN)

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

                                  usersgroupid=gr_name,

                                  userid=TEST_USER_ADMIN_LOGIN)

        response = self.app.post(API_URL, content_type='application/json',

                                 params=params)

        expected = {

                    'msg': 'removed member `%s` from users group `%s`' % (

                                TEST_USER_ADMIN_LOGIN, gr_name

                            ),

                    'success': True}

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

        UsersGroupModel().delete(users_group=gr_name)

    @mock.patch.object(UsersGroupModel, 'remove_user_from_group', crash)

    def test_api_remove_user_from_users_group_exception_occurred(self):

        gr_name = 'test_group_3'

        gr = UsersGroupModel().create(gr_name)

        UsersGroupModel().add_user_to_group(gr, user=TEST_USER_ADMIN_LOGIN)

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

                                  usersgroupid=gr_name,

                                  userid=TEST_USER_ADMIN_LOGIN)

        response = self.app.post(API_URL, content_type='application/json',

                                 params=params)

        expected = 'failed to remove member from users group `%s`' % gr_name

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

        UsersGroupModel().delete(users_group=gr_name)

    @parameterized.expand([('none', 'repository.none'),

                           ('read', 'repository.read'),

                           ('write', 'repository.write'),

                           ('admin', 'repository.admin')])

    def test_api_grant_user_permission(self, name, perm):

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

                                  repoid=self.REPO,

                                  userid=TEST_USER_ADMIN_LOGIN,

                                  perm=perm)

        response = self.app.post(API_URL, content_type='application/json',

                                 params=params)

        response = self.app.post(API_URL, content_type='application/json',

                                 params=params)

        expected = 'failed to edit permission for users group: `%s` in repo: `%s`' % (

                    TEST_USERS_GROUP, self.REPO

                )

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

    def test_api_revoke_users_group_permission(self):

        RepoModel().grant_users_group_permission(repo=self.REPO,

                                                 group_name=TEST_USERS_GROUP,

                                                 perm='repository.read')

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

                                  repoid=self.REPO,

                                  usersgroupid=TEST_USERS_GROUP,)

        response = self.app.post(API_URL, content_type='application/json',

                                 params=params)

        expected = {

            'msg': 'Revoked perm for users group: `%s` in repo: `%s`' % (

                TEST_USERS_GROUP, self.REPO

            ),

            'success': True

        }

from rhodecode.tests import *

from rhodecode.model.user import UserModel

from rhodecode.model.notification import NotificationModel

class TestNotificationsController(TestController):

    def tearDown(self):

        for n in Notification.query().all():

            inst = Notification.get(n.notification_id)

    def test_index(self):

        self.log_user()

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

                                               email='u1@rhodecode.org',

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

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

#

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

                                               email='u1@rhodecode.org',

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

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

                                               email='u2@rhodecode.org',

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

        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=

# -*- coding: utf-8 -*-

import os

from rhodecode.lib import vcs

from rhodecode.tests import *

class TestAdminReposController(TestController):

    def __make_repo(self):

        pass

    def test_index(self):

        self.log_user()

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

        # Test response...

        private = False

        response = self.app.post(url('repos'), {'repo_name': repo_name,

                                                'repo_type': 'hg',

                                                'clone_uri': '',

                                                'repo_group': '',

                                                'description': description,

                                                'private': private,

                                                'landing_rev': 'tip'})

        self.checkSessionFlash(response,

                               'created repository %s' % (repo_name))

        #test if the repo was created in the database

            .filter(Repository.repo_name == repo_name).one()

        self.assertEqual(new_repo.repo_name, repo_name)

        self.assertEqual(new_repo.description, description)

        #test if repository is visible in the list ?

        response = response.follow()

        response.mustcontain(repo_name)

        #test if repository was created on filesystem

        try:

        private = False

        response = self.app.post(url('repos'), {'repo_name': repo_name,

                                                'repo_type': 'hg',

                                                'clone_uri': '',

                                                'repo_group': '',

                                                'description': description,

                                                'private': private,

                                                'landing_rev': 'tip'})

        self.checkSessionFlash(response,

                               'created repository %s' % (repo_name_unicode))

        #test if the repo was created in the database

            .filter(Repository.repo_name == repo_name_unicode).one()

        self.assertEqual(new_repo.repo_name, repo_name_unicode)

        self.assertEqual(new_repo.description, description_unicode)

        #test if repository is visible in the list ?

        response = response.follow()

        response.mustcontain(repo_name)

        #test if repository was created on filesystem

        try:

            vcs.get_repo(os.path.join(TESTS_TMP_PATH, repo_name))

        except:

            self.fail('no repo %s in filesystem' % repo_name)

    def test_create_hg_in_group(self):

    def test_create_git(self):

        self.log_user()

        repo_name = NEW_GIT_REPO

        description = 'description for newly created repo'

        private = False

        response = self.app.post(url('repos'), {'repo_name': repo_name,