from kallithea.tests.base import *

from kallithea.model.db import ChangesetComment, Notification, \

    UserNotification

from kallithea.model.meta import Session

class TestChangeSetCommentsController(TestController):

    def setup_method(self, method):

        for x in ChangesetComment.query().all():

            Session().delete(x)

        Session().commit()

        self.remove_all_notifications()

    def test_create(self):

        self.log_user()

        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'

        text = u'CommentOnRevision'

        params = {'text': text, '_authentication_token': self.authentication_token()}

        response = self.app.post(url(controller='changeset', action='comment',

                                     repo_name=HG_REPO, revision=rev),

                                     params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})

        # Test response...

        assert response.status == '200 OK'

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

                                repo_name=HG_REPO, revision=rev))

        # test DB

        assert ChangesetComment.query().count() == 1

        response.mustcontain(

            '''<div class="comments-number">'''

            ''' 1 comment (0 inline, 1 general)'''

        )

        assert Notification.query().count() == 1

        assert ChangesetComment.query().count() == 1

        notification = Notification.query().all()[0]

        assert notification.type_ == Notification.TYPE_CHANGESET_COMMENT

        sbj = (u'/%s/changeset/'

        print "%s vs %s" % (sbj, notification.subject)

        assert sbj in notification.subject

    def test_create_inline(self):

        self.log_user()

        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'

        text = u'CommentOnRevision'

        f_path = 'vcs/web/simplevcs/views/repository.py'

        line = 'n1'

        params = {'text': text, 'f_path': f_path, 'line': line, '_authentication_token': self.authentication_token()}

        response = self.app.post(url(controller='changeset', action='comment',

                                     repo_name=HG_REPO, revision=rev),

                                     params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})

        # Test response...

        assert response.status == '200 OK'

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

                                repo_name=HG_REPO, revision=rev))

        # test DB

        assert ChangesetComment.query().count() == 1

        response.mustcontain(

            '''<div class="comments-number">'''

            ''' 1 comment (1 inline, 0 general)'''

        )

        response.mustcontain(

            '''<div class="comments-list-chunk" '''

            '''data-f_path="vcs/web/simplevcs/views/repository.py" '''

            '''data-line_no="n1" data-target-id="vcswebsimplevcsviewsrepositorypy_n1">'''

        )

        assert Notification.query().count() == 1

        assert ChangesetComment.query().count() == 1

        notification = Notification.query().all()[0]

        assert notification.type_ == Notification.TYPE_CHANGESET_COMMENT

        sbj = (u'/%s/changeset/'

        print "%s vs %s" % (sbj, notification.subject)

        assert sbj in notification.subject

    def test_create_with_mention(self):

        self.log_user()

        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'

        text = u'@%s check CommentOnRevision' % TEST_USER_REGULAR_LOGIN

        params = {'text': text, '_authentication_token': self.authentication_token()}

        response = self.app.post(url(controller='changeset', action='comment',

                                     repo_name=HG_REPO, revision=rev),

                                     params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})

        # Test response...

        assert response.status == '200 OK'

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

                                repo_name=HG_REPO, revision=rev))

        # test DB

        assert ChangesetComment.query().count() == 1

        response.mustcontain(

            '''<div class="comments-number">'''

            ''' 1 comment (0 inline, 1 general)'''

        )

        assert Notification.query().count() == 2

        users = [x.user.username for x in UserNotification.query().all()]

        # test_regular gets notification by @mention

        assert sorted(users) == [TEST_USER_ADMIN_LOGIN, TEST_USER_REGULAR_LOGIN]

    def test_delete(self):

        self.log_user()

        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'

        text = u'CommentOnRevision'

        params = {'text': text, '_authentication_token': self.authentication_token()}

        response = self.app.post(url(controller='changeset', action='comment',

                                     repo_name=HG_REPO, revision=rev),

                                     params=params, extra_environ={'HTTP_X_PARTIAL_XHR': '1'})

        comments = ChangesetComment.query().all()

        assert len(comments) == 1

        comment_id = comments[0].comment_id

        self.app.post(url("changeset_comment_delete",

                                    repo_name=HG_REPO,

                                    comment_id=comment_id),

            params={'_authentication_token': self.authentication_token()})

        comments = ChangesetComment.query().all()

        assert len(comments) == 0

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

                                repo_name=HG_REPO, revision=rev))

        response.mustcontain(

            '''<div class="comments-number">'''

            ''' 0 comments (0 inline, 0 general)'''

        )

# -*- 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.tests.other.test_libs

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

Package for testing various lib/helper functions in 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: Jun 9, 2011

:author: marcink

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

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

"""

import datetime

import hashlib

import mock

from kallithea.tests.base import *

from kallithea.lib.utils2 import AttributeDict

from kallithea.model.db import Repository

from tg.util.webtest import test_context

proto = 'http'

TEST_URLS = [

    ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

     '%s://127.0.0.1' % proto),

    ('%s://username@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

     '%s://127.0.0.1' % proto),

    ('%s://username:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

     '%s://127.0.0.1' % proto),

    ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],

     '%s://127.0.0.1:8080' % proto),

    ('%s://example.com' % proto, ['%s://' % proto, 'example.com'],

     '%s://example.com' % proto),

    ('%s://user:pass@example.com:8080' % proto, ['%s://' % proto, 'example.com',

                                                '8080'],

     '%s://example.com:8080' % proto),

]

proto = 'https'

TEST_URLS += [

    ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

     '%s://127.0.0.1' % proto),

    ('%s://username@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

     '%s://127.0.0.1' % proto),

    ('%s://username:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

     '%s://127.0.0.1' % proto),

    ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],

     '%s://127.0.0.1:8080' % proto),

    ('%s://example.com' % proto, ['%s://' % proto, 'example.com'],

     '%s://example.com' % proto),

    ('%s://user:pass@example.com:8080' % proto, ['%s://' % proto, 'example.com',

                                                '8080'],

     '%s://example.com:8080' % proto),

]

class FakeUrlGenerator(object):

    def __init__(self, current_url=None, default_route=None, **routes):

        """Initialize using specified 'current' URL template,

        default route template, and all other aguments describing known

        routes (format: route=template)"""

        self.current_url = current_url

        self.default_route = default_route

        self.routes = routes

    def __call__(self, route_name, *args, **kwargs):

        if route_name in self.routes:

            return self.routes[route_name] % kwargs

        return self.default_route % kwargs

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

        return self.current_url % kwargs

class TestLibs(TestController):

    @parametrize('test_url,expected,expected_creds', TEST_URLS)

    def test_uri_filter(self, test_url, expected, expected_creds):

        from kallithea.lib.utils2 import uri_filter

        assert uri_filter(test_url) == expected

    @parametrize('test_url,expected,expected_creds', TEST_URLS)

    def test_credentials_filter(self, test_url, expected, expected_creds):

        from kallithea.lib.utils2 import credentials_filter

        assert credentials_filter(test_url) == expected_creds

    @parametrize('str_bool,expected', [

                           ('t', True),

                           ('true', True),

                           ('y', True),

                           ('yes', True),

                           ('on', True),

                           ('1', True),

                           ('Y', True),

                           ('yeS', True),

                           ('Y', True),

                           ('TRUE', True),

                           ('T', True),

                           ('False', False),

                           ('F', False),

                           ('FALSE', False),

                           ('0', False),

                           ('-1', False),

                           ('', False)

    ])

    def test_str2bool(self, str_bool, expected):

        from kallithea.lib.utils2 import str2bool

        assert str2bool(str_bool) == expected

    def test_mention_extractor(self):

        from kallithea.lib.utils2 import extract_mentioned_usernames

        sample = (

            "@first hi there @world here's my email username@example.com "

            "@lukaszb check @one_more22 it pls @ ttwelve @D[] @one@two@three "

            "@UPPER    @cAmEL @2one_more22 @john please see this http://org.pl "

            "@marian.user just do it @marco-polo and next extract @marco_polo "

            "user.dot  hej ! not-needed maril@example.com"

        )

        expected = set([

            '2one_more22', 'first', 'lukaszb', 'one', 'one_more22', 'UPPER', 'cAmEL', 'john',

            'marian.user', 'marco-polo', 'marco_polo', 'world'])

        assert expected == set(extract_mentioned_usernames(sample))

    @parametrize('age_args,expected', [

        (dict(), u'just now'),

        (dict(seconds= -1), u'1 second ago'),

        (dict(seconds= -60 * 2), u'2 minutes ago'),

        (dict(hours= -1), u'1 hour ago'),

        (dict(hours= -24), u'1 day ago'),

        (dict(hours= -24 * 5), u'5 days ago'),

        (dict(months= -1), u'1 month ago'),

        (dict(months= -1, days= -2), u'1 month and 2 days ago'),

        (dict(months= -1, days= -20), u'1 month and 19 days ago'),

        (dict(years= -1, months= -1), u'1 year and 1 month ago'),

        (dict(years= -1, months= -10), u'1 year and 10 months ago'),

        (dict(years= -2, months= -4), u'2 years and 4 months ago'),

        (dict(years= -2, months= -11), u'2 years and 11 months ago'),

        (dict(years= -3, months= -2), u'3 years and 2 months ago'),

    ])

    def test_age(self, age_args, expected):

        from kallithea.lib.utils2 import age

        from dateutil import relativedelta

        with test_context(self.app):

            n = datetime.datetime(year=2012, month=5, day=17)

            delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)

            assert age(n + delt(**age_args), now=n) == expected

    @parametrize('age_args,expected', [

        (dict(), u'just now'),

        (dict(seconds= -1), u'1 second ago'),

        (dict(seconds= -60 * 2), u'2 minutes ago'),

        (dict(hours= -1), u'1 hour ago'),

        (dict(hours= -24), u'1 day ago'),

        (dict(hours= -24 * 5), u'5 days ago'),

        (dict(months= -1), u'1 month ago'),

        (dict(months= -1, days= -2), u'1 month ago'),

        (dict(months= -1, days= -20), u'1 month ago'),

        (dict(years= -1, months= -1), u'13 months ago'),

        (dict(years= -1, months= -10), u'22 months ago'),

        (dict(years= -2, months= -4), u'2 years ago'),

        (dict(years= -2, months= -11), u'3 years ago'),

        (dict(years= -3, months= -2), u'3 years ago'),

        (dict(years= -4, months= -8), u'5 years ago'),

    ])

    def test_age_short(self, age_args, expected):

        from kallithea.lib.utils2 import age

        from dateutil import relativedelta

        with test_context(self.app):

            n = datetime.datetime(year=2012, month=5, day=17)

            delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)

            assert age(n + delt(**age_args), show_short_version=True, now=n) == expected

    @parametrize('age_args,expected', [

        (dict(), u'just now'),

        (dict(seconds=1), u'in 1 second'),

        (dict(seconds=60 * 2), u'in 2 minutes'),

        (dict(hours=1), u'in 1 hour'),

        (dict(hours=24), u'in 1 day'),

        (dict(hours=24 * 5), u'in 5 days'),

        (dict(months=1), u'in 1 month'),

        (dict(months=1, days=1), u'in 1 month and 1 day'),

        (dict(years=1, months=1), u'in 1 year and 1 month')

    ])

    def test_age_in_future(self, age_args, expected):

        from kallithea.lib.utils2 import age

        from dateutil import relativedelta

        with test_context(self.app):

            n = datetime.datetime(year=2012, month=5, day=17)

            delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)

            assert age(n + delt(**age_args), now=n) == expected

    def test_tag_extractor(self):

        sample = (

            "hello pta[tag] gog [[]] [[] sda ero[or]d [me =>>< sa]"

            "[requires] [stale] [see<>=>] [see => http://example.com]"

            "[requires => url] [lang => python] [just a tag]"

            "[,d] [ => ULR ] [obsolete] [desc]]"

        )

        from kallithea.lib.helpers import urlify_text

        res = urlify_text(sample, stylize=True)

        assert '<div class="metatag" data-tag="tag">tag</div>' in res

        assert '<div class="metatag" data-tag="obsolete">obsolete</div>' in res

        assert '<div class="metatag" data-tag="stale">stale</div>' in res

        assert '<div class="metatag" data-tag="lang">python</div>' in res

        assert '<div class="metatag" data-tag="requires">requires =&gt; <a href="/url">url</a></div>' in res

        assert '<div class="metatag" data-tag="tag">tag</div>' in res

    def test_alternative_gravatar(self):

        from kallithea.lib.helpers import gravatar_url

        _md5 = lambda s: hashlib.md5(s).hexdigest()

        # mock tg.tmpl_context

        def fake_tmpl_context(_url):

            _c = AttributeDict()

            _c.visual = AttributeDict()

            _c.visual.use_gravatar = True

            _c.visual.gravatar_url = _url

            return _c

        fake_url = FakeUrlGenerator(current_url='https://example.com')

        with mock.patch('kallithea.config.routing.url', fake_url):

            fake = fake_tmpl_context(_url='http://example.com/{email}')

            with mock.patch('tg.tmpl_context', fake):

                    from kallithea.config.routing import url

                    assert url.current() == 'https://example.com'

                    grav = gravatar_url(email_address='test@example.com', size=24)

                    assert grav == 'http://example.com/test@example.com'

            fake = fake_tmpl_context(_url='http://example.com/{email}')

            with mock.patch('tg.tmpl_context', fake):

                grav = gravatar_url(email_address='test@example.com', size=24)

                assert grav == 'http://example.com/test@example.com'

            fake = fake_tmpl_context(_url='http://example.com/{md5email}')

            with mock.patch('tg.tmpl_context', fake):

                em = 'test@example.com'

                grav = gravatar_url(email_address=em, size=24)

                assert grav == 'http://example.com/%s' % (_md5(em))

            fake = fake_tmpl_context(_url='http://example.com/{md5email}/{size}')

            with mock.patch('tg.tmpl_context', fake):

                em = 'test@example.com'

                grav = gravatar_url(email_address=em, size=24)

                assert grav == 'http://example.com/%s/%s' % (_md5(em), 24)

            fake = fake_tmpl_context(_url='{scheme}://{netloc}/{md5email}/{size}')

            with mock.patch('tg.tmpl_context', fake):

                em = 'test@example.com'

                grav = gravatar_url(email_address=em, size=24)

                assert grav == 'https://example.com/%s/%s' % (_md5(em), 24)

    @parametrize('tmpl,repo_name,overrides,prefix,expected', [

        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '', 'http://vps1:8000/group/repo1'),

        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/group/repo1'),

        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '/prefix', 'http://vps1:8000/prefix/group/repo1'),

        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/prefix', 'http://user@vps1:8000/prefix/group/repo1'),

        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '/prefix', 'http://username@vps1:8000/prefix/group/repo1'),

        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/prefix/', 'http://user@vps1:8000/prefix/group/repo1'),

        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '/prefix/', 'http://username@vps1:8000/prefix/group/repo1'),

        ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {}, '', 'http://vps1:8000/_23'),

        ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/_23'),

        ('http://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/_23'),

        ('http://{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://vps1:8000/_23'),

        ('https://{user}@proxy1.example.com/{repo}', 'group/repo1', {'user': 'username'}, '', 'https://username@proxy1.example.com/group/repo1'),

        ('https://{user}@proxy1.example.com/{repo}', 'group/repo1', {}, '', 'https://proxy1.example.com/group/repo1'),

        ('https://proxy1.example.com/{user}/{repo}', 'group/repo1', {'user': 'username'}, '', 'https://proxy1.example.com/username/group/repo1'),

    ])

    def test_clone_url_generator(self, tmpl, repo_name, overrides, prefix, expected):

        from kallithea.lib.utils2 import get_clone_url

        clone_url = get_clone_url(uri_tmpl=tmpl, qualified_home_url='http://vps1:8000'+prefix,

                                  repo_name=repo_name, repo_id=23, **overrides)

        assert clone_url == expected

    def _quick_url(self, text, tmpl="""<a class="changeset_hash" href="%s">%s</a>""", url_=None):

        """

        Changes `some text url[foo]` => `some text <a href="/">foo</a>

        :param text:

        """

        import re

        # quickly change expected url[] into a link

        def url_func(match_obj):

            _url = match_obj.groups()[0]

            return tmpl % (url_ or '/repo_name/changeset/%s' % _url, _url)

    @parametrize('sample,expected', [

      ("",

       ""),

      ("git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68",

       """git-svn-id: <a href="https://svn.apache.org/repos/asf/libcloud/trunk@1441655">https://svn.apache.org/repos/asf/libcloud/trunk@1441655</a> 13f79535-47bb-0310-9956-ffa450edef68"""),

      ("from rev 000000000000",

       """from rev url[000000000000]"""),

      ("from rev 000000000000123123 also rev 000000000000",

       """from rev url[000000000000123123] also rev url[000000000000]"""),

      ("this should-000 00",

       """this should-000 00"""),

      ("longtextffffffffff rev 123123123123",

       """longtextffffffffff rev url[123123123123]"""),

      ("rev ffffffffffffffffffffffffffffffffffffffffffffffffff",

       """rev ffffffffffffffffffffffffffffffffffffffffffffffffff"""),

      ("ffffffffffff some text traalaa",

       """url[ffffffffffff] some text traalaa"""),

       ("""Multi line

       123123123123

       some text 123123123123

       sometimes !

       """,

       """Multi line<br/>"""

       """       url[123123123123]<br/>"""

       """       some text url[123123123123]<br/>"""

       """       sometimes !"""),

    ])

    def test_urlify_text(self, sample, expected):

        expected = self._quick_url(expected)

        fake_url = FakeUrlGenerator(changeset_home='/%(repo_name)s/changeset/%(revision)s')

        with mock.patch('kallithea.config.routing.url', fake_url):

            from kallithea.lib.helpers import urlify_text

            assert urlify_text(sample, 'repo_name') == expected

    @parametrize('sample,expected,url_', [

      ("",

       "",

       ""),

      ("https://svn.apache.org/repos",

       """url[https://svn.apache.org/repos]""",

       "https://svn.apache.org/repos"),

      ("http://svn.apache.org/repos",

       """url[http://svn.apache.org/repos]""",

       "http://svn.apache.org/repos"),

      ("from rev a also rev http://google.com",

       """from rev a also rev url[http://google.com]""",

       "http://google.com"),

      ("http://imgur.com/foo.gif inline http://imgur.com/foo.gif ending http://imgur.com/foo.gif",

       """url[http://imgur.com/foo.gif] inline url[http://imgur.com/foo.gif] ending url[http://imgur.com/foo.gif]""",

       "http://imgur.com/foo.gif"),

      ("""Multi line

       https://foo.bar.example.com

       some text lalala""",

       """Multi line<br/>"""

       """       url[https://foo.bar.example.com]<br/>"""

       """       some text lalala""",

       "https://foo.bar.example.com"),

      ("@mention @someone",

       """<b>@mention</b> <b>@someone</b>""",

       ""),

      ("deadbeefcafe 123412341234",

       """<a class="changeset_hash" href="/repo_name/changeset/deadbeefcafe">deadbeefcafe</a> <a class="changeset_hash" href="/repo_name/changeset/123412341234">123412341234</a>""",

       ""),

      ("We support * markup for *bold* markup of *single or multiple* words, "

       "*a bit @like http://slack.com*. "

       "The first * must come after whitespace and not be followed by whitespace, "

       "contain anything but * and newline until the next *, "

       "which must not come after whitespace "

       "and not be followed by * or alphanumerical *characters*.",

       """We support * markup for <b>*bold*</b> markup of <b>*single or multiple*</b> words, """

       """<b>*a bit <b>@like</b> <a href="http://slack.com">http://slack.com</a>*</b>. """

       """The first * must come after whitespace and not be followed by whitespace, """

       """contain anything but * and newline until the next *, """

       """which must not come after whitespace """

       """and not be followed by * or alphanumerical <b>*characters*</b>.""",

       "-"),

      # tags are covered by test_tag_extractor

    ])

    def test_urlify_test(self, sample, expected, url_):

        expected = self._quick_url(expected,

                                   tmpl="""<a href="%s">%s</a>""", url_=url_)

        fake_url = FakeUrlGenerator(changeset_home='/%(repo_name)s/changeset/%(revision)s')

        with mock.patch('kallithea.config.routing.url', fake_url):

            from kallithea.lib.helpers import urlify_text

            assert urlify_text(sample, 'repo_name', stylize=True) == expected

    @parametrize('sample,expected', [

      ("deadbeefcafe @mention, and http://foo.bar/ yo",

       """<a class="changeset_hash" href="/repo_name/changeset/deadbeefcafe">deadbeefcafe</a>"""

       """<a class="message-link" href="#the-link"> <b>@mention</b>, and </a>"""

       """<a href="http://foo.bar/">http://foo.bar/</a>"""

       """<a class="message-link" href="#the-link"> yo</a>"""),

    ])

    def test_urlify_link(self, sample, expected):

        fake_url = FakeUrlGenerator(changeset_home='/%(repo_name)s/changeset/%(revision)s')

        with mock.patch('kallithea.config.routing.url', fake_url):

            from kallithea.lib.helpers import urlify_text

            assert urlify_text(sample, 'repo_name', link_='#the-link') == expected

    @parametrize('test,expected', [

      ("", None),

      ("/_2", '2'),

      ("_2", '2'),

      ("/_2/", '2'),

      ("_2/", '2'),

      ("/_21", '21'),

      ("_21", '21'),

      ("/_21/", '21'),

      ("_21/", '21'),

      ("/_21/foobar", '21'),

      ("_21/121", '21'),

      ("/_21/_12", '21'),

      ("_21/prefix/foo", '21'),

    ])

    def test_get_repo_by_id(self, test, expected):

        from kallithea.lib.utils import _extract_id_from_repo_name

        _test = _extract_id_from_repo_name(test)

        assert _test == expected, 'url:%s, got:`%s` expected: `%s`' % (test, _test, expected)

# -*- 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 os

import re

import tempfile

import time

import pytest

from tempfile import _RandomNameSequence

from subprocess import Popen, PIPE

from kallithea.tests.base import *

from kallithea.tests.fixture import Fixture

from kallithea.model.db import User, Repository, UserIpMap, CacheInvalidation

from kallithea.model.meta import Session

from kallithea.model.repo import RepoModel

from kallithea.model.user import UserModel

DEBUG = True

HOST = '127.0.0.1:4999'  # test host

fixture = Fixture()

class Command(object):

    def __init__(self, cwd):

        self.cwd = cwd

    def execute(self, cmd, *args, **environ):

        """

        Runs command on the system with given ``args``.

        """

        command = cmd + ' ' + ' '.join(args)

        ignoreReturnCode = environ.pop('ignoreReturnCode', False)

        if DEBUG:

            print '*** CMD %s ***' % command

        testenv = dict(os.environ)

        testenv['LANG'] = 'en_US.UTF-8'

        testenv['LANGUAGE'] = 'en_US:en'

        testenv['HGPLAIN'] = ''

        testenv['HGRCPATH'] = ''

        testenv.update(environ)

        p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd, env=testenv)

        stdout, stderr = p.communicate()

        if DEBUG:

            if stdout:

                print 'stdout:', repr(stdout)

            if stderr:

                print 'stderr:', repr(stderr)

        if not ignoreReturnCode:

            assert p.returncode == 0

        return stdout, stderr

def _get_tmp_dir(prefix='vcs_operations-', suffix=''):

    return tempfile.mkdtemp(dir=TESTS_TMP_PATH, prefix=prefix, suffix=suffix)

                        clone_url=None, username=TEST_USER_ADMIN_LOGIN, password=TEST_USER_ADMIN_PASS):

    """

    vcs is git or hg and defines what VCS we want to make those files for

    :param vcs:

    """

    # commit some stuff into this repo

    #added_file = '%ssetupążźć.py' % _RandomNameSequence().next()

    added_file = '%ssetup.py' % _RandomNameSequence().next()

    open(os.path.join(cwd, added_file), 'a').close()

    Command(cwd).execute('%s add %s' % (vcs, added_file))

    email = 'me@example.com'

    if os.name == 'nt':

        author_str = 'User <%s>' % email

    else:

        author_str = 'User ǝɯɐᴎ <%s>' % email

    for i in xrange(files_no):

        cmd = """echo "added_line%s" >> %s""" % (i, added_file)

        Command(cwd).execute(cmd)

        if vcs == 'hg':

            cmd = """hg commit -m "committed new %s" -u "%s" "%s" """ % (

                i, author_str, added_file

            )

        elif vcs == 'git':

            cmd = """git commit -m "committed new %s" --author "%s" "%s" """ % (

                i, author_str, added_file

            )

        # git commit needs EMAIL on some machines

        Command(cwd).execute(cmd, EMAIL=email)

    # PUSH it back

    _REPO = None

    if vcs == 'hg':

        _REPO = HG_REPO

    elif vcs == 'git':

        _REPO = GIT_REPO

    if clone_url is None:

        clone_url = webserver.repo_url(_REPO, username=username, password=password)

    stdout = stderr = None

    if vcs == 'hg':

        stdout, stderr = Command(cwd).execute('hg push --verbose', clone_url, ignoreReturnCode=ignoreReturnCode)

    elif vcs == 'git':

        stdout, stderr = Command(cwd).execute('git push --verbose', clone_url, "master", ignoreReturnCode=ignoreReturnCode)

    return stdout, stderr

def set_anonymous_access(enable=True):

    user = User.get_default_user()

    user.active = enable

    Session().commit()

    print '\tanonymous access is now:', enable

    if enable != User.get_default_user().active:

        raise Exception('Cannot set anonymous access')

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

# TESTS

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

def _check_proper_git_push(stdout, stderr):

    # WTF Git stderr is output ?!

    assert 'fatal' not in stderr

    assert 'rejected' not in stderr

    assert 'Pushing to' in stderr

    assert 'master -> master' in stderr

@pytest.mark.usefixtures("test_context_fixture")

class TestVCSOperations(TestController):

    @classmethod

    def setup_class(cls):

        # DISABLE ANONYMOUS ACCESS

        set_anonymous_access(False)

    def setup_method(self, method):

        r = Repository.get_by_repo_name(GIT_REPO)

        Repository.unlock(r)

        r.enable_locking = False

        Session().commit()

        r = Repository.get_by_repo_name(HG_REPO)

        Repository.unlock(r)

        r.enable_locking = False

        Session().commit()

    def test_clone_hg_repo_by_admin(self, webserver):

        clone_url = webserver.repo_url(HG_REPO)

        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir())

        assert 'requesting all changes' in stdout

        assert 'adding changesets' in stdout

        assert 'adding manifests' in stdout

        assert 'adding file changes' in stdout

        assert stderr == ''

    def test_clone_git_repo_by_admin(self, webserver):

        clone_url = webserver.repo_url(GIT_REPO)

        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir())

        assert 'Cloning into' in stdout + stderr

        assert stderr == '' or stdout == ''

    def test_clone_wrong_credentials_hg(self, webserver):

        clone_url = webserver.repo_url(HG_REPO, password='bad!')

        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        assert 'abort: authorization failed' in stderr

    def test_clone_wrong_credentials_git(self, webserver):

        clone_url = webserver.repo_url(GIT_REPO, password='bad!')

        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        assert 'fatal: Authentication failed' in stderr

    def test_clone_git_dir_as_hg(self, webserver):

        clone_url = webserver.repo_url(GIT_REPO)

        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        assert 'HTTP Error 404: Not Found' in stderr

    def test_clone_hg_repo_as_git(self, webserver):

        clone_url = webserver.repo_url(HG_REPO)

        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        assert 'not found' in stderr

    def test_clone_non_existing_path_hg(self, webserver):

        clone_url = webserver.repo_url('trololo')

        stdout, stderr = Command(TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        assert 'HTTP Error 404: Not Found' in stderr

    def test_clone_non_existing_path_git(self, webserver):

        clone_url = webserver.repo_url('trololo')

        stdout, stderr = Command(TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        assert 'not found' in stderr

    def test_push_new_file_hg(self, webserver):

        clone_url = webserver.repo_url(HG_REPO)

        fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())

        fixture.create_fork(HG_REPO, fork_name)

        clone_url = webserver.repo_url(fork_name)

        assert 'pushing to' in stdout

        assert 'Repository size' in stdout

        assert 'Last revision is now' in stdout

    def test_push_new_file_git(self, webserver):

        clone_url = webserver.repo_url(GIT_REPO)

        # commit some stuff into this repo

        fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())

        fixture.create_fork(GIT_REPO, fork_name)

        clone_url = webserver.repo_url(fork_name)

        print [(x.repo_full_path,x.repo_path) for x in Repository.query()] # TODO: what is this for

        _check_proper_git_push(stdout, stderr)

    def test_push_invalidates_cache_hg(self, webserver):

        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

                                               == HG_REPO).scalar()

        if not key:

            key = CacheInvalidation(HG_REPO, HG_REPO)

            Session().add(key)

        key.cache_active = True

        Session().commit()

        clone_url = webserver.repo_url(HG_REPO)

        fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())

        fixture.create_fork(HG_REPO, fork_name)

        clone_url = webserver.repo_url(fork_name)