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

<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

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)

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