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

"""

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 os.path import join as jn

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.tests.parameterized import parameterized

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__ = [

    'ldap_lib_installed', 'pam_lib_installed',

    '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 = jn(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 = jn(TESTS_TMP_PATH, GIT_REPO)

TEST_GIT_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)

TEST_GIT_REPO_PULL = jn(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)

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

TEST_HG_REPO = jn(TESTS_TMP_PATH, HG_REPO)

TEST_HG_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)

TEST_HG_REPO_PULL = jn(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 = jn(TESTS_TMP_PATH, GIT_REPO)

HG_REMOTE_REPO = jn(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

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

        Notification.query().delete()

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

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

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

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

from kallithea.tests.api.api_base import _BaseTestApi

    REPO = GIT_REPO

    REPO_TYPE = 'git'

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

from kallithea.tests.api.api_base import _BaseTestApi

    REPO = HG_REPO

    REPO_TYPE = 'hg'

import os

import csv

import datetime

from kallithea.tests import *

from kallithea.model.db import UserLog

from kallithea.model.meta import Session

from kallithea.lib.utils2 import safe_unicode

dn = os.path.dirname

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

    @classmethod

    def setup_class(cls):

        UserLog.query().delete()

        Session().commit()

        def strptime(val):

            fmt = '%Y-%m-%d %H:%M:%S'

            if '.' not in val:

                return datetime.datetime.strptime(val, fmt)

            nofrag, frag = val.split(".")

            date = datetime.datetime.strptime(nofrag, fmt)

            frag = frag[:6]  # truncate to microseconds

            frag += (6 - len(frag)) * '0'  # add 0s

            return date.replace(microsecond=int(frag))

        with open(os.path.join(FIXTURES, 'journal_dump.csv')) as f:

            for row in csv.DictReader(f):

                ul = UserLog()

                for k, v in row.iteritems():

                    v = safe_unicode(v)

                    if k == 'action_date':

                        v = strptime(v)

                    if k in ['user_id', 'repository_id']:

                        # nullable due to FK problems

                        v = None

                    setattr(ul, k, v)

                Session().add(ul)

            Session().commit()

    @classmethod

    def teardown_class(cls):

        UserLog.query().delete()

        Session().commit()

    def test_index(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index'))

        response.mustcontain('Admin Journal')

    def test_filter_all_entries(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index',))

        response.mustcontain('2034 Entries')

    def test_filter_journal_filter_exact_match_on_repository(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index',

                                    filter='repository:xxx'))

        response.mustcontain('3 Entries')

    def test_filter_journal_filter_exact_match_on_repository_CamelCase(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index',

                                    filter='repository:XxX'))

        response.mustcontain('3 Entries')

    def test_filter_journal_filter_wildcard_on_repository(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index',

                                    filter='repository:*test*'))

        response.mustcontain('862 Entries')

    def test_filter_journal_filter_prefix_on_repository(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index',

                                    filter='repository:test*'))

        response.mustcontain('257 Entries')

    def test_filter_journal_filter_prefix_on_repository_CamelCase(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index',

                                    filter='repository:Test*'))

        response.mustcontain('257 Entries')

    def test_filter_journal_filter_prefix_on_repository_and_user(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index',

                                    filter='repository:test* AND username:demo'))

        response.mustcontain('130 Entries')

    def test_filter_journal_filter_prefix_on_repository_or_other_repo(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index',

                                    filter='repository:test* OR repository:xxx'))

        response.mustcontain('260 Entries')  # 257 + 3

    def test_filter_journal_filter_exact_match_on_username(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index',

                                    filter='username:demo'))

        response.mustcontain('1087 Entries')

    def test_filter_journal_filter_exact_match_on_username_camelCase(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index',

                                    filter='username:DemO'))

        response.mustcontain('1087 Entries')

    def test_filter_journal_filter_wildcard_on_username(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index',

                                    filter='username:*test*'))

        response.mustcontain('100 Entries')

    def test_filter_journal_filter_prefix_on_username(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index',

                                    filter='username:demo*'))

        response.mustcontain('1101 Entries')

    def test_filter_journal_filter_prefix_on_user_or_other_user(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index',

                                    filter='username:demo OR username:volcan'))

        response.mustcontain('1095 Entries')  # 1087 + 8

    def test_filter_journal_filter_wildcard_on_action(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index',

                                    filter='action:*pull_request*'))

        response.mustcontain('187 Entries')

    def test_filter_journal_filter_on_date(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index',

                                    filter='date:20121010'))

        response.mustcontain('47 Entries')

    def test_filter_journal_filter_on_date_2(self):

        self.log_user()

        response = self.app.get(url(controller='admin/admin', action='index',

                                    filter='date:20121020'))

        response.mustcontain('17 Entries')

from kallithea.tests import *

from kallithea.model.db import Setting

    def _enable_plugins(self, plugins_list):

        test_url = url(controller='admin/auth_settings',

                       action='auth_settings')

        params={'auth_plugins': plugins_list, '_authentication_token': self.authentication_token()}

        for plugin in plugins_list.split(','):

            enable = plugin.partition('kallithea.lib.auth_modules.')[-1]

            params.update({'%s_enabled' % enable: True})

        response = self.app.post(url=test_url, params=params)

        return params

        #self.checkSessionFlash(response, 'Auth settings updated successfully')

    def test_index(self):

        self.log_user()

        response = self.app.get(url(controller='admin/auth_settings',

                                    action='index'))

        response.mustcontain('Authentication Plugins')

    @skipif(not ldap_lib_installed, reason='skipping due to missing ldap lib')

    def test_ldap_save_settings(self):

        self.log_user()

        params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_ldap')

        params.update({'auth_ldap_host': u'dc.example.com',

                       'auth_ldap_port': '999',

                       'auth_ldap_tls_kind': 'PLAIN',

                       'auth_ldap_tls_reqcert': 'NEVER',

                       'auth_ldap_dn_user': 'test_user',

                       'auth_ldap_dn_pass': 'test_pass',

                       'auth_ldap_base_dn': 'test_base_dn',

                       'auth_ldap_filter': 'test_filter',

                       'auth_ldap_search_scope': 'BASE',

                       'auth_ldap_attr_login': 'test_attr_login',

                       'auth_ldap_attr_firstname': 'ima',

                       'auth_ldap_attr_lastname': 'tester',

                       'auth_ldap_attr_email': 'test@example.com'})

        test_url = url(controller='admin/auth_settings',

                       action='auth_settings')

        response = self.app.post(url=test_url, params=params)

        self.checkSessionFlash(response, 'Auth settings updated successfully')

        new_settings = Setting.get_auth_settings()

        assert new_settings['auth_ldap_host'] == u'dc.example.com', 'fail db write compare'

    @skipif(not ldap_lib_installed, reason='skipping due to missing ldap lib')

    def test_ldap_error_form_wrong_port_number(self):

        self.log_user()

        params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_ldap')

        params.update({'auth_ldap_host': '',

                       'auth_ldap_port': 'i-should-be-number',  # bad port num

                       'auth_ldap_tls_kind': 'PLAIN',

                       'auth_ldap_tls_reqcert': 'NEVER',

                       'auth_ldap_dn_user': '',

                       'auth_ldap_dn_pass': '',

                       'auth_ldap_base_dn': '',

                       'auth_ldap_filter': '',

                       'auth_ldap_search_scope': 'BASE',

                       'auth_ldap_attr_login': '',

                       'auth_ldap_attr_firstname': '',

                       'auth_ldap_attr_lastname': '',

                       'auth_ldap_attr_email': ''})

        test_url = url(controller='admin/auth_settings',

                       action='auth_settings')

        response = self.app.post(url=test_url, params=params)

        response.mustcontain("""<span class="error-message">"""

                             """Please enter a number</span>""")

    @skipif(not ldap_lib_installed, reason='skipping due to missing ldap lib')

    def test_ldap_error_form(self):

        self.log_user()

        params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_ldap')

        params.update({'auth_ldap_host': 'Host',

                       'auth_ldap_port': '123',

                       'auth_ldap_tls_kind': 'PLAIN',

                       'auth_ldap_tls_reqcert': 'NEVER',

                       'auth_ldap_dn_user': '',

                       'auth_ldap_dn_pass': '',

                       'auth_ldap_base_dn': '',

                       'auth_ldap_filter': '',

                       'auth_ldap_search_scope': 'BASE',

                       'auth_ldap_attr_login': '',  # <----- missing required input

                       'auth_ldap_attr_firstname': '',

                       'auth_ldap_attr_lastname': '',

                       'auth_ldap_attr_email': ''})

        test_url = url(controller='admin/auth_settings',

                       action='auth_settings')

        response = self.app.post(url=test_url, params=params)

        response.mustcontain("""<span class="error-message">The LDAP Login"""

                             """ attribute of the CN must be specified""")

    def test_ldap_login(self):

        pass

    def test_ldap_login_incorrect(self):

        pass

    def _container_auth_setup(self, **settings):

        self.log_user()

        params = self._enable_plugins('kallithea.lib.auth_modules.auth_internal,kallithea.lib.auth_modules.auth_container')

        params.update(settings)

        test_url = url(controller='admin/auth_settings',

                       action='auth_settings')

        response = self.app.post(url=test_url, params=params)

        response = response.follow()

        response.click('Log Out') # end admin login session

    def _container_auth_verify_login(self, resulting_username, **get_kwargs):

        response = self.app.get(

            url=url(controller='admin/my_account', action='my_account'),

            **get_kwargs

        )

        response.mustcontain('My Account %s' % resulting_username)

    def test_container_auth_login_header(self):

        self._container_auth_setup(

            auth_container_header='THE_USER_NAME',

            auth_container_email_header='',

            auth_container_firstname_header='',

            auth_container_lastname_header='',

            auth_container_fallback_header='',

            auth_container_clean_username='False',

        )

        self._container_auth_verify_login(

            extra_environ={'THE_USER_NAME': 'john@example.org'},

            resulting_username='john@example.org',

        )

    def test_container_auth_login_header_attr(self):

        self._container_auth_setup(

            auth_container_header='THE_USER_NAME',

            auth_container_email_header='THE_USER_EMAIL',

            auth_container_firstname_header='THE_USER_FIRSTNAME',

            auth_container_lastname_header='THE_USER_LASTNAME',

            auth_container_fallback_header='',

            auth_container_clean_username='False',

        )

        response = self.app.get(

            url=url(controller='admin/my_account', action='my_account'),

            extra_environ={'THE_USER_NAME': 'johnd',

                           'THE_USER_EMAIL': 'john@example.org',

                           'THE_USER_FIRSTNAME': 'John',

                           'THE_USER_LASTNAME': 'Doe',

                           }

        )

        assert response.form['email'].value == 'john@example.org'

        assert response.form['firstname'].value == 'John'

        assert response.form['lastname'].value == 'Doe'

    def test_container_auth_login_fallback_header(self):

        self._container_auth_setup(

            auth_container_header='THE_USER_NAME',

            auth_container_email_header='',

            auth_container_firstname_header='',

            auth_container_lastname_header='',

            auth_container_fallback_header='HTTP_X_YZZY',

            auth_container_clean_username='False',

        )

        self._container_auth_verify_login(

            headers={'X-Yzzy': r'foo\bar'},

            resulting_username=r'foo\bar',

        )

    def test_container_auth_clean_username_at(self):

        self._container_auth_setup(

            auth_container_header='REMOTE_USER',

            auth_container_email_header='',

            auth_container_firstname_header='',

            auth_container_lastname_header='',

            auth_container_fallback_header='',

            auth_container_clean_username='True',

        )

        self._container_auth_verify_login(

            extra_environ={'REMOTE_USER': 'john@example.org'},

            resulting_username='john',

        )

    def test_container_auth_clean_username_backslash(self):

        self._container_auth_setup(

            auth_container_header='REMOTE_USER',

from kallithea.tests import *

from kallithea.model.db import Setting

    def test_index(self):

        self.log_user()

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

        response.mustcontain('default_repo_private')

        response.mustcontain('default_repo_enable_statistics')

        response.mustcontain('default_repo_enable_downloads')

        response.mustcontain('default_repo_enable_locking')

    def test_index_as_xml(self):

        response = self.app.get(url('formatted_defaults', format='xml'))

    def test_create(self):

        response = self.app.post(url('defaults'),

            {'_authentication_token': self.authentication_token()})

    def test_new(self):

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

    def test_new_as_xml(self):

        response = self.app.get(url('formatted_new_default', format='xml'))

    def test_update_params_true_hg(self):

        self.log_user()

        params = {

            'default_repo_enable_locking': True,

            'default_repo_enable_downloads': True,

            'default_repo_enable_statistics': True,

            'default_repo_private': True,

            'default_repo_type': 'hg',

            '_authentication_token': self.authentication_token(),

        }

        response = self.app.put(url('default', id='default'), params=params)

        self.checkSessionFlash(response, 'Default settings updated successfully')

        params.pop('_authentication_token')

        defs = Setting.get_default_repo_settings()

        assert params == defs

    def test_update_params_false_git(self):

        self.log_user()

        params = {

            'default_repo_enable_locking': False,

            'default_repo_enable_downloads': False,

            'default_repo_enable_statistics': False,

            'default_repo_private': False,

            'default_repo_type': 'git',

            '_authentication_token': self.authentication_token(),

        }

        response = self.app.put(url('default', id='default'), params=params)

        self.checkSessionFlash(response, 'Default settings updated successfully')

        params.pop('_authentication_token')

        defs = Setting.get_default_repo_settings()

        assert params == defs

    def test_update_browser_fakeout(self):

        response = self.app.post(url('default', id=1), params=dict(_method='put', _authentication_token=self.authentication_token()))

    def test_delete(self):

        # Not possible due to CSRF protection.

        response = self.app.delete(url('default', id=1), status=405)

    def test_delete_browser_fakeout(self):

        response = self.app.post(url('default', id=1), params=dict(_method='delete', _authentication_token=self.authentication_token()))

    def test_show(self):

        response = self.app.get(url('default', id=1))

    def test_show_as_xml(self):

        response = self.app.get(url('formatted_default', id=1, format='xml'))

    def test_edit(self):

        response = self.app.get(url('edit_default', id=1))

    def test_edit_as_xml(self):

        response = self.app.get(url('formatted_edit_default', id=1, format='xml'))

from kallithea.tests import *

from kallithea.model.gist import GistModel

from kallithea.model.meta import Session

from kallithea.model.db import User, Gist

def _create_gist(f_name, content='some gist', lifetime=-1,

                 description=u'gist-desc', gist_type='public',

                 owner=TEST_USER_ADMIN_LOGIN):

    gist_mapping = {

        f_name: {'content': content}

    }

    user = User.get_by_username(owner)

    gist = GistModel().create(description, owner=user,

                       gist_mapping=gist_mapping, gist_type=gist_type,

                       lifetime=lifetime)

    Session().commit()

    return gist

    def teardown_method(self, method):

        for g in Gist.get_all():

            GistModel().delete(g)

        Session().commit()

    def test_index(self):

        self.log_user()

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

        # Test response...

        response.mustcontain('There are no gists yet')

        g1 = _create_gist('gist1').gist_access_id

        g2 = _create_gist('gist2', lifetime=1400).gist_access_id

        g3 = _create_gist('gist3', description=u'gist3-desc').gist_access_id

        g4 = _create_gist('gist4', gist_type='private').gist_access_id

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

        # Test response...

        response.mustcontain('gist: %s' % g1)

        response.mustcontain('gist: %s' % g2)

        response.mustcontain('Expires: in 23 hours')  # we don't care about the end

        response.mustcontain('gist: %s' % g3)

        response.mustcontain('gist3-desc')

        response.mustcontain(no=['gist: %s' % g4])

    def test_index_private_gists(self):

        self.log_user()

        gist = _create_gist('gist5', gist_type='private')

        response = self.app.get(url('gists', private=1))

        # Test response...

        #and privates

        response.mustcontain('gist: %s' % gist.gist_access_id)

    def test_create_missing_description(self):

        self.log_user()

        response = self.app.post(url('gists'),

                                 params={'lifetime': -1, '_authentication_token': self.authentication_token()},

                                 status=200)

        response.mustcontain('Missing value')

    def test_create(self):

        self.log_user()

        response = self.app.post(url('gists'),

                                 params={'lifetime': -1,

                                         'content': 'gist test',

                                         'filename': 'foo',

                                         'public': 'public',

                                         '_authentication_token': self.authentication_token()},

                                 status=302)

        response = response.follow()

        response.mustcontain('added file: foo')

        response.mustcontain('gist test')

        response.mustcontain('<div class="btn btn-mini btn-success disabled">Public Gist</div>')

    def test_create_with_path_with_dirs(self):

        self.log_user()

        response = self.app.post(url('gists'),

                                 params={'lifetime': -1,

                                         'content': 'gist test',

                                         'filename': '/home/foo',

                                         'public': 'public',

                                         '_authentication_token': self.authentication_token()},

                                 status=200)

        response.mustcontain('Filename cannot be inside a directory')

    def test_access_expired_gist(self):

        self.log_user()

        gist = _create_gist('never-see-me')

        gist.gist_expires = 0  # 1970

        Session().add(gist)

        Session().commit()

        response = self.app.get(url('gist', gist_id=gist.gist_access_id), status=404)

    def test_create_private(self):

        self.log_user()

        response = self.app.post(url('gists'),

                                 params={'lifetime': -1,

                                         'content': 'private gist test',

                                         'filename': 'private-foo',

                                         'private': 'private',

                                         '_authentication_token': self.authentication_token()},

                                 status=302)

        response = response.follow()

        response.mustcontain('added file: private-foo<')