Improved testing scenarios. Made test env creator

Fixed hg_model error message
some other tweeks and fixes
Models fixe for uniq email, and removed some extra not needed imports from model main module

pylons_app/config/environment.py

@@ -1,74 +1,77 b''
 """Pylons environment configuration"""
 from mako.lookup import TemplateLookup
 from pylons.configuration import PylonsConfig
 from pylons.error import handle_mako_error
 from pylons_app.config.routing import make_map
 from pylons_app.lib.auth import set_available_permissions, set_base_path
 from pylons_app.lib.utils import repo2db_mapper, make_ui, set_hg_app_config
 from pylons_app.model import init_model
 from pylons_app.model.hg_model import _get_repos_cached_initial
 from sqlalchemy import engine_from_config
 import logging
 import os
 import pylons_app.lib.app_globals as app_globals
 import pylons_app.lib.helpers
 
 log = logging.getLogger(__name__)
 
 def load_environment(global_conf, app_conf, initial=False):
     """Configure the Pylons environment via the ``pylons.config``
     object
     """
     config = PylonsConfig()
     
     # Pylons paths
     root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
     paths = dict(root=root,
                  controllers=os.path.join(root, 'controllers'),
                  static_files=os.path.join(root, 'public'),
                  templates=[os.path.join(root, 'templates')])
 
     # Initialize config with the basic options
     config.init_app(global_conf, app_conf, package='pylons_app', paths=paths)
 
     config['routes.map'] = make_map(config)
     config['pylons.app_globals'] = app_globals.Globals(config)
     config['pylons.h'] = pylons_app.lib.helpers
     
     # Setup cache object as early as possible
     import pylons
     pylons.cache._push_object(config['pylons.app_globals'].cache)
     
     # Create the Mako TemplateLookup, with the default auto-escaping
     config['pylons.app_globals'].mako_lookup = TemplateLookup(
         directories=paths['templates'],
         error_handler=handle_mako_error,
         module_directory=os.path.join(app_conf['cache_dir'], 'templates'),
         input_encoding='utf-8', default_filters=['escape'],
         imports=['from webhelpers.html import escape'])
 
     #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] == 'tests.ini'
+    test = os.path.split(config['__file__'])[-1] == 'test.ini'
+    if test:
+        from pylons_app.lib.utils import make_test_env
+        make_test_env()
     #MULTIPLE DB configs
     # Setup the SQLAlchemy database engine
     if config['debug'] and not test:
         #use query time debugging.
         from pylons_app.lib.timerproxy import TimerProxy
         sa_engine_db1 = engine_from_config(config, 'sqlalchemy.db1.',
                                                             proxy=TimerProxy())
     else:
         sa_engine_db1 = engine_from_config(config, 'sqlalchemy.db1.')
 
     init_model(sa_engine_db1)
     #init baseui
     config['pylons.app_globals'].baseui = make_ui('db')
     
     repo2db_mapper(_get_repos_cached_initial(config['pylons.app_globals'], initial))
     set_available_permissions(config)
     set_base_path(config)
     set_hg_app_config(config)
     # CONFIGURATION OPTIONS HERE (note: all config options will override
     # any Pylons config options)
     
     return config

pylons_app/controllers/admin/settings.py

@@ -1,293 +1,293 b''
 #!/usr/bin/env python
 # encoding: utf-8
 # settings controller for pylons
 # 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 July 14, 2010
 settings controller for pylons
 @author: marcink
 """
 from formencode import htmlfill
 from pylons import request, session, tmpl_context as c, url, app_globals as g, \
     config
 from pylons.controllers.util import abort, redirect
 from pylons.i18n.translation import _
 from pylons_app.lib import helpers as h
 from pylons_app.lib.auth import LoginRequired, HasPermissionAllDecorator, \
     HasPermissionAnyDecorator
 from pylons_app.lib.base import BaseController, render
 from pylons_app.lib.utils import repo2db_mapper, invalidate_cache, \
     set_hg_app_config, get_hg_settings, get_hg_ui_settings, make_ui
 from pylons_app.model.db import User, UserLog, HgAppSettings, HgAppUi
 from pylons_app.model.forms import UserForm, ApplicationSettingsForm, \
     ApplicationUiSettingsForm
 from pylons_app.model.hg_model import HgModel
 from pylons_app.model.user_model import UserModel
 from pylons_app.lib.celerylib import tasks,run_task
 import formencode
 import logging
 import traceback
  
 log = logging.getLogger(__name__)
 
 
 class SettingsController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
     # To properly map this controller, ensure your config/routing.py
     # file has a resource setup:
     #     map.resource('setting', 'settings', controller='admin/settings', 
     #         path_prefix='/admin', name_prefix='admin_')
 
 
     @LoginRequired()
     def __before__(self):
         c.admin_user = session.get('admin_user')
         c.admin_username = session.get('admin_username')
         super(SettingsController, self).__before__()
     
     
     @HasPermissionAllDecorator('hg.admin')    
     def index(self, format='html'):
         """GET /admin/settings: All items in the collection"""
         # url('admin_settings')
 
         defaults = get_hg_settings()
         defaults.update(get_hg_ui_settings())
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
         )  
     
     @HasPermissionAllDecorator('hg.admin')
     def create(self):
         """POST /admin/settings: Create a new item"""
         # url('admin_settings')
     
     @HasPermissionAllDecorator('hg.admin')
     def new(self, format='html'):
         """GET /admin/settings/new: Form to create a new item"""
         # url('admin_new_setting')
         
     @HasPermissionAllDecorator('hg.admin')
     def update(self, setting_id):
         """PUT /admin/settings/setting_id: Update an existing item"""
         # Forms posted to this method should contain a hidden field:
         #    <input type="hidden" name="_method" value="PUT" />
         # Or using helpers:
         #    h.form(url('admin_setting', setting_id=ID),
         #           method='put')
         # url('admin_setting', setting_id=ID)
         if setting_id == 'mapping':
             rm_obsolete = request.POST.get('destroy', False)
             log.debug('Rescanning directories with destroy=%s', rm_obsolete)
 
             initial = HgModel.repo_scan(g.paths[0][0], g.paths[0][1], g.baseui)
             repo2db_mapper(initial, rm_obsolete)
             invalidate_cache('cached_repo_list')
             h.flash(_('Repositories sucessfully rescanned'), category='success')            
         
         if setting_id == 'whoosh':
             repo_location = get_hg_ui_settings()['paths_root_path']
             full_index = request.POST.get('full_index',False)
-            task = run_task(tasks.whoosh_index,True,repo_location,full_index)
+            task = run_task(tasks.whoosh_index,repo_location,full_index)
             
             h.flash(_('Whoosh reindex task scheduled'), category='success')
         if setting_id == 'global':
             
             application_form = ApplicationSettingsForm()()
             try:
                 form_result = application_form.to_python(dict(request.POST))
             
                 try:
                     hgsettings1 = self.sa.query(HgAppSettings)\
                     .filter(HgAppSettings.app_settings_name == 'title').one()
                     hgsettings1.app_settings_value = form_result['hg_app_title'] 
                     
                     hgsettings2 = self.sa.query(HgAppSettings)\
                     .filter(HgAppSettings.app_settings_name == 'realm').one()
                     hgsettings2.app_settings_value = form_result['hg_app_realm'] 
                     
                     
                     self.sa.add(hgsettings1)
                     self.sa.add(hgsettings2)
                     self.sa.commit()
                     set_hg_app_config(config)
                     h.flash(_('Updated application settings'),
                             category='success')
                                     
                 except:
                     log.error(traceback.format_exc())
                     h.flash(_('error occured during updating application settings'),
                             category='error')
                                 
                     self.sa.rollback()
                     
 
             except formencode.Invalid as errors:
                 return htmlfill.render(
                      render('admin/settings/settings.html'),
                      defaults=errors.value,
                      errors=errors.error_dict or {},
                      prefix_error=False,
                      encoding="UTF-8") 
         
         if setting_id == 'mercurial':
             application_form = ApplicationUiSettingsForm()()
             try:
                 form_result = application_form.to_python(dict(request.POST))
             
                 try:
                     
                     hgsettings1 = self.sa.query(HgAppUi)\
                     .filter(HgAppUi.ui_key == 'push_ssl').one()
                     hgsettings1.ui_value = form_result['web_push_ssl']
                     
                     hgsettings2 = self.sa.query(HgAppUi)\
                     .filter(HgAppUi.ui_key == '/').one()
                     hgsettings2.ui_value = form_result['paths_root_path']                    
                     
                     
                     #HOOKS
                     hgsettings3 = self.sa.query(HgAppUi)\
                     .filter(HgAppUi.ui_key == 'changegroup.update').one()
                     hgsettings3.ui_active = bool(form_result['hooks_changegroup_update'])  
                     
                     hgsettings4 = self.sa.query(HgAppUi)\
                     .filter(HgAppUi.ui_key == 'changegroup.repo_size').one()
                     hgsettings4.ui_active = bool(form_result['hooks_changegroup_repo_size'])                                          
                     
                     
                     
                     
                     self.sa.add(hgsettings1)
                     self.sa.add(hgsettings2)
                     self.sa.add(hgsettings3)
                     self.sa.add(hgsettings4)
                     self.sa.commit()
                     
                     h.flash(_('Updated mercurial settings'),
                             category='success')
                                     
                 except:
                     log.error(traceback.format_exc())
                     h.flash(_('error occured during updating application settings'),
                             category='error')
                                 
                     self.sa.rollback()
                     
 
             except formencode.Invalid as errors:
                 return htmlfill.render(
                      render('admin/settings/settings.html'),
                      defaults=errors.value,
                      errors=errors.error_dict or {},
                      prefix_error=False,
                      encoding="UTF-8") 
                 
                 
                         
         return redirect(url('admin_settings'))
     
     @HasPermissionAllDecorator('hg.admin')
     def delete(self, setting_id):
         """DELETE /admin/settings/setting_id: Delete an existing item"""
         # Forms posted to this method should contain a hidden field:
         #    <input type="hidden" name="_method" value="DELETE" />
         # Or using helpers:
         #    h.form(url('admin_setting', setting_id=ID),
         #           method='delete')
         # url('admin_setting', setting_id=ID)
     
     @HasPermissionAllDecorator('hg.admin')
     def show(self, setting_id, format='html'):
         """GET /admin/settings/setting_id: Show a specific item"""
         # url('admin_setting', setting_id=ID)
     
     @HasPermissionAllDecorator('hg.admin')         
     def edit(self, setting_id, format='html'):
         """GET /admin/settings/setting_id/edit: Form to edit an existing item"""
         # url('admin_edit_setting', setting_id=ID)
 
 
     def my_account(self):
         """
         GET /_admin/my_account Displays info about my account 
         """
         # url('admin_settings_my_account')
         c.user = self.sa.query(User).get(c.hg_app_user.user_id)
         c.user_repos = []
         for repo in c.cached_repo_list.values():
             if repo.dbrepo.user.username == c.user.username:
                 c.user_repos.append(repo)
                 
         if c.user.username == 'default':
             h.flash(_("You can't edit this user since it's" 
               " crucial for entire application"), category='warning')
             return redirect(url('users'))
         
         defaults = c.user.__dict__
         return htmlfill.render(
             render('admin/users/user_edit_my_account.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
         ) 
 
     def my_account_update(self):
         """PUT /_admin/my_account_update: Update an existing item"""
         # Forms posted to this method should contain a hidden field:
         #    <input type="hidden" name="_method" value="PUT" />
         # Or using helpers:
         #    h.form(url('admin_settings_my_account_update'),
         #           method='put')
         # url('admin_settings_my_account_update', id=ID)
         user_model = UserModel()
         uid = c.hg_app_user.user_id
         _form = UserForm(edit=True, old_data={'user_id':uid})()
         form_result = {}
         try:
             form_result = _form.to_python(dict(request.POST))
             user_model.update_my_account(uid, form_result)
             h.flash(_('Your account was updated succesfully'),
                     category='success')
                            
         except formencode.Invalid as errors:
             #c.user = self.sa.query(User).get(c.hg_app_user.user_id)
             return htmlfill.render(
                 render('admin/users/user_edit_my_account.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8")
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('error occured during update of user %s') \
                     % form_result.get('username'), category='error')
                     
         return redirect(url('my_account'))
     
     @HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')
     def create_repository(self):
         """GET /_admin/create_repository: Form to create a new item"""
         new_repo = request.GET.get('repo', '')
         c.new_repo = h.repo_name_slug(new_repo)
 
         return render('admin/repos/repo_add_create_repository.html')
         

pylons_app/controllers/summary.py

@@ -1,70 +1,70 b''
 #!/usr/bin/env python
 # encoding: utf-8
 # summary controller for pylons
 # 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 April 18, 2010
 summary controller for pylons
 @author: marcink
 """
 from pylons import tmpl_context as c, request,url
 from pylons_app.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator
 from pylons_app.lib.base import BaseController, render
 from pylons_app.lib.utils import OrderedDict
 from pylons_app.model.hg_model import HgModel
 from webhelpers.paginate import Page
 from pylons_app.lib.celerylib import run_task
 from pylons_app.lib.celerylib.tasks import get_commits_stats
 import logging
 
 log = logging.getLogger(__name__)
 
 class SummaryController(BaseController):
     
     @LoginRequired()
     @HasRepoPermissionAnyDecorator('repository.read', 'repository.write',
                                    'repository.admin')           
     def __before__(self):
         super(SummaryController, self).__before__()
                 
     def index(self):
         hg_model = HgModel()
         c.repo_info = hg_model.get_repo(c.repo_name)
         c.repo_changesets = Page(list(c.repo_info[:10]), page=1, items_per_page=20)
         e = request.environ
         uri = u'%(protocol)s://%(user)s@%(host)s/%(repo_name)s' % {
                                         'protocol': e.get('wsgi.url_scheme'),
                                         'user':str(c.hg_app_user.username),
                                         'host':e.get('HTTP_HOST'),
                                         'repo_name':c.repo_name, }
         c.clone_repo_url = uri
         c.repo_tags = OrderedDict()
         for name, hash in c.repo_info.tags.items()[:10]:
             c.repo_tags[name] = c.repo_info.get_changeset(hash)
         
         c.repo_branches = OrderedDict()
         for name, hash in c.repo_info.branches.items()[:10]:
             c.repo_branches[name] = c.repo_info.get_changeset(hash)
         
-        task = run_task(get_commits_stats,False,c.repo_info.name)
+        task = run_task(get_commits_stats,c.repo_info.name)
         c.ts_min = task.result[0]
         c.ts_max = task.result[1]
         c.commit_data = task.result[2]
         
         return render('summary/summary.html')
 

pylons_app/lib/utils.py

@@ -1,364 +1,431 b''
 #!/usr/bin/env python
 # encoding: utf-8
 # Utilities 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.
+import shutil
 
 """
 Created on April 18, 2010
 Utilities for hg app
 @author: marcink
 """
 from beaker.cache import cache_region
 from mercurial import ui, config, hg
 from mercurial.error import RepoError
 from pylons_app.model import meta
 from pylons_app.model.db import Repository, User, HgAppUi, HgAppSettings
 from vcs.backends.base import BaseChangeset
 from vcs.utils.lazy import LazyProperty
 import logging
 import os
+from os.path import dirname as dn, join as jn
+import tarfile
 log = logging.getLogger(__name__)
 
 
 def get_repo_slug(request):
     return request.environ['pylons.routes_dict'].get('repo_name')
 
 def is_mercurial(environ):
     """
     Returns True if request's target is mercurial server - header
     ``HTTP_ACCEPT`` of such request would start with ``application/mercurial``.
     """
     http_accept = environ.get('HTTP_ACCEPT')
     if http_accept and http_accept.startswith('application/mercurial'):
         return True
     return False
 
 def check_repo_dir(paths):
     repos_path = paths[0][1].split('/')
     if repos_path[-1] in ['*', '**']:
         repos_path = repos_path[:-1]
     if repos_path[0] != '/':
         repos_path[0] = '/'
     if not os.path.isdir(os.path.join(*repos_path)):
         raise Exception('Not a valid repository in %s' % paths[0][1])
 
 def check_repo_fast(repo_name, base_path):
     if os.path.isdir(os.path.join(base_path, repo_name)):return False
     return True
 
 def check_repo(repo_name, base_path, verify=True):
 
     repo_path = os.path.join(base_path, repo_name)
 
     try:
         if not check_repo_fast(repo_name, base_path):
             return False
         r = hg.repository(ui.ui(), repo_path)
         if verify:
             hg.verify(r)
         #here we hnow that repo exists it was verified
         log.info('%s repo is already created', repo_name)
         return False
     except RepoError:
         #it means that there is no valid repo there...
         log.info('%s repo is free for creation', repo_name)
         return True
 
 def ask_ok(prompt, retries=4, complaint='Yes or no, please!'):
     while True:
         ok = raw_input(prompt)
         if ok in ('y', 'ye', 'yes'): return True
         if ok in ('n', 'no', 'nop', 'nope'): return False
         retries = retries - 1
         if retries < 0: raise IOError
         print complaint
         
 @cache_region('super_short_term', 'cached_hg_ui')
 def get_hg_ui_cached():
     try:
         sa = meta.Session
         ret = sa.query(HgAppUi).all()
     finally:
         meta.Session.remove()
     return ret
 
 
 def get_hg_settings():
     try:
         sa = meta.Session
         ret = sa.query(HgAppSettings).all()
     finally:
         meta.Session.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():
     try:
         sa = meta.Session
         ret = sa.query(HgAppUi).all()
     finally:
         meta.Session.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
 
 #propagated from mercurial documentation
 ui_sections = ['alias', 'auth',
                 'decode/encode', 'defaults',
                 'diff', 'email',
                 'extensions', 'format',
                 'merge-patterns', 'merge-tools',
                 'hooks', 'http_proxy',
                 'smtp', 'patch',
                 'paths', 'profiling',
                 'server', 'trusted',
                 'ui', 'web', ]
         
 def make_ui(read_from='file', path=None, checkpaths=True):        
     """
     A function that will read python rc files or database
     and make an mercurial ui object from read options
     
     @param path: path to mercurial config file
     @param checkpaths: check the path
     @param read_from: read from 'file' or 'db'
     """
 
     baseui = ui.ui()
 
     if read_from == 'file':
         if not os.path.isfile(path):
             log.warning('Unable to read config file %s' % path)
             return False
         log.debug('reading hgrc from %s', path)
         cfg = config.config()
         cfg.read(path)
         for section in ui_sections:
             for k, v in cfg.items(section):
                 baseui.setconfig(section, k, v)
                 log.debug('settings ui from file[%s]%s:%s', section, k, v)
         if checkpaths:check_repo_dir(cfg.items('paths'))                
               
         
     elif read_from == 'db':
         hg_ui = get_hg_ui_cached()
         for ui_ in hg_ui:
             if ui_.ui_active:
                 log.debug('settings ui from db[%s]%s:%s', ui_.ui_section, ui_.ui_key, ui_.ui_value)
                 baseui.setconfig(ui_.ui_section, ui_.ui_key, ui_.ui_value)
         
     
     return baseui
 
 
 def set_hg_app_config(config):
     hgsettings = get_hg_settings()
     
     for k, v in hgsettings.items():
         config[k] = v
 
 def invalidate_cache(name, *args):
     """Invalidates given name cache"""
     
     from beaker.cache import region_invalidate
     log.info('INVALIDATING CACHE FOR %s', name)
     
     """propagate our arguments to make sure invalidation works. First
     argument has to be the name of cached func name give to cache decorator
     without that the invalidation would not work"""
     tmp = [name]
     tmp.extend(args)
     args = tuple(tmp)
     
     if name == 'cached_repo_list':
         from pylons_app.model.hg_model import _get_repos_cached
         region_invalidate(_get_repos_cached, None, *args)
         
     if name == 'full_changelog':
         from pylons_app.model.hg_model import _full_changelog_cached
         region_invalidate(_full_changelog_cached, None, *args)
         
 class EmptyChangeset(BaseChangeset):
     
     revision = -1
     message = ''
     
     @LazyProperty
     def raw_id(self):
         """
         Returns raw string identifing this changeset, useful for web
         representation.
         """
         return '0' * 12
 
 
 def repo2db_mapper(initial_repo_list, remove_obsolete=False):
     """
     maps all found repositories into db
     """
     from pylons_app.model.repo_model import RepoModel
     
     sa = meta.Session
     user = sa.query(User).filter(User.admin == True).first()
     
     rm = RepoModel()
     
     for name, repo in initial_repo_list.items():
         if not sa.query(Repository).filter(Repository.repo_name == name).scalar():
             log.info('repository %s not found creating default', name)
                 
             form_data = {
                          'repo_name':name,
                          'description':repo.description if repo.description != 'unknown' else \
                                         'auto description for %s' % name,
                          'private':False
                          }
             rm.create(form_data, user, just_db=True)
 
 
     if remove_obsolete:
         #remove from database those repositories that are not in the filesystem
         for repo in sa.query(Repository).all():
             if repo.repo_name not in initial_repo_list.keys():
                 sa.delete(repo)
                 sa.commit()
 
     
     meta.Session.remove()
 
 from UserDict import DictMixin
 
 class OrderedDict(dict, DictMixin):
 
     def __init__(self, *args, **kwds):
         if len(args) > 1:
             raise TypeError('expected at most 1 arguments, got %d' % len(args))
         try:
             self.__end
         except AttributeError:
             self.clear()
         self.update(*args, **kwds)
 
     def clear(self):
         self.__end = end = []
         end += [None, end, end]         # sentinel node for doubly linked list
         self.__map = {}                 # key --> [key, prev, next]
         dict.clear(self)
 
     def __setitem__(self, key, value):
         if key not in self:
             end = self.__end
             curr = end[1]
             curr[2] = end[1] = self.__map[key] = [key, curr, end]
         dict.__setitem__(self, key, value)
 
     def __delitem__(self, key):
         dict.__delitem__(self, key)
         key, prev, next = self.__map.pop(key)
         prev[2] = next
         next[1] = prev
 
     def __iter__(self):
         end = self.__end
         curr = end[2]
         while curr is not end:
             yield curr[0]
             curr = curr[2]
 
     def __reversed__(self):
         end = self.__end
         curr = end[1]
         while curr is not end:
             yield curr[0]
             curr = curr[1]
 
     def popitem(self, last=True):
         if not self:
             raise KeyError('dictionary is empty')
         if last:
             key = reversed(self).next()
         else:
             key = iter(self).next()
         value = self.pop(key)
         return key, value
 
     def __reduce__(self):
         items = [[k, self[k]] for k in self]
         tmp = self.__map, self.__end
         del self.__map, self.__end
         inst_dict = vars(self).copy()
         self.__map, self.__end = tmp
         if inst_dict:
             return (self.__class__, (items,), inst_dict)
         return self.__class__, (items,)
 
     def keys(self):
         return list(self)
 
     setdefault = DictMixin.setdefault
     update = DictMixin.update
     pop = DictMixin.pop
     values = DictMixin.values
     items = DictMixin.items
     iterkeys = DictMixin.iterkeys
     itervalues = DictMixin.itervalues
     iteritems = DictMixin.iteritems
 
     def __repr__(self):
         if not self:
             return '%s()' % (self.__class__.__name__,)
         return '%s(%r)' % (self.__class__.__name__, self.items())
 
     def copy(self):
         return self.__class__(self)
 
     @classmethod
     def fromkeys(cls, iterable, value=None):
         d = cls()
         for key in iterable:
             d[key] = value
         return d
 
     def __eq__(self, other):
         if isinstance(other, OrderedDict):
             return len(self) == len(other) and self.items() == other.items()
         return dict.__eq__(self, other)
 
     def __ne__(self, other):
         return not self == other
+
+def make_test_env():
+    """Makes a fresh database from base64+zlib dump and 
+    install test repository into tmp dir
+    """
+    new_db_dump = """
+eJztXN1vE8sVn9nxR+wAIXDpFiiXjSAXfEOc2ElwQkVLPjYf5NNOAklUydrYG3tv1t5ldx0nuUJV\noL
+cPrVr1X7jSfUJ96nMfK1Xty23VqlWlPlRIlahUXbXqFUL0pTNjx5614xAoKEDmJ3t2zpkzM2fO\neHe+
+zno+PqU5qrRmWDnFkXqAB0AIbkkSAKANf8+BKprwFzI0G28ECXQ+PufFEYT+Tehz6L/oaSnK\nwcFxGP
+igFQfHjuMg4CehH7UA9Af0Y2ShWdSPLmOSg+N9x7U9eKf9PiC2nIWm4mTtri4nZ3Z9DE/5\nfOD0+RZY
+VFdXFVstWHoXPOPFvDbKU3TdKCbNgp39GLZ5MPtKW5WtWKmstqFmtqVtzZRWt6NQRFjk\ngkhESJ6kbe
+trim6rcFTAdcfuwqxhrNuprJLPqBnLKJhhSzWNpK1tq+aWkzXyN8wt3cjbScU0w7q2\nGqbyVSHYAXE5
+kSv15RTMtOKo2YxUikjf+SgKg4Dc/38C6Dn6Gn2FnqDH6K+Y5ODgeGfhRRD6/ST0\n+Ujo9ZLQ4yEhQi
+QUBBJCeFy4BLywHaCfCEXM+AJHOWpx39sMrux4IbzQ3gMc1XaSlpop6IoVvRxV\nLke6L4/cmx7vjedG
+4qmVmXvTW5nl7PDaSmFEXR6ejC+YVrpnsNi1fn17fHldj06p6YH84tzaGKBF\n5ZWcSq66Uorn8Iih5W
+/ZBolqejhl5O57mkEPqf6sOFCq3lRsu2hYaayHrTplJeJD/Uu3p7u3Er19\nS4sb26PmemQiE54vLKfn
+I8Wx2/Nd+XurmbH4TOpupHdk25I/sYbmCgDQstK0oHLdpWGmc1U3MqR6\nbICF123RHb/QDNpIm1rFnk
+HaJiWd0/Llpgzq41lzIJMrjMXi2/JmdyGxMDKnjs1FR9WMcduMb3TZ\nfZuZTXVs1uiS53NxY9yan4Vw
+PDNICqEl3dKNlKJnDdshbYh2R7o7uwc6I1EpGr3RHbvREwn3D/T3\nd/fuBFAzaHdpUu7csi6Tw4ou94
+zOLt3JxTNZo7g8muvV1Lg6sNj/SX4dD7srqenpfCJ6d3g5vKRM\njq/Ob3VHIXgJXaKx8PWBvoHrvfdg
+MzhPVDl/vgek1TWloO927tbUdsqeNzfurK5Frq+v5NbHZ1bG\nCnZxdnxxbGStmOudnwub6+rQYNxZku
+Wh28Ph9Nos2C3EfblVvhJlyPjvRY+Z8f91dzUHB8fhYf/x\nv3T/PwL47v87+iX6I45ycHC8dWhFV6Br
+7ukVUQ/cYzroOYnaXZLoBGqD1TmW0IzOw/IUAJL9v6Dg\nA+jP6Ofo+yiBelFA+IvwC2EFMzmOCBJBD/
+huMZsJ41+MZjuqFVYKjpFUUo62oThqosyV8mpRKtg4\nUtScrJTNdCqmSeNGwZFIFqmcRTPydwIeMPwp
+W2ZOyRcU/SVLLWViym1v8oDOLrbcvJGvFpbWbGVV\nV9NhvweEZCyWslRcWVnINGzNMawtiXJxaRX5kM
+8D+rqq8lZFtjaX+i2vB1zoxKL0dhrPSHSmj6u3\nFCzV4cH6fbuavSTFFEJp3KCUatsdqEa4aGkOqyel
+y8IhwQM6BhhhrE2akSVkWfQKxKJ9jGhN8/NG\nWZCM/0H0q5r9P/Q79FvM5ODgeOtBZvLBIAkDARI2Nb
+3E/h/O7wdDAAzBj+Cy8IXwpfAc/eZlat9R\noF+8eBE+bHXIgzSbIQcTyYJWiQjDCXlwQZYWBoemZKnC
+lq4GAwUtqaWliZkFeUxOSDOzC9LM4tTU\nNYmm2GqKPqEX5KWFMmtd3WLJDUUvqCyDjhKqNDQ7OyUPzh
+DmXGJiejCxLE3Ky9JVWl2IsBdnJuKL\nMssZHpeHJymjXMjEjHS1+5oUCYWCoYjgE+WLEGj5tLpp39Px
+MzlJhjtKJytNSkYqUfRgHPlFUYQ/\nMKhZyPhm08DjMgdlUVPgSENj4DSyN1hp6u6Er8Kob3hplGEYrg
+J2dxsrDLrZ6EpO6kYGlzCCdV2Y\nmJbrjVlS2G1Ohlc2aJ012TSqozuJLYpoiK0f8vjEm2Ij61MLJiP0
+4g15XywapRffzpTPL166BB8k\naQeZqpXTbBv/4Gwm6nd1FpNAuqxKNuo4RsLdf1W+buQzrjSXkV1VuO
+zjTgmG+vw+ceJSo5Yzmicj\nDNFE7n8BfQnQ33DAwcHxLqMFLxHEs47mkIGYrKM+xAsBMYZXBnquvLDC
+D4Wsmne0FF3/kPm/gL6m\n8//DVp6Dg+PNo3b+7wOPAHgEH8F/CFfRT9GvD1u/vbFzv8kvdnTAhxF2nW
+GrjqPlM3YNGdxrzbGb\nSOZuLN1o9uaScc3RXCnuVYhr+lZTi2sCd+C08iz4ZsAnxjtesAapZIrUMJpv
+Bl8me7SGcfxBqtkv\ntrfDzwLU+pWdJU212fgJl93ZFGJ06qPWwNg0rWLkuuVPwxm2RfcS2YVOWrVTlm
+a61o6uXimr4bJ4\npfp67r6So7MJeWJshhRcWf1ICXlUTsgzw/L87vpuj4XRrubsOjN2zCdOtjfqJNac
+yQhLtcSOHzhj\nlKVOlsb/fwL0FAccHBzvLQJIhHRpIJAYXRPQ8R+i3wP84eDgeNfRCX3gAoRjGyk7Sc
+78BUDPZdlJ\n0ZphSbvJZPyH6D8Afzg4ON5/HEMX4O7tD0v3/3OAPxwcHEcG1f0/hJ4A9Az9C184ODje
+Q/gQ+WcP\nKPgEevX5IL0GyPiP0Fdl/7/D1pKDg+PNYe/3f+j4/wSP/88OWz8ODo43Ab+H3O0CKl19Qu
+kaoPN/\nD/gcgM+FD4W7ws8OW886PNg+UTp4jlX8aJOOQR0a2XhrnVftbkrFubZM7+dkewA/zgYS9a6x
+1erq\nXWRr0thDZLdfJ3uU7PI+rXcMfYWT6Bq33WtSrVNprGW/Y2VXUyIsdSp28sAZoyx1+kGulXqTfx
+aq\ndrduZOxK5Ex9RxN2pZcx8So9XEozKw4D1Vdn6v0RFLdfeolM0r/U2d9buqRbvekZ/iv0IpulqrYr
+\nl9sRo+rBEAyR+x8/ADg4OI4gyPyf3/8cHEcTJf+fpwB/ODg4jgSaoBfQ/QB+/s/BcSRR3f+H6Bng\n
+e/8cHEcHpf1/CI+jHwEP3AToLtx8e9/9e//w8Hun6bHGDz+tvE+3uwfOxsW69+nYYw2WfjPHGtX9\n5A
+MdfNQo9P+eS7youNdyVuJq4ot2zRsdnLgLCYYip/b7w5jKqUX51IREv4F/FJ7YBy96ja963sJS\n34yd
+OXDGKEud/R8efZUt\n
+"""    
+    newdb = open('test.db','wb')
+    newdb.write(new_db_dump.decode('base64').decode('zlib'))
+    newdb.close()
+    
+    
+    #PART TWO make test repo
+    if os.path.isdir('/tmp/vcs_test'):
+        shutil.rmtree('/tmp/vcs_test')
+        
+    cur_dir = dn(dn(os.path.abspath(__file__)))
+    tar = tarfile.open(jn(cur_dir,'tests',"vcs_test.tar.gz"))
+    tar.extractall('/tmp')
+    tar.close()
+    
+    

pylons_app/model/__init__.py

@@ -1,30 +1,23 b''
 """The application's model objects"""
 import logging
-import sqlalchemy as sa
-from sqlalchemy import orm
 from pylons_app.model import meta
-from pylons_app.model.meta import Session
 log = logging.getLogger(__name__)
 
-# Add these two imports:
-import datetime
-from sqlalchemy import schema, types
-
 def init_model(engine):
     """Call me before using any of the tables or classes in the model"""
     log.info("INITIALIZING DB MODELS")
     meta.Base.metadata.bind = engine
     #meta.Base2.metadata.bind = engine2
 
 #THIS IS A TEST FOR EXECUTING SCRIPT AND LOAD PYLONS APPLICATION GLOBALS
 #from paste.deploy import appconfig
 #from pylons import config
 #from sqlalchemy import engine_from_config
 #from pylons_app.config.environment import load_environment
 #
 #conf = appconfig('config:development.ini', relative_to = './../../')
 #load_environment(conf.global_conf, conf.local_conf)
 #
 #engine = engine_from_config(config, 'sqlalchemy.')
 #init_model(engine)
 # DO SOMETHING

pylons_app/model/db.py

@@ -1,125 +1,125 b''
 from pylons_app.model.meta import Base
 from sqlalchemy import *
 from sqlalchemy.orm import relation, backref
 from sqlalchemy.orm.session import Session
 from vcs.utils.lazy import LazyProperty
 import logging
 
 log = logging.getLogger(__name__)
 
 class HgAppSettings(Base):
     __tablename__ = 'hg_app_settings'
     __table_args__ = (UniqueConstraint('app_settings_name'), {'useexisting':True})
     app_settings_id = Column("app_settings_id", INTEGER(), nullable=False, unique=True, default=None, primary_key=True)
     app_settings_name = Column("app_settings_name", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     app_settings_value = Column("app_settings_value", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
 
 class HgAppUi(Base):
     __tablename__ = 'hg_app_ui'
     __table_args__ = {'useexisting':True}
     ui_id = Column("ui_id", INTEGER(), nullable=False, unique=True, default=None, primary_key=True)
     ui_section = Column("ui_section", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     ui_key = Column("ui_key", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     ui_value = Column("ui_value", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     ui_active = Column("ui_active", BOOLEAN(), nullable=True, unique=None, default=True)
     
     
 class User(Base): 
     __tablename__ = 'users'
-    __table_args__ = (UniqueConstraint('username'), {'useexisting':True})
+    __table_args__ = (UniqueConstraint('username'), UniqueConstraint('email'), {'useexisting':True})
     user_id = Column("user_id", INTEGER(), nullable=False, unique=True, default=None, primary_key=True)
     username = Column("username", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     password = Column("password", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     active = Column("active", BOOLEAN(), nullable=True, unique=None, default=None)
     admin = Column("admin", BOOLEAN(), nullable=True, unique=None, default=False)
     name = Column("name", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     lastname = Column("lastname", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     email = Column("email", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     last_login = Column("last_login", DATETIME(timezone=False), nullable=True, unique=None, default=None)
     
     user_log = relation('UserLog')
     user_perms = relation('UserToPerm', primaryjoin="User.user_id==UserToPerm.user_id")
     
     @LazyProperty
     def full_contact(self):
         return '%s %s <%s>' % (self.name, self.lastname, self.email)
         
     def __repr__(self):
         return "<User('id:%s:%s')>" % (self.user_id, self.username)
     
     def update_lastlogin(self):
         """Update user lastlogin"""
         import datetime
         
         try:
             session = Session.object_session(self)
             self.last_login = datetime.datetime.now()
             session.add(self)
             session.commit()
             log.debug('updated user %s lastlogin',self.username)
         except Exception:
             session.rollback()        
     
       
 class UserLog(Base): 
     __tablename__ = 'user_logs'
     __table_args__ = {'useexisting':True}
     user_log_id = Column("user_log_id", INTEGER(), nullable=False, unique=True, default=None, primary_key=True)
     user_id = Column("user_id", INTEGER(), ForeignKey(u'users.user_id'), nullable=False, unique=None, default=None)
     user_ip = Column("user_ip", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None) 
     repository = Column("repository", TEXT(length=None, convert_unicode=False, assert_unicode=None), ForeignKey(u'repositories.repo_name'), nullable=False, unique=None, default=None)
     action = Column("action", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     action_date = Column("action_date", DATETIME(timezone=False), nullable=True, unique=None, default=None)
     
     user = relation('User')
     
 class Repository(Base):
     __tablename__ = 'repositories'
     __table_args__ = (UniqueConstraint('repo_name'), {'useexisting':True},)
     repo_id = Column("repo_id", INTEGER(), nullable=False, unique=True, default=None, primary_key=True)
     repo_name = Column("repo_name", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=False, unique=True, default=None)
     user_id = Column("user_id", INTEGER(), ForeignKey(u'users.user_id'), nullable=False, unique=False, default=None)
     private = Column("private", BOOLEAN(), nullable=True, unique=None, default=None)
     description = Column("description", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     
     user = relation('User')
     repo_to_perm = relation('RepoToPerm', cascade='all')
     
     def __repr__(self):
         return "<Repository('id:%s:%s')>" % (self.repo_id, self.repo_name)
         
 class Permission(Base):
     __tablename__ = 'permissions'
     __table_args__ = {'useexisting':True}
     permission_id = Column("permission_id", INTEGER(), nullable=False, unique=True, default=None, primary_key=True)
     permission_name = Column("permission_name", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     permission_longname = Column("permission_longname", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     
     def __repr__(self):
         return "<Permission('%s:%s')>" % (self.permission_id, self.permission_name)
 
 class RepoToPerm(Base):
     __tablename__ = 'repo_to_perm'
     __table_args__ = (UniqueConstraint('user_id', 'repository_id'), {'useexisting':True})
     repo_to_perm_id = Column("repo_to_perm_id", INTEGER(), nullable=False, unique=True, default=None, primary_key=True)
     user_id = Column("user_id", INTEGER(), ForeignKey(u'users.user_id'), nullable=False, unique=None, default=None)
     permission_id = Column("permission_id", INTEGER(), ForeignKey(u'permissions.permission_id'), nullable=False, unique=None, default=None)
     repository_id = Column("repository_id", INTEGER(), ForeignKey(u'repositories.repo_id'), nullable=False, unique=None, default=None) 
     
     user = relation('User')
     permission = relation('Permission')
     repository = relation('Repository')
 
 class UserToPerm(Base):
     __tablename__ = 'user_to_perm'
     __table_args__ = (UniqueConstraint('user_id', 'permission_id'), {'useexisting':True})
     user_to_perm_id = Column("user_to_perm_id", INTEGER(), nullable=False, unique=True, default=None, primary_key=True)
     user_id = Column("user_id", INTEGER(), ForeignKey(u'users.user_id'), nullable=False, unique=None, default=None)
     permission_id = Column("permission_id", INTEGER(), ForeignKey(u'permissions.permission_id'), nullable=False, unique=None, default=None)
     
     user = relation('User')
     permission = relation('Permission')
 
 
 
 

pylons_app/model/hg_model.py

@@ -1,174 +1,174 b''
 #!/usr/bin/env python
 # encoding: utf-8
 # Model 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 April 9, 2010
 Model for hg app
 @author: marcink
 """
 from beaker.cache import cache_region
 from mercurial import ui
 from mercurial.hgweb.hgwebdir_mod import findrepos
 from pylons.i18n.translation import _
 from pylons_app.lib.auth import HasRepoPermissionAny
 from pylons_app.model import meta
 from pylons_app.model.db import Repository, User
 from pylons_app.lib import helpers as h
 from vcs.exceptions import RepositoryError, VCSError
 import logging
 import os
 import sys
 log = logging.getLogger(__name__)
 
 try:
     from vcs.backends.hg import MercurialRepository
 except ImportError:
     sys.stderr.write('You have to import vcs module')
     raise Exception('Unable to import vcs')
 
 def _get_repos_cached_initial(app_globals, initial):
     """
     return cached dict with repos
     """
     g = app_globals
     return HgModel.repo_scan(g.paths[0][0], g.paths[0][1], g.baseui, initial)
 
 @cache_region('long_term', 'cached_repo_list')
 def _get_repos_cached():
     """
     return cached dict with repos
     """
     log.info('getting all repositories list')
     from pylons import app_globals as g
     return HgModel.repo_scan(g.paths[0][0], g.paths[0][1], g.baseui)
 
 @cache_region('super_short_term', 'cached_repos_switcher_list')
 def _get_repos_switcher_cached(cached_repo_list):
     repos_lst = []
     for repo in sorted(x.name.lower() for x in cached_repo_list.values()):
         if HasRepoPermissionAny('repository.write', 'repository.read', 'repository.admin')(repo, 'main page check'):
             repos_lst.append(repo)
     
     return repos_lst
 
 @cache_region('long_term', 'full_changelog')
 def _full_changelog_cached(repo_name):
     log.info('getting full changelog for %s', repo_name)
     return list(reversed(list(HgModel().get_repo(repo_name))))
 
 class HgModel(object):
     """
     Mercurial Model
     """
 
     def __init__(self):
         """
         Constructor
         """
     
     @staticmethod
     def repo_scan(repos_prefix, repos_path, baseui, initial=False):
         """
         Listing of repositories in given path. This path should not be a 
         repository itself. Return a dictionary of repository objects
         :param repos_path: path to directory it could take syntax with 
         * or ** for deep recursive displaying repositories
         """
         sa = meta.Session()
         def check_repo_dir(path):
             """
             Checks the repository
             :param path:
             """
             repos_path = path.split('/')
             if repos_path[-1] in ['*', '**']:
                 repos_path = repos_path[:-1]
             if repos_path[0] != '/':
                 repos_path[0] = '/'
             if not os.path.isdir(os.path.join(*repos_path)):
-                raise RepositoryError('Not a valid repository in %s' % path[0][1])        
+                raise RepositoryError('Not a valid repository in %s' % path)        
         if not repos_path.endswith('*'):
             raise VCSError('You need to specify * or ** at the end of path '
                             'for recursive scanning')
             
         check_repo_dir(repos_path)
         log.info('scanning for repositories in %s', repos_path)
         repos = findrepos([(repos_prefix, repos_path)])
         if not isinstance(baseui, ui.ui):
             baseui = ui.ui()
     
         repos_list = {}
         for name, path in repos:
             try:
                 #name = name.split('/')[-1]
                 if repos_list.has_key(name):
                     raise RepositoryError('Duplicate repository name %s found in'
                                     ' %s' % (name, path))
                 else:
                     
                     repos_list[name] = MercurialRepository(path, baseui=baseui)
                     repos_list[name].name = name
                     
                     dbrepo = None
                     if not initial:
                         dbrepo = sa.query(Repository)\
                             .filter(Repository.repo_name == name).scalar()
                             
                     if dbrepo:
                         log.info('Adding db instance to cached list')
                         repos_list[name].dbrepo = dbrepo
                         repos_list[name].description = dbrepo.description
                         if dbrepo.user:
                             repos_list[name].contact = dbrepo.user.full_contact
                         else:
                             repos_list[name].contact = sa.query(User)\
                             .filter(User.admin == True).first().full_contact
             except OSError:
                 continue
         meta.Session.remove()
         return repos_list
         
     def get_repos(self):
         for name, repo in _get_repos_cached().items():
             if repo._get_hidden():
                 #skip hidden web repository
                 continue
             
             last_change = repo.last_change
             tip = h.get_changeset_safe(repo, 'tip')
                 
             tmp_d = {}
             tmp_d['name'] = repo.name
             tmp_d['name_sort'] = tmp_d['name'].lower()
             tmp_d['description'] = repo.description
             tmp_d['description_sort'] = tmp_d['description']
             tmp_d['last_change'] = last_change
             tmp_d['last_change_sort'] = last_change[1] - last_change[0]
             tmp_d['tip'] = tip.raw_id
             tmp_d['tip_sort'] = tip.revision 
             tmp_d['rev'] = tip.revision
             tmp_d['contact'] = repo.contact
             tmp_d['contact_sort'] = tmp_d['contact']
             tmp_d['repo_archives'] = list(repo._get_archives())
             tmp_d['last_msg'] = tip.message
             tmp_d['repo'] = repo
             yield tmp_d
 
     def get_repo(self, repo_name):
         return _get_repos_cached()[repo_name]

pylons_app/tests/__init__.py

@@ -1,45 +1,48 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
+import logging
+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__']])
+#SetupCommand('setup-app').run([pylons.test.pylonsapp.config['__file__']])
 
 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
+        return response.follow()

pylons_app/tests/functional/test_admin.py

@@ -1,7 +1,9 b''
 from pylons_app.tests import *
 
 class TestAdminController(TestController):
 
     def test_index(self):
+        self.log_user()
         response = self.app.get(url(controller='admin/admin', action='index'))
+        assert 'Admin dashboard - journal' in response.body,'No proper title in dashboard'
         # Test response...

pylons_app/tests/functional/test_branches.py

@@ -1,7 +1,8 b''
 from pylons_app.tests import *
 
 class TestBranchesController(TestController):
 
     def test_index(self):
+        self.log_user()
         response = self.app.get(url(controller='branches', action='index',repo_name='vcs_test'))
         # Test response...

pylons_app/tests/functional/test_changelog.py

@@ -1,7 +1,8 b''
 from pylons_app.tests import *
 
 class TestChangelogController(TestController):
 
     def test_index(self):
+        self.log_user()
         response = self.app.get(url(controller='changelog', action='index',repo_name='vcs_test'))
         # Test response...

pylons_app/tests/functional/test_feed.py

@@ -1,13 +1,15 b''
 from pylons_app.tests import *
 
 class TestFeedController(TestController):
 
     def test_rss(self):
+        self.log_user()
         response = self.app.get(url(controller='feed', action='rss',
                                     repo_name='vcs_test'))
         # Test response...
 
     def test_atom(self):
+        self.log_user()
         response = self.app.get(url(controller='feed', action='atom',
                                     repo_name='vcs_test'))
         # Test response...
  No newline at end of file

pylons_app/tests/functional/test_files.py

@@ -1,10 +1,11 b''
 from pylons_app.tests import *
 
 class TestFilesController(TestController):
 
     def test_index(self):
+        self.log_user()
         response = self.app.get(url(controller='files', action='index',
                                     repo_name='vcs_test',
                                     revision='tip',
                                     f_path='/'))
         # Test response...

pylons_app/tests/functional/test_login.py

@@ -1,111 +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_regular2'
+        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_regular2').one()
+        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 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]',})
+        
+        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,})
+        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_settings.py

@@ -1,8 +1,9 b''
 from pylons_app.tests import *
 
 class TestSettingsController(TestController):
 
     def test_index(self):
+        self.log_user()
         response = self.app.get(url(controller='settings', action='index',
                                     repo_name='vcs_test'))
         # Test response...

pylons_app/tests/functional/test_shortlog.py

@@ -1,7 +1,8 b''
 from pylons_app.tests import *
 
 class TestShortlogController(TestController):
 
     def test_index(self):
+        self.log_user()
         response = self.app.get(url(controller='shortlog', action='index',repo_name='vcs_test'))
         # Test response...

pylons_app/tests/functional/test_summary.py

@@ -1,7 +1,8 b''
 from pylons_app.tests import *
 
 class TestSummaryController(TestController):
 
     def test_index(self):
+        self.log_user()
         response = self.app.get(url(controller='summary', action='index',repo_name='vcs_test'))
         # Test response...

pylons_app/tests/functional/test_tags.py

@@ -1,7 +1,8 b''
 from pylons_app.tests import *
 
 class TestTagsController(TestController):
 
     def test_index(self):
+        self.log_user()
         response = self.app.get(url(controller='tags', action='index',repo_name='vcs_test'))
         # Test response...

pylons_app/websetup.py

@@ -1,47 +1,45 b''
 """Setup the pylons_app application"""
 
-from os.path import dirname as dn, join as jn
+from os.path import dirname as dn
 from pylons_app.config.environment import load_environment
 from pylons_app.lib.db_manage import DbManage
-import datetime
-from time import mktime
 import logging
 import os
 import sys
-import tarfile
 
 log = logging.getLogger(__name__)
 
 ROOT = dn(dn(os.path.realpath(__file__)))
 sys.path.append(ROOT)
 
+
 def setup_app(command, conf, vars):
     """Place any commands to setup pylons_app here"""
     log_sql = True
     tests = False
+    REPO_TEST_PATH = None
     
     dbname = os.path.split(conf['sqlalchemy.db1.url'])[-1]
-    filename = os.path.split(conf.filename)[-1]
-    
-    if filename == 'tests.ini':
-        uniq_suffix = str(int(mktime(datetime.datetime.now().timetuple())))
-        REPO_TEST_PATH = '/tmp/hg_app_test_%s' % uniq_suffix
-        
-        if not os.path.isdir(REPO_TEST_PATH):
-            os.mkdir(REPO_TEST_PATH)
-            cur_dir = dn(os.path.abspath(__file__))
-            tar = tarfile.open(jn(cur_dir,'tests',"vcs_test.tar.gz"))
-            tar.extractall(REPO_TEST_PATH)
-            tar.close()
-            
-        tests = True    
+#    filename = os.path.split(conf.filename)[-1]
+#    if filename == 'test.ini':
+#        uniq_suffix = str(int(mktime(datetime.datetime.now().timetuple())))
+#        REPO_TEST_PATH = '/tmp/hg_app_test_%s' % uniq_suffix
+#        
+#        if not os.path.isdir(REPO_TEST_PATH):
+#            os.mkdir(REPO_TEST_PATH)
+#            cur_dir = dn(os.path.abspath(__file__))
+#            tar = tarfile.open(jn(cur_dir,'tests',"vcs_test.tar.gz"))
+#            tar.extractall(REPO_TEST_PATH)
+#            tar.close()
+#            
+#        tests = True    
     
     dbmanage = DbManage(log_sql, dbname, tests)
     dbmanage.create_tables(override=True)
     dbmanage.config_prompt(REPO_TEST_PATH)
     dbmanage.create_default_user()
     dbmanage.admin_prompt()
     dbmanage.create_permissions()
     dbmanage.populate_default_permissions()
     load_environment(conf.global_conf, conf.local_conf, initial=True)
 

setup.cfg

@@ -1,34 +1,34 b''
 [egg_info]
 tag_build = dev
 tag_svn_revision = true
 
 [easy_install]
 find_links = http://www.pylonshq.com/download/
 
 [nosetests]
 verbose=True
 verbosity=2
-with-pylons=tests.ini
+with-pylons=test.ini
 detailed-errors=1
 
 # Babel configuration
 [compile_catalog]
 domain = pylons_app
 directory = pylons_app/i18n
 statistics = true
 
 [extract_messages]
 add_comments = TRANSLATORS:
 output_file = pylons_app/i18n/pylons_app.pot
 width = 80
 
 [init_catalog]
 domain = pylons_app
 input_file = pylons_app/i18n/pylons_app.pot
 output_dir = pylons_app/i18n
 
 [update_catalog]
 domain = pylons_app
 input_file = pylons_app/i18n/pylons_app.pot
 output_dir = pylons_app/i18n
 previous = true