"""Pylons application test package

This package assumes the Pylons environment is already loaded, such as

when this script is imported from the `nosetests --with-pylons=test.ini`

command.

This module initializes the application via ``websetup`` (`paster

setup-app`) and provides the base testing objects.

"""

from unittest import TestCase

from paste.deploy import loadapp

from paste.script.appinstall import SetupCommand

from pylons import config, url

from routes.util import URLGenerator

from webtest import TestApp

import os

from rhodecode.model import meta

import logging

log = logging.getLogger(__name__)

import pylons.test

__all__ = ['environ', 'url', 'TestController', 'TESTS_TMP_PATH', 'HG_REPO',

           'GIT_REPO', 'NEW_HG_REPO', 'NEW_GIT_REPO', 'HG_FORK', 'GIT_FORK', ]

# Invoke websetup with the current config file

#SetupCommand('setup-app').run([config_file])

##RUNNING DESIRED TESTS

#nosetests -x rhodecode.tests.functional.test_admin_settings:TestSettingsController.test_my_account

environ = {}

#SOME GLOBALS FOR TESTS

TESTS_TMP_PATH = '/tmp'

HG_REPO = 'vcs_test_hg'

GIT_REPO = 'vcs_test_git'

NEW_HG_REPO = 'vcs_test_hg_new'

NEW_GIT_REPO = 'vcs_test_git_new'

HG_FORK = 'vcs_test_hg_fork'

GIT_FORK = 'vcs_test_git_fork'

class TestController(TestCase):

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

        wsgiapp = pylons.test.pylonsapp

        config = wsgiapp.config

        self.app = TestApp(wsgiapp)

        url._push_object(URLGenerator(config['routes.map'], environ))

        self.sa = meta.Session

        self.index_location = config['app_conf']['index_dir']

        TestCase.__init__(self, *args, **kwargs)

    def log_user(self, username='test_admin', password='test12'):

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

                                 {'username':username,

                                  'password':password})

        if 'invalid user name' in response.body:

            assert False, 'could not login using %s %s' % (username, password)

        assert response.status == '302 Found', 'Wrong response code from login got %s' % response.status

        assert response.session['rhodecode_user'].username == username, 'wrong logged in user got %s expected %s' % (response.session['rhodecode_user'].username, username)

        return response.follow()

                                    revision='27cd5cce30c96924232dffcd24178a07ffeb5dfc',

                                    f_path='vcs/nodes.py'))

        print response.body

        assert """<optgroup label="Changesets">

<option selected="selected" value="8911406ad776fdd3d0b9932a2e89677e57405a48">r167:8911406ad776</option>

<option value="aa957ed78c35a1541f508d2ec90e501b0a9e3167">r165:aa957ed78c35</option>

<option value="48e11b73e94c0db33e736eaeea692f990cb0b5f1">r140:48e11b73e94c</option>

<option value="adf3cbf483298563b968a6c673cd5bde5f7d5eea">r126:adf3cbf48329</option>

<option value="6249fd0fb2cfb1411e764129f598e2cf0de79a6f">r113:6249fd0fb2cf</option>

<option value="75feb4c33e81186c87eac740cee2447330288412">r109:75feb4c33e81</option>

<option value="9a4dc232ecdc763ef2e98ae2238cfcbba4f6ad8d">r108:9a4dc232ecdc</option>

<option value="595cce4efa21fda2f2e4eeb4fe5f2a6befe6fa2d">r107:595cce4efa21</option>

<option value="4a8bd421fbc2dfbfb70d85a3fe064075ab2c49da">r104:4a8bd421fbc2</option>

<option value="57be63fc8f85e65a0106a53187f7316f8c487ffa">r102:57be63fc8f85</option>

<option value="5530bd87f7e2e124a64d07cb2654c997682128be">r101:5530bd87f7e2</option>

<option value="e516008b1c93f142263dc4b7961787cbad654ce1">r99:e516008b1c93</option>

<option value="41f43fc74b8b285984554532eb105ac3be5c434f">r93:41f43fc74b8b</option>

<option value="cc66b61b8455b264a7a8a2d8ddc80fcfc58c221e">r92:cc66b61b8455</option>

<option value="73ab5b616b3271b0518682fb4988ce421de8099f">r91:73ab5b616b32</option>

<option value="e0da75f308c0f18f98e9ce6257626009fdda2b39">r82:e0da75f308c0</option>

<option value="fb2e41e0f0810be4d7103bc2a4c7be16ee3ec611">r81:fb2e41e0f081</option>

<option value="602ae2f5e7ade70b3b66a58cdd9e3e613dc8a028">r76:602ae2f5e7ad</option>

<option value="a066b25d5df7016b45a41b7e2a78c33b57adc235">r73:a066b25d5df7</option>

<option value="637a933c905958ce5151f154147c25c1c7b68832">r61:637a933c9059</option>

<option value="0c21004effeb8ce2d2d5b4a8baf6afa8394b6fbc">r60:0c21004effeb</option>

<option value="a1f39c56d3f1d52d5fb5920370a2a2716cd9a444">r59:a1f39c56d3f1</option>

<option value="97d32df05c715a3bbf936bf3cc4e32fb77fe1a7f">r58:97d32df05c71</option>

<option value="08eaf14517718dccea4b67755a93368341aca919">r57:08eaf1451771</option>

<option value="22f71ad265265a53238359c883aa976e725aa07d">r56:22f71ad26526</option>

<option value="97501f02b7b4330924b647755663a2d90a5e638d">r49:97501f02b7b4</option>

<option value="86ede6754f2b27309452bb11f997386ae01d0e5a">r47:86ede6754f2b</option>

<option value="014c40c0203c423dc19ecf94644f7cac9d4cdce0">r45:014c40c0203c</option>

<option value="ee87846a61c12153b51543bf860e1026c6d3dcba">r30:ee87846a61c1</option>

<option value="9bb326a04ae5d98d437dece54be04f830cf1edd9">r26:9bb326a04ae5</option>

<option value="536c1a19428381cfea92ac44985304f6a8049569">r24:536c1a194283</option>

<option value="dc5d2c0661b61928834a785d3e64a3f80d3aad9c">r8:dc5d2c0661b6</option>

<option value="3803844fdbd3b711175fc3da9bdacfcd6d29a6fb">r7:3803844fdbd3</option>

</optgroup>

<optgroup label="Branches">

<option value="27cd5cce30c96924232dffcd24178a07ffeb5dfc">default</option>

<option value="97e8b885c04894463c51898e14387d80c30ed1ee">git</option>

<option value="2e6a2bf9356ca56df08807f4ad86d480da72a8f4">web</option>

</optgroup>

<optgroup label="Tags">

<option value="27cd5cce30c96924232dffcd24178a07ffeb5dfc">tip</option>

<option value="fd4bdb5e9b2a29b4393a4ac6caef48c17ee1a200">0.1.4</option>

<option value="17544fbfcd33ffb439e2b728b5d526b1ef30bfcf">0.1.3</option>

<option value="a7e60bff65d57ac3a1a1ce3b12a70f8a9e8a7720">0.1.2</option>

<option value="eb3a60fc964309c1a318b8dfe26aa2d1586c85ae">0.1.1</option>

</optgroup>""" in response.body, 'missing or wrong history in annotation'

        assert """<span style="text-transform: uppercase;"><a href="#">branch: default</a></span>""" in response.body, 'missing or wrong branch info'

    def test_archival(self):

        self.log_user()

        for arch_ext, info in ARCHIVE_SPECS.items():

            fname = '27cd5cce30c96924232dffcd24178a07ffeb5dfc%s' % arch_ext

            filename = '%s-%s' % (HG_REPO, fname)

            response = self.app.get(url(controller='files', action='archivefile',

                                        repo_name=HG_REPO,

                                        fname=fname))

            assert response.status == '200 OK', 'wrong response code'

            assert response.response._headers.items() == [('Pragma', 'no-cache'),

                                                  ('Cache-Control', 'no-cache'),

                                                  ('Content-Type', '%s; charset=utf-8' % info[0]),

                                                  ('Content-Disposition', 'attachment; filename=%s' % filename), ], 'wrong headers'

    def test_archival_wrong_ext(self):

        self.log_user()

        for arch_ext in ['tar', 'rar', 'x', '..ax', '.zipz']:

            fname = '27cd5cce30c96924232dffcd24178a07ffeb5dfc%s' % arch_ext

            response = self.app.get(url(controller='files', action='archivefile',

                                        repo_name=HG_REPO,

                                        fname=fname))

            assert 'Unknown archive type' in response.body

    def test_archival_wrong_revision(self):

        self.log_user()

        for rev in ['00x000000', 'tar', 'wrong', '@##$@$424213232', '232dffcd']:

            fname = '%s.zip' % rev

            response = self.app.get(url(controller='files', action='archivefile',

                                        repo_name=HG_REPO,

                                        fname=fname))

            assert 'Unknown revision' in response.body

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

"""

    rhodecode.tests.test_hg_operations

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

    Test suite for making push/pull operations

    :created_on: Dec 30, 2010

    :copyright: (c) 2010 by marcink.

    :license: LICENSE_NAME, see LICENSE_FILE for more details.

"""

import os

import shutil

import logging

from os.path import join as jn

from tempfile import _RandomNameSequence

from subprocess import Popen, PIPE

from paste.deploy import appconfig

from pylons import config

from sqlalchemy import engine_from_config

from rhodecode.lib.utils import add_cache

from rhodecode.model import init_model

from rhodecode.model import meta

from rhodecode.model.db import User, Repository

from rhodecode.lib.auth import get_crypt_password

from rhodecode.tests import TESTS_TMP_PATH, NEW_HG_REPO, HG_REPO

from rhodecode.config.environment import load_environment

conf = appconfig('config:development.ini', relative_to='./../../')

load_environment(conf.global_conf, conf.local_conf)

add_cache(conf)

USER = 'test_admin'

PASS = 'test12'

HOST = '127.0.0.1:5000'

DEBUG = True

log = logging.getLogger(__name__)

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)

        log.debug('Executing %s' % command)

        if DEBUG:

            print command

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

        stdout, stderr = p.communicate()

        if DEBUG:

            print stdout, stderr

        return stdout, stderr

def get_session():

    engine = engine_from_config(conf, 'sqlalchemy.db1.')

    init_model(engine)

    sa = meta.Session()

    return sa

def create_test_user(force=True):

    print 'creating test user'

    sa = get_session()

    user = sa.query(User).filter(User.username == USER).scalar()

    if force and user is not None:

        print 'removing current user'

        sa.delete(user)

        sa.commit()

    if user is None or force:

        print 'creating new one'

        new_usr = User()

        new_usr.username = USER

        new_usr.password = get_crypt_password(PASS)

        new_usr.email = 'mail@mail.com'

        new_usr.name = 'test'

        new_usr.lastname = 'lasttestname'

        new_usr.active = True

        new_usr.admin = True

        sa.add(new_usr)

        sa.commit()

    print 'done'

def create_test_repo(force=True):

    from rhodecode.model.repo import RepoModel

    sa = get_session()

    user = sa.query(User).filter(User.username == USER).scalar()

    if user is None:

        raise Exception('user not found')

    repo = sa.query(Repository).filter(Repository.repo_name == HG_REPO).scalar()

    if repo is None:

        print 'repo not found creating'

        form_data = {'repo_name':HG_REPO,

                     'repo_type':'hg',

                     'private':False, }

        rm = RepoModel(sa)

        rm.base_path = '/home/hg'

        rm.create(form_data, user)

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

# TESTS

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

def test_clone(no_errors=False):

    cwd = path = jn(TESTS_TMP_PATH, HG_REPO)

    try:

        shutil.rmtree(path, ignore_errors=True)

        os.makedirs(path)

        #print 'made dirs %s' % jn(path)

    except OSError:

        raise

    clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s %(dest)s' % \

                  {'user':USER,

                   'pass':PASS,

                   'host':HOST,

                   'cloned_repo':HG_REPO,

                   'dest':path}

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

    if no_errors is False:

        assert """adding file changes""" in stdout, 'no messages about cloning'

        assert """abort""" not in stderr , 'got error from clone'

def test_clone_anonymous_ok():

    cwd = path = jn(TESTS_TMP_PATH, HG_REPO)

    try:

        shutil.rmtree(path, ignore_errors=True)

        os.makedirs(path)

        #print 'made dirs %s' % jn(path)

    except OSError:

        raise

    clone_url = 'http://%(host)s/%(cloned_repo)s %(dest)s' % \

                  {'user':USER,

                   'pass':PASS,

                   'host':HOST,

                   'cloned_repo':HG_REPO,

                   'dest':path}

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

    print stdout, stderr

    assert """adding file changes""" in stdout, 'no messages about cloning'

    assert """abort""" not in stderr , 'got error from clone'

def test_clone_wrong_credentials():

    cwd = path = jn(TESTS_TMP_PATH, HG_REPO)

    try:

        shutil.rmtree(path, ignore_errors=True)

        os.makedirs(path)

        #print 'made dirs %s' % jn(path)

    except OSError:

        raise

    clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s %(dest)s' % \

                  {'user':USER + 'error',

                   'pass':PASS,

                   'host':HOST,

                   'cloned_repo':HG_REPO,

                   'dest':path}

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

    assert """abort: authorization failed""" in stderr , 'no error from wrong credentials'

def test_pull():

    pass

def test_push_modify_file(f_name='setup.py'):

    cwd = path = jn(TESTS_TMP_PATH, HG_REPO)

    modified_file = jn(TESTS_TMP_PATH, HG_REPO, f_name)

    for i in xrange(5):

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

        Command(cwd).execute(cmd)

        cmd = """hg ci -m 'changed file %s' %s """ % (i, modified_file)

        Command(cwd).execute(cmd)

    Command(cwd).execute('hg push %s' % jn(TESTS_TMP_PATH, HG_REPO))

def test_push_new_file(commits=15):

    test_clone(no_errors=True)

    cwd = path = jn(TESTS_TMP_PATH, HG_REPO)

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

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

    Command(cwd).execute('hg add %s' % added_file)

    for i in xrange(commits):

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

        Command(cwd).execute(cmd)

        cmd = """hg ci -m 'commited new %s' %s """ % (i, added_file)

        Command(cwd).execute(cmd)

    push_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s' % \

                  {'user':USER,

                   'pass':PASS,

                   'host':HOST,

                   'cloned_repo':HG_REPO,

                   'dest':jn(TESTS_TMP_PATH, HG_REPO)}

    Command(cwd).execute('hg push --verbose --debug %s' % push_url)

def test_push_wrong_credentials():

    cwd = path = jn(TESTS_TMP_PATH, HG_REPO)

    clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s' % \

                  {'user':USER + 'xxx',

                   'pass':PASS,

                   'host':HOST,

                   'cloned_repo':HG_REPO,

                   'dest':jn(TESTS_TMP_PATH, HG_REPO)}

    modified_file = jn(TESTS_TMP_PATH, HG_REPO, 'setup.py')

    for i in xrange(5):

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

        Command(cwd).execute(cmd)

        cmd = """hg ci -m 'commited %s' %s """ % (i, modified_file)

        Command(cwd).execute(cmd)

    Command(cwd).execute('hg push %s' % clone_url)

def test_push_wrong_path():

    cwd = path = jn(TESTS_TMP_PATH, HG_REPO)

    added_file = jn(path, 'somefile.py')

    try:

        shutil.rmtree(path, ignore_errors=True)

        os.makedirs(path)

        print 'made dirs %s' % jn(path)

    except OSError:

        raise

    Command(cwd).execute("""echo '' > %s""" % added_file)

    Command(cwd).execute("""hg init %s""" % path)

    Command(cwd).execute("""hg add %s""" % added_file)

    for i in xrange(2):

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

        Command(cwd).execute(cmd)

        cmd = """hg ci -m 'commited new %s' %s """ % (i, added_file)

        Command(cwd).execute(cmd)

    clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s' % \

                  {'user':USER,

                   'pass':PASS,

                   'host':HOST,

                   'cloned_repo':HG_REPO + '_error',

                   'dest':jn(TESTS_TMP_PATH, HG_REPO)}

    stdout, stderr = Command(cwd).execute('hg push %s' % clone_url)

    assert """abort: HTTP Error 403: Forbidden"""  in stderr

if __name__ == '__main__':

    create_test_user(force=False)

    create_test_repo()

    #test_push_modify_file()

    #test_clone()

    #test_clone_anonymous_ok()

    #test_clone_wrong_credentials()

    #test_pull()

    #test_push_wrong_path()

    #test_push_wrong_credentials()