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

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']})

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...

        #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 == ''