_revset_predicates = {

                'branch': 'branch',

                'book': 'bookmark',

                'tag': 'tag',

                'rev': 'id',

            }

        # avoid expensive branch(x) iteration over whole repo

        rev_spec = "%%s & %s(%%s)" % _revset_predicates[ref_type]

        try:

            revs = self._repo.revs(rev_spec, ref_name, ref_name)

        except LookupError:

            msg = ("Ambiguous identifier %s:%s for %s" % (ref_type, ref_name, self.name))

            raise ChangesetDoesNotExistError(msg)

        except RepoLookupError:

            msg = ("Revision %s:%s does not exist for %s" % (ref_type, ref_name, self.name))

            raise ChangesetDoesNotExistError(msg)

        if revs:

            try:

                revision = revs.last()

            except AttributeError:

                # removed in hg 3.2

                revision = revs[-1]

        else:

            # TODO: just report 'not found'?

            revision = ref_name

        return self._get_revision(revision)

    def _get_archives(self, archive_name='tip'):

        allowed = self.baseui.configlist("web", "allow_archive",

                                         untrusted=True)

        for i in [('zip', '.zip'), ('gz', '.tar.gz'), ('bz2', '.tar.bz2')]:

            if i[0] in allowed or self._repo.ui.configbool("web",

                                                           "allow" + i[0],

                                                           untrusted=True):

                yield {"type": i[0], "extension": i[1], "node": archive_name}

    def _get_url(self, url):

        """

        Returns normalized url. If schema is not given, would fall

        to filesystem

        (``file:///``) schema.

        """

        url = safe_str(url)

        if url != 'default' and '://' not in url:

            url = "file:" + urllib.pathname2url(url)

        return url

    def get_hook_location(self):

        """

        returns absolute path to location where hooks are stored

        """

        return os.path.join(self.path, '.hg', '.hgrc')

    def get_changeset(self, revision=None):

        """

        Returns ``MercurialChangeset`` object representing repository's

        changeset at the given ``revision``.

        """

        revision = self._get_revision(revision)

        changeset = MercurialChangeset(repository=self, revision=revision)

        return changeset

    def get_changesets(self, start=None, end=None, start_date=None,

                       end_date=None, branch_name=None, reverse=False):

        """

        Returns iterator of ``MercurialChangeset`` objects from start to end

        (both are inclusive)

        :param start: None, str, int or mercurial lookup format

        :param end:  None, str, int or mercurial lookup format

        :param start_date:

        :param end_date:

        :param branch_name:

        :param reversed: return changesets in reversed order

        """

        start_raw_id = self._get_revision(start)

        start_pos = self.revisions.index(start_raw_id) if start else None

        end_raw_id = self._get_revision(end)

        end_pos = self.revisions.index(end_raw_id) if end else None

        if None not in [start, end] and start_pos > end_pos:

            raise RepositoryError("Start revision '%s' cannot be "

                                  "after end revision '%s'" % (start, end))

        if branch_name and branch_name not in self.allbranches.keys():

            msg = ("Branch %s not found in %s" % (branch_name, self))

            raise BranchDoesNotExistError(msg)

        if end_pos is not None:

            end_pos += 1

        # filter branches

        filter_ = []

        if branch_name:

            filter_.append('branch("%s")' % (branch_name))

            filter_.append('date(">%s")' % start_date)

            filter_.append('date("<%s")' % end_date)

        if filter_:

        else:

            revisions = self.revisions

        # this is very much a hack to turn this into a list; a better solution

        # would be to get rid of this function entirely and use revsets

        revs = list(revisions)[start_pos:end_pos]

        if reverse:

            revs = reversed(revs)

        return CollectionGenerator(self, revs)

    def pull(self, url):

        """

        Tries to pull changes from external location.

        """

        url = self._get_url(url)

        other = peer(self._repo, {}, url)

        try:

            # hg 3.2 moved push / pull to exchange module

            from mercurial import exchange

            exchange.pull(self._repo, other, heads=None, force=None)

        except ImportError:

            self._repo.pull(other, heads=None, force=None)

        except Abort as err:

            # Propagate error but with vcs's type

            raise RepositoryError(str(err))

    @LazyProperty

    def workdir(self):

        """

        Returns ``Workdir`` instance for this repository.

        """

        return MercurialWorkdir(self)

    def get_config_value(self, section, name=None, config_file=None):

        """

        Returns configuration value for a given [``section``] and ``name``.

        :param section: Section we want to retrieve value from

        :param name: Name of configuration we want to retrieve

        :param config_file: A path to file which should be used to retrieve

          configuration from (might also be a list of file paths)

        """

        if config_file is None:

            config_file = []

        elif isinstance(config_file, basestring):

            config_file = [config_file]

        config = self._repo.ui

        for path in config_file:

            config.readconfig(path)

        return config.config(section, name)

    def get_user_name(self, config_file=None):

        """

        Returns user's name from global configuration file.

        :param config_file: A path to file which should be used to retrieve

          configuration from (might also be a list of file paths)

        """

        username = self.get_config_value('ui', 'username')

        if username:

            return author_name(username)

        return None

    def get_user_email(self, config_file=None):

        """

        Returns user's email from global configuration file.

        :param config_file: A path to file which should be used to retrieve

          configuration from (might also be a list of file paths)

        """

        username = self.get_config_value('ui', 'username')

        if username:

            return author_email(username)

        return None

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

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

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

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

    def test_api_get_gists_regular_user_with_different_userid(self):

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

                                  userid=TEST_USER_ADMIN_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_create_gist(self):

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

                                  lifetime=10,

                                  description='foobar-gist',

                                  gist_type='public',

                                  files={'foobar': {'content': 'foo'}})

        response = api_call(self, params)

        response_json = response.json

        expected = {

            'gist': {

                'access_id': response_json['result']['gist']['access_id'],

                'created_on': response_json['result']['gist']['created_on'],

                'description': 'foobar-gist',

                'expires': response_json['result']['gist']['expires'],

                'gist_id': response_json['result']['gist']['gist_id'],

                'type': 'public',

                'url': response_json['result']['gist']['url']

            },

            'msg': 'created new gist'

        }

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

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

    def test_api_create_gist_exception_occurred(self):

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

                                  files={})

        response = api_call(self, params)

        expected = 'failed to create gist'

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

    def test_api_delete_gist(self):

        gist_id = fixture.create_gist().gist_access_id

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

                                  gistid=gist_id)

        response = api_call(self, params)

        expected = {'gist': None, 'msg': 'deleted gist ID:%s' % gist_id}

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

    def test_api_delete_gist_regular_user(self):

        gist_id = fixture.create_gist(owner=self.TEST_USER_LOGIN).gist_access_id

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

                                  gistid=gist_id)

        response = api_call(self, params)

        expected = {'gist': None, 'msg': 'deleted gist ID:%s' % gist_id}

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

    def test_api_delete_gist_regular_user_no_permission(self):

        gist_id = fixture.create_gist().gist_access_id

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

                                  gistid=gist_id)

        response = api_call(self, params)

        expected = 'gist `%s` does not exist' % (gist_id,)

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

    @mock.patch.object(GistModel, 'delete', crash)

    def test_api_delete_gist_exception_occurred(self):

        gist_id = fixture.create_gist().gist_access_id

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

                                  gistid=gist_id)

        response = api_call(self, params)

        expected = 'failed to delete gist ID:%s' % (gist_id,)

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

    def test_api_get_ip(self):

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

        response = api_call(self, params)

        expected = {

            'server_ip_addr': '0.0.0.0',

            'user_ips': []

        }

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

    def test_api_get_server_info(self):

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

        response = api_call(self, params)

        expected = Setting.get_server_info()

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

    def test_api_get_changesets(self):

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

                                  repoid=self.REPO, start=0, end=2)

        response = api_call(self, params)

        result = json.loads(response.body)["result"]

        assert len(result) == 3

        assert 'message' in result[0]

        assert 'added' not in result[0]

    def test_api_get_changesets_with_file_list(self):

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

                                  repoid=self.REPO, start_date="2010-04-07T23:30:30", end_date="2010-04-08T00:31:14", with_file_list=True)

        response = api_call(self, params)

        result = json.loads(response.body)["result"]

        assert len(result) == 3

        assert 'message' in result[0]

        assert 'added' in result[0]

    def test_api_get_changeset(self):

        review = fixture.review_changeset(self.REPO, self.TEST_REVISION, "approved")

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

                                  repoid=self.REPO, raw_id=self.TEST_REVISION)

        response = api_call(self, params)

        result = json.loads(response.body)["result"]

        assert result["raw_id"] == self.TEST_REVISION

        assert "reviews" not in result

    def test_api_get_changeset_with_reviews(self):

        reviewobjs = fixture.review_changeset(self.REPO, self.TEST_REVISION, "approved")

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

                                  repoid=self.REPO, raw_id=self.TEST_REVISION,

                                  with_reviews=True)

        response = api_call(self, params)

        result = json.loads(response.body)["result"]

        assert result["raw_id"] == self.TEST_REVISION

        assert "reviews" in result

        assert len(result["reviews"]) == 1

        review = result["reviews"][0]

        expected = {

            'status': 'approved',

            'modified_at': reviewobjs[0].modified_at.replace(microsecond=0).isoformat(),

            'reviewer': 'test_admin',

        }

        assert review == expected

    def test_api_get_changeset_that_does_not_exist(self):

        """ Fetch changeset status for non-existant changeset.

        revision id is the above git hash used in the test above with the

        last 3 nibbles replaced with 0xf.  Should not exist for git _or_ hg.

        """

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

                                  repoid=self.REPO, raw_id = '7ab37bc680b4aa72c34d07b230c866c28e9fcfff')

        response = api_call(self, params)

        expected = u'Changeset %s does not exist' % ('7ab37bc680b4aa72c34d07b230c866c28e9fcfff',)

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

    def test_api_get_changeset_without_permission(self):

        review = fixture.review_changeset(self.REPO, self.TEST_REVISION, "approved")

        RepoModel().revoke_user_permission(repo=self.REPO, user=self.TEST_USER_LOGIN)

        RepoModel().revoke_user_permission(repo=self.REPO, user="default")

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

                                  repoid=self.REPO, raw_id=self.TEST_REVISION)

        response = api_call(self, params)

        expected = u'Access denied to repo %s' % self.REPO

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

    def test_api_get_pullrequest(self):

        pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, u'get test')

        random_id = random.randrange(1, 9999)

        params = json.dumps({

            "id": random_id,

            "api_key": self.apikey,

            "method": 'get_pullrequest',

            "args": {"pullrequest_id": pull_request_id},

        })

        response = api_call(self, params)

        pullrequest = PullRequest().get(pull_request_id)

        expected = {

            "status": "new",

            "pull_request_id": pull_request_id,

            "description": "No description",

            "url": "/%s/pull-request/%s/_/%s" % (self.REPO, pull_request_id, "stable"),

            "reviewers": [{"username": "test_regular"}],

            "org_repo_url": "http://localhost:80/%s" % self.REPO,

            "org_ref_parts": ["branch", "stable", self.TEST_PR_SRC],

            "other_ref_parts": ["branch", "default", self.TEST_PR_DST],

            "comments": [{"username": TEST_USER_ADMIN_LOGIN, "text": "",

                         "comment_id": pullrequest.comments[0].comment_id}],

            "owner": TEST_USER_ADMIN_LOGIN,

            "statuses": [{"status": "under_review", "reviewer": TEST_USER_ADMIN_LOGIN, "modified_at": "2000-01-01T00:00:00"} for i in range(0, len(self.TEST_PR_REVISIONS))],

            "title": "get test",

            "revisions": self.TEST_PR_REVISIONS,

        }

        self._compare_ok(random_id, expected,

                         given=re.sub("\d\d\d\d\-\d\d\-\d\dT\d\d\:\d\d\:\d\d",

                                      "2000-01-01T00:00:00", response.body))

    def test_api_close_pullrequest(self):

        pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, u'close test')

        random_id = random.randrange(1, 9999)

        params = json.dumps({

            "id": random_id,

            "api_key": self.apikey,

            "method": "comment_pullrequest",

            "args": {"pull_request_id": pull_request_id, "close_pr": True},