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

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

Test suite for making push/pull operations.

Run it in two terminals::

 paster serve test.ini

 KALLITHEA_WHOOSH_TEST_DISABLE=1 KALLITHEA_NO_TMP_PATH=1 nosetests kallithea/tests/other/manual_test_vcs_operations.py

You must have git > 1.8.1 for tests to work fine

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 tempfile

import time

from os.path import join as jn

from tempfile import _RandomNameSequence

from subprocess import Popen, PIPE

from kallithea.tests import *

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

class Command(object):

    def __init__(self, cwd):

        self.cwd = cwd

    def execute(self, cmd, *args):

        """

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

        """

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

        if DEBUG:

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

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

        stdout, stderr = p.communicate()

        if DEBUG:

            print 'stdout:', repr(stdout)

            print 'stderr:', repr(stderr)

        return stdout, stderr

def _get_tmp_dir():

    return tempfile.mkdtemp(prefix='rc_integration_test')

def _construct_url(repo, dest=None, **kwargs):

    if dest is None:

        #make temp clone

        dest = _get_tmp_dir()

    params = {

        'user': TEST_USER_ADMIN_LOGIN,

        'passwd': TEST_USER_ADMIN_PASS,

        'host': HOST,

        'cloned_repo': repo,

        'dest': dest

    }

    params.update(**kwargs)

    if params['user'] and params['passwd']:

        _url = 'http://%(user)s:%(passwd)s@%(host)s/%(cloned_repo)s %(dest)s' % params

    else:

        _url = 'http://(host)s/%(cloned_repo)s %(dest)s' % params

    return _url

def _add_files_and_push(vcs, DEST, **kwargs):

    """

    Generate some files, add it to DEST repo and push back

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

    :param vcs:

    :param DEST:

    """

    # commit some stuff into this repo

    cwd = path = jn(DEST)

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

    added_file = jn(path, '%ssetup.py' % _RandomNameSequence().next())

    Command(cwd).execute('touch %s' % added_file)

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

    for i in xrange(kwargs.get('files_no', 3)):

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

        Command(cwd).execute(cmd)

        author_str = 'User ǝɯɐᴎ <me@email.com>'

        if vcs == 'hg':

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

                i, author_str, added_file

            )

        elif vcs == 'git':

            cmd = """EMAIL="me@email.com" git commit -m 'commited new %s' --author '%s' %s """ % (

                i, author_str, added_file

            )

        Command(cwd).execute(cmd)

    # PUSH it back

    _REPO = None

    if vcs == 'hg':

        _REPO = HG_REPO

    elif vcs == 'git':

        _REPO = GIT_REPO

    kwargs['dest'] = ''

    clone_url = _construct_url(_REPO, **kwargs)

    if 'clone_url' in kwargs:

        clone_url = kwargs['clone_url']

    stdout = stderr = None

    if vcs == 'hg':

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

    elif vcs == 'git':

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

    return stdout, stderr

def set_anonymous_access(enable=True):

    def test_clone_non_existing_path_git(self):

        clone_url = _construct_url('trololo')

        stdout, stderr = Command('/tmp').execute('git clone', clone_url)

        assert 'not found' in stderr

    def test_push_new_file_hg(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(HG_REPO, dest=DEST)

        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)

        stdout, stderr = _add_files_and_push('hg', DEST)

        assert 'pushing to' in stdout

        assert 'Repository size' in stdout

        assert 'Last revision is now' in stdout

    def test_push_new_file_git(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO, dest=DEST)

        stdout, stderr = Command('/tmp').execute('git clone', clone_url)

        # commit some stuff into this repo

        stdout, stderr = _add_files_and_push('git', DEST)

        print [(x.repo_full_path,x.repo_path) for x in Repository.get_all()]

        _check_proper_git_push(stdout, stderr)

    def test_push_invalidates_cache_hg(self):

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

                                               ==HG_REPO).scalar()

        if not key:

            key = CacheInvalidation(HG_REPO, HG_REPO)

        key.cache_active = True

        Session().add(key)

        Session().commit()

        DEST = _get_tmp_dir()

        clone_url = _construct_url(HG_REPO, dest=DEST)

        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)

        stdout, stderr = _add_files_and_push('hg', DEST, files_no=1)

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

                                               ==HG_REPO).one()

        self.assertEqual(key.cache_active, False)

    def test_push_invalidates_cache_git(self):

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

                                               ==GIT_REPO).scalar()

        if not key:

            key = CacheInvalidation(GIT_REPO, GIT_REPO)

        key.cache_active = True

        Session().add(key)

        Session().commit()

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO, dest=DEST)

        stdout, stderr = Command('/tmp').execute('git clone', clone_url)

        # commit some stuff into this repo

        stdout, stderr = _add_files_and_push('git', DEST, files_no=1)

        _check_proper_git_push(stdout, stderr)

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

                                               ==GIT_REPO).one()

        print CacheInvalidation.get_all()

        self.assertEqual(key.cache_active, False)

    def test_push_wrong_credentials_hg(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(HG_REPO, dest=DEST)

        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)

        stdout, stderr = _add_files_and_push('hg', DEST, user='bad',

                                             passwd='name')

        assert 'abort: authorization failed' in stderr

    def test_push_wrong_credentials_git(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO, dest=DEST)

        stdout, stderr = Command('/tmp').execute('git clone', clone_url)

        stdout, stderr = _add_files_and_push('git', DEST, user='bad',

                                             passwd='name')

        assert 'fatal: Authentication failed' in stderr

    def test_push_back_to_wrong_url_hg(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(HG_REPO, dest=DEST)

        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)

        stdout, stderr = _add_files_and_push('hg', DEST,

        assert 'HTTP Error 404: Not Found' in stderr

    def test_push_back_to_wrong_url_git(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO, dest=DEST)

        stdout, stderr = Command('/tmp').execute('git clone', clone_url)

        stdout, stderr = _add_files_and_push('git', DEST,

        assert 'not found' in stderr

    def test_clone_and_create_lock_hg(self):

        # enable locking

        r = Repository.get_by_repo_name(HG_REPO)

        r.enable_locking = True

        Session().add(r)

        Session().commit()

        # clone

        clone_url = _construct_url(HG_REPO)

        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)

        #check if lock was made

        r = Repository.get_by_repo_name(HG_REPO)

        assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id

    def test_clone_and_create_lock_git(self):

        # enable locking

        r = Repository.get_by_repo_name(GIT_REPO)

        r.enable_locking = True

        Session().add(r)

        Session().commit()

        # clone

        clone_url = _construct_url(GIT_REPO)

        stdout, stderr = Command('/tmp').execute('git clone', clone_url)

        #check if lock was made

        r = Repository.get_by_repo_name(GIT_REPO)

        assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id

    def test_clone_after_repo_was_locked_hg(self):

        #lock repo

        r = Repository.get_by_repo_name(HG_REPO)

        Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)

        #pull fails since repo is locked

        clone_url = _construct_url(HG_REPO)

        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)

        msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""

                % (HG_REPO, TEST_USER_ADMIN_LOGIN))

        assert msg in stderr

    def test_clone_after_repo_was_locked_git(self):

        #lock repo

        r = Repository.get_by_repo_name(GIT_REPO)

        Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)

        #pull fails since repo is locked

        clone_url = _construct_url(GIT_REPO)

        stdout, stderr = Command('/tmp').execute('git clone', clone_url)

        msg = ("""The requested URL returned error: 423""")

        assert msg in stderr

    def test_push_on_locked_repo_by_other_user_hg(self):

        #clone some temp

        DEST = _get_tmp_dir()

        clone_url = _construct_url(HG_REPO, dest=DEST)

        stdout, stderr = Command('/tmp').execute('hg clone', clone_url)

        #lock repo

        r = Repository.get_by_repo_name(HG_REPO)

        # let this user actually push !

        RepoModel().grant_user_permission(repo=r, user=TEST_USER_REGULAR_LOGIN,

                                          perm='repository.write')

        Session().commit()

        Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)

        #push fails repo is locked by other user !

        stdout, stderr = _add_files_and_push('hg', DEST,

                                             user=TEST_USER_REGULAR_LOGIN,

                                             passwd=TEST_USER_REGULAR_PASS)

        msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""

                % (HG_REPO, TEST_USER_ADMIN_LOGIN))

        assert msg in stderr

    def test_push_on_locked_repo_by_other_user_git(self):

        #clone some temp

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO, dest=DEST)

        stdout, stderr = Command('/tmp').execute('git clone', clone_url)

        #lock repo

        r = Repository.get_by_repo_name(GIT_REPO)

        # let this user actually push !

        RepoModel().grant_user_permission(repo=r, user=TEST_USER_REGULAR_LOGIN,

                                          perm='repository.write')

        Session().commit()

        Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)

        #push fails repo is locked by other user !

        stdout, stderr = _add_files_and_push('git', DEST,

                                             user=TEST_USER_REGULAR_LOGIN,

                                             passwd=TEST_USER_REGULAR_PASS)

        err = 'Repository `%s` locked by user `%s`' % (GIT_REPO, TEST_USER_ADMIN_LOGIN)

        assert err in stderr

        #TODO: fix this somehow later on Git, Git is stupid and even if we throw

## PASTE ##

#use = egg:Paste#http

## nr of worker threads to spawn

#threadpool_workers = 5

## max request before thread respawn

#threadpool_max_requests = 10

## option to use threads of process

#use_threadpool = true

## WAITRESS ##

use = egg:waitress#main

## number of worker threads

threads = 5

## MAX BODY SIZE 100GB

max_request_body_size = 107374182400

## use poll instead of select, fixes fd limits, may not work on old

## windows systems.

#asyncore_use_poll = True

## GUNICORN ##

#use = egg:gunicorn#main

## number of process workers. You must set `instance_id = *` when this option

## is set to more than one worker

#workers = 1

## process name

#proc_name = kallithea

## type of worker class, one of sync, eventlet, gevent, tornado

## recommended for bigger setup is using of of other than sync one

#worker_class = sync

#max_requests = 1000

## ammount of time a worker can handle request before it gets killed and

## restarted

#timeout = 3600

## UWSGI ##

## run with uwsgi --ini-paste-logged <inifile.ini>

#[uwsgi]

#socket = /tmp/uwsgi.sock

#master = true

#http = 127.0.0.1:5000

## set as deamon and redirect all output to file

#daemonize = ./uwsgi_kallithea.log

## master process PID

#pidfile = ./uwsgi_kallithea.pid

## stats server with workers statistics, use uwsgitop

## for monitoring, `uwsgitop 127.0.0.1:1717`

#stats = 127.0.0.1:1717

#memory-report = true

## log 5XX errors

#log-5xx = true

## Set the socket listen queue size.

#listen = 256

## Gracefully Reload workers after the specified amount of managed requests

## (avoid memory leaks).

#max-requests = 1000

## enable large buffers

#buffer-size = 65535

## socket and http timeouts ##

#http-timeout = 3600

#socket-timeout = 3600

## Log requests slower than the specified number of milliseconds.

#log-slow = 10

## Exit if no app can be loaded.

#need-app = true

## Set lazy mode (load apps in workers instead of master).

#lazy = true

## scaling ##

## set cheaper algorithm to use, if not set default will be used

#cheaper-algo = spare

## minimum number of workers to keep at all times

#cheaper = 1

## number of workers to spawn at startup

#cheaper-initial = 1

## maximum number of workers that can be spawned

#workers = 4

## how many workers should be spawned at a time

#cheaper-step = 1

## COMMON ##

host = 127.0.0.1

## middleware for hosting the WSGI application under a URL prefix

#[filter:proxy-prefix]

#use = egg:PasteDeploy#prefix

#prefix = /<your-prefix>

[app:main]

use = egg:kallithea

## enable proxy prefix middleware

#filter-with = proxy-prefix

full_stack = true

static_files = true

## Available Languages:

## cs de fr hu ja nl_BE pl pt_BR ru sk zh_CN zh_TW

lang =

cache_dir = %(here)s/data

index_dir = %(here)s/data/index

## perform a full repository scan on each server start, this should be

## set to false after first startup, to allow faster server restarts.

#initial_repo_scan = false

initial_repo_scan = true

## uncomment and set this path to use archive download cache

archive_cache_dir = %(here)s/tarballcache

## change this to unique ID for security

app_instance_uuid = test

## cut off limit for large diffs (size in bytes)

cut_off_limit = 256000

## use cache version of scm repo everywhere

#vcs_full_cache = true

vcs_full_cache = false

## force https in Kallithea, fixes https redirects, assumes it's always https

force_https = false

## use Strict-Transport-Security headers

use_htsts = false

## number of commits stats will parse on each iteration

commit_parse_limit = 25

## path to git executable

git_path = git

## git rev filter option, --all is the default filter, if you need to

## hide all refs in changelog switch this to --branches --tags

#git_rev_filter = --branches --tags

## RSS feed options

rss_cut_off_limit = 256000

rss_items_per_page = 10

rss_include_diff = false

## options for showing and identifying changesets

show_sha_length = 12

show_revision_number = true

## gist URL alias, used to create nicer urls for gist. This should be an

## url that does rewrites to _admin/gists/<gistid>.

## example: http://gist.kallithea.server/{gistid}. Empty means use the internal

## Kallithea url, ie. http[s]://your.kallithea.server/_admin/gists/<gistid>

gist_alias_url =

## white list of API enabled controllers. This allows to add list of

## controllers to which access will be enabled by api_key. eg: to enable

## api access to raw_files put `FilesController:raw`, to enable access to patches

## add `ChangesetController:changeset_patch`. This list should be "," separated

## Syntax is <ControllerClass>:<function>. Check debug logs for generated names

## Recommended settings below are commented out:

api_access_controllers_whitelist =

#    ChangesetController:changeset_patch,

#    ChangesetController:changeset_raw,

#    FilesController:raw,

#    FilesController:archivefile

## default encoding used to convert from and to unicode

## can be also a comma seperated list of encoding in case of mixed encodings

default_encoding = utf8

## issue tracker for Kallithea (leave blank to disable, absent for default)

#bugtracker = https://bitbucket.org/conservancy/kallithea/issues

## issue tracking mapping for commits messages

## comment out issue_pat, issue_server, issue_prefix to enable

## pattern to get the issues from commit messages

## default one used here is #<numbers> with a regex passive group for `#`

## {id} will be all groups matched from this pattern

issue_pat = (?:\s*#)(\d+)