# You should have received a copy of the GNU General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""

kallithea.model.repo

~~~~~~~~~~~~~~~~~~~~

Repository model for kallithea

This file was forked by the Kallithea project in July 2014.

Original author and date, and relevant copyright and licensing information is below:

:created_on: Jun 5, 2010

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import os

import shutil

import logging

import traceback

from datetime import datetime

from sqlalchemy.orm import subqueryload

from kallithea.lib.vcs.backends import get_backend

from kallithea.lib.compat import json

from kallithea.lib.utils2 import LazyProperty, safe_str, safe_unicode, \

    remove_prefix, obfuscate_url_pw, get_current_authuser

from kallithea.lib.caching_query import FromCache

from kallithea.lib.hooks import log_delete_repository

from kallithea.model import BaseModel

from kallithea.model.db import Repository, UserRepoToPerm, UserGroupRepoToPerm, \

    UserRepoGroupToPerm, UserGroupRepoGroupToPerm, User, Permission, \

    Statistics, UserGroup, Ui, RepoGroup, RepositoryField

from kallithea.lib import helpers as h

from kallithea.lib.auth import HasRepoPermissionAny, HasUserGroupPermissionAny

from kallithea.lib.exceptions import AttachedForksError

from kallithea.model.scm import UserGroupList

log = logging.getLogger(__name__)

class RepoModel(BaseModel):

    cls = Repository

    URL_SEPARATOR = Repository.url_sep()

        return defaults

    def update(self, repo, **kwargs):

        try:

            cur_repo = self._get_repo(repo)

            org_repo_name = cur_repo.repo_name

            if 'user' in kwargs:

                cur_repo.user = User.get_by_username(kwargs['user'])

            if 'repo_group' in kwargs:

                cur_repo.group = RepoGroup.get(kwargs['repo_group'])

                cur_repo.repo_name = cur_repo.get_new_name(cur_repo.just_name)

            log.debug('Updating repo %s with params:%s', cur_repo, kwargs)

            for k in ['repo_enable_downloads',

                      'repo_description',

                      'repo_enable_locking',

                      'repo_landing_rev',

                      'repo_private',

                      'repo_enable_statistics',

                      ]:

                if k in kwargs:

                    setattr(cur_repo, remove_prefix(k, 'repo_'), kwargs[k])

            clone_uri = kwargs.get('clone_uri')

            if clone_uri is not None and clone_uri != cur_repo.clone_uri_hidden:

                cur_repo.clone_uri = clone_uri

            if 'repo_name' in kwargs:

                cur_repo.repo_name = cur_repo.get_new_name(kwargs['repo_name'])

            #if private flag is set, reset default permission to NONE

            if kwargs.get('repo_private'):

                EMPTY_PERM = 'repository.none'

                RepoModel().grant_user_permission(

                    repo=cur_repo, user='default', perm=EMPTY_PERM

                )

                #handle extra fields

            for field in filter(lambda k: k.startswith(RepositoryField.PREFIX),

                                kwargs):

                k = RepositoryField.un_prefix_key(field)

                ex_field = RepositoryField.get_by_key_name(key=k, repo=cur_repo)

                if ex_field:

                    ex_field.field_value = kwargs[field]

                    self.sa.add(ex_field)

            self.sa.add(cur_repo)

            if org_repo_name != cur_repo.repo_name:

                # rename repository

                self._rename_filesystem_repo(old=org_repo_name, new=cur_repo.repo_name)

        """

        from kallithea.model.scm import ScmModel

        owner = self._get_user(owner)

        fork_of = self._get_repo(fork_of)

        repo_group = self._get_repo_group(repo_group)

        try:

            repo_name = safe_unicode(repo_name)

            description = safe_unicode(description)

            # 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_full = repo_name

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

            new_repo = Repository()

            new_repo.repo_state = state

            new_repo.enable_statistics = False

            new_repo.repo_name = repo_name_full

            new_repo.repo_type = repo_type

            new_repo.user = owner

            new_repo.group = repo_group

            new_repo.description = description or repo_name

            new_repo.private = private

            new_repo.clone_uri = clone_uri

            new_repo.landing_rev = landing_rev

            new_repo.enable_statistics = enable_statistics

            new_repo.enable_locking = enable_locking

            new_repo.enable_downloads = enable_downloads

            if repo_group:

                new_repo.enable_locking = repo_group.enable_locking

            if fork_of:

                parent_repo = fork_of

                new_repo.fork = parent_repo

            self.sa.add(new_repo)

            if fork_of and copy_fork_permissions:

                repo = fork_of

                user_perms = UserRepoToPerm.query() \

                    .filter(UserRepoToPerm.repository == repo).all()

                group_perms = UserGroupRepoToPerm.query() \

                    .filter(UserGroupRepoToPerm.repository == repo).all()

                for perm in user_perms:

            log.debug('Revoked perm to %s on %s', repo, group_name)

    def delete_stats(self, repo_name):

        """

        removes stats for given repo

        :param repo_name:

        """

        repo = self._get_repo(repo_name)

        try:

            obj = self.sa.query(Statistics) \

                .filter(Statistics.repository == repo).scalar()

            if obj is not None:

                self.sa.delete(obj)

        except Exception:

            log.error(traceback.format_exc())

            raise

    def _create_filesystem_repo(self, repo_name, repo_type, repo_group,

                                clone_uri=None, repo_store_location=None):

        """

        Makes repository on filesystem. Operation is group aware, meaning that it will create

        a repository within a group, and alter the paths accordingly to the group location.

        """

        from kallithea.lib.utils import is_valid_repo, is_valid_repo_group

        from kallithea.model.scm import ScmModel

        if '/' in repo_name:

            raise ValueError('repo_name must not contain groups got `%s`' % repo_name)

        if isinstance(repo_group, RepoGroup):

            new_parent_path = os.sep.join(repo_group.full_path_splitted)

        else:

            new_parent_path = repo_group or ''

        if repo_store_location:

            _paths = [repo_store_location]

        else:

            _paths = [self.repos_path, new_parent_path, repo_name]

            # we need to make it str for mercurial

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

        # 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

        expected = ret

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

    def test_api_get_user_without_giving_userid_non_admin(self):

        id_, params = _build_data(self.apikey_regular, 'get_user')

        response = api_call(self, params)

        usr = User.get_by_username(self.TEST_USER_LOGIN)

        ret = usr.get_api_data()

        ret['permissions'] = AuthUser(dbuser=usr).permissions

        expected = ret

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

    def test_api_get_user_with_giving_userid_non_admin(self):

        id_, params = _build_data(self.apikey_regular, 'get_user',

                                  userid=self.TEST_USER_LOGIN)

        response = api_call(self, params)

        expected = 'userid is not the same as your user'

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

    def test_api_pull(self):

        repo_name = 'test_pull'

        r = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)

        r.clone_uri = os.path.join(TESTS_TMP_PATH, self.REPO)

        Session.add(r)

        Session.commit()

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

                                  repoid=repo_name,)

        response = api_call(self, params)

        expected = {'msg': 'Pulled from `%s`' % repo_name,

                    'repository': repo_name}

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

        fixture.destroy_repo(repo_name)

    def test_api_pull_error(self):

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

                                  repoid=self.REPO, )

        response = api_call(self, params)

        expected = 'Unable to pull changes from `%s`' % self.REPO

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

    def test_api_rescan_repos(self):

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

        finally:

            RepoModel().revoke_user_permission(self.REPO, self.TEST_USER_LOGIN)

    def test_api_create_repo(self):

        repo_name = 'api-repo'

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

                                  repo_name=repo_name,

                                  owner=TEST_USER_ADMIN_LOGIN,

                                  repo_type=self.REPO_TYPE,

        )

        response = api_call(self, params)

        repo = RepoModel().get_by_repo_name(repo_name)

        self.assertNotEqual(repo, None)

        ret = {

            'msg': 'Created new repository `%s`' % repo_name,

            'success': True,

            'task': None,

        }

        expected = ret

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

        fixture.destroy_repo(repo_name)

    def test_api_create_repo_clone_uri_local(self):

        clone_uri = os.path.join(TESTS_TMP_PATH, self.REPO)

        repo_name = u'api-repo'

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

                                  repo_name=repo_name,

                                  owner=TEST_USER_ADMIN_LOGIN,

                                  repo_type=self.REPO_TYPE,

                                  clone_uri=clone_uri,

        )

        response = api_call(self, params)

        fixture.destroy_repo(repo_name)

    def test_api_create_repo_and_repo_group(self):

        repo_name = 'my_gr/api-repo'

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

                                  repo_name=repo_name,

                                  owner=TEST_USER_ADMIN_LOGIN,

                                  repo_type=self.REPO_TYPE,)

        response = api_call(self, params)

        print params

        repo = RepoModel().get_by_repo_name(repo_name)

        self.assertNotEqual(repo, None)

        ret = {

            'msg': 'Created new repository `%s`' % repo_name,

            'success': True,

            'task': None,

        }

        expected = ret

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

        fixture.destroy_repo(repo_name)

        fixture.destroy_repo_group('my_gr')

    def test_api_create_repo_in_repo_group_without_permission(self):

        repo_group_name = '%s/api-repo-repo' % TEST_REPO_GROUP

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

                                  repo_name=repo_name,

                                  owner=TEST_USER_ADMIN_LOGIN,

                                  repo_type=self.REPO_TYPE,)

        response = api_call(self, params)

        expected = "repo `%s` already exist" % repo_name

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

    @mock.patch.object(RepoModel, 'create', crash)

    def test_api_create_repo_exception_occurred(self):

        repo_name = 'api-repo'

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

                                  repo_name=repo_name,

                                  owner=TEST_USER_ADMIN_LOGIN,

                                  repo_type=self.REPO_TYPE,)

        response = api_call(self, params)

        expected = 'failed to create repository `%s`' % repo_name

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

    @parameterized.expand([

        ('owner', {'owner': TEST_USER_REGULAR_LOGIN}),

        ('description', {'description': 'new description'}),

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

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

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

        ('landing_rev', {'landing_rev': 'branch:master'}),

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

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

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

        ('name', {'name': 'new_repo_name'}),

        ('repo_group', {'group': 'test_group_for_update'}),

    ])

    def test_api_update_repo(self, changing_attr, updates):

        repo_name = 'api_update_me'

        repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)

        if changing_attr == 'repo_group':

            fixture.create_repo_group(updates['group'])

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

                                  repoid=repo_name, **updates)

        response = api_call(self, params)

        if changing_attr == 'name':

            repo_name = updates['name']

        if changing_attr == 'repo_group':

            repo_name = '/'.join([updates['group'], repo_name])

        try:

        finally:

            fixture.destroy_repo(repo_name)

            if changing_attr == 'repo_group':

                fixture.destroy_repo_group(updates['group'])

    @parameterized.expand([

        ('owner', {'owner': TEST_USER_REGULAR_LOGIN}),

        ('description', {'description': u'new description'}),

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

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

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

        ('landing_rev', {'landing_rev': 'branch:master'}),

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

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

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

        ('name', {'name': u'new_repo_name'}),

        ('repo_group', {'group': u'test_group_for_update'}),

    ])

    def test_api_update_group_repo(self, changing_attr, updates):

        group_name = u'lololo'

        fixture.create_repo_group(group_name)

        repo_name = u'%s/api_update_me' % group_name

        repo = fixture.create_repo(repo_name, repo_group=group_name, repo_type=self.REPO_TYPE)

        if changing_attr == 'repo_group':

            fixture.create_repo_group(updates['group'])

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

                                  repoid=repo_name, **updates)

        response = api_call(self, params)

        if changing_attr == 'name':

            repo_name = u'%s/%s' % (group_name, updates['name'])

        if changing_attr == 'repo_group':

            repo_name = u'/'.join([updates['group'], repo_name.rsplit('/', 1)[-1]])

        try:

        finally:

            fixture.destroy_repo(repo_name)

            if changing_attr == 'repo_group':

                fixture.destroy_repo_group(updates['group'])

        fixture.destroy_repo_group(group_name)

    def test_api_update_repo_repo_group_does_not_exist(self):

        repo_name = 'admin_owned'

        fixture.create_repo(repo_name)

        updates = {'group': 'test_group_for_update'}

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

                                  repoid=repo_name, **updates)

        response = api_call(self, params)

        try:

            expected = 'repository group `%s` does not exist' % updates['group']

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

        finally:

            fixture.destroy_repo(repo_name)

    def test_api_update_repo_regular_user_not_allowed(self):

        repo_name = 'admin_owned'

        fixture.create_repo(repo_name)

        updates = {'active': False}

        id_, params = _build_data(self.apikey_regular, 'update_repo',