# -*- 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 re

import string

from typing import Sized

import mock

import pytest

from webtest import TestApp

from kallithea.lib import ext_json

from kallithea.lib.auth import AuthUser

from kallithea.lib.utils2 import ascii_bytes

from kallithea.model import db, meta

from kallithea.model.changeset_status import ChangesetStatusModel

from kallithea.model.gist import GistModel

from kallithea.model.pull_request import PullRequestModel

from kallithea.model.repo import RepoModel

from kallithea.model.repo_group import RepoGroupModel

from kallithea.model.scm import ScmModel

from kallithea.model.user import UserModel

from kallithea.model.user_group import UserGroupModel

from kallithea.tests import base

from kallithea.tests.fixture import Fixture, raise_exception

API_URL = '/_admin/api'

TEST_USER_GROUP = 'test_user_group'

TEST_REPO_GROUP = 'test_repo_group'

fixture = Fixture()

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

    """

    Builds API data with given random ID

    For convenience, the json is returned as str

    """

    random_id = random.randrange(1, 9999)

    return random_id, ext_json.dumps({

        "id": random_id,

        "api_key": apikey,

        "method": method,

        "args": kw

    })

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

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=base.TEST_USER_ADMIN_LOGIN)

    UserGroupModel().add_user_to_group(user_group=gr,

                                       user=base.TEST_USER_ADMIN_LOGIN)

    meta.Session().commit()

    return gr

def make_repo_group(name=TEST_REPO_GROUP):

    gr = fixture.create_repo_group(name, cur_user=base.TEST_USER_ADMIN_LOGIN)

    meta.Session().commit()

    return gr

class _BaseTestApi(object):

    app: TestApp  # assigned by app_fixture in subclass TestController mixin

    # assigned in subclass:

    REPO: str

    REPO_TYPE: str

    TEST_REVISION: str

    TEST_PR_SRC: str

    TEST_PR_DST: str

    TEST_PR_REVISIONS: Sized

    @classmethod

    def setup_class(cls):

        cls.usr = db.User.get_by_username(base.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='first',

            lastname='last'

        )

        meta.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 = ext_json.loads(given)

        assert expected == given, (expected, given)

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

        expected = jsonify({

            'id': id_,

            'error': expected,

            'result': None

        })

        given = ext_json.loads(given)

        assert expected == given, (expected, given)

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

        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 = db.User.query().filter_by(is_default_user=False) \

            .order_by(db.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)

        ret['followers'] = followers

        expected = ret

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

        fixture.destroy_user_group(new_group)

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

                                  with_revision_names=True,

                                  with_pullrequests=True)

        response = api_call(self, params)

        assert "v0.2.0" in response.json['result']['tags']

        assert 'pull_requests' in response.json['result']

    @base.parametrize('grant_perm', [

        ('repository.admin'),

        ('repository.write'),

        ('repository.read'),

    ])

    def test_api_get_repo_by_non_admin(self, grant_perm):

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

                                          user=self.TEST_USER_LOGIN,

                                          perm=grant_perm)

        meta.Session().commit()

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

                                  repoid=self.REPO)

        response = api_call(self, params)

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

        assert len(repo.repo_to_perm) >= 2  # make sure we actually are testing something - probably the default 2 permissions, possibly more

        expected = repo.get_api_data()

        members = []

        for user in repo.repo_to_perm:

            perm = user.permission.permission_name

            user_obj = user.user

            user_data = {'name': user_obj.username, 'type': "user",

                         'permission': perm}

            members.append(user_data)

        for user_group in repo.users_group_to_perm:

            perm = user_group.permission.permission_name

            user_group_obj = user_group.users_group

            user_group_data = {'name': user_group_obj.users_group_name,

                               'type': "user_group", 'permission': perm}

            members.append(user_group_data)

        expected['members'] = members

        followers = []

        for user in repo.followers:

            followers.append(user.user.get_api_data())

        expected['followers'] = followers

        try:

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

        finally:

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

    def test_api_get_repo_by_non_admin_no_permission_to_repo(self):

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

                                          user=db.User.DEFAULT_USER_NAME,

                                          perm='repository.none')

        try:

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

                                              user=self.TEST_USER_LOGIN,

                                              perm='repository.none')

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

                                      repoid=self.REPO)

            response = api_call(self, params)

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

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

        finally:

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

                                              user=db.User.DEFAULT_USER_NAME,

                                              perm='repository.read')

    def test_api_get_repo_that_doesn_not_exist(self):

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

                                  repoid='no-such-repo')

        response = api_call(self, params)

        ret = 'repository `%s` does not exist' % 'no-such-repo'

        expected = ret

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

    def test_api_get_repos(self):

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

        response = api_call(self, params)

        expected = jsonify([

            repo.get_api_data()

            for repo in db.Repository.query()

        ])

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

    def test_api_get_repos_non_admin(self):

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

        response = api_call(self, params)

        expected = jsonify([

            repo.get_api_data()

            for repo in AuthUser(dbuser=db.User.get_by_username(self.TEST_USER_LOGIN)).get_all_user_repos()

        ])

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

    @base.parametrize('name,ret_type', [

        ('all', 'all'),

        ('dirs', 'dirs'),

        ('files', 'files'),

    ])

    def test_api_get_repo_nodes(self, name, ret_type):

        rev = 'tip'

        path = '/'

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

                                  repoid=self.REPO, revision=rev,

                                  root_path=path,

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

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

    def test_api_get_repo_nodes_bad_revisions(self):

        rev = 'i-dont-exist'

        path = '/'

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

                                  repoid=self.REPO, revision=rev,

                                  root_path=path, )

        response = api_call(self, params)

        expected = 'failed to get repo: `%s` nodes' % self.REPO

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

    def test_api_get_repo_nodes_bad_path(self):

        rev = 'tip'

        path = '/idontexits'

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

                                  repoid=self.REPO, revision=rev,

                                  root_path=path, )

        response = api_call(self, params)

        expected = 'failed to get repo: `%s` nodes' % self.REPO

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

    def test_api_get_repo_nodes_bad_ret_type(self):

        rev = 'tip'

        path = '/'

        ret_type = 'error'

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

                                  repoid=self.REPO, revision=rev,

                                  root_path=path,

                                  ret_type=ret_type)

        response = api_call(self, params)

        expected = ('ret_type must be one of %s'

                    % (','.join(sorted(['files', 'dirs', 'all']))))

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

    @base.parametrize('name,ret_type,grant_perm', [

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

        ('dirs', 'dirs', 'repository.admin'),

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

    ])

    def test_api_get_repo_nodes_by_regular_user(self, name, ret_type, grant_perm):

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

                                          user=self.TEST_USER_LOGIN,

                                          perm=grant_perm)

        meta.Session().commit()

        rev = 'tip'

        path = '/'

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

                                  repoid=self.REPO, revision=rev,

                                  root_path=path,

                                  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)

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

        response = api_call(self, params)

    @base.parametrize('repo_name', [

        '',

        '.',

        '..',

        ':',

        '/',

        '<test>',

    ])

    def test_api_create_repo_bad_names(self, repo_name):

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

                                  repo_name=repo_name,

                                  owner=base.TEST_USER_ADMIN_LOGIN,

                                  repo_type=self.REPO_TYPE,

        )

        response = api_call(self, params)

        if repo_name == '/':

            expected = "repo group `` not found"

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

        else:

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

        # cloning from local repos was a misfeature - it would bypass access control

        # TODO: introduce other test coverage of actual remote cloning

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

        repo_name = 'api-repo'

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

                                  repo_name=repo_name,

                                  owner=base.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):

        repo_group_name = 'my_gr'

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

        # repo creation can no longer also create repo group

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

                                  repo_name=repo_name,

                                  owner=base.TEST_USER_ADMIN_LOGIN,

                                  repo_type=self.REPO_TYPE,)

        response = api_call(self, params)

        expected = 'repo group `%s` not found' % repo_group_name

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

        assert RepoModel().get_by_repo_name(repo_name) is None

        # create group before creating repo

        rg = fixture.create_repo_group(repo_group_name)

        meta.Session().commit()

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

                                  repo_name=repo_name,

                                  owner=base.TEST_USER_ADMIN_LOGIN,

                                  repo_type=self.REPO_TYPE,)

        response = api_call(self, params)

        expected = {

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

            'success': True,

        }

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

        repo = RepoModel().get_by_repo_name(repo_name)

        assert repo is not None

        fixture.destroy_repo(repo_name)

        fixture.destroy_repo_group(repo_group_name)

    def test_api_create_repo_in_repo_group_without_permission(self):

        repo_group_basename = 'api-repo-repo'

        repo_group_name = '%s/%s' % (TEST_REPO_GROUP, repo_group_basename)

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

        top_group = db.RepoGroup.get_by_group_name(TEST_REPO_GROUP)

        assert top_group

        rg = fixture.create_repo_group(repo_group_basename, parent_group_id=top_group)

        meta.Session().commit()

        RepoGroupModel().grant_user_permission(repo_group_name,

                                               self.TEST_USER_LOGIN,

                                               'group.none')

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

        # API access control match Web access control:

        expected = 'no permission to create repo in test_repo_group/api-repo-repo'

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

        fixture.destroy_repo(repo_name)

        fixture.destroy_repo_group(repo_group_name)

    def test_api_create_repo_unknown_owner(self):

        repo_name = '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 = 'api-repo'

        owner = 'i-dont-exist'

        id_, params = _build_data(self.apikey, '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,

        }

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

        }

        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 = '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=base.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)

    def test_api_create_repo_dot_dot(self):

        # it is only possible to create repositories in existing repo groups - and '..' can't be used

        group_name = '%s/..' % TEST_REPO_GROUP

        repo_name = '%s/%s' % (group_name, 'could-be-outside')

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

                                  repo_name=repo_name,

                                  owner=base.TEST_USER_ADMIN_LOGIN,

                                  repo_type=self.REPO_TYPE,)

        response = api_call(self, params)

        expected = 'repo group `%s` not found' % group_name

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

        fixture.destroy_repo(repo_name)

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

    def test_api_create_repo_exception_occurred(self):

        repo_name = 'api-repo'

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

                                  repo_name=repo_name,