Fixed issue #501 error on setting set_as_fork to same repo

- also fixed html fill select set
- wrote set of tests for that option

rhodecode/controllers/admin/repos.py

@@ -1,446 +1,448 b''
 # -*- coding: utf-8 -*-
 """
     rhodecode.controllers.admin.repos
     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
     Repositories controller for RhodeCode
 
     :created_on: Apr 7, 2010
     :author: marcink
     :copyright: (C) 2010-2012 Marcin Kuzminski <[e-mail]>
     :license: GPLv3, see COPYING for more details.
 """
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 
 from webob.exc import HTTPInternalServerError
 from pylons import request, session, tmpl_context as c, url
 from pylons.controllers.util import redirect
 from pylons.i18n.translation import _
 from sqlalchemy.exc import IntegrityError
 
 from rhodecode.lib import helpers as h
 from rhodecode.lib.auth import LoginRequired, HasPermissionAllDecorator, \
     HasPermissionAnyDecorator, HasRepoPermissionAllDecorator
 from rhodecode.lib.base import BaseController, render
 from rhodecode.lib.utils import invalidate_cache, action_logger, repo_name_slug
 from rhodecode.lib.helpers import get_token
 from rhodecode.model.meta import Session
 from rhodecode.model.db import User, Repository, UserFollowing, RepoGroup
 from rhodecode.model.forms import RepoForm
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.repo import RepoModel
 
 log = logging.getLogger(__name__)
 
 
 class ReposController(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('repo', 'repos')
 
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')
     def __before__(self):
         c.admin_user = session.get('admin_user')
         c.admin_username = session.get('admin_username')
         super(ReposController, self).__before__()
 
     def __load_defaults(self):
         c.repo_groups = RepoGroup.groups_choices()
         c.repo_groups_choices = map(lambda k: unicode(k[0]), c.repo_groups)
 
         repo_model = RepoModel()
         c.users_array = repo_model.get_users_js()
         c.users_groups_array = repo_model.get_users_groups_js()
         choices, c.landing_revs = ScmModel().get_repo_landing_revs()
         c.landing_revs_choices = choices
 
     def __load_data(self, repo_name=None):
         """
         Load defaults settings for edit, and update
 
         :param repo_name:
         """
         self.__load_defaults()
 
         c.repo_info = db_repo = Repository.get_by_repo_name(repo_name)
         repo = db_repo.scm_instance
 
         if c.repo_info is None:
             h.flash(_('%s repository is not mapped to db perhaps'
                       ' it was created or renamed from the filesystem'
                       ' please run the application again'
                       ' in order to rescan repositories') % repo_name,
                       category='error')
 
             return redirect(url('repos'))
 
         choices, c.landing_revs = ScmModel().get_repo_landing_revs(c.repo_info)
         c.landing_revs_choices = choices
 
         c.default_user_id = User.get_by_username('default').user_id
         c.in_public_journal = UserFollowing.query()\
             .filter(UserFollowing.user_id == c.default_user_id)\
             .filter(UserFollowing.follows_repository == c.repo_info).scalar()
 
         if c.repo_info.stats:
             # this is on what revision we ended up so we add +1 for count
             last_rev = c.repo_info.stats.stat_on_revision + 1
         else:
             last_rev = 0
         c.stats_revision = last_rev
 
         c.repo_last_rev = repo.count() if repo.revisions else 0
 
         if last_rev == 0 or c.repo_last_rev == 0:
             c.stats_percentage = 0
         else:
             c.stats_percentage = '%.2f' % ((float((last_rev)) /
                                             c.repo_last_rev) * 100)
 
         defaults = RepoModel()._get_defaults(repo_name)
 
         c.repos_list = [('', _('--REMOVE FORK--'))]
         c.repos_list += [(x.repo_id, x.repo_name) for x in
-                   Repository.query().order_by(Repository.repo_name).all()]
+                    Repository.query().order_by(Repository.repo_name).all()
+                    if x.repo_id != c.repo_info.repo_id]
 
+        defaults['id_fork_of'] = db_repo.fork.repo_id if db_repo.fork else ''
         return defaults
 
     @HasPermissionAllDecorator('hg.admin')
     def index(self, format='html'):
         """GET /repos: All items in the collection"""
         # url('repos')
 
         c.repos_list = ScmModel().get_repos(Repository.query()
                                             .order_by(Repository.repo_name)
                                             .all(), sort_key='name_sort')
         return render('admin/repos/repos.html')
 
     @HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')
     def create(self):
         """
         POST /repos: Create a new item"""
         # url('repos')
 
         self.__load_defaults()
         form_result = {}
         try:
             form_result = RepoForm(repo_groups=c.repo_groups_choices,
                                    landing_revs=c.landing_revs_choices)()\
                             .to_python(dict(request.POST))
             RepoModel().create(form_result, self.rhodecode_user.user_id)
             if form_result['clone_uri']:
                 h.flash(_('created repository %s from %s') \
                     % (form_result['repo_name'], form_result['clone_uri']),
                     category='success')
             else:
                 h.flash(_('created repository %s') % form_result['repo_name'],
                     category='success')
 
             if request.POST.get('user_created'):
                 # created by regular non admin user
                 action_logger(self.rhodecode_user, 'user_created_repo',
                               form_result['repo_name_full'], self.ip_addr,
                               self.sa)
             else:
                 action_logger(self.rhodecode_user, 'admin_created_repo',
                               form_result['repo_name_full'], self.ip_addr,
                               self.sa)
             Session.commit()
         except formencode.Invalid, errors:
 
             c.new_repo = errors.value['repo_name']
 
             if request.POST.get('user_created'):
                 r = render('admin/repos/repo_add_create_repository.html')
             else:
                 r = render('admin/repos/repo_add.html')
 
             return htmlfill.render(
                 r,
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8")
 
         except Exception:
             log.error(traceback.format_exc())
             msg = _('error occurred during creation of repository %s') \
                     % form_result.get('repo_name')
             h.flash(msg, category='error')
         if request.POST.get('user_created'):
             return redirect(url('home'))
         return redirect(url('repos'))
 
     @HasPermissionAllDecorator('hg.admin')
     def new(self, format='html'):
         """GET /repos/new: Form to create a new item"""
         new_repo = request.GET.get('repo', '')
         c.new_repo = repo_name_slug(new_repo)
         self.__load_defaults()
         return render('admin/repos/repo_add.html')
 
     @HasPermissionAllDecorator('hg.admin')
     def update(self, repo_name):
         """
         PUT /repos/repo_name: 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('repo', repo_name=ID),
         #           method='put')
         # url('repo', repo_name=ID)
         self.__load_defaults()
         repo_model = RepoModel()
         changed_name = repo_name
         #override the choices with extracted revisions !
         choices, c.landing_revs = ScmModel().get_repo_landing_revs(repo_name)
         c.landing_revs_choices = choices
 
         _form = RepoForm(edit=True, old_data={'repo_name': repo_name},
                          repo_groups=c.repo_groups_choices,
                          landing_revs=c.landing_revs_choices)()
         try:
             form_result = _form.to_python(dict(request.POST))
             repo = repo_model.update(repo_name, form_result)
             invalidate_cache('get_repo_cached_%s' % repo_name)
             h.flash(_('Repository %s updated successfully') % repo_name,
                     category='success')
             changed_name = repo.repo_name
             action_logger(self.rhodecode_user, 'admin_updated_repo',
                               changed_name, self.ip_addr, self.sa)
             Session.commit()
         except formencode.Invalid, errors:
             defaults = self.__load_data(repo_name)
             defaults.update(errors.value)
             return htmlfill.render(
                 render('admin/repos/repo_edit.html'),
                 defaults=defaults,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8")
 
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('error occurred during update of repository %s') \
                     % repo_name, category='error')
         return redirect(url('edit_repo', repo_name=changed_name))
 
     @HasPermissionAllDecorator('hg.admin')
     def delete(self, repo_name):
         """
         DELETE /repos/repo_name: 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('repo', repo_name=ID),
         #           method='delete')
         # url('repo', repo_name=ID)
 
         repo_model = RepoModel()
         repo = repo_model.get_by_repo_name(repo_name)
         if not repo:
             h.flash(_('%s repository is not mapped to db perhaps'
                       ' it was moved or renamed  from the filesystem'
                       ' please run the application again'
                       ' in order to rescan repositories') % repo_name,
                       category='error')
 
             return redirect(url('repos'))
         try:
             action_logger(self.rhodecode_user, 'admin_deleted_repo',
                               repo_name, self.ip_addr, self.sa)
             repo_model.delete(repo)
             invalidate_cache('get_repo_cached_%s' % repo_name)
             h.flash(_('deleted repository %s') % repo_name, category='success')
             Session.commit()
         except IntegrityError, e:
             if e.message.find('repositories_fork_id_fkey') != -1:
                 log.error(traceback.format_exc())
                 h.flash(_('Cannot delete %s it still contains attached '
                           'forks') % repo_name,
                         category='warning')
             else:
                 log.error(traceback.format_exc())
                 h.flash(_('An error occurred during '
                           'deletion of %s') % repo_name,
                         category='error')
 
         except Exception, e:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during deletion of %s') % repo_name,
                     category='error')
 
         return redirect(url('repos'))
 
     @HasRepoPermissionAllDecorator('repository.admin')
     def delete_perm_user(self, repo_name):
         """
         DELETE an existing repository permission user
 
         :param repo_name:
         """
         try:
             RepoModel().revoke_user_permission(repo=repo_name,
                                                user=request.POST['user_id'])
             Session.commit()
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during deletion of repository user'),
                     category='error')
             raise HTTPInternalServerError()
 
     @HasRepoPermissionAllDecorator('repository.admin')
     def delete_perm_users_group(self, repo_name):
         """
         DELETE an existing repository permission users group
 
         :param repo_name:
         """
 
         try:
             RepoModel().revoke_users_group_permission(
                 repo=repo_name, group_name=request.POST['users_group_id']
             )
             Session.commit()
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('An error occurred during deletion of repository'
                       ' users groups'),
                     category='error')
             raise HTTPInternalServerError()
 
     @HasPermissionAllDecorator('hg.admin')
     def repo_stats(self, repo_name):
         """
         DELETE an existing repository statistics
 
         :param repo_name:
         """
 
         try:
             RepoModel().delete_stats(repo_name)
             Session.commit()
         except Exception, e:
             h.flash(_('An error occurred during deletion of repository stats'),
                     category='error')
         return redirect(url('edit_repo', repo_name=repo_name))
 
     @HasPermissionAllDecorator('hg.admin')
     def repo_cache(self, repo_name):
         """
         INVALIDATE existing repository cache
 
         :param repo_name:
         """
 
         try:
             ScmModel().mark_for_invalidation(repo_name)
             Session.commit()
         except Exception, e:
             h.flash(_('An error occurred during cache invalidation'),
                     category='error')
         return redirect(url('edit_repo', repo_name=repo_name))
 
     @HasPermissionAllDecorator('hg.admin')
     def repo_public_journal(self, repo_name):
         """
         Set's this repository to be visible in public journal,
         in other words assing default user to follow this repo
 
         :param repo_name:
         """
 
         cur_token = request.POST.get('auth_token')
         token = get_token()
         if cur_token == token:
             try:
                 repo_id = Repository.get_by_repo_name(repo_name).repo_id
                 user_id = User.get_by_username('default').user_id
                 self.scm_model.toggle_following_repo(repo_id, user_id)
                 h.flash(_('Updated repository visibility in public journal'),
                         category='success')
                 Session.commit()
             except:
                 h.flash(_('An error occurred during setting this'
                           ' repository in public journal'),
                         category='error')
 
         else:
             h.flash(_('Token mismatch'), category='error')
         return redirect(url('edit_repo', repo_name=repo_name))
 
     @HasPermissionAllDecorator('hg.admin')
     def repo_pull(self, repo_name):
         """
         Runs task to update given repository with remote changes,
         ie. make pull on remote location
 
         :param repo_name:
         """
         try:
             ScmModel().pull_changes(repo_name, self.rhodecode_user.username)
             h.flash(_('Pulled from remote location'), category='success')
         except Exception, e:
             h.flash(_('An error occurred during pull from remote location'),
                     category='error')
 
         return redirect(url('edit_repo', repo_name=repo_name))
 
     @HasPermissionAllDecorator('hg.admin')
     def repo_as_fork(self, repo_name):
         """
         Mark given repository as a fork of another
 
         :param repo_name:
         """
         try:
             fork_id = request.POST.get('id_fork_of')
             repo = ScmModel().mark_as_fork(repo_name, fork_id,
                                     self.rhodecode_user.username)
             fork = repo.fork.repo_name if repo.fork else _('Nothing')
-            Session.commit()
-            h.flash(_('Marked repo %s as fork of %s') % (repo_name,fork),
+            Session().commit()
+            h.flash(_('Marked repo %s as fork of %s') % (repo_name, fork),
                     category='success')
         except Exception, e:
-            raise
+            log.error(traceback.format_exc())
             h.flash(_('An error occurred during this operation'),
                     category='error')
 
         return redirect(url('edit_repo', repo_name=repo_name))
 
     @HasPermissionAllDecorator('hg.admin')
     def show(self, repo_name, format='html'):
         """GET /repos/repo_name: Show a specific item"""
         # url('repo', repo_name=ID)
 
     @HasPermissionAllDecorator('hg.admin')
     def edit(self, repo_name, format='html'):
         """GET /repos/repo_name/edit: Form to edit an existing item"""
         # url('edit_repo', repo_name=ID)
         defaults = self.__load_data(repo_name)
 
         return htmlfill.render(
             render('admin/repos/repo_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
         )

rhodecode/model/scm.py

@@ -1,598 +1,600 b''
 # -*- coding: utf-8 -*-
 """
     rhodecode.model.scm
     ~~~~~~~~~~~~~~~~~~~
 
     Scm model for RhodeCode
 
     :created_on: Apr 9, 2010
     :author: marcink
     :copyright: (C) 2010-2012 Marcin Kuzminski <[e-mail]>
     :license: GPLv3, see COPYING for more details.
 """
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 from __future__ import with_statement
 import os
 import re
 import time
 import traceback
 import logging
 import cStringIO
 import pkg_resources
 from os.path import dirname as dn, join as jn
 
 from sqlalchemy import func
 from pylons.i18n.translation import _
 
 import rhodecode
 from rhodecode.lib.vcs import get_backend
 from rhodecode.lib.vcs.exceptions import RepositoryError
 from rhodecode.lib.vcs.utils.lazy import LazyProperty
 from rhodecode.lib.vcs.nodes import FileNode
 
 from rhodecode import BACKENDS
 from rhodecode.lib import helpers as h
 from rhodecode.lib.utils2 import safe_str, safe_unicode
 from rhodecode.lib.auth import HasRepoPermissionAny, HasReposGroupPermissionAny
 from rhodecode.lib.utils import get_repos as get_filesystem_repos, make_ui, \
     action_logger, EmptyChangeset, REMOVED_REPO_PAT
 from rhodecode.model import BaseModel
 from rhodecode.model.db import Repository, RhodeCodeUi, CacheInvalidation, \
     UserFollowing, UserLog, User, RepoGroup, PullRequest
 
 log = logging.getLogger(__name__)
 
 
 class UserTemp(object):
     def __init__(self, user_id):
         self.user_id = user_id
 
     def __repr__(self):
         return "<%s('id:%s')>" % (self.__class__.__name__, self.user_id)
 
 
 class RepoTemp(object):
     def __init__(self, repo_id):
         self.repo_id = repo_id
 
     def __repr__(self):
         return "<%s('id:%s')>" % (self.__class__.__name__, self.repo_id)
 
 
 class CachedRepoList(object):
     """
     Cached repo list, uses in-memory cache after initialization, that is
     super fast
     """
 
     def __init__(self, db_repo_list, repos_path, order_by=None):
         self.db_repo_list = db_repo_list
         self.repos_path = repos_path
         self.order_by = order_by
         self.reversed = (order_by or '').startswith('-')
 
     def __len__(self):
         return len(self.db_repo_list)
 
     def __repr__(self):
         return '<%s (%s)>' % (self.__class__.__name__, self.__len__())
 
     def __iter__(self):
         # pre-propagated cache_map to save executing select statements
         # for each repo
         cache_map = CacheInvalidation.get_cache_map()
 
         for dbr in self.db_repo_list:
             scmr = dbr.scm_instance_cached(cache_map)
             # check permission at this level
             if not HasRepoPermissionAny(
                 'repository.read', 'repository.write', 'repository.admin'
             )(dbr.repo_name, 'get repo check'):
                 continue
 
             if scmr is None:
                 log.error(
                     '%s this repository is present in database but it '
                     'cannot be created as an scm instance' % dbr.repo_name
                 )
                 continue
 
             last_change = scmr.last_change
             tip = h.get_changeset_safe(scmr, 'tip')
 
             tmp_d = {}
             tmp_d['name'] = dbr.repo_name
             tmp_d['name_sort'] = tmp_d['name'].lower()
             tmp_d['description'] = dbr.description
             tmp_d['description_sort'] = tmp_d['description'].lower()
             tmp_d['last_change'] = last_change
             tmp_d['last_change_sort'] = time.mktime(last_change.timetuple())
             tmp_d['tip'] = tip.raw_id
             tmp_d['tip_sort'] = tip.revision
             tmp_d['rev'] = tip.revision
             tmp_d['contact'] = dbr.user.full_contact
             tmp_d['contact_sort'] = tmp_d['contact']
             tmp_d['owner_sort'] = tmp_d['contact']
             tmp_d['repo_archives'] = list(scmr._get_archives())
             tmp_d['last_msg'] = tip.message
             tmp_d['author'] = tip.author
             tmp_d['dbrepo'] = dbr.get_dict()
             tmp_d['dbrepo_fork'] = dbr.fork.get_dict() if dbr.fork else {}
             yield tmp_d
 
 
 class SimpleCachedRepoList(CachedRepoList):
     """
     Lighter version of CachedRepoList without the scm initialisation
     """
 
     def __iter__(self):
         for dbr in self.db_repo_list:
             # check permission at this level
             if not HasRepoPermissionAny(
                 'repository.read', 'repository.write', 'repository.admin'
             )(dbr.repo_name, 'get repo check'):
                 continue
 
             tmp_d = {}
             tmp_d['name'] = dbr.repo_name
             tmp_d['name_sort'] = tmp_d['name'].lower()
             tmp_d['description'] = dbr.description
             tmp_d['description_sort'] = tmp_d['description'].lower()
             tmp_d['dbrepo'] = dbr.get_dict()
             tmp_d['dbrepo_fork'] = dbr.fork.get_dict() if dbr.fork else {}
             yield tmp_d
 
 
 class GroupList(object):
 
     def __init__(self, db_repo_group_list):
         self.db_repo_group_list = db_repo_group_list
 
     def __len__(self):
         return len(self.db_repo_group_list)
 
     def __repr__(self):
         return '<%s (%s)>' % (self.__class__.__name__, self.__len__())
 
     def __iter__(self):
         for dbgr in self.db_repo_group_list:
             # check permission at this level
             if not HasReposGroupPermissionAny(
                 'group.read', 'group.write', 'group.admin'
             )(dbgr.group_name, 'get group repo check'):
                 continue
 
             yield dbgr
 
 
 class ScmModel(BaseModel):
     """
     Generic Scm Model
     """
 
     def __get_repo(self, instance):
         cls = Repository
         if isinstance(instance, cls):
             return instance
         elif isinstance(instance, int) or str(instance).isdigit():
             return cls.get(instance)
         elif isinstance(instance, basestring):
             return cls.get_by_repo_name(instance)
         elif instance:
             raise Exception('given object must be int, basestr or Instance'
                             ' of %s got %s' % (type(cls), type(instance)))
 
     @LazyProperty
     def repos_path(self):
         """
         Get's the repositories root path from database
         """
 
         q = self.sa.query(RhodeCodeUi).filter(RhodeCodeUi.ui_key == '/').one()
 
         return q.ui_value
 
     def repo_scan(self, repos_path=None):
         """
         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 containing repositories
         """
 
         if repos_path is None:
             repos_path = self.repos_path
 
         log.info('scanning for repositories in %s' % repos_path)
 
         baseui = make_ui('db')
         repos = {}
 
         for name, path in get_filesystem_repos(repos_path, recursive=True):
             # skip removed repos
             if REMOVED_REPO_PAT.match(name):
                 continue
 
             # name need to be decomposed and put back together using the /
             # since this is internal storage separator for rhodecode
             name = Repository.url_sep().join(name.split(os.sep))
 
             try:
                 if name in repos:
                     raise RepositoryError('Duplicate repository name %s '
                                           'found in %s' % (name, path))
                 else:
 
                     klass = get_backend(path[0])
 
                     if path[0] == 'hg' and path[0] in BACKENDS.keys():
                         repos[name] = klass(safe_str(path[1]), baseui=baseui)
 
                     if path[0] == 'git' and path[0] in BACKENDS.keys():
                         repos[name] = klass(path[1])
             except OSError:
                 continue
 
         return repos
 
     def get_repos(self, all_repos=None, sort_key=None, simple=False):
         """
         Get all repos from db and for each repo create it's
         backend instance and fill that backed with information from database
 
         :param all_repos: list of repository names as strings
             give specific repositories list, good for filtering
 
         :param sort_key: initial sorting of repos
         :param simple: use SimpleCachedList - one without the SCM info
         """
         if all_repos is None:
             all_repos = self.sa.query(Repository)\
                         .filter(Repository.group_id == None)\
                         .order_by(func.lower(Repository.repo_name)).all()
         if simple:
             repo_iter = SimpleCachedRepoList(all_repos,
                                              repos_path=self.repos_path,
                                              order_by=sort_key)
         else:
             repo_iter = CachedRepoList(all_repos,
                                        repos_path=self.repos_path,
                                        order_by=sort_key)
 
         return repo_iter
 
     def get_repos_groups(self, all_groups=None):
         if all_groups is None:
             all_groups = RepoGroup.query()\
                 .filter(RepoGroup.group_parent_id == None).all()
         group_iter = GroupList(all_groups)
 
         return group_iter
 
     def mark_for_invalidation(self, repo_name):
         """
         Puts cache invalidation task into db for
         further global cache invalidation
 
         :param repo_name: this repo that should invalidation take place
         """
         CacheInvalidation.set_invalidate(repo_name)
 
     def toggle_following_repo(self, follow_repo_id, user_id):
 
         f = self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_repo_id == follow_repo_id)\
             .filter(UserFollowing.user_id == user_id).scalar()
 
         if f is not None:
             try:
                 self.sa.delete(f)
                 action_logger(UserTemp(user_id),
                               'stopped_following_repo',
                               RepoTemp(follow_repo_id))
                 return
             except:
                 log.error(traceback.format_exc())
                 raise
 
         try:
             f = UserFollowing()
             f.user_id = user_id
             f.follows_repo_id = follow_repo_id
             self.sa.add(f)
 
             action_logger(UserTemp(user_id),
                           'started_following_repo',
                           RepoTemp(follow_repo_id))
         except:
             log.error(traceback.format_exc())
             raise
 
     def toggle_following_user(self, follow_user_id, user_id):
         f = self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_user_id == follow_user_id)\
             .filter(UserFollowing.user_id == user_id).scalar()
 
         if f is not None:
             try:
                 self.sa.delete(f)
                 return
             except:
                 log.error(traceback.format_exc())
                 raise
 
         try:
             f = UserFollowing()
             f.user_id = user_id
             f.follows_user_id = follow_user_id
             self.sa.add(f)
         except:
             log.error(traceback.format_exc())
             raise
 
     def is_following_repo(self, repo_name, user_id, cache=False):
         r = self.sa.query(Repository)\
             .filter(Repository.repo_name == repo_name).scalar()
 
         f = self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_repository == r)\
             .filter(UserFollowing.user_id == user_id).scalar()
 
         return f is not None
 
     def is_following_user(self, username, user_id, cache=False):
         u = User.get_by_username(username)
 
         f = self.sa.query(UserFollowing)\
             .filter(UserFollowing.follows_user == u)\
             .filter(UserFollowing.user_id == user_id).scalar()
 
         return f is not None
 
     def get_followers(self, repo):
         repo = self._get_repo(repo)
 
         return self.sa.query(UserFollowing)\
                 .filter(UserFollowing.follows_repository == repo).count()
 
     def get_forks(self, repo):
         repo = self._get_repo(repo)
         return self.sa.query(Repository)\
                 .filter(Repository.fork == repo).count()
 
     def get_pull_requests(self, repo):
         repo = self._get_repo(repo)
         return self.sa.query(PullRequest)\
                 .filter(PullRequest.other_repo == repo).count()
 
     def mark_as_fork(self, repo, fork, user):
         repo = self.__get_repo(repo)
         fork = self.__get_repo(fork)
+        if fork and repo.repo_id == fork.repo_id:
+            raise Exception("Cannot set repository as fork of itself")
         repo.fork = fork
         self.sa.add(repo)
         return repo
 
     def pull_changes(self, repo, username):
         dbrepo = self.__get_repo(repo)
         clone_uri = dbrepo.clone_uri
         if not clone_uri:
             raise Exception("This repository doesn't have a clone uri")
 
         repo = dbrepo.scm_instance
         try:
             extras = {
                 'ip': '',
                 'username': username,
                 'action': 'push_remote',
                 'repository': dbrepo.repo_name,
                 'scm': repo.alias,
             }
             Repository.inject_ui(repo, extras=extras)
 
             if repo.alias == 'git':
                 repo.fetch(clone_uri)
             else:
                 repo.pull(clone_uri)
             self.mark_for_invalidation(dbrepo.repo_name)
         except:
             log.error(traceback.format_exc())
             raise
 
     def commit_change(self, repo, repo_name, cs, user, author, message,
                       content, f_path):
 
         if repo.alias == 'hg':
             from rhodecode.lib.vcs.backends.hg import \
                 MercurialInMemoryChangeset as IMC
         elif repo.alias == 'git':
             from rhodecode.lib.vcs.backends.git import \
                 GitInMemoryChangeset as IMC
 
         # decoding here will force that we have proper encoded values
         # in any other case this will throw exceptions and deny commit
         content = safe_str(content)
         path = safe_str(f_path)
         # message and author needs to be unicode
         # proper backend should then translate that into required type
         message = safe_unicode(message)
         author = safe_unicode(author)
         m = IMC(repo)
         m.change(FileNode(path, content))
         tip = m.commit(message=message,
                        author=author,
                        parents=[cs], branch=cs.branch)
 
         new_cs = tip.short_id
         action = 'push_local:%s' % new_cs
 
         action_logger(user, action, repo_name)
 
         self.mark_for_invalidation(repo_name)
 
     def create_node(self, repo, repo_name, cs, user, author, message, content,
                       f_path):
         if repo.alias == 'hg':
             from rhodecode.lib.vcs.backends.hg import MercurialInMemoryChangeset as IMC
         elif repo.alias == 'git':
             from rhodecode.lib.vcs.backends.git import GitInMemoryChangeset as IMC
         # decoding here will force that we have proper encoded values
         # in any other case this will throw exceptions and deny commit
 
         if isinstance(content, (basestring,)):
             content = safe_str(content)
         elif isinstance(content, (file, cStringIO.OutputType,)):
             content = content.read()
         else:
             raise Exception('Content is of unrecognized type %s' % (
                 type(content)
             ))
 
         message = safe_unicode(message)
         author = safe_unicode(author)
         path = safe_str(f_path)
         m = IMC(repo)
 
         if isinstance(cs, EmptyChangeset):
             # EmptyChangeset means we we're editing empty repository
             parents = None
         else:
             parents = [cs]
 
         m.add(FileNode(path, content=content))
         tip = m.commit(message=message,
                        author=author,
                        parents=parents, branch=cs.branch)
         new_cs = tip.short_id
         action = 'push_local:%s' % new_cs
 
         action_logger(user, action, repo_name)
 
         self.mark_for_invalidation(repo_name)
 
     def get_nodes(self, repo_name, revision, root_path='/', flat=True):
         """
         recursive walk in root dir and return a set of all path in that dir
         based on repository walk function
 
         :param repo_name: name of repository
         :param revision: revision for which to list nodes
         :param root_path: root path to list
         :param flat: return as a list, if False returns a dict with decription
 
         """
         _files = list()
         _dirs = list()
         try:
             _repo = self.__get_repo(repo_name)
             changeset = _repo.scm_instance.get_changeset(revision)
             root_path = root_path.lstrip('/')
             for topnode, dirs, files in changeset.walk(root_path):
                 for f in files:
                     _files.append(f.path if flat else {"name": f.path,
                                                        "type": "file"})
                 for d in dirs:
                     _dirs.append(d.path if flat else {"name": d.path,
                                                       "type": "dir"})
         except RepositoryError:
             log.debug(traceback.format_exc())
             raise
 
         return _dirs, _files
 
     def get_unread_journal(self):
         return self.sa.query(UserLog).count()
 
     def get_repo_landing_revs(self, repo=None):
         """
         Generates select option with tags branches and bookmarks (for hg only)
         grouped by type
 
         :param repo:
         :type repo:
         """
 
         hist_l = []
         choices = []
         repo = self.__get_repo(repo)
         hist_l.append(['tip', _('latest tip')])
         choices.append('tip')
         if not repo:
             return choices, hist_l
 
         repo = repo.scm_instance
 
         branches_group = ([(k, k) for k, v in
                            repo.branches.iteritems()], _("Branches"))
         hist_l.append(branches_group)
         choices.extend([x[0] for x in branches_group[0]])
 
         if repo.alias == 'hg':
             bookmarks_group = ([(k, k) for k, v in
                                 repo.bookmarks.iteritems()], _("Bookmarks"))
             hist_l.append(bookmarks_group)
             choices.extend([x[0] for x in bookmarks_group[0]])
 
         tags_group = ([(k, k) for k, v in
                        repo.tags.iteritems()], _("Tags"))
         hist_l.append(tags_group)
         choices.extend([x[0] for x in tags_group[0]])
 
         return choices, hist_l
 
     def install_git_hook(self, repo, force_create=False):
         """
         Creates a rhodecode hook inside a git repository
 
         :param repo: Instance of VCS repo
         :param force_create: Create even if same name hook exists
         """
 
         loc = jn(repo.path, 'hooks')
         if not repo.bare:
             loc = jn(repo.path, '.git', 'hooks')
         if not os.path.isdir(loc):
             os.makedirs(loc)
 
         tmpl = pkg_resources.resource_string(
             'rhodecode', jn('config', 'post_receive_tmpl.py')
         )
 
         _hook_file = jn(loc, 'post-receive')
         _rhodecode_hook = False
         log.debug('Installing git hook in repo %s' % repo)
         if os.path.exists(_hook_file):
             # let's take a look at this hook, maybe it's rhodecode ?
             log.debug('hook exists, checking if it is from rhodecode')
             _HOOK_VER_PAT = re.compile(r'^RC_HOOK_VER')
             with open(_hook_file, 'rb') as f:
                 data = f.read()
                 matches = re.compile(r'(?:%s)\s*=\s*(.*)'
                                      % 'RC_HOOK_VER').search(data)
                 if matches:
                     try:
                         ver = matches.groups()[0]
                         log.debug('got %s it is rhodecode' % (ver))
                         _rhodecode_hook = True
                     except:
                         log.error(traceback.format_exc())
 
         if _rhodecode_hook or force_create:
             log.debug('writing hook file !')
             with open(_hook_file, 'wb') as f:
                 tmpl = tmpl.replace('_TMPL_', rhodecode.__version__)
                 f.write(tmpl)
             os.chmod(_hook_file, 0755)
         else:
             log.debug('skipping writing hook file')
  No newline at end of file

rhodecode/tests/__init__.py

@@ -1,164 +1,164 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.
 """
 import os
 import time
 import logging
 import datetime
 import hashlib
 import tempfile
 from os.path import join as jn
 
 from unittest import TestCase
 from tempfile import _RandomNameSequence
 
 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
 
 from rhodecode import is_windows
 from rhodecode.model.meta import Session
 from rhodecode.model.db import User
 from rhodecode.tests.nose_parametrized import parameterized
 
 import pylons.test
 
 
 os.environ['TZ'] = 'UTC'
 if not is_windows:
     time.tzset()
 
 log = logging.getLogger(__name__)
 
 __all__ = [
     'parameterized', 'environ', 'url', 'get_new_dir', 'TestController',
     'TESTS_TMP_PATH', 'HG_REPO', 'GIT_REPO', 'NEW_HG_REPO', 'NEW_GIT_REPO',
     'HG_FORK', 'GIT_FORK', 'TEST_USER_ADMIN_LOGIN', 'TEST_USER_REGULAR_LOGIN',
     'TEST_USER_REGULAR_PASS', 'TEST_USER_REGULAR_EMAIL',
     'TEST_USER_REGULAR2_LOGIN', 'TEST_USER_REGULAR2_PASS',
     'TEST_USER_REGULAR2_EMAIL', 'TEST_HG_REPO', 'TEST_HG_REPO_CLONE',
     'TEST_HG_REPO_PULL', 'TEST_GIT_REPO', 'TEST_GIT_REPO_CLONE',
     'TEST_GIT_REPO_PULL', 'HG_REMOTE_REPO', 'GIT_REMOTE_REPO', 'SCM_TESTS',
 ]
 
 # Invoke websetup with the current config file
 # SetupCommand('setup-app').run([config_file])
 
 ##RUNNING DESIRED TESTS
 # nosetests -x rhodecode.tests.functional.test_admin_settings:TestSettingsController.test_my_account
 # nosetests --pdb --pdb-failures
 # nosetests --with-coverage --cover-package=rhodecode.model.validators rhodecode.tests.test_validators
 environ = {}
 
 #SOME GLOBALS FOR TESTS
 
 TESTS_TMP_PATH = jn('/', 'tmp', 'rc_test_%s' % _RandomNameSequence().next())
 TEST_USER_ADMIN_LOGIN = 'test_admin'
 TEST_USER_ADMIN_PASS = 'test12'
 TEST_USER_ADMIN_EMAIL = '[e-mail]'
 
 TEST_USER_REGULAR_LOGIN = 'test_regular'
 TEST_USER_REGULAR_PASS = 'test12'
 TEST_USER_REGULAR_EMAIL = '[e-mail]'
 
 TEST_USER_REGULAR2_LOGIN = 'test_regular2'
 TEST_USER_REGULAR2_PASS = 'test12'
 TEST_USER_REGULAR2_EMAIL = '[e-mail]'
 
 HG_REPO = 'vcs_test_hg'
 GIT_REPO = 'vcs_test_git'
 
 NEW_HG_REPO = 'vcs_test_hg_new'
 NEW_GIT_REPO = 'vcs_test_git_new'
 
 HG_FORK = 'vcs_test_hg_fork'
 GIT_FORK = 'vcs_test_git_fork'
 
 ## VCS
 SCM_TESTS = ['hg', 'git']
 uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))
 
 GIT_REMOTE_REPO = 'git://github.com/codeinn/vcs.git'
 
 TEST_GIT_REPO = jn(TESTS_TMP_PATH, GIT_REPO)
 TEST_GIT_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)
 TEST_GIT_REPO_PULL = jn(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)
 
 
 HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'
 
 TEST_HG_REPO = jn(TESTS_TMP_PATH, HG_REPO)
 TEST_HG_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)
 TEST_HG_REPO_PULL = jn(TESTS_TMP_PATH, 'vcshgpull%s' % uniq_suffix)
 
 TEST_DIR = tempfile.gettempdir()
 TEST_REPO_PREFIX = 'vcs-test'
 
 # cached repos if any !
 # comment out to get some other repos from bb or github
 GIT_REMOTE_REPO = jn(TESTS_TMP_PATH, GIT_REPO)
 HG_REMOTE_REPO = jn(TESTS_TMP_PATH, HG_REPO)
 
 
 def get_new_dir(title):
     """
     Returns always new directory path.
     """
     from rhodecode.tests.vcs.utils import get_normalized_path
     name = TEST_REPO_PREFIX
     if title:
         name = '-'.join((name, title))
     hex = hashlib.sha1(str(time.time())).hexdigest()
     name = '-'.join((name, hex))
     path = os.path.join(TEST_DIR, name)
     return get_normalized_path(path)
 
 
 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.Session = Session
         self.index_location = config['app_conf']['index_dir']
         TestCase.__init__(self, *args, **kwargs)
 
     def log_user(self, username=TEST_USER_ADMIN_LOGIN,
                  password=TEST_USER_ADMIN_PASS):
         self._logged_username = username
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': username,
                                   'password': password})
 
         if 'invalid user name' in response.body:
             self.fail('could not login using %s %s' % (username, password))
 
         self.assertEqual(response.status, '302 Found')
         ses = response.session['rhodecode_user']
         self.assertEqual(ses.get('username'), username)
         response = response.follow()
         self.assertEqual(ses.get('is_authenticated'), True)
 
         return response.session['rhodecode_user']
 
     def _get_logged_user(self):
         return User.get_by_username(self._logged_username)
 
     def checkSessionFlash(self, response, msg):
         self.assertTrue('flash' in response.session)
         if not msg in response.session['flash'][0][1]:
             self.fail(
                 'msg `%s` not found in session flash: got `%s` instead' % (
-                      msg, response.session['flash'][0][1])
+                      msg, response.session['flash'])
             )

rhodecode/tests/functional/test_admin_settings.py

@@ -1,213 +1,266 b''
 # -*- coding: utf-8 -*-
 
 from rhodecode.lib.auth import get_crypt_password, check_password
-from rhodecode.model.db import User, RhodeCodeSetting
+from rhodecode.model.db import User, RhodeCodeSetting, Repository
 from rhodecode.tests import *
 from rhodecode.lib import helpers as h
 from rhodecode.model.user import UserModel
+from rhodecode.model.scm import ScmModel
 
 
 class TestAdminSettingsController(TestController):
 
     def test_index(self):
         response = self.app.get(url('admin_settings'))
         # Test response...
 
     def test_index_as_xml(self):
         response = self.app.get(url('formatted_admin_settings', format='xml'))
 
     def test_create(self):
         response = self.app.post(url('admin_settings'))
 
     def test_new(self):
         response = self.app.get(url('admin_new_setting'))
 
     def test_new_as_xml(self):
         response = self.app.get(url('formatted_admin_new_setting', format='xml'))
 
     def test_update(self):
         response = self.app.put(url('admin_setting', setting_id=1))
 
     def test_update_browser_fakeout(self):
         response = self.app.post(url('admin_setting', setting_id=1), params=dict(_method='put'))
 
     def test_delete(self):
         response = self.app.delete(url('admin_setting', setting_id=1))
 
     def test_delete_browser_fakeout(self):
         response = self.app.post(url('admin_setting', setting_id=1), params=dict(_method='delete'))
 
     def test_show(self):
         response = self.app.get(url('admin_setting', setting_id=1))
 
     def test_show_as_xml(self):
         response = self.app.get(url('formatted_admin_setting', setting_id=1, format='xml'))
 
     def test_edit(self):
         response = self.app.get(url('admin_edit_setting', setting_id=1))
 
     def test_edit_as_xml(self):
         response = self.app.get(url('formatted_admin_edit_setting',
                                     setting_id=1, format='xml'))
 
     def test_ga_code_active(self):
         self.log_user()
         old_title = 'RhodeCode'
         old_realm = 'RhodeCode authentication'
         new_ga_code = 'ga-test-123456789'
         response = self.app.post(url('admin_setting', setting_id='global'),
                                      params=dict(
                                                  _method='put',
                                                  rhodecode_title=old_title,
                                                  rhodecode_realm=old_realm,
                                                  rhodecode_ga_code=new_ga_code
                                                  ))
 
         self.checkSessionFlash(response, 'Updated application settings')
 
         self.assertEqual(RhodeCodeSetting
                          .get_app_settings()['rhodecode_ga_code'], new_ga_code)
 
         response = response.follow()
         response.mustcontain("""_gaq.push(['_setAccount', '%s']);""" % new_ga_code)
 
     def test_ga_code_inactive(self):
         self.log_user()
         old_title = 'RhodeCode'
         old_realm = 'RhodeCode authentication'
         new_ga_code = ''
         response = self.app.post(url('admin_setting', setting_id='global'),
                                      params=dict(
                                                  _method='put',
                                                  rhodecode_title=old_title,
                                                  rhodecode_realm=old_realm,
                                                  rhodecode_ga_code=new_ga_code
                                                  ))
 
         self.assertTrue('Updated application settings' in
                         response.session['flash'][0][1])
         self.assertEqual(RhodeCodeSetting
                         .get_app_settings()['rhodecode_ga_code'], new_ga_code)
 
         response = response.follow()
         self.assertFalse("""_gaq.push(['_setAccount', '%s']);""" % new_ga_code
                          in response.body)
 
     def test_title_change(self):
         self.log_user()
         old_title = 'RhodeCode'
         new_title = old_title + '_changed'
         old_realm = 'RhodeCode authentication'
 
         for new_title in ['Changed', 'Żółwik', old_title]:
             response = self.app.post(url('admin_setting', setting_id='global'),
                                          params=dict(
                                                      _method='put',
                                                      rhodecode_title=new_title,
                                                      rhodecode_realm=old_realm,
                                                      rhodecode_ga_code=''
                                                      ))
 
             self.checkSessionFlash(response, 'Updated application settings')
             self.assertEqual(RhodeCodeSetting
                              .get_app_settings()['rhodecode_title'],
                              new_title.decode('utf-8'))
 
             response = response.follow()
             response.mustcontain("""<h1><a href="/">%s</a></h1>""" % new_title)
 
     def test_my_account(self):
         self.log_user()
         response = self.app.get(url('admin_settings_my_account'))
 
         self.assertTrue('value="test_admin' in response.body)
 
     @parameterized.expand([('firstname', 'new_username'),
                            ('lastname', 'new_username'),
                            ('admin', True),
                            ('admin', False),
                            ('ldap_dn', 'test'),
                            ('ldap_dn', None),
                            ('active', False),
                            ('active', True),
                            ('email', '[e-mail]'),
                            ])
     def test_my_account_update(self, name, expected):
         uname = 'testme'
         usr = UserModel().create_or_update(username=uname, password='qweqwe',
                                            email='[e-mail]')
         self.Session().commit()
         params = usr.get_api_data()
         user_id = usr.user_id
         self.log_user(username=uname, password='qweqwe')
         params.update({name: expected})
         params.update({'password_confirmation': ''})
         params.update({'new_password': ''})
 
         try:
             response = self.app.put(url('admin_settings_my_account_update',
                                         id=user_id), params)
 
             self.checkSessionFlash(response,
                                    'Your account was updated successfully')
 
             updated_user = User.get_by_username(uname)
             updated_params = updated_user.get_api_data()
             updated_params.update({'password_confirmation': ''})
             updated_params.update({'new_password': ''})
 
             params['last_login'] = updated_params['last_login']
             if name == 'email':
                 params['emails'] = [expected]
             if name == 'ldap_dn':
                 #cannot update this via form
                 params['ldap_dn'] = None
             if name == 'active':
                 #my account cannot deactivate account
                 params['active'] = True
             if name == 'admin':
                 #my account cannot make you an admin !
                 params['admin'] = False
 
             self.assertEqual(params, updated_params)
 
         finally:
             UserModel().delete('testme')
 
     def test_my_account_update_err_email_exists(self):
         self.log_user()
 
         new_email = '[e-mail]'  # already exisitn email
         response = self.app.put(url('admin_settings_my_account_update'),
                                 params=dict(
                                     username='test_admin',
                                     new_password='test12',
                                     password_confirmation='test122',
                                     firstname='NewName',
                                     lastname='NewLastname',
                                     email=new_email,)
                                 )
 
         response.mustcontain('This e-mail address is already taken')
 
     def test_my_account_update_err(self):
         self.log_user('test_regular2', 'test12')
 
         new_email = 'newmail.pl'
         response = self.app.post(url('admin_settings_my_account_update'),
                                  params=dict(
                                             _method='put',
                                             username='test_admin',
                                             new_password='test12',
                                             password_confirmation='test122',
                                             firstname='NewName',
                                             lastname='NewLastname',
                                             email=new_email,)
                                  )
 
         response.mustcontain('An email address must contain a single @')
         from rhodecode.model import validators
         msg = validators.ValidUsername(edit=False,
                                     old_data={})._messages['username_exists']
         msg = h.html_escape(msg % {'username': 'test_admin'})
         response.mustcontain(u"%s" % msg)
+
+    def test_set_repo_fork_has_no_self_id(self):
+        self.log_user()
+        repo = Repository.get_by_repo_name(HG_REPO)
+        response = self.app.get(url('edit_repo', repo_name=HG_REPO))
+        opt = """<option value="%s">vcs_test_git</option>""" % repo.repo_id
+        assert opt not in response.body
+
+    def test_set_fork_of_repo(self):
+        self.log_user()
+        repo = Repository.get_by_repo_name(HG_REPO)
+        repo2 = Repository.get_by_repo_name(GIT_REPO)
+        response = self.app.put(url('repo_as_fork', repo_name=HG_REPO),
+                                 params=dict(
+                                    id_fork_of=repo2.repo_id
+                                 ))
+        repo = Repository.get_by_repo_name(HG_REPO)
+        repo2 = Repository.get_by_repo_name(GIT_REPO)
+        self.checkSessionFlash(response,
+        'Marked repo %s as fork of %s' % (repo.repo_name, repo2.repo_name))
+
+        assert repo.fork == repo2
+        response = response.follow()
+        # check if given repo is selected
+
+        opt = """<option value="%s" selected="selected">%s</option>""" % (
+                                                repo2.repo_id, repo2.repo_name)
+        response.mustcontain(opt)
+
+        # clean session flash
+        #response = self.app.get(url('edit_repo', repo_name=HG_REPO))
+
+        ## mark it as None
+        response = self.app.put(url('repo_as_fork', repo_name=HG_REPO),
+                                 params=dict(
+                                    id_fork_of=None
+                                 ))
+        repo = Repository.get_by_repo_name(HG_REPO)
+        repo2 = Repository.get_by_repo_name(GIT_REPO)
+        self.checkSessionFlash(response,
+        'Marked repo %s as fork of %s' % (repo.repo_name, "Nothing"))
+        assert repo.fork == None
+
+    def test_set_fork_of_same_repo(self):
+        self.log_user()
+        repo = Repository.get_by_repo_name(HG_REPO)
+        response = self.app.put(url('repo_as_fork', repo_name=HG_REPO),
+                                 params=dict(
+                                    id_fork_of=repo.repo_id
+                                 ))
+        self.checkSessionFlash(response,
+                               'An error occurred during this operation')
  No newline at end of file