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

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

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

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

"""

Tests for the JSON-RPC web api.

"""

import os

import random

import mock

import re

import pytest

from kallithea.tests.base import *

from kallithea.tests.fixture import Fixture

from kallithea.lib.compat import json

from kallithea.lib.auth import AuthUser

from kallithea.model.user import UserModel

from kallithea.model.user_group import UserGroupModel

from kallithea.model.repo import RepoModel

from kallithea.model.repo_group import RepoGroupModel

from kallithea.model.meta import Session

from kallithea.model.scm import ScmModel

from kallithea.model.gist import GistModel

from kallithea.model.changeset_status import ChangesetStatusModel

from kallithea.lib.utils2 import time_to_datetime

API_URL = '/_admin/api'

TEST_USER_GROUP = u'test_user_group'

TEST_REPO_GROUP = u'test_repo_group'

fixture = Fixture()

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

    """

    Builds API data with given random ID

    :param random_id:

    """

    random_id = random.randrange(1, 9999)

    return random_id, json.dumps({

        "id": random_id,

        "api_key": apikey,

        "method": method,

        "args": kw

    })

jsonify = lambda obj: json.loads(json.dumps(obj))

def crash(*args, **kwargs):

    raise Exception('Total Crash !')

def api_call(test_obj, params):

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

                                 params=params)

    return response

## helpers

def make_user_group(name=TEST_USER_GROUP):

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

                                  ret_type=ret_type)

        response = api_call(self, params)

        # we don't the actual return types here since it's tested somewhere

        # else

        expected = response.json['result']

        try:

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

        finally:

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

    def test_api_create_repo(self):

        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,

        )

        response = api_call(self, params)

        repo = RepoModel().get_by_repo_name(repo_name)

        assert repo is not 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):

        # cloning from local repo was a mis-feature - it would bypass access control

        # TODO: introduce other test coverage of actual remote cloning

        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)

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

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

        fixture.destroy_repo(repo_name)

    def test_api_create_repo_and_repo_group(self):

        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)

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

            'success': True,

            'task': None,

        }

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

        fixture.destroy_repo(repo_name)

    def test_api_create_repo_in_repo_group_without_permission(self):

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

        repo_name = u'%s/api-repo' % repo_group_name

        rg = fixture.create_repo_group(repo_group_name)

        Session().commit()

        RepoGroupModel().grant_user_permission(repo_group_name,

                                               self.TEST_USER_LOGIN,

                                               'group.none')

        Session().commit()

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

                                  repo_name=repo_name,

                                  repo_type=self.REPO_TYPE,

        )

        response = api_call(self, params)

        # Current result when API access control is different from Web:

        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)

        # Expected and arguably more correct result:

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

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

        fixture.destroy_repo_group(repo_group_name)

    def test_api_create_repo_unknown_owner(self):

        repo_name = u'api-repo'

        owner = 'i-dont-exist'

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

                                  repo_name=repo_name,

                                  owner=owner,

                                  repo_type=self.REPO_TYPE,

        )

        response = api_call(self, params)

        expected = 'user `%s` does not exist' % owner

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

    def test_api_create_repo_dont_specify_owner(self):

        repo_name = u'api-repo'

        }

        expected = ret

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

        fixture.destroy_repo(repo_name)

    def test_api_create_repo_by_non_admin(self):

        repo_name = u'api-repo'

        owner = 'i-dont-exist'

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

                                  repo_name=repo_name,

                                  repo_type=self.REPO_TYPE,

        )

        response = api_call(self, params)

        repo = RepoModel().get_by_repo_name(repo_name)

        assert repo is not 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_by_non_admin_specify_owner(self):

        repo_name = u'api-repo'

        owner = 'i-dont-exist'

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

                                  repo_name=repo_name,

                                  repo_type=self.REPO_TYPE,

                                  owner=owner)

        response = api_call(self, params)

        expected = 'Only Kallithea admin can specify `owner` param'

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

        fixture.destroy_repo(repo_name)

    def test_api_create_repo_exists(self):

        repo_name = self.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 = "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 = 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,)

        response = api_call(self, params)

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

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

    @parametrize('changing_attr,updates', [

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

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

        ('clone_uri', {'clone_uri': 'http://example.com/repo'}), # will fail - pulling from non-existing repo should fail

        ('clone_uri', {'clone_uri': '/repo'}), # will fail - pulling from local repo was a mis-feature - it would bypass access control

        ('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_repo(self, changing_attr, updates):

        repo_name = u'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 = u'/'.join([updates['group'], repo_name])

        try:

            if changing_attr == 'clone_uri' and updates['clone_uri']:

                expected = u'failed to update repo `%s`' % repo_name

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

            else:

                expected = {

                    'msg': 'updated repo ID:%s %s' % (repo.repo_id, repo_name),

                    'repository': repo.get_api_data()

                }

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

        finally: