"""

Pylons application test package

This package assumes the Pylons environment is already loaded.

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

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

Refer to docs/contributing.rst for details on running the test suite.

"""

import os

import re

import time

import logging

import datetime

import hashlib

import tempfile

from tempfile import _RandomNameSequence

import pylons

import pylons.test

from pylons import config, url

from pylons.i18n.translation import _get_translator

from pylons.util import ContextObj

from routes.util import URLGenerator

from webtest import TestApp

import pytest

from kallithea.lib.compat import unittest

from kallithea import is_windows

from kallithea.model.db import Notification, User, UserNotification

from kallithea.model.meta import Session

from kallithea.lib.utils2 import safe_str

os.environ['TZ'] = 'UTC'

if not is_windows:

    time.tzset()

log = logging.getLogger(__name__)

skipif = pytest.mark.skipif

parametrize = pytest.mark.parametrize

__all__ = [

    'skipif', 'parametrize', 'environ', 'url', 'TestController',

    'TESTS_TMP_PATH', 'HG_REPO', 'GIT_REPO', 'NEW_HG_REPO', 'NEW_GIT_REPO',

    'HG_FORK', 'GIT_FORK', 'TEST_USER_ADMIN_LOGIN', 'TEST_USER_ADMIN_PASS',

    'TEST_USER_ADMIN_EMAIL', 'TEST_USER_REGULAR_LOGIN', 'TEST_USER_REGULAR_PASS',

    'TEST_USER_REGULAR_EMAIL', 'TEST_USER_REGULAR2_LOGIN',

    'TEST_USER_REGULAR2_PASS', 'TEST_USER_REGULAR2_EMAIL', 'TEST_HG_REPO',

    'TEST_HG_REPO_CLONE', 'TEST_HG_REPO_PULL', 'TEST_GIT_REPO',

    'TEST_GIT_REPO_CLONE', 'TEST_GIT_REPO_PULL', 'HG_REMOTE_REPO',

    'GIT_REMOTE_REPO', 'SCM_TESTS',

]

# Invoke websetup with the current config file

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

environ = {}

#SOME GLOBALS FOR TESTS

TESTS_TMP_PATH = os.path.join(tempfile.gettempdir(), 'rc_test_%s' % _RandomNameSequence().next())

TEST_USER_ADMIN_LOGIN = 'test_admin'

TEST_USER_ADMIN_PASS = 'test12'

TEST_USER_ADMIN_EMAIL = 'test_admin@example.com'

TEST_USER_REGULAR_LOGIN = 'test_regular'

TEST_USER_REGULAR_PASS = 'test12'

TEST_USER_REGULAR_EMAIL = 'test_regular@example.com'

TEST_USER_REGULAR2_LOGIN = 'test_regular2'

TEST_USER_REGULAR2_PASS = 'test12'

TEST_USER_REGULAR2_EMAIL = 'test_regular2@example.com'

HG_REPO = u'vcs_test_hg'

GIT_REPO = u'vcs_test_git'

NEW_HG_REPO = u'vcs_test_hg_new'

NEW_GIT_REPO = u'vcs_test_git_new'

HG_FORK = u'vcs_test_hg_fork'

GIT_FORK = u'vcs_test_git_fork'

## VCS

SCM_TESTS = ['hg', 'git']

uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))

GIT_REMOTE_REPO = 'git://github.com/codeinn/vcs.git'

TEST_GIT_REPO = os.path.join(TESTS_TMP_PATH, GIT_REPO)

TEST_GIT_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)

TEST_GIT_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)

HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'

TEST_HG_REPO = os.path.join(TESTS_TMP_PATH, HG_REPO)

TEST_HG_REPO_CLONE = os.path.join(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)

TEST_HG_REPO_PULL = os.path.join(TESTS_TMP_PATH, 'vcshgpull%s' % uniq_suffix)

TEST_DIR = tempfile.gettempdir()

TEST_REPO_PREFIX = 'vcs-test'

# cached repos if any !

# comment out to get some other repos from bb or github

GIT_REMOTE_REPO = os.path.join(TESTS_TMP_PATH, GIT_REPO)

HG_REMOTE_REPO = os.path.join(TESTS_TMP_PATH, HG_REPO)

#skip ldap tests if LDAP lib is not installed

ldap_lib_installed = False

try:

    import ldap

    ldap.API_VERSION

    ldap_lib_installed = True

except ImportError:

    # means that python-ldap is not installed

    pass

try:

    import pam

    pam.PAM_TEXT_INFO

    pam_lib_installed = True

except ImportError:

    pam_lib_installed = False

class NullHandler(logging.Handler):

    def emit(self, record):

        pass

class TestController(object):

    """Pytest-style test controller"""

    # Note: pytest base classes cannot have an __init__ method

    @pytest.fixture(autouse=True)

    def app_fixture(self):

        self.wsgiapp = pylons.test.pylonsapp

        self.init_stack(self.wsgiapp.config)

        self.app = TestApp(self.wsgiapp)

        self.maxDiff = None

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

        return self.app

    def init_stack(self, config=None):

        if not config:

            config = pylons.test.pylonsapp.config

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

        pylons.app_globals._push_object(config['pylons.app_globals'])

        pylons.config._push_object(config)

        pylons.tmpl_context._push_object(ContextObj())

        # Initialize a translator for tests that utilize i18n

        translator = _get_translator(pylons.config.get('lang'))

        pylons.translator._push_object(translator)

        h = NullHandler()

        logging.getLogger("kallithea").addHandler(h)

    def remove_all_notifications(self):

        # query().delete() does not (by default) trigger cascades

        # ( http://docs.sqlalchemy.org/en/rel_0_7/orm/collections.html#passive-deletes )

        # so delete the UserNotification first to ensure referential integrity.

        UserNotification.query().delete()

        Notification.query().delete()

        Session().commit()

    def log_user(self, username=TEST_USER_ADMIN_LOGIN,

                 password=TEST_USER_ADMIN_PASS):

        self._logged_username = username

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

                                 {'username': username,

                                  'password': password})

        if 'Invalid username or password' in response.body:

"""

Helpers for fixture generation

"""

import os

import time

from kallithea.tests import *

from kallithea.model.db import Repository, User, RepoGroup, UserGroup, Gist

from kallithea.model.meta import Session

from kallithea.model.repo import RepoModel

from kallithea.model.user import UserModel

from kallithea.model.repo_group import RepoGroupModel

from kallithea.model.user_group import UserGroupModel

from kallithea.model.gist import GistModel

from kallithea.model.scm import ScmModel

from kallithea.lib.vcs.backends.base import EmptyChangeset

from os.path import dirname

FIXTURES = os.path.join(dirname(dirname(os.path.abspath(__file__))), 'tests', 'fixtures')

def error_function(*args, **kwargs):

    raise Exception('Total Crash !')

class Fixture(object):

    def __init__(self):

        pass

    def anon_access(self, status):

        """

        Context manager for controlling anonymous access.

        Anon access will be set and committed, but restored again when exiting the block.

        Usage:

        fixture = Fixture()

        with fixture.anon_access(False):

            stuff

        """

        class context(object):

            def __enter__(self):

                anon = User.get_default_user()

                self._before = anon.active

                anon.active = status

                Session().add(anon)

                Session().commit()

            def __exit__(self, exc_type, exc_val, exc_tb):

                anon = User.get_default_user()

                anon.active = self._before

                Session().add(anon)

                Session().commit()

        return context()

    def _get_repo_create_params(self, **custom):

        defs = dict(

            repo_name=None,

            repo_type='hg',

            clone_uri='',

            repo_group=u'-1',

            repo_description=u'DESC',

            repo_private=False,

            repo_landing_rev='rev:tip',

            repo_copy_permissions=False,

            repo_state=Repository.STATE_CREATED,

        )

        defs.update(custom)

        if 'repo_name_full' not in custom:

            defs.update({'repo_name_full': defs['repo_name']})

        # fix the repo name if passed as repo_name_full

        if defs['repo_name']:

            defs['repo_name'] = defs['repo_name'].split('/')[-1]

        return defs

    def _get_group_create_params(self, **custom):

        defs = dict(

            group_name=None,

            group_description=u'DESC',

            group_parent_id=None,

            perms_updates=[],

            perms_new=[],

            enable_locking=False,

            recursive=False

        )

        defs.update(custom)

        return defs

    def _get_user_create_params(self, name, **custom):

        defs = dict(

            username=name,

import time

from kallithea.model.db import User, UserIpMap

from kallithea.tests import *

class TestAdminPermissionsController(TestController):

    def test_index(self):

        self.log_user()

        response = self.app.get(url('admin_permissions'))

        # Test response...

    def test_index_ips(self):

        self.log_user()

        response = self.app.get(url('admin_permissions_ips'))

        # Test response...

        response.mustcontain('All IP addresses are allowed')

    def test_add_ips(self):

        self.log_user()

        default_user_id = User.get_default_user().user_id

        response = self.app.post(url('edit_user_ips_update', id=default_user_id),

                                 params=dict(new_ip='127.0.0.0/24',

                                 _authentication_token=self.authentication_token()))

        self.app.get(url('admin_permissions_ips'), status=302)

        # REMOTE_ADDR must match 127.0.0.0/24

        response = self.app.get(url('admin_permissions_ips'),

                                extra_environ={'REMOTE_ADDR': '127.0.0.1'})

        response.mustcontain('127.0.0.0/24')

        response.mustcontain('127.0.0.0 - 127.0.0.255')

        ## delete

        default_user_id = User.get_default_user().user_id

        del_ip_id = UserIpMap.query().filter(UserIpMap.user_id ==

                                             default_user_id).first().ip_id

        response = self.app.post(url('edit_user_ips_delete', id=default_user_id),

                                 params=dict(del_ip_id=del_ip_id,

                                             _authentication_token=self.authentication_token()),

                                 extra_environ={'REMOTE_ADDR': '127.0.0.1'})

        response = self.app.get(url('admin_permissions_ips'))

        response.mustcontain('All IP addresses are allowed')

        response.mustcontain(no=['127.0.0.0/24'])

        response.mustcontain(no=['127.0.0.0 - 127.0.0.255'])

    def test_index_overview(self):

        self.log_user()

        response = self.app.get(url('admin_permissions_perms'))

        # Test response...

        assert str('remote: Released lock on repo `%s`' % fork_name) in stdout

        #we need to cleanup the Session Here !

        Session.remove()

        r = Repository.get_by_repo_name(fork_name)

        assert r.locked == [None, None]

    #TODO: fix me ! somehow during tests hooks don't get called on Git

    def test_push_unlocks_repository_git(self):

        # enable locking

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

        fixture.create_fork(GIT_REPO, fork_name)

        r = Repository.get_by_repo_name(fork_name)

        r.enable_locking = True

        Session().add(r)

        Session().commit()

        #clone some temp

        DEST = _get_tmp_dir()

        clone_url = _construct_url(fork_name, dest=DEST)

        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)

        #check for lock repo after clone

        r = Repository.get_by_repo_name(fork_name)

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

        #push is ok and repo is now unlocked

        stdout, stderr = _add_files_and_push('git', DEST, clone_url=clone_url.split()[0])

        _check_proper_git_push(stdout, stderr)

        assert ('remote: Released lock on repo `%s`' % fork_name) in stderr

        #we need to cleanup the Session Here !

        Session.remove()

        r = Repository.get_by_repo_name(fork_name)

        assert r.locked == [None, None]

    def test_ip_restriction_hg(self):

        user_model = UserModel()

        try:

            user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')

            Session().commit()

            clone_url = _construct_url(HG_REPO)

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

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

        finally:

            #release IP restrictions

            for ip in UserIpMap.get_all():

                UserIpMap.delete(ip.ip_id)

            Session().commit()

        clone_url = _construct_url(HG_REPO)

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

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

        user_model = UserModel()

        try:

            user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')

            Session().commit()

            clone_url = _construct_url(GIT_REPO)

            stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)

            # The message apparently changed in Git 1.8.3, so match it loosely.

            assert re.search(r'\b403\b', stderr)

        finally:

            #release IP restrictions

            for ip in UserIpMap.get_all():

                UserIpMap.delete(ip.ip_id)

            Session().commit()

        clone_url = _construct_url(GIT_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url)

        assert 'Cloning into' in stdout + stderr

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