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

    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)

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)

                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

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

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('&', '&')

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

    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

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

    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)

import json

import os

import re

import tempfile

import time

import urllib.request

from subprocess import PIPE, Popen

from tempfile import _RandomNameSequence

import pytest

from kallithea.lib.utils2 import ascii_bytes, safe_str

from kallithea.model.db import Repository, Ui, User, UserIpMap, UserLog

from kallithea.model.meta import Session

from kallithea.model.ssh_key import SshKeyModel

from kallithea.model.user import UserModel

from kallithea.tests import base

from kallithea.tests.fixture import Fixture

DEBUG = True

HOST = '127.0.0.1:4999'  # test host

class HgHttpVcsTest(HgVcsTest, HttpVcsTest):

    pass

class GitHttpVcsTest(GitVcsTest, HttpVcsTest):

    pass

class HgSshVcsTest(HgVcsTest, SshVcsTest):

    @staticmethod

    def _ssh_param(repo_name, user, ssh_key, client_ip):

        # Specify a custom ssh command on the command line

        return r"""--config ui.ssh="bash -c 'SSH_ORIGINAL_COMMAND=\"\$2\" SSH_CONNECTION=\"%s 1024 127.0.0.1 22\" kallithea-cli ssh-serve -c %s %s %s' --" ssh://someuser@somehost/%s""" % (

            client_ip,

            user.user_id,

            ssh_key.user_ssh_key_id,

            repo_name)

class GitSshVcsTest(GitVcsTest, SshVcsTest):

    @staticmethod

    def _ssh_param(repo_name, user, ssh_key, client_ip):

        # Set a custom ssh command in the global environment

        os.environ['GIT_SSH_COMMAND'] = r"""bash -c 'SSH_ORIGINAL_COMMAND="$2" SSH_CONNECTION="%s 1024 127.0.0.1 22" kallithea-cli ssh-serve -c %s %s %s' --""" % (

            client_ip,

            user.user_id,

            ssh_key.user_ssh_key_id)

        return "ssh://someuser@somehost/%s""" % repo_name

parametrize_vcs_test = base.parametrize('vt', [

    HgHttpVcsTest,

    GitHttpVcsTest,

    HgSshVcsTest,

    GitSshVcsTest,

])

parametrize_vcs_test_hg = base.parametrize('vt', [

    HgHttpVcsTest,