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

    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

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

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