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

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import datetime

import logging

import os

import pytest

import re

import tempfile

import time

from tg import config

from webtest import TestApp

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

from kallithea.model.meta import Session

from kallithea.lib.utils2 import safe_str

log = logging.getLogger(__name__)

skipif = pytest.mark.skipif

parametrize = pytest.mark.parametrize

# Hack: These module global values MUST be set to actual values before running any tests. This is currently done by conftest.py.

url = None

testapp = None

__all__ = [

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

    'ldap_lib_installed', 'pam_lib_installed', 'invalidate_all_caches',

    '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' % tempfile._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

def invalidate_all_caches():

    """Invalidate all beaker caches currently configured.

import os

import sys

import logging

import pkg_resources

from paste.deploy import loadwsgi

from routes.util import URLGenerator

import pytest

from pytest_localserver.http import WSGIServer

from kallithea.controllers.root import RootController

from kallithea.lib.utils import repo2db_mapper

from kallithea.model.user import UserModel

from kallithea.model.meta import Session

from kallithea.model.db import Setting, User, UserIpMap

from kallithea.model.scm import ScmModel

from kallithea.tests.base import invalidate_all_caches, TEST_USER_REGULAR_LOGIN, TESTS_TMP_PATH, \

    TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS

import kallithea.tests.base # FIXME: needed for setting testapp instance!!!

from tg.util.webtest import test_context

def pytest_configure():

    path = os.getcwd()

    sys.path.insert(0, path)

    pkg_resources.working_set.add_entry(path)

    # Disable INFO logging of test database creation, restore with NOTSET

    logging.disable(logging.INFO)

    context = loadwsgi.loadcontext(loadwsgi.APP, 'config:kallithea/tests/test.ini', relative_to=path)

    if os.environ.get('TEST_DB'):

        # swap config if we pass environment variable

        context.local_conf['sqlalchemy.url'] = os.environ.get('TEST_DB')

    from kallithea.tests.fixture import create_test_env, create_test_index

    # set KALLITHEA_NO_TMP_PATH=1 to disable re-creating the database and test repos

    if not int(os.environ.get('KALLITHEA_NO_TMP_PATH', 0)):

        create_test_env(TESTS_TMP_PATH, context.config())

    # set KALLITHEA_WHOOSH_TEST_DISABLE=1 to disable whoosh index during tests

    if not int(os.environ.get('KALLITHEA_WHOOSH_TEST_DISABLE', 0)):

        create_test_index(TESTS_TMP_PATH, context.config(), True)

    kallithea.tests.base.testapp = context.create()

    # do initial repo scan

    repo2db_mapper(ScmModel().repo_scan(TESTS_TMP_PATH))

    logging.disable(logging.NOTSET)

    kallithea.tests.base.url = URLGenerator(RootController().mapper, kallithea.tests.base.environ)

@pytest.fixture

def create_test_user():

    """Provide users that automatically disappear after test is over."""

    test_user_ids = []

    def _create_test_user(user_form):

        user = UserModel().create(user_form)

        test_user_ids.append(user.user_id)

        return user

    yield _create_test_user

    for user_id in test_user_ids:

        UserModel().delete(user_id)

    Session().commit()

def _set_settings(*kvtseq):

    session = Session()

    for kvt in kvtseq:

        assert len(kvt) in (2, 3)

        k = kvt[0]

        v = kvt[1]

        t = kvt[2] if len(kvt) == 3 else 'unicode'

        Setting.create_or_update(k, v, t)

    session.commit()

@pytest.fixture

def set_test_settings():

    """Restore settings after test is over."""

    # Save settings.

    settings_snapshot = [

        (s.app_settings_name, s.app_settings_value, s.app_settings_type)

        for s in Setting.query().all()]

    yield _set_settings

    # Restore settings.

    session = Session()

    keys = frozenset(k for (k, v, t) in settings_snapshot)

    for s in Setting.query().all():

        if s.app_settings_name not in keys:

            session.delete(s)

    for k, v, t in settings_snapshot:

        if t == 'list' and hasattr(v, '__iter__'):

            v = ','.join(v) # Quirk: must format list value manually.

        Setting.create_or_update(k, v, t)

    session.commit()

@pytest.fixture

def auto_clear_ip_permissions():

    """Fixture that provides nothing but clearing IP permissions upon test

    exit. This clearing is needed to avoid other test failing to make fake http

    accesses."""

    yield

    # cleanup

    user_model = UserModel()

    user_ids = []

    user_ids.append(User.get_default_user().user_id)

    user_ids.append(User.get_by_username(TEST_USER_REGULAR_LOGIN).user_id)

    for user_id in user_ids:

        for ip in UserIpMap.query().filter(UserIpMap.user_id == user_id):

            user_model.delete_extra_ip(user_id, ip.ip_id)

    # IP permissions are cached, need to invalidate this cache explicitly

    invalidate_all_caches()