# -*- 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.model.db import Repository, User, Setting, Ui, PullRequest, ChangesetStatus, RepoGroup

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

    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 = User.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',

            email='test@example.com',

            firstname=u'first',

            lastname=u'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_method(self, method):

        make_user_group()

        make_repo_group()

    def teardown_method(self, method):

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

            'error': None,

            'result': expected

        })

        given = json.loads(given)

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

        expected = jsonify({

            'id': id_,

            'error': expected,

            'result': None

        })

        given = json.loads(given)

    def test_Optional_object(self):

        from kallithea.controllers.api.api import Optional

        option1 = Optional(None)

        assert '<Optional:%s>' % None == repr(option1)

        assert option1() is None

        assert 1 == Optional.extract(Optional(1))

        assert 'trololo' == Optional.extract('trololo')

    def test_Optional_OAttr(self):

        from kallithea.controllers.api.api import Optional, OAttr

        option1 = Optional(OAttr('apiuser'))

        assert 'apiuser' == Optional.extract(option1)

    def test_OAttr_object(self):

        from kallithea.controllers.api.api import OAttr

        oattr1 = OAttr('apiuser')

        assert '<OptionalAttr:apiuser>' == repr(oattr1)

        assert oattr1() == oattr1

    def test_api_wrong_key(self):

        id_, params = _build_data('trololo', 'get_user')

        response = api_call(self, params)

        expected = 'Invalid API key'

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

    def test_api_missing_non_optional_param(self):

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

        response = api_call(self, params)

        expected = 'Missing non optional `repoid` arg in JSON DATA'

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

    def test_api_missing_non_optional_param_args_null(self):

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

        params = params.replace('"args": {}', '"args": null')

        response = api_call(self, params)

        expected = 'Missing non optional `repoid` arg in JSON DATA'

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

    def test_api_missing_non_optional_param_args_bad(self):

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

        params = params.replace('"args": {}', '"args": 1')

        response = api_call(self, params)

        expected = 'Missing non optional `repoid` arg in JSON DATA'

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

    def test_api_args_is_null(self):

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

        params = params.replace('"args": {}', '"args": null')

        response = api_call(self, params)

        assert response.status == '200 OK'

    def test_api_args_is_bad(self):

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

        params = params.replace('"args": {}', '"args": 1')

        response = api_call(self, params)

        assert response.status == '200 OK'

    def test_api_args_different_args(self):

        import string

        expected = {

            'ascii_letters': string.ascii_letters,

            'ws': string.whitespace,

            'printables': string.printable

        }

        id_, params = _build_data(self.apikey, 'test', args=expected)

        response = api_call(self, params)

        assert response.status == '200 OK'

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

    def test_api_get_users(self):

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

        response = api_call(self, params)

        ret_all = []

        _users = User.query().filter_by(is_default_user=False) \

            .order_by(User.username).all()

        for usr in _users:

            ret = usr.get_api_data()

            ret_all.append(jsonify(ret))

        expected = ret_all

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

    def test_api_get_user(self):

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

                                  userid=TEST_USER_ADMIN_LOGIN)

        response = api_call(self, params)

        usr = User.get_by_username(TEST_USER_ADMIN_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_that_does_not_exist(self):

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

                                  userid='trololo')

        response = api_call(self, params)

        expected = "user `%s` does not exist" % 'trololo'

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

    def test_api_get_user_without_giving_userid(self):

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

        response = api_call(self, params)

        usr = User.get_by_username(TEST_USER_ADMIN_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_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_remote(self):

        # Note: pulling from local repos is a mis-feature - it will bypass access control

        # ... but ok, if the path already has been set in the database

        repo_name = u'test_pull'

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

        # hack around that clone_uri can't be set to to a local path

        # (as shown by test_api_create_repo_clone_uri_local)

        r.clone_uri = os.path.join(Ui.get_by_key('paths', '/').ui_value, self.REPO)

        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_fork(self):

        fork_name = u'fork'

        fixture.create_fork(self.REPO, fork_name)

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

                                  repoid=fork_name,)

        response = api_call(self, params)

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

                    'repository': fork_name}

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

        fixture.destroy_repo(fork_name)

    def test_api_pull_error_no_remote_no_fork(self):

        # should fail because no clone_uri is set

        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_pull_custom_remote(self):

        repo_name = u'test_pull_custom_remote'

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

        custom_remote_path = os.path.join(Ui.get_by_key('paths', '/').ui_value, self.REPO)

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

                                  repoid=repo_name,

                                  clone_uri=custom_remote_path)

        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_rescan_repos(self):

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

        response = api_call(self, params)

        expected = {'added': [], 'removed': []}

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

    @mock.patch.object(ScmModel, 'repo_scan', crash)

    def test_api_rescann_error(self):

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

        response = api_call(self, params)

        expected = 'Error occurred during rescan repositories action'

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

    def test_api_invalidate_cache(self):

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

        repo.scm_instance_cached()  # seed cache

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

                                  repoid=self.REPO)

        response = api_call(self, params)

        expected = {

            'msg': "Cache for repository `%s` was invalidated" % (self.REPO,),

            'repository': self.REPO

        }

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

    @mock.patch.object(ScmModel, 'mark_for_invalidation', crash)

    def test_api_invalidate_cache_error(self):

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

                                  repoid=self.REPO)

        response = api_call(self, params)

        expected = 'Error occurred during cache invalidation action'

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

    def test_api_invalidate_cache_regular_user_no_permission(self):

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

        repo.scm_instance_cached() # seed cache

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

                                  repoid=self.REPO)

        response = api_call(self, params)

        expected = "repository `%s` does not exist" % (self.REPO,)

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

    def test_api_lock_repo_lock_acquire(self):

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

                                  userid=TEST_USER_ADMIN_LOGIN,

                                  repoid=self.REPO,

                                  locked=True)

        response = api_call(self, params)

        expected = {

            'repo': self.REPO, 'locked': True,

            'locked_since': response.json['result']['locked_since'],

            'locked_by': TEST_USER_ADMIN_LOGIN,

            'lock_state_changed': True,

            'msg': ('User `%s` set lock state for repo `%s` to `%s`'

                    % (TEST_USER_ADMIN_LOGIN, self.REPO, True))

        }

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

    def test_api_lock_repo_lock_acquire_by_non_admin(self):

        repo_name = u'api_delete_me'

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

                            cur_user=self.TEST_USER_LOGIN)

        try:

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

                                      repoid=repo_name,

                                      locked=True)

            response = api_call(self, params)

            expected = {

                'repo': repo_name,

                'locked': True,

                'locked_since': response.json['result']['locked_since'],

                'locked_by': self.TEST_USER_LOGIN,

                'lock_state_changed': True,

                'msg': ('User `%s` set lock state for repo `%s` to `%s`'

                        % (self.TEST_USER_LOGIN, repo_name, True))

            }

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

        finally:

            fixture.destroy_repo(repo_name)

    def test_api_lock_repo_lock_acquire_non_admin_with_userid(self):

        repo_name = u'api_delete_me'

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

                            cur_user=self.TEST_USER_LOGIN)

        try:

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

                                      userid=TEST_USER_ADMIN_LOGIN,

                                      repoid=repo_name,

                                      locked=True)

            response = api_call(self, params)

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

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

        finally:

            fixture.destroy_repo(repo_name)

    def test_api_lock_repo_lock_acquire_non_admin_not_his_repo(self):

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

                                  repoid=self.REPO,

                                  locked=True)

        response = api_call(self, params)

        expected = 'repository `%s` does not exist' % (self.REPO)

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

    def test_api_lock_repo_lock_release(self):

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

                                  userid=TEST_USER_ADMIN_LOGIN,

                                  repoid=self.REPO,

                                  locked=False)

        response = api_call(self, params)

        expected = {

            'repo': self.REPO,

            'locked': False,

            'locked_since': None,

            'locked_by': TEST_USER_ADMIN_LOGIN,

            'lock_state_changed': True,

            'msg': ('User `%s` set lock state for repo `%s` to `%s`'

                    % (TEST_USER_ADMIN_LOGIN, self.REPO, False))

        }

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

    def test_api_lock_repo_lock_acquire_optional_userid(self):

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

                                  repoid=self.REPO,

                                  locked=True)

        response = api_call(self, params)

        time_ = response.json['result']['locked_since']

        expected = {

            'repo': self.REPO,

            'locked': True,

            'locked_since': time_,

            'locked_by': TEST_USER_ADMIN_LOGIN,

            'lock_state_changed': True,

            'msg': ('User `%s` set lock state for repo `%s` to `%s`'

                    % (TEST_USER_ADMIN_LOGIN, self.REPO, True))

        }

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

    def test_api_lock_repo_lock_optional_locked(self):

        try:

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

                                      repoid=self.REPO)

            response = api_call(self, params)

            time_ = response.json['result']['locked_since']

            expected = {

                'repo': self.REPO,

                'locked': True,

                'locked_since': time_,

                'locked_by': TEST_USER_ADMIN_LOGIN,

                'lock_state_changed': False,

                'msg': ('Repo `%s` locked by `%s` on `%s`.'

                        % (self.REPO, TEST_USER_ADMIN_LOGIN,

                           json.dumps(time_to_datetime(time_))))

            }

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

        finally:

            # cleanup

            Repository.unlock(RepoModel().get_by_repo_name(self.REPO))

    def test_api_lock_repo_lock_optional_not_locked(self):

        repo_name = u'api_not_locked'

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

                            cur_user=self.TEST_USER_LOGIN)

        assert repo.locked == [None, None]

        try:

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

                                      repoid=repo.repo_id)

            response = api_call(self, params)

            expected = {

                'repo': repo_name,

                'locked': False,

                'locked_since': None,

                'locked_by': None,

                'lock_state_changed': False,

                'msg': ('Repo `%s` not locked.' % (repo_name,))

            }

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

        finally: