import time

import traceback

import logging

from sqlalchemy import or_

from kallithea import EXTERN_TYPE_INTERNAL

from kallithea.controllers.api import JSONRPCController, JSONRPCError

from kallithea.lib.auth import (

    PasswordGenerator, AuthUser, HasPermissionAnyDecorator,

    HasPermissionAnyDecorator, HasPermissionAnyApi, HasRepoPermissionAnyApi,

    HasRepoGroupPermissionAnyApi, HasUserGroupPermissionAny)

from kallithea.lib.utils import map_groups, repo2db_mapper

from kallithea.lib.utils2 import (

    str2bool, time_to_datetime, safe_int, Optional, OAttr)

from kallithea.model.meta import Session

from kallithea.model.repo_group import RepoGroupModel

from kallithea.model.scm import ScmModel, UserGroupList

from kallithea.model.repo import RepoModel

from kallithea.model.user import UserModel

from kallithea.model.user_group import UserGroupModel

from kallithea.model.gist import GistModel

from kallithea.model.db import (

    Repository, Setting, UserIpMap, Permission, User, Gist,

from kallithea.lib.compat import json

from kallithea.lib.exceptions import (

    DefaultUserException, UserGroupsAssignedException)

log = logging.getLogger(__name__)

def store_update(updates, attr, name):

    """

    Stores param in updates dict if it's not instance of Optional

    allows easy updates of passed in params

    """

    if not isinstance(attr, Optional):

        updates[name] = attr

def get_user_or_error(userid):

    """

    Get user by id or name or return JsonRPCError if not found

    :param userid:

    """

    user = UserModel().get_user(userid)

    if user is None:

        data = user_group.get_api_data()

        return data

    # permission check inside

    def get_user_groups(self, apiuser):

        """

        Lists all existing user groups. This command can be executed only using

        api_key belonging to user with admin rights or user who has at least

        read access to user group.

        :param apiuser: filled automatically from apikey

        :type apiuser: AuthUser

        OUTPUT::

            id : <id_given_in_input>

            result : [<user_group_obj>,...]

            error : null

        """

        result = []

        _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)

        extras = {'user': apiuser}

                                        perm_set=_perms, extra_kwargs=extras):

            result.append(user_group.get_api_data())

        return result

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

    def create_user_group(self, apiuser, group_name, description=Optional(''),

                          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 apiuser: filled automatically from apikey

        :type apiuser: AuthUser

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

        :param active: group is active

        :type active: Optional(bool)

        OUTPUT::

            result: [

                      {

                        "repo_id" :          "<repo_id>",

                        "repo_name" :        "<reponame>"

                        "repo_type" :        "<repo_type>",

                        "clone_uri" :        "<clone_uri>",

                        "private": :         "<bool>",

                        "created_on" :       "<datetimecreated>",

                        "description" :      "<description>",

                        "landing_rev":       "<landing_rev>",

                        "owner":             "<repo_owner>",

                        "fork_of":           "<name_of_fork_parent>",

                        "enable_downloads":  "<bool>",

                        "enable_locking":    "<bool>",

                        "enable_statistics": "<bool>",

                      },

                      …

                    ]

            error:  null

        """

        result = []

        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

            repos = RepoModel().get_all_user_repos(user=apiuser)

        else:

        for repo in repos:

            result.append(repo.get_api_data())

        return result

    # permission check inside

    def get_repo_nodes(self, apiuser, repoid, revision, root_path,

                       ret_type=Optional('all')):

        """

        returns a list of nodes and it's children in a flat list for a given path

        at given revision. It's possible to specify ret_type to show only `files` or

        `dirs`.  This command can be executed only using api_key belonging to

        user with admin rights or regular user that have at least read access to repository.

        :param apiuser: filled automatically from apikey

        :type apiuser: AuthUser

        :param repoid: repository name or repository id

        :type repoid: str or int

        :param revision: revision for which listing should be done

        :type revision: str

        :param root_path: path from which start displaying

        :type root_path: str

        :param ret_type: return type 'all|files|dirs' nodes

        :type ret_type: Optional(str)

        """

        from kallithea.model.db import User

        return self._get_instance(User, user,

                                  callback=User.get_by_username)

    def _get_repo(self, repository):

        """

        Helper method to get repository by ID, or repository name

        :param repository: RepoID, repository name or Repository Instance

        """

        from kallithea.model.db import Repository

        return self._get_instance(Repository, repository,

                                  callback=Repository.get_by_repo_name)

    def _get_perm(self, permission):

        """

        Helper method to get permission by ID, or permission name

        :param permission: PermissionID, permission_name or Permission instance

        """

        from kallithea.model.db import Permission

        return self._get_instance(Permission, permission,

                                  callback=Permission.get_by_key)

                                          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)

    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)

        result = []

            result.append(repo.get_api_data())

        ret = jsonify(result)

        expected = ret

        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)

        result = []

        for repo in RepoModel().get_all_user_repos(self.TEST_USER_LOGIN):

            result.append(repo.get_api_data())

        ret = jsonify(result)

        expected = ret

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

    @parametrize('name,ret_type', [

        ('all', 'all'),

        ('dirs', 'dirs'),

        ('files', 'files'),

    ])

    def test_api_get_repo_nodes(self, name, ret_type):

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

"""

Helpers for fixture generation

"""

import os

import time

from kallithea.tests import *

from kallithea.model.meta import Session

from kallithea.model.repo import RepoModel

from kallithea.model.user import UserModel

from kallithea.model.repo_group import RepoGroupModel

from kallithea.model.user_group import UserGroupModel

from kallithea.model.gist import GistModel

from kallithea.model.scm import ScmModel

from kallithea.lib.vcs.backends.base import EmptyChangeset

from os.path import dirname

FIXTURES = os.path.join(dirname(dirname(os.path.abspath(__file__))), 'tests', 'fixtures')

def error_function(*args, **kwargs):

    raise Exception('Total Crash !')

class Fixture(object):

    def __init__(self):

        pass

    def anon_access(self, status):

        """

    def destroy_user_group(self, usergroupid):

        UserGroupModel().delete(user_group=usergroupid, force=True)

        Session().commit()

    def create_gist(self, **kwargs):

        form_data = {

            'description': u'new-gist',

            'owner': TEST_USER_ADMIN_LOGIN,

            'gist_type': GistModel.cls.GIST_PUBLIC,

            'lifetime': -1,

            'gist_mapping': {'filename1.txt':{'content':'hello world'},}

        }

        form_data.update(kwargs)

        gist = GistModel().create(

            description=form_data['description'],owner=form_data['owner'],

            gist_mapping=form_data['gist_mapping'], gist_type=form_data['gist_type'],

            lifetime=form_data['lifetime']

        )

        Session().commit()

        return gist

    def destroy_gists(self, gistid=None):

            if gistid:

                if gistid == g.gist_access_id:

                    GistModel().delete(g)

            else:

                GistModel().delete(g)

        Session().commit()

    def load_resource(self, resource_name, strip=True):

        with open(os.path.join(FIXTURES, resource_name), 'rb') as f:

            source = f.read()

            if strip:

                source = source.strip()

        return source

    def commit_change(self, repo, filename, content, message, vcs_type, parent=None, newfile=False):

        repo = Repository.get_by_repo_name(repo)

        _cs = parent

        if not parent:

            _cs = EmptyChangeset(alias=vcs_type)

        if newfile:

            nodes = {

                filename: {

from kallithea.tests import *

from kallithea.tests.fixture import Fixture

from kallithea.model.user_group import UserGroupModel

from kallithea.model.meta import Session

fixture = Fixture()

class TestUserGroups(TestController):

    def teardown_method(self, method):

        # delete all groups

            fixture.destroy_user_group(gr)

        Session().commit()

    @parametrize('pre_existing,regular_should_be,external_should_be,groups,expected', [

        ([], [], [], [], []),

        ([], [u'regular'], [], [], [u'regular']),  # no changes of regular

        ([u'some_other'], [], [], [u'some_other'], []),   # not added to regular group

        ([], [u'regular'], [u'container'], [u'container'], [u'regular', u'container']),

        ([], [u'regular'], [], [u'container', u'container2'], [u'regular', u'container', u'container2']),

        ([], [u'regular'], [u'other'], [], [u'regular']),  # remove not used

        ([u'some_other'], [u'regular'], [u'other', u'container'], [u'container', u'container2'], [u'regular', u'container', u'container2']),

    ])

    def test_enforce_groups(self, pre_existing, regular_should_be,

                            external_should_be, groups, expected):

        # delete all groups

            fixture.destroy_user_group(gr)

        Session().commit()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        for gr in pre_existing:

            gr = fixture.create_user_group(gr)

        Session().commit()

        # make sure use is just in those groups

        for gr in regular_should_be:

            gr = fixture.create_user_group(gr)

            Session().commit()

            UserGroupModel().add_user_to_group(gr, user)

            Session().commit()

        # now special external groups created by auth plugins

        for gr in external_should_be:

            gr = fixture.create_user_group(gr, user_group_data={'extern_type': 'container'})

            Session().commit()

            UserGroupModel().add_user_to_group(gr, user)

            Session().commit()

        UserGroupModel().enforce_groups(user, groups, 'container')

        Session().commit()