Removed config names from whoosh and celery,

celery is now configured based on the config name it's using
on celeryconfig. And whoosh uses it's own logger configured just for whoosh
Test creates a fresh whoosh index now, for more accurate checks
fixed tests for searching

celeryconfig.py

@@ -1,33 +1,45 b''
 # List of modules to import when celery starts.
 import sys
 import os
+import ConfigParser
+
+PYLONS_CONFIG_NAME = 'test.ini'
+
+root = os.getcwd()
+config = ConfigParser.ConfigParser({'here':root})
+config.read('%s/%s' % (root, PYLONS_CONFIG_NAME))
+PYLONS_CONFIG = config
+
+
+print config.items('app:main')
+
 sys.path.append(os.getcwd())
-CELERY_IMPORTS = ("pylons_app.lib.celerylib.tasks", )
+CELERY_IMPORTS = ("pylons_app.lib.celerylib.tasks",)
 
 ## Result store settings.
 CELERY_RESULT_BACKEND = "database"
 CELERY_RESULT_DBURI = "sqlite:///hg_app.db"
 
 BROKER_CONNECTION_MAX_RETRIES = 30
 
 ## Broker settings.
 BROKER_HOST = "localhost"
 BROKER_PORT = 5672
 BROKER_VHOST = "rabbitmqhost"
 BROKER_USER = "rabbitmq"
 BROKER_PASSWORD = "qweqwe"
 
 ## Worker settings
 ## If you're doing mostly I/O you can have more processes,
 ## but if mostly spending CPU, try to keep it close to the
 ## number of CPUs on your machine. If not set, the number of CPUs/cores
 ## available will be used.
 CELERYD_CONCURRENCY = 2
 # CELERYD_LOG_FILE = "celeryd.log"
 CELERYD_LOG_LEVEL = "DEBUG"
 CELERYD_MAX_TASKS_PER_CHILD = 1
 
 #CELERY_ALWAYS_EAGER = True
 #rabbitmqctl add_user rabbitmq qweqwe
 #rabbitmqctl add_vhost rabbitmqhost
-#rabbitmqctl set_permissions -p rabbitmqhost rabbitmq ".*" ".*" ".*"
  No newline at end of file
+#rabbitmqctl set_permissions -p rabbitmqhost rabbitmq ".*" ".*" ".*"

pylons_app/lib/celerylib/tasks.py

@@ -1,215 +1,208 b''
 from celery.decorators import task
 from celery.task.sets import subtask
+from celeryconfig import PYLONS_CONFIG as config
 from datetime import datetime, timedelta
-from os.path import dirname as dn
 from pylons.i18n.translation import _
 from pylons_app.lib.celerylib import run_task
 from pylons_app.lib.helpers import person
 from pylons_app.lib.smtp_mailer import SmtpMailer
 from pylons_app.lib.utils import OrderedDict
 from time import mktime
 from vcs.backends.hg import MercurialRepository
-import ConfigParser
 import calendar
-import os
 import traceback
 
-
-root = dn(dn(dn(dn(os.path.realpath(__file__)))))
-config = ConfigParser.ConfigParser({'here':root})
-config.read('%s/development.ini' % root)
-
 __all__ = ['whoosh_index', 'get_commits_stats',
            'reset_user_password', 'send_email']
 
 def get_session():
     from sqlalchemy import engine_from_config
     from sqlalchemy.orm import sessionmaker, scoped_session
     engine = engine_from_config(dict(config.items('app:main')), 'sqlalchemy.db1.')
     sa = scoped_session(sessionmaker(bind=engine))
     return sa
 
 def get_hg_settings():
     from pylons_app.model.db import HgAppSettings
     try:
         sa = get_session()
         ret = sa.query(HgAppSettings).all()
     finally:
         sa.remove()
         
     if not ret:
         raise Exception('Could not get application settings !')
     settings = {}
     for each in ret:
         settings['hg_app_' + each.app_settings_name] = each.app_settings_value    
     
     return settings
 
 def get_hg_ui_settings():
     from pylons_app.model.db import HgAppUi
     try:
         sa = get_session()
         ret = sa.query(HgAppUi).all()
     finally:
         sa.remove()
         
     if not ret:
         raise Exception('Could not get application ui settings !')
     settings = {}
     for each in ret:
         k = each.ui_key
         v = each.ui_value
         if k == '/':
             k = 'root_path'
         
         if k.find('.') != -1:
             k = k.replace('.', '_')
         
         if each.ui_section == 'hooks':
             v = each.ui_active
         
         settings[each.ui_section + '_' + k] = v  
     
     return settings   
 
 @task
 def whoosh_index(repo_location, full_index):
     log = whoosh_index.get_logger()
     from pylons_app.lib.indexers import DaemonLock
     from pylons_app.lib.indexers.daemon import WhooshIndexingDaemon, LockHeld
     try:
         l = DaemonLock()
         WhooshIndexingDaemon(repo_location=repo_location)\
             .run(full_index=full_index)
         l.release()
         return 'Done'
     except LockHeld:
         log.info('LockHeld')
         return 'LockHeld'    
 
 @task
 def get_commits_stats(repo):
     log = get_commits_stats.get_logger()
     aggregate = OrderedDict()
-    repos_path = get_hg_ui_settings()['paths_root_path'].replace('*','')
+    repos_path = get_hg_ui_settings()['paths_root_path'].replace('*', '')
     repo = MercurialRepository(repos_path + repo)
     #graph range
     td = datetime.today() + timedelta(days=1) 
     y, m, d = td.year, td.month, td.day
     ts_min = mktime((y, (td - timedelta(days=calendar.mdays[m])).month,
                         d, 0, 0, 0, 0, 0, 0,))
     ts_max = mktime((y, m, d, 0, 0, 0, 0, 0, 0,))
     
     def author_key_cleaner(k):
         k = person(k)
         k = k.replace('"', "'") #for js data compatibilty
         return k
             
     for cs in repo[:200]:#added limit 200 until fix #29 is made
         k = '%s-%s-%s' % (cs.date.timetuple()[0], cs.date.timetuple()[1],
                           cs.date.timetuple()[2])
         timetupple = [int(x) for x in k.split('-')]
         timetupple.extend([0 for _ in xrange(6)])
         k = mktime(timetupple)
         if aggregate.has_key(author_key_cleaner(cs.author)):
             if aggregate[author_key_cleaner(cs.author)].has_key(k):
                 aggregate[author_key_cleaner(cs.author)][k]["commits"] += 1
                 aggregate[author_key_cleaner(cs.author)][k]["added"] += len(cs.added)
                 aggregate[author_key_cleaner(cs.author)][k]["changed"] += len(cs.changed)
                 aggregate[author_key_cleaner(cs.author)][k]["removed"] += len(cs.removed)
                 
             else:
                 #aggregate[author_key_cleaner(cs.author)].update(dates_range)
                 if k >= ts_min and k <= ts_max:
                     aggregate[author_key_cleaner(cs.author)][k] = {}
                     aggregate[author_key_cleaner(cs.author)][k]["commits"] = 1
                     aggregate[author_key_cleaner(cs.author)][k]["added"] = len(cs.added)
                     aggregate[author_key_cleaner(cs.author)][k]["changed"] = len(cs.changed)
                     aggregate[author_key_cleaner(cs.author)][k]["removed"] = len(cs.removed) 
                                         
         else:
             if k >= ts_min and k <= ts_max:
                 aggregate[author_key_cleaner(cs.author)] = OrderedDict()
                 #aggregate[author_key_cleaner(cs.author)].update(dates_range)
                 aggregate[author_key_cleaner(cs.author)][k] = {}
                 aggregate[author_key_cleaner(cs.author)][k]["commits"] = 1
                 aggregate[author_key_cleaner(cs.author)][k]["added"] = len(cs.added)
                 aggregate[author_key_cleaner(cs.author)][k]["changed"] = len(cs.changed)
                 aggregate[author_key_cleaner(cs.author)][k]["removed"] = len(cs.removed)                 
     
     d = ''
     tmpl0 = u""""%s":%s"""
     tmpl1 = u"""{label:"%s",data:%s,schema:["commits"]},"""
     for author in aggregate:
         
         d += tmpl0 % (author,
                       tmpl1 \
                       % (author,
                     [{"time":x,
                       "commits":aggregate[author][x]['commits'],
                       "added":aggregate[author][x]['added'],
                       "changed":aggregate[author][x]['changed'],
                       "removed":aggregate[author][x]['removed'],
                       } for x in aggregate[author]]))
     if d == '':
         d = '"%s":{label:"%s",data:[[0,1],]}' \
             % (author_key_cleaner(repo.contact),
                author_key_cleaner(repo.contact))
     return (ts_min, ts_max, d)    
 
 @task
 def reset_user_password(user_email):
     log = reset_user_password.get_logger()
     from pylons_app.lib import auth
     from pylons_app.model.db import User
     
     try:
         
         try:
             sa = get_session()
             user = sa.query(User).filter(User.email == user_email).scalar()
             new_passwd = auth.PasswordGenerator().gen_password(8,
                              auth.PasswordGenerator.ALPHABETS_BIG_SMALL)
             user.password = auth.get_crypt_password(new_passwd)
             sa.add(user)
             sa.commit()
             log.info('change password for %s', user_email)
             if new_passwd is None:
                 raise Exception('unable to generate new password')
             
         except:
             log.error(traceback.format_exc())
             sa.rollback()
         
         run_task(send_email, user_email,
                  "Your new hg-app password",
                  'Your new hg-app password:%s' % (new_passwd))
         log.info('send new password mail to %s', user_email)
         
         
     except:
         log.error('Failed to update user password')
         log.error(traceback.format_exc())
     return True
 
 @task    
 def send_email(recipients, subject, body):
     log = send_email.get_logger()
     email_config = dict(config.items('DEFAULT')) 
     mail_from = email_config.get('app_email_from')
     user = email_config.get('smtp_username')
     passwd = email_config.get('smtp_password')
     mail_server = email_config.get('smtp_server')
     mail_port = email_config.get('smtp_port')
     tls = email_config.get('smtp_use_tls')
     ssl = False
     
     try:
-        m = SmtpMailer(mail_from, user, passwd, mail_server, 
+        m = SmtpMailer(mail_from, user, passwd, mail_server,
                        mail_port, ssl, tls)
         m.send(recipients, subject, body)  
     except:
         log.error('Mail sending failed')
         log.error(traceback.format_exc())
         return False
     return True

pylons_app/lib/indexers/daemon.py

@@ -1,226 +1,238 b''
 #!/usr/bin/env python
 # encoding: utf-8
 # whoosh indexer daemon for hg-app
 # Copyright (C) 2009-2010 Marcin Kuzminski <[e-mail]>
 #
 # 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; version 2
 # of the License or (at your opinion) any later version of the license.
 # 
 # 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, write to the Free Software
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
 # MA  02110-1301, USA.
 """
 Created on Jan 26, 2010
 
 @author: marcink
 A deamon will read from task table and run tasks
 """
 import sys
 import os
 from os.path import dirname as dn
 from os.path import join as jn
 
 #to get the pylons_app import
 project_path = dn(dn(dn(dn(os.path.realpath(__file__)))))
 sys.path.append(project_path)
 
 from pidlock import LockHeld, DaemonLock
-import traceback
-from pylons_app.config.environment import load_environment
 from pylons_app.model.hg_model import HgModel
 from pylons_app.lib.helpers import safe_unicode
 from whoosh.index import create_in, open_dir
 from shutil import rmtree
-from pylons_app.lib.indexers import ANALYZER, INDEX_EXTENSIONS, IDX_LOCATION, \
-SCHEMA, IDX_NAME
+from pylons_app.lib.indexers import INDEX_EXTENSIONS, IDX_LOCATION, SCHEMA, IDX_NAME
 
 import logging
-import logging.config
-logging.config.fileConfig(jn(project_path, 'development.ini'))
+
 log = logging.getLogger('whooshIndexer')
+# create logger
+log.setLevel(logging.DEBUG)
+
+# create console handler and set level to debug
+ch = logging.StreamHandler()
+ch.setLevel(logging.DEBUG)
+
+# create formatter
+formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+
+# add formatter to ch
+ch.setFormatter(formatter)
+
+# add ch to logger
+log.addHandler(ch)
 
 def scan_paths(root_location):
     return HgModel.repo_scan('/', root_location, None, True)
 
 class WhooshIndexingDaemon(object):
     """Deamon for atomic jobs"""
 
     def __init__(self, indexname='HG_INDEX', repo_location=None):
         self.indexname = indexname
         self.repo_location = repo_location
         self.initial = False
         if not os.path.isdir(IDX_LOCATION):
             os.mkdir(IDX_LOCATION)
             log.info('Cannot run incremental index since it does not'
                      ' yet exist running full build')
             self.initial = True
     
     def get_paths(self, root_dir):
         """recursive walk in root dir and return a set of all path in that dir
         excluding files in .hg dir"""
         index_paths_ = set()
         for path, dirs, files in os.walk(root_dir):
             if path.find('.hg') == -1:
                 for f in files:
                     index_paths_.add(jn(path, f))
     
         return index_paths_
     
     def add_doc(self, writer, path, repo):
         """Adding doc to writer"""
         
         ext = unicode(path.split('/')[-1].split('.')[-1].lower())
         #we just index the content of choosen files
         if ext in INDEX_EXTENSIONS:
             log.debug('    >> %s [WITH CONTENT]' % path)
             fobj = open(path, 'rb')
             content = fobj.read()
             fobj.close()
             u_content = safe_unicode(content)
         else:
             log.debug('    >> %s' % path)
             #just index file name without it's content
             u_content = u''
         
         
         
         try:
             os.stat(path)
             writer.add_document(owner=unicode(repo.contact),
                             repository=u"%s" % repo.name,
                             path=u"%s" % path,
                             content=u_content,
                             modtime=os.path.getmtime(path),
                             extension=ext)             
         except OSError, e:
             import errno
             if e.errno == errno.ENOENT:
                 log.debug('path %s does not exist or is a broken symlink' % path)
             else:
                 raise e                 
 
     
     def build_index(self):
         if os.path.exists(IDX_LOCATION):
             log.debug('removing previos index')
             rmtree(IDX_LOCATION)
             
         if not os.path.exists(IDX_LOCATION):
             os.mkdir(IDX_LOCATION)
         
         idx = create_in(IDX_LOCATION, SCHEMA, indexname=IDX_NAME)
         writer = idx.writer()
         
         for cnt, repo in enumerate(scan_paths(self.repo_location).values()):
             log.debug('building index @ %s' % repo.path)
         
             for idx_path in self.get_paths(repo.path):
                 self.add_doc(writer, idx_path, repo)
         writer.commit(merge=True)
                 
         log.debug('>>> FINISHED BUILDING INDEX <<<')
             
     
     def update_index(self):
         log.debug('STARTING INCREMENTAL INDEXING UPDATE')
             
         idx = open_dir(IDX_LOCATION, indexname=self.indexname)
         # The set of all paths in the index
         indexed_paths = set()
         # The set of all paths we need to re-index
         to_index = set()
         
         reader = idx.reader()
         writer = idx.writer()
     
         # Loop over the stored fields in the index
         for fields in reader.all_stored_fields():
             indexed_path = fields['path']
             indexed_paths.add(indexed_path)
     
             if not os.path.exists(indexed_path):
                 # This file was deleted since it was indexed
                 log.debug('removing from index %s' % indexed_path)
                 writer.delete_by_term('path', indexed_path)
     
             else:
                 # Check if this file was changed since it
                 # was indexed
                 indexed_time = fields['modtime']
                 
                 mtime = os.path.getmtime(indexed_path)
     
                 if mtime > indexed_time:
     
                     # The file has changed, delete it and add it to the list of
                     # files to reindex
                     log.debug('adding to reindex list %s' % indexed_path)
                     writer.delete_by_term('path', indexed_path)
                     to_index.add(indexed_path)
                     #writer.commit()
     
         # Loop over the files in the filesystem
         # Assume we have a function that gathers the filenames of the
         # documents to be indexed
         for repo in scan_paths(self.repo_location).values():
             for path in self.get_paths(repo.path):
                 if path in to_index or path not in indexed_paths:
                     # This is either a file that's changed, or a new file
                     # that wasn't indexed before. So index it!
                     self.add_doc(writer, path, repo)
                     log.debug('reindexing %s' % path)
     
         writer.commit(merge=True)
         #idx.optimize()
         log.debug('>>> FINISHED <<<')
         
     def run(self, full_index=False):
         """Run daemon"""
         if full_index or self.initial:
             self.build_index()
         else:
             self.update_index()
         
 if __name__ == "__main__":
     arg = sys.argv[1:]
     if len(arg) != 2:
         sys.stderr.write('Please specify indexing type [full|incremental]' 
                          'and path to repositories as script args \n')
         sys.exit()
     
     
     if arg[0] == 'full':
         full_index = True
     elif arg[0] == 'incremental':
         # False means looking just for changes
         full_index = False
     else:
         sys.stdout.write('Please use [full|incremental]' 
                          ' as script first arg \n')
         sys.exit()
     
     if not os.path.isdir(arg[1]):
         sys.stderr.write('%s is not a valid path \n' % arg[1])
         sys.exit()
     else:
         if arg[1].endswith('/'):
             repo_location = arg[1] + '*'
         else:
             repo_location = arg[1] + '/*'
     
     try:
         l = DaemonLock()
         WhooshIndexingDaemon(repo_location=repo_location)\
             .run(full_index=full_index)
         l.release()
+        reload(logging)
     except LockHeld:
         sys.exit(1)
 

pylons_app/tests/__init__.py

@@ -1,49 +1,70 b''
 """Pylons application test package
 
 This package assumes the Pylons environment is already loaded, such as
 when this script is imported from the `nosetests --with-pylons=test.ini`
 command.
 
 This module initializes the application via ``websetup`` (`paster
 setup-app`) and provides the base testing objects.
 """
 from unittest import TestCase
 
 from paste.deploy import loadapp
 from paste.script.appinstall import SetupCommand
 from pylons import config, url
 from routes.util import URLGenerator
 from webtest import TestApp
 import os
 from pylons_app.model import meta
+from pylons_app.lib.indexers import IDX_LOCATION
 import logging
+import shutil
 log = logging.getLogger(__name__) 
 
 import pylons.test
 
 __all__ = ['environ', 'url', 'TestController']
 
 # Invoke websetup with the current config file
 #SetupCommand('setup-app').run([pylons.test.pylonsapp.config['__file__']])
+def create_index(repo_location, full_index):
+    from pylons_app.lib.indexers import daemon
+    from pylons_app.lib.indexers.daemon import WhooshIndexingDaemon
+    from pylons_app.lib.indexers.pidlock import DaemonLock, LockHeld
+    
+    try:
+        l = DaemonLock()
+        WhooshIndexingDaemon(repo_location=repo_location)\
+            .run(full_index=full_index)
+        l.release()
+    except LockHeld:
+        pass    
+    
+if os.path.exists(IDX_LOCATION):
+    shutil.rmtree(IDX_LOCATION)
+    
+create_index('/tmp/*', True)    
 
 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):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username':'test_admin',
                                   'password':'test'})
         assert response.status == '302 Found', 'Wrong response code from login got %s' % response.status
         assert response.session['hg_app_user'].username == 'test_admin', 'wrong logged in user'
         return response.follow()
-    
  No newline at end of file
+
+ 

pylons_app/tests/functional/test_login.py

@@ -1,139 +1,139 b''
 from pylons_app.tests import *
 from pylons_app.model.db import User
 from pylons_app.lib.auth import check_password
 
 
 class TestLoginController(TestController):
 
     def test_index(self):
         response = self.app.get(url(controller='login', action='index'))
         assert response.status == '200 OK', 'Wrong response from login page got %s' % response.status
         # Test response...
 
     def test_login_admin_ok(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username':'test_admin',
                                   'password':'test'})
         assert response.status == '302 Found', 'Wrong response code from login got %s' % response.status
         assert response.session['hg_app_user'].username == 'test_admin', 'wrong logged in user'
         response = response.follow()
         assert 'auto description for vcs_test' in response.body
     
     def test_login_regular_ok(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username':'test_regular',
                                   'password':'test'})
         assert response.status == '302 Found', 'Wrong response code from login got %s' % response.status
         assert response.session['hg_app_user'].username == 'test_regular', 'wrong logged in user'
         response = response.follow()
         assert 'auto description for vcs_test' in response.body
         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',
                                   'password':'test'})
         assert response.status == '302 Found', 'Wrong response code from came from redirection'
         response = response.follow()
         
         assert response.status == '200 OK', 'Wrong response from login page got %s' % response.status
         assert 'Users administration' in response.body, 'No proper title in response'
         
                 
     def test_login_wrong(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username':'error',
                                   'password':'test'})
         assert response.status == '200 OK', 'Wrong response from login page'
         
         assert 'invalid user name' in response.body, 'No error username message in response'
         assert 'invalid password' in response.body, 'No error password message in response'
         
         
     def test_register(self):
         response = self.app.get(url(controller='login', action='register'))
         assert 'Sign Up to hg-app' in response.body, 'wrong page for user registration'
         
     def test_register_err_same_username(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username':'test_admin',
                                              'password':'test',
                                              'email':'[e-mail]',
                                              'name':'test',
                                              'lastname':'test'})
         
         assert response.status == '200 OK', 'Wrong response from register page got %s' % response.status
         assert 'This username already exists' in response.body 
         
     def test_register_err_wrong_data(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username':'xs',
                                              'password':'',
                                              'email':'goodmailm',
                                              'name':'test',
                                              'lastname':'test'})
         
         assert response.status == '200 OK', 'Wrong response from register page got %s' % response.status
         assert 'An email address must contain a single @' in response.body
         assert 'Enter a value 3 characters long or more' in response.body
         assert 'Please enter a value<' in response.body
         
         
         
     def test_register_ok(self):
         username = 'test_regular4'
         password = 'qweqwe'
-        email = '[e-mail]'
+        email = '[e-mail]'
         name = 'testname'
         lastname = 'testlastname'
         
         response = self.app.post(url(controller='login', action='register'),
                                             {'username':username,
                                              'password':password,
                                              'email':email,
                                              'name':name,
                                              'lastname':lastname})
         print response.body
         assert response.status == '302 Found', 'Wrong response from register page got %s' % response.status        
         assert 'You have successfully registered into hg-app' in response.session['flash'][0], 'No flash message about user registration'
         
         ret = self.sa.query(User).filter(User.username == 'test_regular4').one()
         assert ret.username == username , 'field mismatch %s %s' % (ret.username, username)
-        assert check_password(password,ret.password) == True , 'password mismatch'
+        assert check_password(password, ret.password) == True , 'password mismatch'
         assert ret.email == email , 'field mismatch %s %s' % (ret.email, email)
         assert ret.name == name , 'field mismatch %s %s' % (ret.name, name)
         assert ret.lastname == lastname , 'field mismatch %s %s' % (ret.lastname, lastname)
     
         
     def test_forgot_password_wrong_mail(self):    
         response = self.app.post(url(controller='login', action='password_reset'),
-                                            {'email':'[e-mail]',})
+                                            {'email':'[e-mail]', })
         
-        assert "That e-mail address doesn't exist" in response.body,'Missing error message about wrong email'
+        assert "That e-mail address doesn't exist" in response.body, 'Missing error message about wrong email'
                 
     def test_forgot_password(self):
         response = self.app.get(url(controller='login', action='password_reset'))
         assert response.status == '200 OK', 'Wrong response from login page got %s' % response.status
 
         username = 'test_password_reset_1'
         password = 'qweqwe'
         email = '[e-mail]'
         name = 'passwd'
         lastname = 'reset'
                 
         response = self.app.post(url(controller='login', action='register'),
                                             {'username':username,
                                              'password':password,
                                              'email':email,
                                              'name':name,
                                              'lastname':lastname})        
         #register new user for email test
         response = self.app.post(url(controller='login', action='password_reset'),
-                                            {'email':email,})
+                                            {'email':email, })
         print response.session['flash']
         assert 'You have successfully registered into hg-app' in response.session['flash'][0], 'No flash message about user registration'
         assert 'Your new password was sent' in response.session['flash'][1], 'No flash message about password reset'
         
         
         

pylons_app/tests/functional/test_search.py

@@ -1,29 +1,38 b''
 from pylons_app.tests import *
 from pylons_app.lib.indexers import IDX_LOCATION
 import os
 from nose.plugins.skip import SkipTest
 
 class TestSearchController(TestController):
 
     def test_index(self):
         self.log_user()
         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'
+        assert 'class="small" id="q" name="q" type="text"' in response.body, 'Search box content error'
         # Test response...
 
     def test_empty_search(self):
         
         if os.path.isdir(IDX_LOCATION):
             raise SkipTest('skipped due to existing index')
         else:
             self.log_user()
-            response = self.app.get(url(controller='search', action='index'),{'q':'vcs_test'})
-            assert 'There is no index to search in. Please run whoosh indexer' in response.body,'No error message about empty index'
+            response = self.app.get(url(controller='search', action='index'), {'q':'vcs_test'})
+            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'})
+        response = self.app.get(url(controller='search', action='index'), {'q':'def repo'})
         print response.body
-        assert '9 results' in response.body,'no message about proper search results'
+        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()
+        response = self.app.get(url(controller='search', action='index'), {'q':'repository:vcs_test def test'})
+        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'
+