Tests for the JSON-RPC web api.

"""

import os

import random

import mock

from kallithea.tests.base import *

from kallithea.tests.fixture import Fixture

from kallithea.lib.compat import json

from kallithea.lib.auth import AuthUser

from kallithea.model.user import UserModel

from kallithea.model.user_group import UserGroupModel

from kallithea.model.repo import RepoModel

from kallithea.model.repo_group import RepoGroupModel

from kallithea.model.meta import Session

from kallithea.model.scm import ScmModel

from kallithea.model.gist import GistModel

from kallithea.model.db import Repository, User, Setting, Ui

from kallithea.lib.utils2 import time_to_datetime

API_URL = '/_admin/api'

TEST_USER_GROUP = u'test_user_group'

TEST_REPO_GROUP = u'test_repo_group'

fixture = Fixture()

def _build_data(apikey, method, **kw):

    """

    Builds API data with given random ID

    :param random_id:

    """

    random_id = random.randrange(1, 9999)

    return random_id, json.dumps({

        "id": random_id,

        "api_key": apikey,

        "method": method,

        "args": kw

    })

jsonify = lambda obj: json.loads(json.dumps(obj))

def crash(*args, **kwargs):

    raise Exception('Total Crash !')

def api_call(test_obj, params):

    response = test_obj.app.post(API_URL, content_type='application/json',

                                 params=params)

    return response

## helpers

def make_user_group(name=TEST_USER_GROUP):

    gr = fixture.create_user_group(name, cur_user=TEST_USER_ADMIN_LOGIN)

    UserGroupModel().add_user_to_group(user_group=gr,

                                       user=TEST_USER_ADMIN_LOGIN)

    Session().commit()

    return gr

def make_repo_group(name=TEST_REPO_GROUP):

    gr = fixture.create_repo_group(name, cur_user=TEST_USER_ADMIN_LOGIN)

    Session().commit()

    return gr

class _BaseTestApi(object):

    REPO = None

    REPO_TYPE = None

    @classmethod

    def setup_class(cls):

        cls.usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)

        cls.apikey = cls.usr.api_key

        cls.test_user = UserModel().create_or_update(

            username='test-api',

            password='test',

            email='test@example.com',

            firstname=u'first',

            lastname=u'last'

        )

        Session().commit()

        cls.TEST_USER_LOGIN = cls.test_user.username

        cls.apikey_regular = cls.test_user.api_key

    @classmethod

    def teardown_class(cls):

        pass

    def setup_method(self, method):

        make_user_group()

        make_repo_group()

    def teardown_method(self, method):

        fixture.destroy_user_group(TEST_USER_GROUP)

        fixture.destroy_gists()

        fixture.destroy_repo_group(TEST_REPO_GROUP)

    def _compare_ok(self, id_, expected, given):

        expected = jsonify({

            'id': id_,

            'error': None,

            'result': expected

        })

        given = json.loads(given)

        assert expected == given

    def _compare_error(self, id_, expected, given):

        expected = jsonify({

            'id': id_,

            'error': expected,

            'result': None

        })

        given = json.loads(given)

        assert expected == given

    def test_Optional_object(self):

        from kallithea.controllers.api.api import Optional

        option1 = Optional(None)

        assert '<Optional:%s>' % None == repr(option1)

        assert option1() == None

        assert 1 == Optional.extract(Optional(1))

        assert 'trololo' == Optional.extract('trololo')

    def test_Optional_OAttr(self):

        from kallithea.controllers.api.api import Optional, OAttr

        option1 = Optional(OAttr('apiuser'))

        assert 'apiuser' == Optional.extract(option1)

    def test_OAttr_object(self):

        from kallithea.controllers.api.api import OAttr

        oattr1 = OAttr('apiuser')

        assert '<OptionalAttr:apiuser>' == repr(oattr1)

        assert oattr1() == oattr1

    def test_api_wrong_key(self):

        id_, params = _build_data('trololo', 'get_user')

        response = api_call(self, params)

        expected = 'Invalid API key'

        self._compare_error(id_, expected, given=response.body)

    def test_api_missing_non_optional_param(self):

        id_, params = _build_data(self.apikey, 'get_repo')

        response = api_call(self, params)

        expected = 'Missing non optional `repoid` arg in JSON DATA'

        self._compare_error(id_, expected, given=response.body)

    def test_api_missing_non_optional_param_args_null(self):

        id_, params = _build_data(self.apikey, 'get_repo')

        params = params.replace('"args": {}', '"args": null')

        response = api_call(self, params)

        expected = 'Missing non optional `repoid` arg in JSON DATA'

        self._compare_error(id_, expected, given=response.body)

    def test_api_missing_non_optional_param_args_bad(self):

        id_, params = _build_data(self.apikey, 'get_repo')

        params = params.replace('"args": {}', '"args": 1')

        response = api_call(self, params)

        expected = 'Missing non optional `repoid` arg in JSON DATA'

        self._compare_error(id_, expected, given=response.body)

    def test_api_args_is_null(self):

        id_, params = _build_data(self.apikey, 'get_users', )

        params = params.replace('"args": {}', '"args": null')

        response = api_call(self, params)

        assert response.status == '200 OK'

    def test_api_args_is_bad(self):

        id_, params = _build_data(self.apikey, 'get_users', )

        params = params.replace('"args": {}', '"args": 1')

        response = api_call(self, params)

        assert response.status == '200 OK'

    def test_api_args_different_args(self):

        import string

        expected = {

            'ascii_letters': string.ascii_letters,

            'ws': string.whitespace,

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

    Useful when manipulating IP permissions in a test and changes need to take

    effect immediately.

    Note: Any use of this function is probably a workaround - it should be

    replaced with a more specific cache invalidation in code or test."""

    from beaker.cache import cache_managers

    for cache in cache_managers.values():

        cache.clear()

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

            pytest.fail('could not login using %s %s' % (username, password))

        assert response.status == '302 Found'

        self.assert_authenticated_user(response, username)

        response = response.follow()

        return response.session['authuser']

    def _get_logged_user(self):

        return User.get_by_username(self._logged_username)

    def assert_authenticated_user(self, response, expected_username):

        cookie = response.session.get('authuser')

        user = cookie and cookie.get('user_id')

        user = user and User.get(user)

        user = user and user.username

        assert user == expected_username

    def authentication_token(self):

        return self.app.get(url('authentication_token')).body

    def checkSessionFlash(self, response, msg=None, skip=0, _matcher=lambda msg, m: msg in m):

        if 'flash' not in response.session:

            pytest.fail(safe_str(u'msg `%s` not found - session has no flash:\n%s' % (msg, response)))

        try:

            level, m = response.session['flash'][-1 - skip]

            if _matcher(msg, m):

                return

        except IndexError:

            pass

        pytest.fail(safe_str(u'msg `%s` not found in session flash (skipping %s): %s' %

                           (msg, skip,

                            ', '.join('`%s`' % m for level, m in response.session['flash']))))

    def checkSessionFlashRegex(self, response, regex, skip=0):

        self.checkSessionFlash(response, regex, skip=skip, _matcher=re.search)