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

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

    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)