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

"""

kallithea.controllers.feed

~~~~~~~~~~~~~~~~~~~~~~~~~~

Feed controller for Kallithea

This file was forked by the Kallithea project in July 2014.

Original author and date, and relevant copyright and licensing information is below:

:created_on: Apr 23, 2010

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import logging

from beaker.cache import cache_region

from tg import response

from tg import tmpl_context as c

from tg.i18n import ugettext as _

from kallithea.lib import feeds

from kallithea.lib import helpers as h

from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired

from kallithea.lib.base import BaseRepoController

from kallithea.lib.diffs import DiffProcessor

from kallithea.lib.utils2 import asbool, safe_int, safe_str

log = logging.getLogger(__name__)

class FeedController(BaseRepoController):

    @LoginRequired(allow_default_user=True)

    @HasRepoPermissionLevelDecorator('read')

    def _before(self, *args, **kwargs):

        super(FeedController, self)._before(*args, **kwargs)

    def _get_title(self, cs):

        return h.shorter(cs.message, 160)

    def __get_desc(self, cs):

        desc_msg = [(_('%s committed on %s')

                     % (h.person(cs.author), h.fmt_date(cs.date))) + '<br/>']

        # branches, tags, bookmarks

        for branch in cs.branches:

            desc_msg.append('branch: %s<br/>' % branch)

        for book in cs.bookmarks:

            desc_msg.append('bookmark: %s<br/>' % book)

        for tag in cs.tags:

            desc_msg.append('tag: %s<br/>' % tag)

        changes = []

        raw_diff = cs.diff()

        diff_processor = DiffProcessor(raw_diff,

                                       diff_limit=diff_limit,

                                       inline_diff=False)

        for st in diff_processor.parsed:

            st.update({'added': st['stats']['added'],

                       'removed': st['stats']['deleted']})

            changes.append('\n %(operation)s %(filename)s '

                           '(%(added)s lines added, %(removed)s lines removed)'

                            % st)

        if diff_processor.limited_diff:

            changes = changes + ['\n ' +

                                 _('Changeset was too big and was cut off...')]

        # rev link

        _url = h.canonical_url('changeset_home', repo_name=c.db_repo.repo_name,

                   revision=cs.raw_id)

        desc_msg.append('changeset: <a href="%s">%s</a>' % (_url, cs.raw_id[:8]))

        desc_msg.append('<pre>')

        desc_msg.append(h.urlify_text(cs.message))

        desc_msg.append('\n')

        desc_msg.extend(changes)

            desc_msg.append('\n\n')

            desc_msg.append(safe_str(raw_diff))

        desc_msg.append('</pre>')

        return desc_msg

    def _feed(self, repo_name, feeder):

        """Produce a simple feed"""

        @cache_region('long_term_file', '_get_feed_from_cache')

        def _get_feed_from_cache(*_cache_keys):  # parameters are not really used - only as caching key

            header = dict(

                title=_('%s %s feed') % (c.site_name, repo_name),

                link=h.canonical_url('summary_home', repo_name=repo_name),

                description=_('Changes on %s repository') % repo_name,

            )

            entries=[]

            for cs in reversed(list(c.db_repo_scm_instance[-rss_items_per_page:])):

                entries.append(dict(

                    title=self._get_title(cs),

                    link=h.canonical_url('changeset_home', repo_name=repo_name, revision=cs.raw_id),

                    author_email=cs.author_email,

                    author_name=cs.author_name,

                    description=''.join(self.__get_desc(cs)),

                    pubdate=cs.date,

                ))

            return feeder.render(header, entries)

        response.content_type = feeder.content_type

        return _get_feed_from_cache(repo_name, feeder.__name__)

    def atom(self, repo_name):

        """Produce a simple atom-1.0 feed"""

        return self._feed(repo_name, feeds.AtomFeed)

    def rss(self, repo_name):

        """Produce a simple rss2 feed"""

        return self._feed(repo_name, feeds.RssFeed)

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

"""

kallithea.controllers.files

~~~~~~~~~~~~~~~~~~~~~~~~~~~

Files controller for Kallithea

This file was forked by the Kallithea project in July 2014.

Original author and date, and relevant copyright and licensing information is below:

:created_on: Apr 21, 2010

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import logging

import os

import posixpath

import shutil

import tempfile

import traceback

from collections import OrderedDict

from tg import request, response

from tg import tmpl_context as c

from tg.i18n import ugettext as _

from webob.exc import HTTPFound, HTTPNotFound

from kallithea.config.routing import url

from kallithea.lib import diffs

from kallithea.lib import helpers as h

from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired

from kallithea.lib.base import BaseRepoController, jsonify, render

from kallithea.lib.exceptions import NonRelativePathError

from kallithea.lib.utils import action_logger

from kallithea.lib.utils2 import asbool, convert_line_endings, detect_mode, safe_str

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

from kallithea.lib.vcs.conf import settings

from kallithea.lib.vcs.exceptions import (ChangesetDoesNotExistError, ChangesetError, EmptyRepositoryError, ImproperArchiveTypeError, NodeAlreadyExistsError,

                                          NodeDoesNotExistError, NodeError, RepositoryError, VCSError)

from kallithea.lib.vcs.nodes import FileNode

from kallithea.model import db

from kallithea.model.repo import RepoModel

from kallithea.model.scm import ScmModel

log = logging.getLogger(__name__)

class FilesController(BaseRepoController):

    def _before(self, *args, **kwargs):

        super(FilesController, self)._before(*args, **kwargs)

    def __get_cs(self, rev, silent_empty=False):

        """

        Safe way to get changeset if error occur it redirects to tip with

        proper message

        :param rev: revision to fetch

        :silent_empty: return None if repository is empty

        """

        try:

            return c.db_repo_scm_instance.get_changeset(rev)

        except EmptyRepositoryError as e:

            if silent_empty:

                return None

            url_ = url('files_add_home',

                       repo_name=c.repo_name,

                       revision=0, f_path='', anchor='edit')

            add_new = h.link_to(_('Click here to add new file'), url_, class_="alert-link")

            h.flash(_('There are no files yet.') + ' ' + add_new, category='warning')

            raise HTTPNotFound()

        except (ChangesetDoesNotExistError, LookupError):

            msg = _('Such revision does not exist for this repository')

                          'contain .. in path'), category='warning')

                raise HTTPFound(location=url('changeset_home', repo_name=c.repo_name,

                                    revision='tip'))

            except (NodeError, NodeAlreadyExistsError) as e:

                h.flash(_(e), category='error')

            except Exception:

                log.error(traceback.format_exc())

                h.flash(_('Error occurred during commit'), category='error')

            raise HTTPFound(location=url('changeset_home',

                                repo_name=c.repo_name, revision='tip'))

        return render('files/files_add.html')

    @LoginRequired(allow_default_user=True)

    @HasRepoPermissionLevelDecorator('read')

    def archivefile(self, repo_name, fname):

        fileformat = None

        revision = None

        ext = None

        subrepos = request.GET.get('subrepos') == 'true'

        for a_type, ext_data in settings.ARCHIVE_SPECS.items():

            archive_spec = fname.split(ext_data[1])

            if len(archive_spec) == 2 and archive_spec[1] == '':

                fileformat = a_type or ext_data[1]

                revision = archive_spec[0]

                ext = ext_data[1]

        try:

            dbrepo = RepoModel().get_by_repo_name(repo_name)

            if not dbrepo.enable_downloads:

                return _('Downloads disabled') # TODO: do something else?

            if c.db_repo_scm_instance.alias == 'hg':

                # patch and reset hooks section of UI config to not run any

                # hooks on fetching archives with subrepos

                for k, v in c.db_repo_scm_instance._repo.ui.configitems('hooks'):

                    c.db_repo_scm_instance._repo.ui.setconfig('hooks', k, None)

            cs = c.db_repo_scm_instance.get_changeset(revision)

            content_type = settings.ARCHIVE_SPECS[fileformat][0]

        except ChangesetDoesNotExistError:

            return _('Unknown revision %s') % revision

        except EmptyRepositoryError:

            return _('Empty repository')

        except (ImproperArchiveTypeError, KeyError):

            return _('Unknown archive type')

        rev_name = cs.raw_id[:12]

        archive_name = '%s-%s%s' % (repo_name.replace('/', '_'), rev_name, ext)

        archive_path = None

        cached_archive_path = None

        if archive_cache_dir and not subrepos: # TODO: subrepo caching?

            if not os.path.isdir(archive_cache_dir):

                os.makedirs(archive_cache_dir)

            cached_archive_path = os.path.join(archive_cache_dir, archive_name)

            if os.path.isfile(cached_archive_path):

                log.debug('Found cached archive in %s', cached_archive_path)

                archive_path = cached_archive_path

            else:

                log.debug('Archive %s is not yet cached', archive_name)

        if archive_path is None:

            # generate new archive

            fd, archive_path = tempfile.mkstemp()

            log.debug('Creating new temp archive in %s', archive_path)

            with os.fdopen(fd, 'wb') as stream:

                cs.fill_archive(stream=stream, kind=fileformat, subrepos=subrepos)

                # stream (and thus fd) has been closed by cs.fill_archive

            if cached_archive_path is not None:

                # we generated the archive - move it to cache

                log.debug('Storing new archive in %s', cached_archive_path)

                shutil.move(archive_path, cached_archive_path)

                archive_path = cached_archive_path

        def get_chunked_archive(archive_path):

            stream = open(archive_path, 'rb')

            while True:

                data = stream.read(16 * 1024)

                if not data:

                    break

                yield data

            stream.close()

            if archive_path != cached_archive_path:

                log.debug('Destroying temp archive %s', archive_path)

                os.remove(archive_path)

        action_logger(user=request.authuser,

                      action='user_downloaded_archive:%s' % (archive_name),

                      repo=repo_name, ipaddr=request.ip_addr, commit=True)

        response.content_disposition = str('attachment; filename=%s' % (archive_name))

        response.content_type = str(content_type)

        return get_chunked_archive(archive_path)

    @LoginRequired(allow_default_user=True)

    @HasRepoPermissionLevelDecorator('read')

    def diff(self, repo_name, f_path):

        ignore_whitespace_diff = h.get_ignore_whitespace_diff(request.GET)

        diff_context_size = h.get_diff_context_size(request.GET)

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

"""

Helper functions

Consists of functions to typically be used within templates, but also

available to Controllers. This module is available to both as 'h'.

"""

import hashlib

import json

import logging

import random

import re

import textwrap

import urllib.parse

from beaker.cache import cache_region

from pygments import highlight as code_highlight

from pygments.formatters.html import HtmlFormatter

from tg.i18n import ugettext as _

from webhelpers2.html import HTML, escape, literal

from webhelpers2.html.tags import NotGiven, Option, Options, _input, _make_safe_id_component, checkbox, end_form

from webhelpers2.html.tags import form as insecure_form

from webhelpers2.html.tags import hidden, link_to, password, radio

from webhelpers2.html.tags import select as webhelpers2_select

from webhelpers2.html.tags import submit, text, textarea

from webhelpers2.number import format_byte_size

from webhelpers2.text import chop_at, truncate, wrap_paragraphs

from kallithea.config.routing import url

from kallithea.lib.annotate import annotate_highlight

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

# PERMS

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

from kallithea.lib.auth import HasPermissionAny, HasRepoGroupPermissionLevel, HasRepoPermissionLevel

from kallithea.lib.markup_renderer import url_re

from kallithea.lib.pygmentsutils import get_custom_lexer

from kallithea.lib.utils2 import MENTIONS_REGEX, AttributeDict

from kallithea.lib.utils2 import age as _age

from kallithea.lib.utils2 import asbool, credentials_filter, safe_bytes, safe_int, safe_str, time_to_datetime

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

from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError

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

# SCM FILTERS available via h.

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

from kallithea.lib.vcs.utils import author_email, author_name

# mute pyflakes "imported but unused"

assert Option

assert checkbox

assert end_form

assert password

assert radio

assert submit

assert text

assert textarea

assert format_byte_size

assert chop_at

assert wrap_paragraphs

assert HasPermissionAny

assert HasRepoGroupPermissionLevel

assert HasRepoPermissionLevel

assert time_to_datetime

assert EmptyChangeset

log = logging.getLogger(__name__)

def canonical_url(*args, **kargs):

    '''Like url(x, qualified=True), but returns url that not only is qualified

    but also canonical, as configured in canonical_url'''

    try:

        kargs['host'] = parts[1]

        kargs['protocol'] = parts[0]

    except IndexError:

        kargs['qualified'] = True

    return url(*args, **kargs)

def canonical_hostname():

    '''Return canonical hostname of system'''

    try:

        return parts[1].split('/', 1)[0]

    except IndexError:

        parts = url('home', qualified=True).split('://', 1)

        return parts[1].split('/', 1)[0]

def html_escape(s):

    """Return string with all html escaped.

    This is also safe for javascript in html but not necessarily correct.

    """

    return (s

        .replace('&', '&')

        .replace(">", ">")

        .replace("<", "&lt;")

        .replace('"', """)

        .replace("'", "'") # Note: this is HTML5 not HTML4 and might not work in mails

        )

def js(value):

    """Convert Python value to the corresponding JavaScript representation.

    This is necessary to safely insert arbitrary values into HTML <script>

    sections e.g. using Mako template expression substitution.

    Note: Rather than using this function, it's preferable to avoid the

    insertion of values into HTML <script> sections altogether. Instead,

    data should (to the extent possible) be passed to JavaScript using

    data attributes or AJAX calls, eliminating the need for JS specific

    escaping.

    Note: This is not safe for use in attributes (e.g. onclick), because

    quotes are not escaped.

    Because the rules for parsing <script> varies between XHTML (where

    normal rules apply for any special characters) and HTML (where

    entities are not interpreted, but the literal string "</script>"

    is forbidden), the function ensures that the result never contains

    '&', '<' and '>', thus making it safe in both those contexts (but

    not in attributes).

    """

    return literal(

        ('(' + json.dumps(value) + ')')

        # In JSON, the following can only appear in string literals.

        .replace('&', r'\x26')

        .replace('<', r'\x3c')

        .replace('>', r'\x3e')

    )

    assert category in ('error', 'success', 'warning'), category

    if hasattr(message, '__html__'):

        # render to HTML for storing in cookie

        safe_message = str(message)

    else:

        # Apply str - the message might be an exception with __str__

        # Escape, so we can trust the result without further escaping, without any risk of injection

        safe_message = html_escape(str(message))

    if logf is None:

        logf = log.info

        if category == 'success':

            logf = log.debug

    logf('Flash %s: %s', category, safe_message)

    _session_flash_messages(append=(category, safe_message))

def pop_flash_messages():

    """Return all accumulated messages and delete them from the session.

    The return value is a list of ``Message`` objects.

    """

    return [_Message(category, message) for category, message in _session_flash_messages(clear=True)]

def age(x, y=False):

    return _age(x, y)

def capitalize(x):

    return x.capitalize()

email = author_email

def short_id(x):

    return x[:12]

def hide_credentials(x):

    return ''.join(credentials_filter(x))

def show_id(cs):

    """

    Configurable function that shows ID

    by default it's r123:fffeeefffeee

    :param cs: changeset instance

    """

    raw_id = cs.raw_id[:def_len]

    if show_rev:

        return 'r%s:%s' % (cs.revision, raw_id)

    else:

        return raw_id

def fmt_date(date):

    if date:

        return date.strftime("%Y-%m-%d %H:%M:%S")

    return ""

def is_git(repository):

    if hasattr(repository, 'alias'):

        _type = repository.alias

    elif hasattr(repository, 'repo_type'):

        _type = repository.repo_type

    else:

        _type = repository

    return _type == 'git'

def is_hg(repository):

    if hasattr(repository, 'alias'):

        _type = repository.alias

    elif hasattr(repository, 'repo_type'):

        _type = repository.repo_type

    else:

        _type = repository

    return _type == 'hg'

@cache_region('long_term', 'user_attr_or_none')

def user_attr_or_none(author, show_attr):

    """Try to match email part of VCS committer string with a local user and return show_attr

    - or return None if user not found"""

    email = author_email(author)

    if email:

        from kallithea.model.db import User

        user = User.get_by_email(email)

        if user is not None:

            return getattr(user, show_attr)

    return None

def email_or_none(author):

        Extract urls from text and make html links out of them

        """

        return _URLIFY_RE.sub(_replace, s)

    if truncate is None:

        s = s.rstrip()

    else:

        s = truncatef(s, truncate, whole_word=True)

    s = html_escape(s)

    s = _urlify(s)

    if repo_name is not None:

        s = urlify_issues(s, repo_name)

    if link_ is not None:

        # make href around everything that isn't a href already

        s = linkify_others(s, link_)

    s = s.replace('\r\n', '<br/>').replace('\n', '<br/>')

    # Turn HTML5 into more valid HTML4 as required by some mail readers.

    # (This is not done in one step in html_escape, because character codes like

    # { risk to be seen as an issue reference due to the presence of '#'.)

    s = s.replace("'", "'")

    return literal(s)

def linkify_others(t, l):

    """Add a default link to html with links.

    HTML doesn't allow nesting of links, so the outer link must be broken up

    in pieces and give space for other links.

    """

    urls = re.compile(r'(\<a.*?\<\/a\>)',)

    links = []

    for e in urls.split(t):

        if e.strip() and not urls.match(e):

            links.append('<a class="message-link" href="%s">%s</a>' % (l, e))

        else:

            links.append(e)

    return ''.join(links)

# Global variable that will hold the actual urlify_issues function body.

# Will be set on first use when the global configuration has been read.

_urlify_issues_f = None

def urlify_issues(newtext, repo_name):

    """Urlify issue references according to .ini configuration"""

    global _urlify_issues_f

    if _urlify_issues_f is None:

        from kallithea.model.db import URL_SEP

        # Build chain of urlify functions, starting with not doing any transformation

        def tmp_urlify_issues_f(s):

            return s

        issue_pat_re = re.compile(r'issue_pat(.*)')

            # Find all issue_pat* settings that also have corresponding server_link and prefix configuration

            m = issue_pat_re.match(k)

            if m is None:

                continue

            suffix = m.group(1)

            if issue_prefix:

                log.error('found unsupported issue_prefix%s = %r - use issue_sub%s instead', suffix, issue_prefix, suffix)

            if not issue_pat:

                log.error('skipping incomplete issue pattern %r: it needs a regexp', k)

                continue

            if not issue_server_link:

                log.error('skipping incomplete issue pattern %r: it needs issue_server_link%s', k, suffix)

                continue

            if issue_sub is None: # issue_sub can be empty but should be present

                log.error('skipping incomplete issue pattern %r: it needs (a potentially empty) issue_sub%s', k, suffix)

                continue

            # Wrap tmp_urlify_issues_f with substitution of this pattern, while making sure all loop variables (and compiled regexpes) are bound

            try:

                issue_re = re.compile(issue_pat)

            except re.error as e:

                log.error('skipping invalid issue pattern %r: %r -> %r %r. Error: %s', k, issue_pat, issue_server_link, issue_sub, str(e))

                continue

            log.debug('issue pattern %r: %r -> %r %r', k, issue_pat, issue_server_link, issue_sub)

            def issues_replace(match_obj,

                               issue_server_link=issue_server_link, issue_sub=issue_sub):

                try:

                    issue_url = match_obj.expand(issue_server_link)

                except (IndexError, re.error) as e:

                    log.error('invalid issue_url setting %r -> %r %r. Error: %s', issue_pat, issue_server_link, issue_sub, str(e))

                    issue_url = issue_server_link

                issue_url = issue_url.replace('{repo}', repo_name)

                issue_url = issue_url.replace('{repo_name}', repo_name.split(URL_SEP)[-1])

                # if issue_sub is empty use the matched issue reference verbatim

                if not issue_sub:

                    issue_text = match_obj.group()

                else:

                    try:

                        issue_text = match_obj.expand(issue_sub)

                    except (IndexError, re.error) as e:

                        log.error('invalid issue_sub setting %r -> %r %r. Error: %s', issue_pat, issue_server_link, issue_sub, str(e))

                        issue_text = match_obj.group()

                return (

                    '<a class="issue-tracker-link" href="%(url)s">'

                    '%(text)s'

                    '</a>'

                    ) % {

                     'url': issue_url,

                     'text': issue_text,

                    }

import mock

from kallithea.config.conf import INDEX_FILENAMES

from kallithea.model.meta import Session

from kallithea.model.repo import RepoModel

from kallithea.model.repo_group import RepoGroupModel

from kallithea.tests import base

from kallithea.tests.fixture import Fixture, create_test_index

fixture = Fixture()

def init_indexing_test(repo):

    prev = fixture.commit_change(repo.repo_name,

                                 filename='this_should_be_unique_filename.txt',

                                 content='this_should_be_unique_content\n',

                                 message='this_should_be_unique_commit_log',

                                 vcs_type='hg',

                                 newfile=True)

def init_stopword_test(repo):

    prev = fixture.commit_change(repo.repo_name,

                                 filename='this/is/it',

                                 content='def test\n',

                                 message='bother to ask where - in folder',

                                 vcs_type='hg',

                                 newfile=True)

    prev = fixture.commit_change(repo.repo_name,

                                 filename='join.us',

                                 content='def test\n',

                                 message='bother to ask where - top level',

                                 author='this is it <this-is-it@foo.bar.com>',

                                 vcs_type='hg',

                                 parent=prev,

                                 newfile=True)

repos = [

    # reponame,              init func or fork base, groupname

    ('indexing_test',       init_indexing_test,     None),

    ('indexing_test-fork',  'indexing_test',       None),

    ('group/indexing_test', 'indexing_test',       'group'),

    ('this-is-it',          'indexing_test',       None),

    ('indexing_test-foo',   'indexing_test',       None),

    ('stopword_test',       init_stopword_test,     None),

]

# map: name => id

repoids = {}

groupids = {}

def rebuild_index(full_index):

    with mock.patch('kallithea.lib.indexers.daemon.log.debug',

                    lambda *args, **kwargs: None):

        # The more revisions managed repositories have, the more

        # memory capturing "log.debug()" output in "indexers.daemon"

        # requires. This may cause unintentional failure of subsequent

        # tests, if ENOMEM at forking "git" prevents from rebuilding

        # index for search.

        # Therefore, "log.debug()" is disabled regardless of logging

        # level while rebuilding index.

        # (FYI, ENOMEM occurs at forking "git" with python 2.7.3,

        # Linux 3.2.78-1 x86_64, 3GB memory, and no ulimit

        # configuration for memory)

class TestSearchControllerIndexing(base.TestController):

    @classmethod

    def setup_class(cls):

        for reponame, init_or_fork, groupname in repos:

            if groupname and groupname not in groupids:

                group = fixture.create_repo_group(groupname)

                groupids[groupname] = group.group_id

            if callable(init_or_fork):

                repo = fixture.create_repo(reponame,

                                           repo_group=groupname)

                init_or_fork(repo)

            else:

                repo = fixture.create_fork(init_or_fork, reponame,

                                           repo_group=groupname)

            repoids[reponame] = repo.repo_id

        # treat "it" as indexable filename

        filenames_mock = list(INDEX_FILENAMES)

        filenames_mock.append('it')

        with mock.patch('kallithea.lib.indexers.daemon.INDEX_FILENAMES',

                        filenames_mock):

            rebuild_index(full_index=False) # only for newly added repos

    @classmethod

    def teardown_class(cls):

        # delete in reversed order, to delete fork destination at first

        for reponame, init_or_fork, groupname in reversed(repos):

            RepoModel().delete(repoids[reponame])

        for reponame, init_or_fork, groupname in reversed(repos):

            if groupname in groupids:

                RepoGroupModel().delete(groupids.pop(groupname),

                                        force_delete=True)

        Session().commit()

        Session.remove()

        rebuild_index(full_index=True) # rebuild fully for subsequent tests

    @base.parametrize('reponame', [

        ('indexing_test'),

        ('indexing_test-fork'),

        ('group/indexing_test'),

        ('this-is-it'),

        ('*-fork'),

        ('group/*'),

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

"""

Test suite for vcs push/pull operations.

The tests need Git > 1.8.1.

This file was forked by the Kallithea project in July 2014.

Original author and date, and relevant copyright and licensing information is below:

:created_on: Dec 30, 2010

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import json

import os

import re

import tempfile

import time

import urllib.request

from subprocess import PIPE, Popen

from tempfile import _RandomNameSequence