some fixups in cache, added fallback and cache invalidation when key not found in cached repos list,

added extra test, some other small fixes

celeryconfig.py

@@ -1,77 +1,77 b''
 # List of modules to import when celery starts.
 import sys
 import os
 import ConfigParser
 root = os.getcwd()
 
-PYLONS_CONFIG_NAME = 'test.ini'
+PYLONS_CONFIG_NAME = 'production.ini'
 
 sys.path.append(root)
 config = ConfigParser.ConfigParser({'here':root})
 config.read('%s/%s' % (root, PYLONS_CONFIG_NAME))
 PYLONS_CONFIG = config
 
 CELERY_IMPORTS = ("pylons_app.lib.celerylib.tasks",)
 
 ## Result store settings.
 CELERY_RESULT_BACKEND = "database"
 CELERY_RESULT_DBURI = dict(config.items('app:main'))['sqlalchemy.db1.url']
 CELERY_RESULT_SERIALIZER = 'json'
 
 
 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 = 3
 
 #Tasks will never be sent to the queue, but executed locally instead.
 CELERY_ALWAYS_EAGER = False
 if PYLONS_CONFIG_NAME == 'test.ini':
     #auto eager for tests
     CELERY_ALWAYS_EAGER = True
 
 #===============================================================================
 # EMAIL SETTINGS
 #===============================================================================
 pylons_email_config = dict(config.items('DEFAULT'))
 
 CELERY_SEND_TASK_ERROR_EMAILS = True
 
 #List of (name, email_address) tuples for the admins that should receive error e-mails.
 ADMINS = [('Administrator', pylons_email_config.get('email_to'))]
 
 #The e-mail address this worker sends e-mails from. Default is "celery@localhost".
 SERVER_EMAIL = pylons_email_config.get('error_email_from')
 
 #The mail server to use. Default is "localhost".
 MAIL_HOST = pylons_email_config.get('smtp_server')
 
 #Username (if required) to log on to the mail server with.
 MAIL_HOST_USER = pylons_email_config.get('smtp_username')
 
 #Password (if required) to log on to the mail server with.
 MAIL_HOST_PASSWORD = pylons_email_config.get('smtp_password')
 
 MAIL_PORT = pylons_email_config.get('smtp_port')
 
 
 #===============================================================================
 # INSTRUCTIONS FOR RABBITMQ
 #===============================================================================
 # rabbitmqctl add_user rabbitmq qweqwe
 # rabbitmqctl add_vhost rabbitmqhost
 # rabbitmqctl set_permissions -p rabbitmqhost rabbitmq ".*" ".*" ".*"

pylons_app/lib/base.py

@@ -1,40 +1,46 b''
 """The base Controller API
 
 Provides the BaseController class for subclassing.
 """
 from pylons import config, tmpl_context as c, request, session
 from pylons.controllers import WSGIController
 from pylons.templating import render_mako as render
 from pylons_app import __version__
 from pylons_app.lib import auth
 from pylons_app.lib.utils import get_repo_slug
 from pylons_app.model import meta
 from pylons_app.model.hg_model import _get_repos_cached, \
     _get_repos_switcher_cached
 
 class BaseController(WSGIController):
     
     def __before__(self):
         c.hg_app_version = __version__
         c.hg_app_name = config['hg_app_title']
         c.repo_name = get_repo_slug(request)
         c.cached_repo_list = _get_repos_cached()
         c.repo_switcher_list = _get_repos_switcher_cached(c.cached_repo_list)
         
         if c.repo_name:
-            c.repository_tags = c.cached_repo_list[c.repo_name].tags
-            c.repository_branches = c.cached_repo_list[c.repo_name].branches
+            cached_repo = c.cached_repo_list.get(c.repo_name)
+            
+            if cached_repo:
+                c.repository_tags = cached_repo.tags
+                c.repository_branches = cached_repo.branches
+            else:
+                c.repository_tags = {}
+                c.repository_branches = {}
                     
         self.sa = meta.Session
     
     def __call__(self, environ, start_response):
         """Invoke the Controller"""
         # WSGIController.__call__ dispatches to the Controller method
         # the request is routed to. This routing information is
         # available in environ['pylons.routes_dict']
         try:
             #putting this here makes sure that we update permissions every time
             c.hg_app_user = auth.get_user(session)
             return WSGIController.__call__(self, environ, start_response)
         finally:
             meta.Session.remove()

pylons_app/lib/celerylib/tasks.py

@@ -1,318 +1,316 b''
 from celery.decorators import task
 from celery.task.sets import subtask
 from celeryconfig import PYLONS_CONFIG as config
 from operator import itemgetter
 from pylons.i18n.translation import _
 from pylons_app.lib.celerylib import run_task, locked_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 json
 import traceback
 
 __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
 @locked_task
 def whoosh_index(repo_location, full_index):
     log = whoosh_index.get_logger()
     from pylons_app.lib.indexers.daemon import WhooshIndexingDaemon
     WhooshIndexingDaemon(repo_location=repo_location).run(full_index=full_index)
 
 @task
 @locked_task
 def get_commits_stats(repo_name, ts_min_y, ts_max_y):
     from pylons_app.model.db import Statistics, Repository
     log = get_commits_stats.get_logger()
     author_key_cleaner = lambda k: person(k).replace('"', "") #for js data compatibilty
     
     commits_by_day_author_aggregate = {}
     commits_by_day_aggregate = {}
     repos_path = get_hg_ui_settings()['paths_root_path'].replace('*', '')
     repo = MercurialRepository(repos_path + repo_name)
 
     skip_date_limit = True
     parse_limit = 350 #limit for single task changeset parsing optimal for
     last_rev = 0
     last_cs = None
     timegetter = itemgetter('time')
     
     sa = get_session()
     
     dbrepo = sa.query(Repository)\
         .filter(Repository.repo_name == repo_name).scalar()
     cur_stats = sa.query(Statistics)\
         .filter(Statistics.repository == dbrepo).scalar()
     if cur_stats:
         last_rev = cur_stats.stat_on_revision
     if not repo.revisions:
         return True
     
     if last_rev == repo.revisions[-1] and len(repo.revisions) > 1:
         #pass silently without any work if we're not on first revision or current
         #state of parsing revision(from db marker) is the last revision
         return True
     
     if cur_stats:
         commits_by_day_aggregate = OrderedDict(
                                        json.loads(
                                         cur_stats.commit_activity_combined))
         commits_by_day_author_aggregate = json.loads(cur_stats.commit_activity)
     
     log.debug('starting parsing %s', parse_limit)
     for cnt, rev in enumerate(repo.revisions[last_rev:]):
         last_cs = cs = repo.get_changeset(rev)
         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 commits_by_day_author_aggregate.has_key(author_key_cleaner(cs.author)):
             try:
                 l = [timegetter(x) for x in commits_by_day_author_aggregate\
                         [author_key_cleaner(cs.author)]['data']]
                 time_pos = l.index(k)
             except ValueError:
                 time_pos = False
                 
             if time_pos >= 0 and time_pos is not False:
                 
                 datadict = commits_by_day_author_aggregate\
                     [author_key_cleaner(cs.author)]['data'][time_pos]
                 
                 datadict["commits"] += 1
                 datadict["added"] += len(cs.added)
                 datadict["changed"] += len(cs.changed)
                 datadict["removed"] += len(cs.removed)
                 #print datadict
                 
             else:
                 #print 'ELSE !!!!'
                 if k >= ts_min_y and k <= ts_max_y or skip_date_limit:
                     
                     datadict = {"time":k,
                                 "commits":1,
                                 "added":len(cs.added),
                                 "changed":len(cs.changed),
                                 "removed":len(cs.removed),
                                }
                     commits_by_day_author_aggregate\
                         [author_key_cleaner(cs.author)]['data'].append(datadict)
                                         
         else:
             #print k, 'nokey ADDING'
             if k >= ts_min_y and k <= ts_max_y or skip_date_limit:
                 commits_by_day_author_aggregate[author_key_cleaner(cs.author)] = {
                                     "label":author_key_cleaner(cs.author),
                                     "data":[{"time":k,
                                              "commits":1,
                                              "added":len(cs.added),
                                              "changed":len(cs.changed),
                                              "removed":len(cs.removed),
                                              }],
                                     "schema":["commits"],
                                     }               
     
 #        #gather all data by day
         if commits_by_day_aggregate.has_key(k):
             commits_by_day_aggregate[k] += 1
         else:
             commits_by_day_aggregate[k] = 1
         
         if cnt >= parse_limit:
             #don't fetch to much data since we can freeze application
             break
 
     overview_data = []
     for k, v in commits_by_day_aggregate.items():
         overview_data.append([k, v])
     overview_data = sorted(overview_data, key=itemgetter(0))
         
     if not commits_by_day_author_aggregate:
         commits_by_day_author_aggregate[author_key_cleaner(repo.contact)] = {
             "label":author_key_cleaner(repo.contact),
             "data":[0, 1],
             "schema":["commits"],
         }
 
     stats = cur_stats if cur_stats else Statistics()
     stats.commit_activity = json.dumps(commits_by_day_author_aggregate)
     stats.commit_activity_combined = json.dumps(overview_data)
 
     log.debug('last revison %s', last_rev)
     leftovers = len(repo.revisions[last_rev:])
     log.debug('revisions to parse %s', leftovers)
     
     if last_rev == 0 or leftovers < parse_limit:    
         stats.languages = json.dumps(__get_codes_stats(repo_name))
         
     stats.repository = dbrepo
     stats.stat_on_revision = last_cs.revision
     
     try:
         sa.add(stats)
         sa.commit()    
     except:
         log.error(traceback.format_exc())
         sa.rollback()
         return False
     if len(repo.revisions) > 1:
         run_task(get_commits_stats, repo_name, ts_min_y, ts_max_y)
                             
     return True
 
 @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)
             if user:
                 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,
                        mail_port, ssl, tls)
         m.send(recipients, subject, body)  
     except:
         log.error('Mail sending failed')
         log.error(traceback.format_exc())
         return False
     return True
 
 @task
 def create_repo_fork(form_data, cur_user):
     import os
-    from pylons_app.lib.utils import invalidate_cache
     from pylons_app.model.repo_model import RepoModel
     sa = get_session()
     rm = RepoModel(sa)
     
     rm.create(form_data, cur_user, just_db=True, fork=True)
     
     repos_path = get_hg_ui_settings()['paths_root_path'].replace('*', '')
     repo_path = os.path.join(repos_path, form_data['repo_name'])
     repo_fork_path = os.path.join(repos_path, form_data['fork_name'])
     
     MercurialRepository(str(repo_fork_path), True, clone_url=str(repo_path))
-    #invalidate_cache('cached_repo_list')
 
     
 def __get_codes_stats(repo_name):
     LANGUAGES_EXTENSIONS = ['action', 'adp', 'ashx', 'asmx', 'aspx', 'asx', 'axd', 'c',
                     'cfg', 'cfm', 'cpp', 'cs', 'diff', 'do', 'el', 'erl',
                     'h', 'java', 'js', 'jsp', 'jspx', 'lisp',
                     'lua', 'm', 'mako', 'ml', 'pas', 'patch', 'php', 'php3',
                     'php4', 'phtml', 'pm', 'py', 'rb', 'rst', 's', 'sh',
                     'tpl', 'txt', 'vim', 'wss', 'xhtml', 'xml', 'xsl', 'xslt',
                     'yaws']
     repos_path = get_hg_ui_settings()['paths_root_path'].replace('*', '')
     repo = MercurialRepository(repos_path + repo_name)
 
     code_stats = {}
     for topnode, dirs, files in repo.walk('/', 'tip'):
         for f in files:
             k = f.mimetype
             if f.extension in LANGUAGES_EXTENSIONS:
                 if code_stats.has_key(k):
                     code_stats[k] += 1
                 else:
                     code_stats[k] = 1
                     
     return code_stats or {}
 
 
             
 
 

pylons_app/model/hg_model.py

@@ -1,172 +1,186 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 import helpers as h
+from pylons_app.lib.utils import invalidate_cache
 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 sqlalchemy.orm import joinedload
 from vcs.exceptions import RepositoryError, VCSError
-from sqlalchemy.orm import joinedload
 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 [x for x in cached_repo_list.values()]:
         if HasRepoPermissionAny('repository.write', 'repository.read',
                     'repository.admin')(repo.name, 'main page check'):
             repos_lst.append((repo.name, repo.dbrepo.private,))
     
     return sorted(repos_lst, key=lambda k:k[0].lower())
 
 @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):
         pass
     
     @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)        
         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:
+                        #for initial scann on application first run we don't
+                        #have db repos yet.
                         dbrepo = sa.query(Repository)\
                             .options(joinedload(Repository.fork))\
                             .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.short_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]
+        try:
+            repo = _get_repos_cached()[repo_name]
+            return repo
+        except KeyError:
+            #i we're here and we got key errors let's try to invalidate the
+            #cahce and try again
+            invalidate_cache('cached_repo_list')
+            repo = _get_repos_cached()[repo_name]
+            return repo
+            
+        
+        

pylons_app/tests/__init__.py

@@ -1,56 +1,58 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([config_file])
 
+##RUNNING DESIRED TESTS
+#nosetests pylons_app.tests.functional.test_admin_settings:TestSettingsController.test_my_account
 
 environ = {}
 
 class TestController(TestCase):
 
     def __init__(self, *args, **kwargs):
         wsgiapp = pylons.test.pylonsapp
         config = wsgiapp.config
         self.app = TestApp(wsgiapp)
         url._push_object(URLGenerator(config['routes.map'], environ))
         self.sa = meta.Session
 
         TestCase.__init__(self, *args, **kwargs)
     
     def log_user(self, username='test_admin', password='test12'):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username':username,
                                   'password':password})
         print response
         
         if 'invalid user name' in response.body:
             assert False, 'could not login using %s %s' % (username, password)
         
         assert response.status == '302 Found', 'Wrong response code from login got %s' % response.status
         assert response.session['hg_app_user'].username == username, 'wrong logged in user got %s expected %s' % (response.session['hg_app_user'].username, username)
         return response.follow()

pylons_app/tests/functional/test_settings.py

@@ -1,46 +1,55 b''
 from pylons_app.model.db import Repository
 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...
     
     def test_fork(self):
         self.log_user()
         response = self.app.get(url(controller='settings', action='fork',
                                     repo_name='vcs_test'))
         
 
     def test_fork_create(self):
         self.log_user()
         fork_name = 'vcs_test_fork'
         description = 'fork of vcs test'
         repo_name = 'vcs_test'
         response = self.app.post(url(controller='settings', action='fork_create',
                                     repo_name=repo_name),
                                     {'fork_name':fork_name,
                                      'description':description,
                                      'private':'False'})
         
         
         print response
         
         #test if we have a message that fork is ok
         assert 'fork %s repository as %s task added' \
                       % (repo_name, fork_name) in response.session['flash'][0], 'No flash message about fork'
                       
         #test if the fork was created in the database
         fork_repo = self.sa.query(Repository).filter(Repository.repo_name == fork_name).one()
         
         assert fork_repo.repo_name == fork_name, 'wrong name of repo name in new db fork repo'
         assert fork_repo.fork.repo_name == repo_name, 'wron fork parrent'
         
         
         #test if fork is visible in the list ?
         response = response.follow()
+
+
+        #check if fork is marked as fork
+        response = self.app.get(url(controller='summary', action='index',
+                                    repo_name=fork_name))
+        
         
         print response
+        
+        assert 'Fork of %s' % repo_name in response.body, 'no message about that this repo is a fork'
+