# 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.bin.ldap_sync

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

LDAP sync script

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: Mar 06, 2013

:author: marcink

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

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

"""

from __future__ import print_function

import urllib2

import uuid

import ldap

from kallithea.lib import ext_json

from kallithea.lib.utils2 import ascii_bytes

config = ConfigParser()

config.read('ldap_sync.conf')

class InvalidResponseIDError(Exception):

    """ Request and response don't have the same UUID. """

class ResponseError(Exception):

    """ Response has an error, something went wrong with request execution. """

class UserAlreadyInGroupError(Exception):

    """ User is already a member of the target group. """

class UserNotInGroupError(Exception):

# 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

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

                '></i>').format(cls=cls, size=size, src=src)

    else:

        # if src is empty then there was no gravatar, so we use a font icon

        html = ("""<i class="icon-user {cls}" style="font-size: {size}px;"></i>"""

            .format(cls=cls, size=size, src=src))

    return literal(html)

def gravatar_url(email_address, size=30, default=''):

    # doh, we need to re-import those to mock it later

    from kallithea.config.routing import url

    from kallithea.model.db import User

    from tg import tmpl_context as c

    if not c.visual.use_gravatar:

        return ""

    _def = 'anonymous@kallithea-scm.org'  # default gravatar

    email_address = email_address or _def

    if email_address == _def:

        return default

    url = (c.visual.gravatar_url or User.DEFAULT_GRAVATAR_URL) \

               .replace('{email}', email_address) \

               .replace('{md5email}', hashlib.md5(safe_bytes(email_address).lower()).hexdigest()) \

               .replace('{netloc}', parsed_url.netloc) \

               .replace('{scheme}', parsed_url.scheme) \

               .replace('{size}', str(size))

    return url

def changed_tooltip(nodes):

    """

    Generates a html string for changed nodes in changeset page.

    It limits the output to 30 entries

    :param nodes: LazyNodesGenerator

    """

    if nodes:

        pref = ': <br/> '

        suf = ''

        if len(nodes) > 30:

            suf = '<br/>' + _(' and %s more') % (len(nodes) - 30)

        return literal(pref + '<br/> '.join([safe_unicode(x.path)

                                             for x in nodes[:30]]) + suf)

    else:

# -*- coding: utf-8 -*-

import re

import time

import mock

from tg.util.webtest import test_context

import kallithea.lib.celerylib.tasks

from kallithea.lib import helpers as h

from kallithea.lib.auth import check_password

from kallithea.lib.utils2 import generate_api_key

from kallithea.model import validators

from kallithea.model.api_key import ApiKeyModel

from kallithea.model.db import User

from kallithea.model.meta import Session

from kallithea.model.user import UserModel

from kallithea.tests import base

from kallithea.tests.fixture import Fixture

fixture = Fixture()

class TestLoginController(base.TestController):

    def test_index(self):

        response = self.app.get(base.url(controller='login', action='index'))

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        response.mustcontain('Invalid username or password')

    def test_login_non_ascii(self):

        response = self.app.post(base.url(controller='login', action='index'),

                                 {'username': base.TEST_USER_REGULAR_LOGIN,

                                  'password': 'blåbærgrød',

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        response.mustcontain('>Invalid username or password<')

    # verify that get arguments are correctly passed along login redirection

    @base.parametrize('args', [

        {'foo':'one', 'bar':'two'},

        {'blue': u'blå', 'green': u'grøn'},

    ])

    def test_redirection_to_login_form_preserves_get_args(self, args):

        with fixture.anon_access(False):

            response = self.app.get(base.url(controller='summary', action='index',

                                        repo_name=base.HG_REPO,

                                        **args))

            assert response.status == '302 Found'

            assert sorted(came_from_qs) == sorted((k, v.encode('utf-8')) for k, v in args.items())

    @base.parametrize('args,args_encoded', [

        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),

        ({'blue': u'blå', 'green':u'grøn'},

             ('blue=bl%C3%A5', 'green=gr%C3%B8n')),

    ])

    def test_login_form_preserves_get_args(self, args, args_encoded):

        response = self.app.get(base.url(controller='login', action='index',

                                    came_from=base.url('/_admin/users', **args)))

        for encoded in args_encoded:

            assert encoded in came_from

    @base.parametrize('args,args_encoded', [

        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),

        ({'blue': u'blå', 'green':u'grøn'},

             ('blue=bl%C3%A5', 'green=gr%C3%B8n')),

    ])

    def test_redirection_after_successful_login_preserves_get_args(self, args, args_encoded):

        response = self.app.post(base.url(controller='login', action='index',

                                     came_from=base.url('/_admin/users', **args)),

                                 {'username': base.TEST_USER_ADMIN_LOGIN,

                                  'password': base.TEST_USER_ADMIN_PASS,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        assert response.status == '302 Found'

        for encoded in args_encoded:

            assert encoded in response.location

    @base.parametrize('args,args_encoded', [

        ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),

        ({'blue': u'blå', 'green':u'grøn'},

             ('blue=bl%C3%A5', 'green=gr%C3%B8n')),

    ])

    def test_login_form_after_incorrect_login_preserves_get_args(self, args, args_encoded):

        response = self.app.post(base.url(controller='login', action='index',

                                     came_from=base.url('/_admin/users', **args)),

                                 {'username': 'error',

                                  'password': 'test12',

                                  '_session_csrf_secret_token': self.session_csrf_secret_token()})

        response.mustcontain('Invalid username or password')

        for encoded in args_encoded:

            assert encoded in came_from

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

    # REGISTRATIONS

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

    def test_register(self):

        response = self.app.get(base.url(controller='login', action='register'))

        response.mustcontain('Sign Up')

    def test_register_err_same_username(self):

        uname = base.TEST_USER_ADMIN_LOGIN

        response = self.app.post(base.url(controller='login', action='register'),

                                            {'username': uname,

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

                                             'email': 'goodmail@example.com',

                                             'firstname': 'test',

                                             'lastname': 'test',

                                             '_session_csrf_secret_token': self.session_csrf_secret_token()})

        with test_context(self.app):

            msg = validators.ValidUsername()._messages['username_exists']

        msg = h.html_escape(msg % {'username': uname})

        token = UserModel().get_reset_password_token(

            User.get_by_username(username), timestamp, self.session_csrf_secret_token())

        collected = []

        def mock_send_email(recipients, subject, body='', html_body='', headers=None, author=None):

            collected.append((recipients, subject, body, html_body))

        with mock.patch.object(kallithea.lib.celerylib.tasks, 'send_email', mock_send_email), \

                mock.patch.object(time, 'time', lambda: timestamp):

            response = self.app.post(base.url(controller='login',

                                         action='password_reset'),

                                     {'email': email,

                                      '_session_csrf_secret_token': self.session_csrf_secret_token()})

        self.checkSessionFlash(response, 'A password reset confirmation code has been sent')

        ((recipients, subject, body, html_body),) = collected

        assert recipients == ['username@example.com']

        assert subject == 'Password reset link'

        assert '\n%s\n' % token in body

        (confirmation_url,) = (line for line in body.splitlines() if line.startswith('http://'))

        assert ' href="%s"' % confirmation_url.replace('&', '&').replace('@', '%40') in html_body

        assert d['token'] == [token]

        assert d['timestamp'] == [str(timestamp)]

        assert d['email'] == [email]

        response = response.follow()

        # BAD TOKEN

        bad_token = "bad"

        response = self.app.post(base.url(controller='login',

                                     action='password_reset_confirmation'),

                                 {'email': email,

                                  'timestamp': timestamp,

                                  'password': "p@ssw0rd",

                                  'password_confirm': "p@ssw0rd",

                                  'token': bad_token,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),

                                 })

        assert response.status == '200 OK'

        response.mustcontain('Invalid password reset token')

        # GOOD TOKEN

# 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.tests.scripts.manual_test_crawler

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

Test for crawling a project for memory usage

This should be runned just as regular script together

with a watch script that will show memory usage.

watch -n1 ./kallithea/tests/mem_watch

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.

"""

from __future__ import print_function

import os

import sys

import tempfile

import time

import urllib

import urllib2

from os.path import dirname

from kallithea.lib import vcs

from kallithea.lib.compat import OrderedSet

from kallithea.lib.vcs.exceptions import RepositoryError

__here__ = os.path.abspath(__file__)

__root__ = dirname(dirname(dirname(__here__)))

sys.path.append(__root__)

PASES = 3

HOST = 'http://127.0.0.1'

PORT = 5000

BASE_URI = '%s:%s/' % (HOST, PORT)

if len(sys.argv) == 2:

    BASE_URI = sys.argv[1]

if not BASE_URI.endswith('/'):

    BASE_URI += '/'

print('Crawling @ %s' % BASE_URI)

BASE_URI += '%s'

PROJECT_PATH = os.path.join('/', 'home', 'username', 'repos')

PROJECTS = [

    # 'linux-magx-pbranch',

    'CPython',

    'kallithea',

]

o = urllib2.build_opener(urllib2.HTTPCookieProcessor(cj))

o.addheaders = [

    ('User-agent', 'kallithea-crawler'),

    ('Accept-Language', 'en - us, en;q = 0.5')

]

urllib2.install_opener(o)

def _get_repo(proj):

    if isinstance(proj, str):

        repo = vcs.get_repo(os.path.join(PROJECT_PATH, proj))

        proj = proj

    else:

        repo = proj

        proj = repo.name

    return repo, proj

def test_changelog_walk(proj, pages=100):

    repo, proj = _get_repo(proj)

    total_time = 0