master_doc = 'index'

# General information about the project.

# The version info for the project you're documenting, acts as replacement for

# |version| and |release|, also used in various other places throughout the

# Grouping the document tree into LaTeX files. List of tuples

# (source start file, target name, title, author, documentclass [howto/manual]).

latex_documents = [

]

# The name of an image file (relative to this directory) to place at the top of

# One entry per manual page. List of tuples

# (source start file, name, description, authors, manual section).

man_pages = [

]

    @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('')):

        ]

    @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

        return pull_request.get_api_data()

    # permission check inside

        """

        Add comment, close and change status of pull request.

        """

                if c.repo_name:

                    # use "repository_rawname:" instead of "repository:"

                    # for case-sensitive matching

                try:

                    query = qp.parse(cur_query)

                    # extract words for highlight

        ImportError -- if we couldn't import the plugin at all

    """

    log.debug("Importing %s", plugin)

        if len(parts) == 2:

            _module, pn = parts

    PLUGIN_CLASS_NAME = "KallitheaAuthPlugin"

    try:

        module = importlib.import_module(plugin)

    def create_user(self, username, password, email='', admin=False):

        log.info('creating user %s', username)

        UserModel().create_or_update(username, password, email,

                                     active=True, admin=admin,

                                     extern_type=User.DEFAULT_AUTH_TYPE)

        user = UserModel().create_or_update(username=User.DEFAULT_USER,

                                            password=str(uuid.uuid1())[:20],

                                            email='anonymous@kallithea-scm.org',

        # based on configuration options activate/deactivate this user which

        # controls anonymous access

        if self.cli_args.get('public_access') is False:

    if date is not None:

        tag = re.sub('/', ',%s:/' % date.strftime('%Y-%m-%d'), tag, 1)

    tag = re.sub('#', '/', tag)

class Attributes(object):

        return res

    def get_short_content(self, res, chunks):

    def get_chunks(self):

        """

            bytes_content = node.content

            if b'\0' in bytes_content:

                log.warning('    >> %s - no text content', path)

            else:

                log.debug('    >> %s', path)

                u_content = safe_str(bytes_content)

        else:

            log.debug('    >> %s - not indexable', path)

            # just index file name without it's content

            indexed += 1

        writer.add_document(

                author=cs.author,

                message=cs.message,

                last=cs.last,

            )

        return indexed

                        continue

                    qp = QueryParser('repository', schema=CHGSETS_SCHEMA)

                    results = searcher.search(q)

                        # delete the docs in the index for the previous

                        # last changeset(s)

                        for hit in results:

                                            (repo_name, hit['raw_id']))

                            writer.delete_by_query(q)

                    log.debug('>> NOTHING TO COMMIT TO CHANGESET INDEX<<')

    def update_file_index(self):

                  'AND REPOS %s', INDEX_EXTENSIONS, ' and '.join(self.repo_paths))

        idx = open_dir(self.index_location, indexname=self.indexname)

    Note: UTF-8 is preferred, but for example ISO-8859-1 or mbcs should also

    work under the right circumstances."""

    try:

    except UnicodeEncodeError:

        log.error("Cannot encode Unicode paths to file system encoding %r", sys.getfilesystemencoding())

        for var in ['LC_ALL', 'LC_CTYPE', 'LANG']:

            log.error(traceback.format_exc())

            raise exc.HTTPExpectationFailed()

            # updating refs manually after each push.

            # Needed for pre-1.7.0.4 git clients using regular HTTP mode.

            from kallithea.lib.vcs import get_repo

        repo_obj = Repository.get_by_repo_name(repo_name)

    else:

        repo_obj = None

    user_log = UserLog()

    user_log.user_id = user_obj.user_id

    rgm = RepoGroupModel()

    owner = User.get_first_admin()

    for lvl, group_name in enumerate(groups):

        group = RepoGroup.get_by_group_name(group_name)

        desc = '%s group' % group_name

    # Format the result

    fmt_funcs = {

    }

    for i, part in enumerate(order):

        return self

    def get_file_content(self, path):

    def get_file_size(self, path):

        return 0

    @LazyProperty

    def contact(self):

        return undefined_contact

    @property

class NodeState:

class NodeGeneratorBase(object):

        then only last part is returned.

        """

        org = self.path.rstrip('/').rsplit('/', 1)[-1]

    DEFAULT_CLONE_URI = '{scheme}://{user}@{netloc}/{repo}'

    DEFAULT_CLONE_SSH = 'ssh://{system_user}@{hostname}/{repo}'

    repo_id = Column(Integer(), primary_key=True)

    repo_name = Column(Unicode(255), nullable=False, unique=True)

        """Return tuple with group_id and name as html literal"""

        from webhelpers2.html import literal

        if repo_group is None:

        return repo_group.group_id, literal(cls.SEP.join(repo_group.full_path_splitted))

    @classmethod

    )

    # values for .status

    pull_request_id = Column(Integer(), primary_key=True)

    title = Column(Unicode(255), nullable=False)

        _table_args_default_dict,

    )

    gist_id = Column(Integer(), primary_key=True)

    gist_access_id = Column(Unicode(250), nullable=False)

        filename = All(v.BasePath()(),

                       v.UnicodeString(strip=True, required=False))

        lifetime = v.OneOf(lifetime_options)

        mimetype = v.UnicodeString(required=False, if_missing=None)

        content = v.UnicodeString(required=True, not_empty=True)

    return _GistForm

    rnd = random.SystemRandom() # use cryptographically secure system PRNG

    alphabet = '23456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghjklmnpqrstuvwxyz'

    length = 20

class GistModel(object):

class NotificationModel(object):

    def create(self, created_by, subject, body, recipients=None,

               type_=TYPE_MESSAGE, with_email=True,

        from kallithea.model.changeset_status import ChangesetStatusModel

        from kallithea.model.comment import ChangesetCommentsModel

        comment = ChangesetCommentsModel().create(

            repo=self.org_repo,

            author=created_by,

            pull_request=pr,

                cur_repo.owner = User.get_by_username(kwargs['owner'])

            if 'repo_group' in kwargs:

                cur_repo.group = RepoGroup.get(kwargs['repo_group'])

                cur_repo.repo_name = cur_repo.get_new_name(cur_repo.just_name)

            log.debug('Updating repo %s with params:%s', cur_repo, kwargs)

                repo_group.parent_group_id = repo_group_args['parent_group_id']

            if 'parent_group_id' in repo_group_args:

                repo_group.parent_group = RepoGroup.get(repo_group_args['parent_group_id'])

            if 'group_name' in repo_group_args:

                group_name = repo_group_args['group_name']

        repo = repo.scm_instance

                           repo.branches.items()], _("Branches"))

        hist_l.append(branches_group)

        choices.extend([x[0] for x in branches_group[0]])

        if repo.alias == 'hg':

                                repo.bookmarks.items()], _("Bookmarks"))

            hist_l.append(bookmarks_group)

            choices.extend([x[0] for x in bookmarks_group[0]])

                       repo.tags.items()], _("Tags"))

        hist_l.append(tags_group)

        choices.extend([x[0] for x in tags_group[0]])

        log_create_user(new_user.get_dict(), cur_user)

        return new_user

                         extern_type=None, extern_name=None, cur_user=None):

        """

        Creates a new instance if not found, or updates current one

        # notification to admins

        subject = _('New user registration')

        body = (

            '---------------------\n'

            '- Username: {user.username}\n'

            '- Full Name: {user.full_name}\n'

        """

        app_secret = config.get('app_instance_uuid')

        return hmac.HMAC(

            digestmod=hashlib.sha1,

        ).hexdigest()

        for gr in set(groups):

            existing_group = UserGroup.get_by_group_name(gr)

            if not existing_group:

                # we use first admin account to set the owner of the group

                existing_group = UserGroupModel().create(gr, desc, owner,

                                        group_data={'extern_type': extern_type})

def ValidAuth():

    class _validator(formencode.validators.FancyValidator):

        messages = {

        }

        def _validate_python(self, value, state):

API_URL = '/_admin/api'

fixture = Fixture()

            username='test-api',

            password='test',

            email='test@example.com',

        )

        Session().commit()

        cls.TEST_USER_LOGIN = cls.test_user.username

    def test_api_pull_remote(self):

        # Note: pulling from local repos is a mis-feature - it will bypass access control

        # ... but ok, if the path already has been set in the database

        r = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)

        # hack around that clone_uri can't be set to to a local path

        # (as shown by test_api_create_repo_clone_uri_local)

        assert pre_cached_tip != post_cached_tip

    def test_api_pull_fork(self):

        fixture.create_fork(self.REPO, fork_name)

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

                                  repoid=fork_name,)

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

    def test_api_pull_custom_remote(self):

        fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)

        custom_remote_path = os.path.join(Ui.get_by_key('paths', '/').ui_value, self.REPO)

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

    def test_api_delete_user(self):

        Session().commit()

        username = usr.username

        email = usr.email

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

    def test_api_delete_user_when_exception_happened(self):

        Session().commit()

        username = usr.username

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

    def test_api_get_repo(self):

        make_user_group(new_group)

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

                                                group_name=new_group,

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

                                  repoid=self.REPO)

        response = api_call(self, params)

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

        ret = repo.get_api_data()

                                  with_revision_names=True,

                                  with_pullrequests=True)

        response = api_call(self, params)

    @base.parametrize('grant_perm', [

        ('repository.admin'),

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

    def test_api_create_repo(self):

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

                                  repo_name=repo_name,

                                  owner=base.TEST_USER_ADMIN_LOGIN,

        fixture.destroy_repo(repo_name)

    @base.parametrize('repo_name', [

    ])

    def test_api_create_repo_bad_names(self, repo_name):

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

        # cloning from local repos was a mis-feature - 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)

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

                                  repo_name=repo_name,

                                  owner=base.TEST_USER_ADMIN_LOGIN,

        fixture.destroy_repo(repo_name)

    def test_api_create_repo_and_repo_group(self):

        # repo creation can no longer also create repo group

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

                                  owner=base.TEST_USER_ADMIN_LOGIN,

                                  repo_type=self.REPO_TYPE,)

        response = api_call(self, params)

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

        assert RepoModel().get_by_repo_name(repo_name) is None

        fixture.destroy_repo_group(repo_group_name)

    def test_api_create_repo_in_repo_group_without_permission(self):

        top_group = RepoGroup.get_by_group_name(TEST_REPO_GROUP)

        assert top_group

        fixture.destroy_repo_group(repo_group_name)

    def test_api_create_repo_unknown_owner(self):

        owner = 'i-dont-exist'

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

                                  repo_name=repo_name,

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

    def test_api_create_repo_dont_specify_owner(self):

        owner = 'i-dont-exist'

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

                                  repo_name=repo_name,

        fixture.destroy_repo(repo_name)

    def test_api_create_repo_by_non_admin(self):

        owner = 'i-dont-exist'

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

                                  repo_name=repo_name,

        fixture.destroy_repo(repo_name)

    def test_api_create_repo_by_non_admin_specify_owner(self):

        owner = 'i-dont-exist'

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

                                  repo_name=repo_name,

                                  owner=base.TEST_USER_ADMIN_LOGIN,

                                  repo_type=self.REPO_TYPE,)

        response = api_call(self, params)

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

        fixture.destroy_repo(repo_name)

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

    def test_api_create_repo_exception_occurred(self):

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

                                  repo_name=repo_name,

                                  owner=base.TEST_USER_ADMIN_LOGIN,

    @base.parametrize('changing_attr,updates', [

        ('owner', {'owner': base.TEST_USER_REGULAR_LOGIN}),

        ('clone_uri', {'clone_uri': 'http://example.com/repo'}), # will fail - pulling from non-existing repo should fail

        ('clone_uri', {'clone_uri': '/repo'}), # will fail - pulling from local repo was a mis-feature - it would bypass access control

        ('clone_uri', {'clone_uri': None}),

        ('landing_rev', {'landing_rev': 'branch:master'}),

        ('enable_statistics', {'enable_statistics': True}),

        ('enable_downloads', {'enable_downloads': True}),

    ])

    def test_api_update_repo(self, changing_attr, updates):

        repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)

        if changing_attr == 'repo_group':

            fixture.create_repo_group(updates['group'])

        if changing_attr == 'name':

            repo_name = updates['name']

        if changing_attr == 'repo_group':

        try:

            if changing_attr == 'clone_uri' and updates['clone_uri']:

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

            else:

                expected = {

    @base.parametrize('changing_attr,updates', [

        ('owner', {'owner': base.TEST_USER_REGULAR_LOGIN}),

        ('clone_uri', {'clone_uri': 'http://example.com/repo'}), # will fail - pulling from non-existing repo should fail

        ('clone_uri', {'clone_uri': '/repo'}), # will fail - pulling from local repo was a mis-feature - it would bypass access control

        ('clone_uri', {'clone_uri': None}),

        ('landing_rev', {'landing_rev': 'branch:master'}),

        ('enable_statistics', {'enable_statistics': True}),

        ('enable_downloads', {'enable_downloads': True}),

    ])

    def test_api_update_group_repo(self, changing_attr, updates):