#sets the c attribute access when don't existing attribute are accessed

    config['pylons.strict_tmpl_context'] = True

    test = os.path.split(config['__file__'])[-1] == 'test.ini'

    if test:

        from rhodecode.lib.utils import create_test_env, create_test_index

    #MULTIPLE DB configs

    # Setup the SQLAlchemy database engine

    if config['debug'] and not test:

        #use query time debugging.

        from rhodecode.lib.timerproxy import TimerProxy

# MA  02110-1301, USA.

"""

Created on Aug 7, 2010

search controller for pylons

@author: marcink

"""

from pylons.controllers.util import abort, redirect

from rhodecode.lib.auth import LoginRequired

from rhodecode.lib.base import BaseController, render

from webhelpers.paginate import Page

from webhelpers.util import update_params

from pylons.i18n.translation import _

from whoosh.index import open_dir, EmptyIndexError

from whoosh.qparser import QueryParser, QueryParserError

from whoosh.query import Phrase

log = logging.getLogger(__name__)

class SearchController(BaseController):

    @LoginRequired()

    def __before__(self):

    def index(self, search_repo=None):

        c.repo_name = search_repo

        c.formated_results = []

        c.runtime = ''

        c.cur_query = request.GET.get('q', None)

        c.cur_search = search_type = {'content':'content',

                                      'commit':'content',

                                      'path':'path',

                                      'repository':'repository'}\

                                      .get(c.cur_type, 'content')

        if c.cur_query:

            cur_query = c.cur_query.lower()

        if c.cur_query:

            p = int(request.params.get('page', 1))

            highlight_items = set()

            try:

                searcher = idx.searcher()

                qp = QueryParser(search_type, schema=SCHEMA)

                if c.repo_name:

                    cur_query = u'repository:%s %s' % (c.repo_name, cur_query)

                try:

                    query = qp.parse(unicode(cur_query))

                    if isinstance(query, Phrase):

                        highlight_items.update(query.words)

                    else:

                        for i in query.all_terms():

                            if i[0] == 'content':

                                highlight_items.add(i[1])

                    matcher = query.matcher(searcher)

                    log.debug(query)

                    log.debug(highlight_items)

                    results = searcher.search(query)

                    res_ln = len(results)

                    c.runtime = '%s results (%.3f seconds)' \

                        % (res_ln, results.runtime)

                    def url_generator(**kw):

                        return update_params("?q=%s&type=%s" \

                                           % (c.cur_query, c.cur_search), **kw)

                    c.formated_results = Page(

                                ResultWrapper(search_type, searcher, matcher,

                                              highlight_items),

                                page=p, item_count=res_ln,

                                items_per_page=10, url=url_generator)

                except QueryParserError:

                    c.runtime = _('Invalid search query. Try quoting it.')

                searcher.close()

            except (EmptyIndexError, IOError):

                log.error(traceback.format_exc())

                log.error('Empty Index data')

                c.runtime = _('There is no index to search in. Please run whoosh indexer')

        # Return a rendered template

        return render('/search/search.html')

    Returns True if request's target is git server. ``HTTP_USER_AGENT`` would

    then have git client version given.

    :param environ:

    """

    http_user_agent = environ.get('HTTP_USER_AGENT')

        return True

    return False

def action_logger(user, action, repo, ipaddr, sa=None):

    """

    Action logger for various action made by users

    """Makes default test index

    :param repo_location:

    :param full_index:

    """

    from rhodecode.lib.indexers.daemon import WhooshIndexingDaemon

    from rhodecode.lib.pidlock import DaemonLock, LockHeld

    import shutil

    try:

        l = DaemonLock()

            .run(full_index=full_index)

        l.release()

    except LockHeld:

        pass

def create_test_env(repos_test_path, config):

    """Makes a fresh database and 

    install test repository into tmp dir

    """

    from rhodecode.lib.db_manage import DbManage

    import tarfile

    import shutil

    from os.path import dirname as dn, join as jn, abspath

    log = logging.getLogger('TestEnvCreator')

    # create logger

    dbmanage.create_default_user()

    dbmanage.admin_prompt()

    dbmanage.create_permissions()

    dbmanage.populate_default_permissions()

    #PART TWO make test repo

    cur_dir = dn(dn(abspath(__file__)))

    tar.close()

class UpgradeDb(command.Command):

    """Command used for paster to upgrade our database to newer version

    """

from webtest import TestApp

import os

from rhodecode.model import meta

import logging

import pylons.test

# Invoke websetup with the current config file

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

##RUNNING DESIRED TESTS

#nosetests rhodecode.tests.functional.test_admin_settings:TestSettingsController.test_my_account

environ = {}

class TestController(TestCase):

    def __init__(self, *args, **kwargs):

        wsgiapp = pylons.test.pylonsapp

        config = wsgiapp.config

        self.app = TestApp(wsgiapp)

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

        self.sa = meta.Session

        TestCase.__init__(self, *args, **kwargs)

    def log_user(self, username='test_admin', password='test12'):

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

                                 {'username':username,

                                  'password':password})

        print response

        if 'invalid user name' in response.body:

            assert False, 'could not login using %s %s' % (username, password)

        assert response.status == '302 Found', 'Wrong response code from login got %s' % response.status

        assert response.session['rhodecode_user'].username == username, 'wrong logged in user got %s expected %s' % (response.session['rhodecode_user'].username, username)

        return response.follow()

from rhodecode.tests import *

class TestBranchesController(TestController):

    def test_index(self):

        self.log_user()

        # Test response...

from rhodecode.tests import *

class TestChangelogController(TestController):

        self.log_user()

        assert """Small update at simplevcs app""" in response.body, 'missing info about commit message'

        #pagination

        # Test response after pagination...

        assert """<div class="date">commit 64: 46ad32a4f974@2010-04-20 00:33:21</div>"""in response.body, 'wrong info about commit 64'

from rhodecode.tests import *

class TestChangesetController(TestController):

    def test_index(self):

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

        # Test response...

class TestFeedController(TestController):

    def test_rss(self):

        self.log_user()

        response = self.app.get(url(controller='feed', action='rss',

        # Test response...

    def test_atom(self):

        self.log_user()

        response = self.app.get(url(controller='feed', action='atom',

        # Test response...

class TestFilesController(TestController):

    def test_index(self):

        self.log_user()

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

                                    revision='tip',

                                    f_path='/'))

        # Test response...

from rhodecode.tests import *

class TestAdminController(TestController):

    def test_index(self):

        self.log_user()

        #if global permission is set

        assert 'ADD NEW REPOSITORY' in response.body, 'Wrong main page'

        # Test response...

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

                                 {'username':'test_admin',

                                  'password':'test12'})

        assert response.status == '302 Found', 'Wrong response code from login got %s' % response.status

        assert response.session['rhodecode_user'].username == 'test_admin', 'wrong logged in user'

        response = response.follow()

    def test_login_regular_ok(self):

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

                                 {'username':'test_regular',

                                  'password':'test12'})

        print response

        assert response.status == '302 Found', 'Wrong response code from login got %s' % response.status

        assert response.session['rhodecode_user'].username == 'test_regular', 'wrong logged in user'

        response = response.follow()

        assert '<a title="Admin" href="/_admin">' not in response.body

    def test_login_ok_came_from(self):

        test_came_from = '/_admin/users'

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

                                 {'username':'test_admin',

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

        # Test response...

    def test_index_as_xml(self):

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

        self.log_user()

        description = 'description for newly created repo'

        private = False

        response = self.app.post(url('repos'), {'repo_name':repo_name,

        print response

        #test if we have a message for that repository

        print '-' * 100

        print response.session

        assert '''created repository %s''' % (repo_name) in response.session['flash'][0], 'No flash message about new repo'

        #test if the fork was created in the database

        new_repo = self.sa.query(Repository).filter(Repository.repo_name == repo_name).one()

        assert new_repo.repo_name == repo_name, 'wrong name of repo name in db'

        assert new_repo.description == description, 'wrong description'

        #test if repository is visible in the list ?

        response = response.follow()

        assert repo_name in response.body, 'missing new repo from the main repos list'

    def test_new(self):

        self.log_user()

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

    def test_new_as_xml(self):

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

    def test_update(self):

    def test_update_browser_fakeout(self):

    def test_delete(self):

        self.log_user()

        repo_name = 'vcs_test_new_to_delete'

        description = 'description for newly created repo'

        private = False

        response = self.app.post(url('repos'), {'repo_name':repo_name,

                                               'description':description,

                                               'private':private})

        print response

        #test if we have a message for that repository

        print '-' * 100

        print response.session

        assert '''created repository %s''' % (repo_name) in response.session['flash'][0], 'No flash message about new repo'

        #test if the repo was created in the database

        new_repo = self.sa.query(Repository).filter(Repository.repo_name == repo_name).one()

        assert new_repo.repo_name == repo_name, 'wrong name of repo name in db'

        assert new_repo.description == description, 'wrong description'

        #test if repository is visible in the list ?

        response = response.follow()

        assert repo_name in response.body, 'missing new repo from the main repos list'

        response = self.app.delete(url('repo', repo_name=repo_name))

        print '-' * 100

        print response.session

        assert '''deleted repository %s''' % (repo_name) in response.session['flash'][0], 'No flash message about delete repo'

        response.follow()

        #check if repo was deleted from db

        deleted_repo = self.sa.query(Repository).filter(Repository.repo_name == repo_name).scalar()

        assert deleted_repo is None, 'Deleted repository was found in db'

    def test_delete_browser_fakeout(self):

    def test_show(self):

        self.log_user()

    def test_show_as_xml(self):

    def test_edit(self):

    def test_edit_as_xml(self):

from rhodecode.tests import *

import os

from nose.plugins.skip import SkipTest

class TestSearchController(TestController):

    def test_index(self):

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

        print response.body

        assert 'class="small" id="q" name="q" type="text"' in response.body, 'Search box content error'

        # Test response...

    def test_empty_search(self):

            raise SkipTest('skipped due to existing index')

        else:

            self.log_user()

            assert 'There is no index to search in. Please run whoosh indexer' in response.body, 'No error message about empty index'

    def test_normal_search(self):

        self.log_user()

        response = self.app.get(url(controller='search', action='index'), {'q':'def repo'})

        print response.body

        assert '10 results' in response.body, 'no message about proper search results'

        assert 'Permission denied' not in response.body, 'Wrong permissions settings for that repo and user'

    def test_repo_search(self):

        self.log_user()

        print response.body

        assert '4 results' in response.body, 'no message about proper search results'

        assert 'Permission denied' not in response.body, 'Wrong permissions settings for that repo and user'

class TestSettingsController(TestController):

    def test_index(self):

        self.log_user()

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

        # Test response...

    def test_fork(self):

        self.log_user()

        response = self.app.get(url(controller='settings', action='fork',

    def test_fork_create(self):

        self.log_user()

        description = 'fork of vcs test'

        response = self.app.post(url(controller='settings', action='fork_create',

                                    repo_name=repo_name),

                                    {'fork_name':fork_name,

                                     'description':description,

                                     'private':'False'})

        #test if we have a message that fork is ok

        assert 'fork %s repository as %s task added' \

                      % (repo_name, fork_name) in response.session['flash'][0], 'No flash message about fork'

        #test if the fork was created in the database

        fork_repo = self.sa.query(Repository).filter(Repository.repo_name == fork_name).one()

        assert fork_repo.repo_name == fork_name, 'wrong name of repo name in new db fork repo'

        assert fork_repo.fork.repo_name == repo_name, 'wrong fork parrent'

        #test if fork is visible in the list ?

        response = response.follow()

        #check if fork is marked as fork

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

                                    repo_name=fork_name))

        assert 'Fork of %s' % repo_name in response.body, 'no message about that this repo is a fork'

from rhodecode.tests import *

class TestShortlogController(TestController):

    def test_index(self):

        self.log_user()

        # Test response...

from rhodecode.tests import *

class TestSummaryController(TestController):

    def test_index(self):

        self.log_user()

from rhodecode.tests import *

class TestTagsController(TestController):

    def test_index(self):

        self.log_user()

        # Test response...

        # Test response...

    def test_index_as_xml(self):

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

    def test_create(self):

    def test_new(self):

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

    def test_new_as_xml(self):

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

[app:main]

use = egg:rhodecode

full_stack = true

static_files = true

lang=en

cache_dir = %(here)s/data

####################################

###         BEAKER CACHE        ####

####################################

beaker.cache.data_dir=/%(here)s/data/cache/data

beaker.cache.lock_dir=/%(here)s/data/cache/lock