error:  null

        """

        return [

            user.get_api_data()

            for user in User.query()

                .order_by(User.username)

                .filter_by(is_default_user=False)

        ]

    @HasPermissionAnyDecorator('hg.admin')

    def create_user(self, username, email, password=Optional(''),

                    active=Optional(True), admin=Optional(False),

                    extern_type=Optional(User.DEFAULT_AUTH_TYPE),

                    extern_name=Optional('')):

        """

        Creates new user. Returns new user object. This command can

        be executed only using api_key belonging to user with admin rights.

        :param username: new username

        :type username: str or int

        :param email: email

        :type email: str

        :param password: password

            id : <id_given_in_input>

            result : [<user_group_obj>,...]

            error : null

        """

        return [

            user_group.get_api_data()

            for user_group in UserGroupList(UserGroup.query().all(), perm_level='read')

        ]

    @HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')

                          owner=Optional(OAttr('apiuser')), active=Optional(True)):

        """

        Creates new user group. This command can be executed only using api_key

        belonging to user with admin rights or an user who has create user group

        permission

        :param group_name: name of new user group

        :type group_name: str

        :param description: group description

        :type description: str

        :param owner: owner of group. If not passed apiuser is the owner

        :type owner: Optional(str or int)

    def get_pullrequest(self, pullrequest_id):

        """

        Get given pull request by id

        """

        pull_request = PullRequest.get(pullrequest_id)

        if pull_request is None:

            raise JSONRPCError('pull request `%s` does not exist' % (pullrequest_id,))

        if not HasRepoPermissionLevel('read')(pull_request.org_repo.repo_name):

            raise JSONRPCError('not allowed')

        return pull_request.get_api_data()

    # permission check inside

        """

        Add comment, close and change status of pull request.

        """

        apiuser = get_user_or_error(request.authuser.user_id)

        pull_request = PullRequest.get(pull_request_id)

        if pull_request is None:

            raise JSONRPCError('pull request `%s` does not exist' % (pull_request_id,))

        if (not HasRepoPermissionLevel('read')(pull_request.org_repo.repo_name)):

            raise JSONRPCError('No permission to add comment. User needs at least reading permissions'

                               ' to the source repository.')

        owner = apiuser.user_id == pull_request.owner_id

        reviewer = apiuser.user_id in [reviewer.user_id for reviewer in pull_request.reviewers]

                                       type_=Notification.TYPE_PULL_REQUEST,

                                       email_kwargs=email_kwargs)

    def mention_from_description(self, user, pr, old_description=''):

        mention_recipients = (extract_mentioned_users(pr.description) -

                              extract_mentioned_users(old_description))

        log.debug("Mentioning %s", mention_recipients)

        self.add_reviewers(user, pr, set(), mention_recipients)

    def remove_reviewers(self, user, pull_request, reviewers):

        """Remove specified users from being reviewers of the PR."""

        PullRequestReviewer.query() \

            .filter_by(pull_request=pull_request) \

            .filter(PullRequestReviewer.user_id.in_(r.user_id for r in reviewers)) \

            .delete(synchronize_session='fetch') # the default of 'evaluate' is not available

    def delete(self, pull_request):

        pull_request = PullRequest.guess_instance(pull_request)

        Session().delete(pull_request)

        if pull_request.org_repo.scm_instance.alias == 'git':

            # remove a ref under refs/pull/ so that commits can be garbage-collected

            try:

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

        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,

            "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.000"} 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\.\d\d\d",

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

    def test_api_close_pullrequest(self):

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

        })

        response = api_call(self, params)

        self._compare_ok(random_id, True, given=response.body)

        pullrequest = PullRequest().get(pull_request_id)

        assert pullrequest.comments[-1].text == ''

        assert pullrequest.status == PullRequest.STATUS_CLOSED

        assert pullrequest.is_closed() == True

    def test_api_status_pullrequest(self):

        random_id = random.randrange(1, 9999)

        params = json.dumps({

            "id": random_id,

            "api_key": User.get_by_username(TEST_USER_REGULAR2_LOGIN).api_key,

            "method": "comment_pullrequest",

            "args": {"pull_request_id": pull_request_id, "status": ChangesetStatus.STATUS_APPROVED},

        })

        response = api_call(self, params)

        pullrequest = PullRequest().get(pull_request_id)

        self._compare_error(random_id, "No permission to change pull request status. User needs to be admin, owner or reviewer.", given=response.body)

        assert ChangesetStatus.STATUS_UNDER_REVIEW == ChangesetStatusModel().calculate_pull_request_result(pullrequest)[2]

        params = json.dumps({

            "id": random_id,

            "api_key": User.get_by_username(TEST_USER_REGULAR_LOGIN).api_key,

            "method": "comment_pullrequest",

            "args": {"pull_request_id": pull_request_id, "status": ChangesetStatus.STATUS_APPROVED},

        })

        response = api_call(self, params)

        self._compare_ok(random_id, True, given=response.body)

        pullrequest = PullRequest().get(pull_request_id)

        assert ChangesetStatus.STATUS_APPROVED == ChangesetStatusModel().calculate_pull_request_result(pullrequest)[2]

    def test_api_comment_pullrequest(self):

        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, "comment_msg": "Looks good to me"},

        })

        response = api_call(self, params)

        self._compare_ok(random_id, True, given=response.body)

        pullrequest = PullRequest().get(pull_request_id)

        assert pullrequest.comments[-1].text == u'Looks good to me'

                message=message,

                content=content,

                f_path=filename

            )

        return cs

    def review_changeset(self, repo, revision, status, author=TEST_USER_ADMIN_LOGIN):

        comment = ChangesetCommentsModel().create(u"review comment", repo, author, revision=revision, send_email=False)

        csm = ChangesetStatusModel().set_status(repo, ChangesetStatus.STATUS_APPROVED, author, comment, revision=revision)

        Session().commit()

        return csm

        org_ref = 'branch:stable:%s' % pr_src_rev

        other_ref = 'branch:default:%s' % pr_dst_rev

        with test_context(testcontroller.app): # needed to be able to mock request user

            org_repo = other_repo = Repository.get_by_repo_name(repo_name)

            owner_user = User.get_by_username(TEST_USER_ADMIN_LOGIN)

            reviewers = [User.get_by_username(TEST_USER_REGULAR_LOGIN)]

            request.authuser = request.user = AuthUser(dbuser=owner_user)

            # creating a PR sends a message with an absolute URL - without routing that requires mocking

            with mock.patch.object(helpers, 'url', (lambda arg, qualified=False, **kwargs: ('https://localhost' if qualified else '') + '/fake/' + arg)):

                pull_request = cmd.execute()

            Session().commit()

        return pull_request.pull_request_id

#==============================================================================

# Global test environment setup

#==============================================================================

def create_test_env(repos_test_path, config):

    """

    Makes a fresh database and

from kallithea.model.meta import Session

from kallithea.model.repo_group import RepoGroupModel

from kallithea.tests.base import TestController, url

from kallithea.tests.fixture import Fixture

fixture = Fixture()

class TestRepoGroupsController(TestController):

    def test_case_insensitivity(self):

        self.log_user()

        response = self.app.post(url('repos_groups'),

                                 fixture._get_repo_group_create_params(group_name=group_name,

                                                                 _authentication_token=self.authentication_token()))

        # try to create repo group with swapped case

        swapped_group_name = group_name.swapcase()

        response = self.app.post(url('repos_groups'),

                                 fixture._get_repo_group_create_params(group_name=swapped_group_name,

                                                                 _authentication_token=self.authentication_token()))

        response.mustcontain('already exists')

        RepoGroupModel().delete(group_name)

        Session().commit()

            email=u'a1@example.com', firstname=u'a1', lastname=u'a1', admin=True

        )

        Session().commit()

    def teardown_method(self, method):

        if hasattr(self, 'test_repo'):

            RepoModel().delete(repo=self.test_repo)

        UserModel().delete(self.u1)

        UserModel().delete(self.u2)

        UserModel().delete(self.u3)

        UserModel().delete(self.a1)

        if hasattr(self, 'g1'):

            RepoGroupModel().delete(self.g1.group_id)

        if hasattr(self, 'g2'):

            RepoGroupModel().delete(self.g2.group_id)

        if hasattr(self, 'ug2'):

            UserGroupModel().delete(self.ug2, force=True)

        if hasattr(self, 'ug1'):

            UserGroupModel().delete(self.ug1, force=True)

        Session().commit()

        ('auth_ldap_enabled', True, 'bool'),

        ('auth_ldap_attr_firstname', 'test_ldap_firstname'),

        ('auth_ldap_attr_lastname', 'test_ldap_lastname'),

        ('auth_ldap_attr_email', 'test_ldap_email'))

class _AuthLdapMock():

    def __init__(self, **kwargs):

        pass

    def authenticate_ldap(self, username, password):

                               test_ldap_email=['spam ldap email'])

def test_update_user_attributes_from_ldap(monkeypatch, create_test_user,

                                          arrange_ldap_auth):

    """Authenticate user with mocked LDAP, verify attributes are updated.

    """

    # Arrange test user.

    uniqifier = uuid.uuid4()

    username = 'test-user-{0}'.format(uniqifier)

    assert User.get_by_username(username) is None

    user_input = dict(username='test-user-{0}'.format(uniqifier),

                      password='spam password',

                      email='spam-email-{0}'.format(uniqifier),

                      active=True,

                      admin=False)

    user = create_test_user(user_input)

    # Arrange LDAP auth.

    monkeypatch.setattr(auth_ldap, 'AuthLdap', _AuthLdapMock)

    # Authenticate with LDAP.

    user_data = authenticate(username, 'password')

    # Verify that authenication succeeded and retrieved correct attributes

    # from LDAP.

    assert user_data is not None

    assert user_data.get('email') == 'spam ldap email'

    # Verify that authentication overwrote user attributes with the ones

    # retrieved from LDAP.

    assert user.email == 'spam ldap email'

def test_init_user_attributes_from_ldap(monkeypatch, arrange_ldap_auth):

    """Authenticate unknown user with mocked LDAP, verify user is created.

    """

    # Arrange test user.

    uniqifier = uuid.uuid4()

    username = 'test-user-{0}'.format(uniqifier)

    assert User.get_by_username(username) is None

    # Arrange LDAP auth.

    monkeypatch.setattr(auth_ldap, 'AuthLdap', _AuthLdapMock)

    # Authenticate with LDAP.

    user_data = authenticate(username, 'password')

    # Verify that authenication succeeded and retrieved correct attributes

    # from LDAP.

    assert user_data is not None

    assert user_data.get('email') == 'spam ldap email'

    # Verify that authentication created new user with attributes

    # retrieved from LDAP.

    new_user = User.get_by_username(username)

    assert new_user is not None

    assert new_user.email == 'spam ldap email'

class _AuthLdapNoEmailMock():

    def __init__(self, **kwargs):

        pass

    def authenticate_ldap(self, username, password):

        return 'spam dn', dict(test_ldap_firstname=['spam ldap first name'],

                               test_ldap_lastname=['spam ldap last name'],

                               test_ldap_email=[''])

    username = 'test-user-{0}'.format(uniqifier)

    assert User.get_by_username(username) is None

    # Arrange LDAP auth.

    monkeypatch.setattr(auth_ldap, 'AuthLdap', _AuthLdapNoEmailMock)

    # Authenticate with LDAP.

    user_data = authenticate(username, 'password')

    # Verify that authenication succeeded and retrieved correct attributes

    # from LDAP, with empty email.

    assert user_data is not None

    assert user_data.get('email') == ''

    # Verify that authentication created new user with attributes

    # retrieved from LDAP, with email == None.

    new_user = User.get_by_username(username)

    assert new_user is not None

    assert new_user.email is None