Changeset - ad41c9a72a88
[Not reviewed]
default
0 6 0
Mads Kiilerich - 9 years ago 2016-08-04 14:23:36
madski@unity3d.com
routing: use POST to 'delete_repo' instead of DELETE
6 files changed with 9 insertions and 10 deletions:
0 comments (0 inline, 0 general)
kallithea/config/routing.py
Show inline comments
 
@@ -29,195 +29,194 @@ def make_map(config):
 
    """Create, configure and return the routes Mapper"""
 
    rmap = Mapper(directory=config['pylons.paths']['controllers'],
 
                  always_scan=config['debug'])
 
    rmap.minimization = False
 
    rmap.explicit = False
 

	
 
    from kallithea.lib.utils import (is_valid_repo, is_valid_repo_group,
 
                                     get_repo_by_id)
 

	
 
    def check_repo(environ, match_dict):
 
        """
 
        check for valid repository for proper 404 handling
 

	
 
        :param environ:
 
        :param match_dict:
 
        """
 
        repo_name = match_dict.get('repo_name')
 

	
 
        if match_dict.get('f_path'):
 
            #fix for multiple initial slashes that causes errors
 
            match_dict['f_path'] = match_dict['f_path'].lstrip('/')
 

	
 
        by_id_match = get_repo_by_id(repo_name)
 
        if by_id_match:
 
            repo_name = by_id_match
 
            match_dict['repo_name'] = repo_name
 

	
 
        return is_valid_repo(repo_name, config['base_path'])
 

	
 
    def check_group(environ, match_dict):
 
        """
 
        check for valid repository group for proper 404 handling
 

	
 
        :param environ:
 
        :param match_dict:
 
        """
 
        repo_group_name = match_dict.get('group_name')
 
        return is_valid_repo_group(repo_group_name, config['base_path'])
 

	
 
    def check_group_skip_path(environ, match_dict):
 
        """
 
        check for valid repository group for proper 404 handling, but skips
 
        verification of existing path
 

	
 
        :param environ:
 
        :param match_dict:
 
        """
 
        repo_group_name = match_dict.get('group_name')
 
        return is_valid_repo_group(repo_group_name, config['base_path'],
 
                                   skip_path_check=True)
 

	
 
    def check_user_group(environ, match_dict):
 
        """
 
        check for valid user group for proper 404 handling
 

	
 
        :param environ:
 
        :param match_dict:
 
        """
 
        return True
 

	
 
    def check_int(environ, match_dict):
 
        return match_dict.get('id').isdigit()
 

	
 
    # The ErrorController route (handles 404/500 error pages); it should
 
    # likely stay at the top, ensuring it can always be resolved
 
    rmap.connect('/error/{action}', controller='error')
 
    rmap.connect('/error/{action}/{id}', controller='error')
 

	
 
    #==========================================================================
 
    # CUSTOM ROUTES HERE
 
    #==========================================================================
 

	
 
    #MAIN PAGE
 
    rmap.connect('home', '/', controller='home', action='index')
 
    rmap.connect('about', '/about', controller='home', action='about')
 
    rmap.connect('repo_switcher_data', '/_repos', controller='home',
 
                 action='repo_switcher_data')
 

	
 
    rmap.connect('rst_help',
 
                 "http://docutils.sourceforge.net/docs/user/rst/quickref.html",
 
                 _static=True)
 
    rmap.connect('kallithea_project_url', "https://kallithea-scm.org/", _static=True)
 
    rmap.connect('issues_url', 'https://bitbucket.org/conservancy/kallithea/issues', _static=True)
 

	
 
    #ADMIN REPOSITORY ROUTES
 
    with rmap.submapper(path_prefix=ADMIN_PREFIX,
 
                        controller='admin/repos') as m:
 
        m.connect("repos", "/repos",
 
                  action="create", conditions=dict(method=["POST"]))
 
        m.connect("repos", "/repos",
 
                  action="index", conditions=dict(method=["GET"]))
 
        m.connect("new_repo", "/create_repository",
 
                  action="create_repository", conditions=dict(method=["GET"]))
 
        m.connect("put_repo", "/repos/{repo_name:.*?}",
 
                  action="update", conditions=dict(method=["PUT"],
 
                  function=check_repo))
 
        m.connect("delete_repo", "/repos/{repo_name:.*?}",
 
                  action="delete", conditions=dict(method=["DELETE"],
 
                  ))
 
        m.connect("delete_repo", "/repos/{repo_name:.*?}/delete",
 
                  action="delete", conditions=dict(method=["POST"]))
 

	
 
    #ADMIN REPOSITORY GROUPS ROUTES
 
    with rmap.submapper(path_prefix=ADMIN_PREFIX,
 
                        controller='admin/repo_groups') as m:
 
        m.connect("repos_groups", "/repo_groups",
 
                  action="create", conditions=dict(method=["POST"]))
 
        m.connect("repos_groups", "/repo_groups",
 
                  action="index", conditions=dict(method=["GET"]))
 
        m.connect("new_repos_group", "/repo_groups/new",
 
                  action="new", conditions=dict(method=["GET"]))
 
        m.connect("update_repos_group", "/repo_groups/{group_name:.*?}",
 
                  action="update", conditions=dict(method=["PUT"],
 
                                                   function=check_group))
 

	
 
        m.connect("repos_group", "/repo_groups/{group_name:.*?}",
 
                  action="show", conditions=dict(method=["GET"],
 
                                                 function=check_group))
 

	
 
        #EXTRAS REPO GROUP ROUTES
 
        m.connect("edit_repo_group", "/repo_groups/{group_name:.*?}/edit",
 
                  action="edit",
 
                  conditions=dict(method=["GET"], function=check_group))
 

	
 
        m.connect("edit_repo_group_advanced", "/repo_groups/{group_name:.*?}/edit/advanced",
 
                  action="edit_repo_group_advanced",
 
                  conditions=dict(method=["GET"], function=check_group))
 

	
 
        m.connect("edit_repo_group_perms", "/repo_groups/{group_name:.*?}/edit/permissions",
 
                  action="edit_repo_group_perms",
 
                  conditions=dict(method=["GET"], function=check_group))
 
        m.connect("edit_repo_group_perms", "/repo_groups/{group_name:.*?}/edit/permissions",
 
                  action="update_perms",
 
                  conditions=dict(method=["PUT"], function=check_group))
 
        m.connect("edit_repo_group_perms_delete", "/repo_groups/{group_name:.*?}/edit/permissions/delete",
 
                  action="delete_perms",
 
                  conditions=dict(method=["POST"], function=check_group))
 

	
 
        m.connect("delete_repo_group", "/repo_groups/{group_name:.*?}/delete",
 
                  action="delete", conditions=dict(method=["POST"],
 
                                                   function=check_group_skip_path))
 

	
 

	
 
    #ADMIN USER ROUTES
 
    with rmap.submapper(path_prefix=ADMIN_PREFIX,
 
                        controller='admin/users') as m:
 
        m.connect("users", "/users",
 
                  action="create", conditions=dict(method=["POST"]))
 
        m.connect("users", "/users",
 
                  action="index", conditions=dict(method=["GET"]))
 
        m.connect("formatted_users", "/users.{format}",
 
                  action="index", conditions=dict(method=["GET"]))
 
        m.connect("new_user", "/users/new",
 
                  action="new", conditions=dict(method=["GET"]))
 
        m.connect("update_user", "/users/{id}",
 
                  action="update", conditions=dict(method=["PUT"]))
 
        m.connect("delete_user", "/users/{id}/delete",
 
                  action="delete", conditions=dict(method=["POST"]))
 
        m.connect("edit_user", "/users/{id}/edit",
 
                  action="edit", conditions=dict(method=["GET"]))
 

	
 
        #EXTRAS USER ROUTES
 
        m.connect("edit_user_advanced", "/users/{id}/edit/advanced",
 
                  action="edit_advanced", conditions=dict(method=["GET"]))
 

	
 
        m.connect("edit_user_api_keys", "/users/{id}/edit/api_keys",
 
                  action="edit_api_keys", conditions=dict(method=["GET"]))
 
        m.connect("edit_user_api_keys", "/users/{id}/edit/api_keys",
 
                  action="add_api_key", conditions=dict(method=["POST"]))
 
        m.connect("edit_user_api_keys_delete", "/users/{id}/edit/api_keys/delete",
 
                  action="delete_api_key", conditions=dict(method=["POST"]))
 

	
 
        m.connect("edit_user_perms", "/users/{id}/edit/permissions",
 
                  action="edit_perms", conditions=dict(method=["GET"]))
 
        m.connect("edit_user_perms", "/users/{id}/edit/permissions",
 
                  action="update_perms", conditions=dict(method=["PUT"]))
 

	
 
        m.connect("edit_user_emails", "/users/{id}/edit/emails",
 
                  action="edit_emails", conditions=dict(method=["GET"]))
 
        m.connect("edit_user_emails", "/users/{id}/edit/emails",
 
                  action="add_email", conditions=dict(method=["PUT"]))
 
        m.connect("edit_user_emails_delete", "/users/{id}/edit/emails/delete",
 
                  action="delete_email", conditions=dict(method=["POST"]))
 

	
 
        m.connect("edit_user_ips", "/users/{id}/edit/ips",
 
                  action="edit_ips", conditions=dict(method=["GET"]))
 
        m.connect("edit_user_ips", "/users/{id}/edit/ips",
 
                  action="add_ip", conditions=dict(method=["PUT"]))
 
        m.connect("edit_user_ips_delete", "/users/{id}/edit/ips/delete",
 
                  action="delete_ip", conditions=dict(method=["POST"]))
 

	
 
    #ADMIN USER GROUPS REST ROUTES
 
    with rmap.submapper(path_prefix=ADMIN_PREFIX,
 
                        controller='admin/user_groups') as m:
 
        m.connect("users_groups", "/user_groups",
 
                  action="create", conditions=dict(method=["POST"]))
 
        m.connect("users_groups", "/user_groups",
kallithea/templates/admin/repos/repo_edit_advanced.html
Show inline comments
 
<h3>${_('Parent')}</h3>
 
${h.form(url('edit_repo_advanced_fork', repo_name=c.repo_info.repo_name), method='put')}
 
<div class="form">
 
       ${h.select('id_fork_of','',c.repos_list,class_="medium")}
 
       ${h.submit('set_as_fork_%s' % c.repo_info.repo_name,_('Set'),class_="btn btn-small")}
 
       <div class="field" style="border:none;color:#888">
 
       <ul>
 
            <li>${_('''Manually set this repository as a fork of another from the list.''')}</li>
 
       </ul>
 
       </div>
 
</div>
 
${h.end_form()}
 

	
 
<script>
 
    $(document).ready(function(){
 
        $("#id_fork_of").select2({
 
            'dropdownAutoWidth': true
 
        });
 
    });
 
</script>
 

	
 
<h3>${_('Public Journal Visibility')}</h3>
 
${h.form(url('edit_repo_advanced_journal', repo_name=c.repo_info.repo_name), method='put')}
 
<div class="form">
 
  <div class="field">
 
  %if c.in_public_journal:
 
    <button class="btn btn-small" type="submit">
 
        <i class="icon-minus"></i>
 
        ${_('Remove from public journal')}
 
    </button>
 
  %else:
 
    <button class="btn btn-small" type="submit">
 
        <i class="icon-plus"></i>
 
        ${_('Add to Public Journal')}
 
    </button>
 
  %endif
 
  </div>
 
 <div class="field" style="border:none;color:#888">
 
 <ul>
 
      <li>${_('All actions done in this repository will be visible to everyone in the public journal.')}</li>
 
 </ul>
 
 </div>
 
</div>
 
${h.end_form()}
 

	
 
<h3>${_('Change Locking')}</h3>
 
${h.form(url('edit_repo_advanced_locking', repo_name=c.repo_info.repo_name), method='put')}
 
<div class="form">
 
      %if c.repo_info.locked[0]:
 
        ${h.hidden('set_unlock', '1')}
 
        <button class="btn btn-small" type="submit"
 
                onclick="return confirm('${_('Confirm to unlock repository.')}');">
 
            <i class="icon-lock-open-alt"></i>
 
            ${_('Unlock Repository')}
 
        </button>
 
       ${_('Locked by %s on %s') % (h.person_by_id(c.repo_info.locked[0]),h.fmt_date(h.time_to_datetime(c.repo_info.locked[1])))}
 
      %else:
 
        ${h.hidden('set_lock', '1')}
 
        <button class="btn btn-small" type="submit"
 
                onclick="return confirm('${_('Confirm to lock repository.')}');">
 
            <i class="icon-lock"></i>
 
            ${_('Lock Repository')}
 
        </button>
 
        ${_('Repository is not locked')}
 
      %endif
 
   <div class="field" style="border:none;color:#888">
 
   <ul>
 
        <li>${_('Force locking on the repository. Works only when anonymous access is disabled. Triggering a pull locks the repository.  The user who is pulling locks the repository; only the user who pulled and locked it can unlock it by doing a push.')}
 
        </li>
 
   </ul>
 
   </div>
 
</div>
 
${h.end_form()}
 

	
 
<h3>${_('Delete')}</h3>
 
${h.form(url('delete_repo', repo_name=c.repo_name), method='delete')}
 
${h.form(url('delete_repo', repo_name=c.repo_name))}
 
<div class="form">
 
    <button class="btn btn-small btn-danger" type="submit"
 
            onclick="return confirm('${_('Confirm to delete this repository: %s') % c.repo_name}');">
 
        <i class="icon-minus-circled"></i>
 
        ${_('Delete this Repository')}
 
    </button>
 
    %if c.repo_info.forks.count():
 
        ${ungettext('This repository has %s fork', 'This repository has %s forks', c.repo_info.forks.count()) % c.repo_info.forks.count()}
 
        <input type="radio" name="forks" value="detach_forks" checked="checked"/> <label for="forks">${_('Detach forks')}</label>
 
        <input type="radio" name="forks" value="delete_forks" /> <label for="forks">${_('Delete forks')}</label>
 
    %endif
 
    <div class="field" style="border:none;color:#888">
 
        <ul>
 
        <li>${_('The deleted repository will be moved away and hidden until the administrator expires it. The administrator can both permanently delete it or restore it.')}</li>
 
        </ul>
 
    </div>
 
</div>
 
${h.end_form()}
kallithea/templates/data_table/_dt_elements.html
Show inline comments
 
@@ -31,187 +31,187 @@
 
           <i class="icon-docs"></i>
 
       </span>
 
       <span>${_('Files')}</span>
 
       </a>
 
    </li>
 
    <li>
 
       <a title="${_('Fork')}" href="${h.url('repo_fork_home',repo_name=repo_name)}">
 
       <span class="icon">
 
           <i class="icon-fork"></i>
 
       </span>
 
       <span>${_('Fork')}</span>
 
       </a>
 
    </li>
 
    <li>
 
       <a title="${_('Settings')}" href="${h.url('edit_repo',repo_name=repo_name)}">
 
       <span class="icon">
 
           <i class="icon-gear"></i>
 
       </span>
 
       <span>${_('Settings')}</span>
 
       </a>
 
    </li>
 
  </ul>
 
</%def>
 

	
 
<%def name="repo_name(name,rtype,rstate,private,fork_of,short_name=False,admin=False)">
 
    <%
 
    def get_name(name,short_name=short_name):
 
      if short_name:
 
        return name.split('/')[-1]
 
      else:
 
        return name
 
    %>
 
  <div class="dt_repo ${'dt_repo_pending' if rstate == 'repo_state_pending' else ''}">
 
    ##NAME
 
    <a href="${h.url('edit_repo' if admin else 'summary_home', repo_name=name)}">
 

	
 
    ##TYPE OF REPO
 
    ${base.repotag(rtype)}
 

	
 
    ##PRIVATE/PUBLIC
 
    %if private and c.visual.show_private_icon:
 
      <i class="icon-keyhole-circled" title="${_('Private repository')}"></i>
 
    %elif not private and c.visual.show_public_icon:
 
      <i class="icon-globe" title="${_('Public repository')}"></i>
 
    %else:
 
      <span style="margin: 0px 8px 0px 8px"></span>
 
    %endif
 
    <span class="dt_repo_name">${get_name(name)}</span>
 
    </a>
 
    %if fork_of:
 
      <a href="${h.url('summary_home',repo_name=fork_of.repo_name)}"><i class="icon-fork"></i></a>
 
    %endif
 
    %if rstate == 'repo_state_pending':
 
      <i class="icon-wrench" title="${_('Repository creation in progress...')}"></i>
 
    %endif
 
  </div>
 
</%def>
 

	
 
<%def name="last_change(last_change)">
 
  <span class="tooltip" date="${last_change}" title="${h.fmt_date(last_change)}">${h.age(last_change)}</span>
 
</%def>
 

	
 
<%def name="revision(name,rev,tip,author,last_msg)">
 
  <div>
 
  %if rev >= 0:
 
      <a title="${'%s:\n\n%s' % (h.escape(author), h.escape(last_msg))}" class="tooltip revision-link safe-html-title" href="${h.url('changeset_home',repo_name=name,revision=tip)}">${'r%s:%s' % (rev,h.short_id(tip))}</a>
 
  %else:
 
      ${_('No changesets yet')}
 
  %endif
 
  </div>
 
</%def>
 

	
 
<%def name="rss(name)">
 
  %if c.authuser.username != 'default':
 
    <a title="${_('Subscribe to %s rss feed')% name}" href="${h.url('rss_feed_home',repo_name=name,api_key=c.authuser.api_key)}"><i class="icon-rss-squared"></i></a>
 
  %else:
 
    <a title="${_('Subscribe to %s rss feed')% name}" href="${h.url('rss_feed_home',repo_name=name)}"><i class="icon-rss-squared"></i></a>
 
  %endif
 
</%def>
 

	
 
<%def name="atom(name)">
 
  %if c.authuser.username != 'default':
 
    <a title="${_('Subscribe to %s atom feed')% name}" href="${h.url('atom_feed_home',repo_name=name,api_key=c.authuser.api_key)}"><i class="icon-rss-squared"></i></a>
 
  %else:
 
    <a title="${_('Subscribe to %s atom feed')% name}" href="${h.url('atom_feed_home',repo_name=name)}"><i class="icon-rss-squared"></i></a>
 
  %endif
 
</%def>
 

	
 
<%def name="repo_actions(repo_name, super_user=True)">
 
  <div>
 
    <div style="float:left; margin-right:5px;" class="grid_edit">
 
      <a href="${h.url('edit_repo',repo_name=repo_name)}" title="${_('Edit')}">
 
        <i class="icon-pencil"></i> ${h.submit('edit_%s' % repo_name,_('Edit'),class_="action_button")}
 
      </a>
 
    </div>
 
    <div style="float:left" class="grid_delete">
 
      ${h.form(h.url('delete_repo', repo_name=repo_name), method='delete')}
 
      ${h.form(h.url('delete_repo', repo_name=repo_name))}
 
        <i class="icon-minus-circled" style="color:#FF4444"></i>
 
        ${h.submit('remove_%s' % repo_name,_('Delete'),class_="action_button",
 
        onclick="return confirm('"+_('Confirm to delete this repository: %s') % repo_name+"');")}
 
      ${h.end_form()}
 
    </div>
 
  </div>
 
</%def>
 

	
 
<%def name="repo_state(repo_state)">
 
  <div>
 
    %if repo_state == u'repo_state_pending':
 
        <div class="btn btn-mini btn-info disabled">${_('Creating')}</div>
 
    %elif repo_state == u'repo_state_created':
 
        <div class="btn btn-mini btn-success disabled">${_('Created')}</div>
 
    %else:
 
        <div class="btn btn-mini btn-danger disabled" title="${repo_state}">invalid</div>
 
    %endif
 
  </div>
 
</%def>
 

	
 
<%def name="user_actions(user_id, username)">
 
 <div style="float:left" class="grid_edit">
 
   <a href="${h.url('edit_user',id=user_id)}" title="${_('Edit')}">
 
     <i class="icon-pencil"></i> ${h.submit('edit_%s' % username,_('Edit'),class_="action_button")}
 
   </a>
 
 </div>
 
 <div style="float:left" class="grid_delete">
 
  ${h.form(h.url('delete_user', id=user_id))}
 
    <i class="icon-minus-circled" style="color:#FF4444"></i>
 
    ${h.submit('remove_',_('Delete'),id="remove_user_%s" % user_id, class_="action_button",
 
    onclick="return confirm('"+_('Confirm to delete this user: %s') % username+"');")}
 
  ${h.end_form()}
 
 </div>
 
</%def>
 

	
 
<%def name="user_group_actions(user_group_id, user_group_name)">
 
 <div style="float:left" class="grid_edit">
 
    <a href="${h.url('edit_users_group', id=user_group_id)}" title="${_('Edit')}">
 
    <i class="icon-pencil"></i>
 
     ${h.submit('edit_%s' % user_group_name,_('Edit'),class_="action_button", id_="submit_user_group_edit")}
 
    </a>
 
 </div>
 
 <div style="float:left" class="grid_delete">
 
    ${h.form(h.url('users_group', id=user_group_id),method='delete')}
 
      <i class="icon-minus-circled" style="color:#FF4444"></i>
 
      ${h.submit('remove_',_('Delete'),id="remove_group_%s" % user_group_id, class_="action_button",
 
      onclick="return confirm('"+_('Confirm to delete this user group: %s') % user_group_name+"');")}
 
    ${h.end_form()}
 
 </div>
 
</%def>
 

	
 
<%def name="repo_group_actions(repo_group_id, repo_group_name, gr_count)">
 
 <div style="float:left" class="grid_edit">
 
    <a href="${h.url('edit_repo_group',group_name=repo_group_name)}" title="${_('Edit')}">
 
    <i class="icon-pencil"></i>
 
     ${h.submit('edit_%s' % repo_group_name, _('Edit'),class_="action_button")}
 
    </a>
 
 </div>
 
 <div style="float:left" class="grid_delete">
 
    ${h.form(h.url('delete_repos_group', group_name=repo_group_name))}
 
        <i class="icon-minus-circled" style="color:#FF4444"></i>
 
        ${h.submit('remove_%s' % repo_group_name,_('Delete'),class_="action_button",
 
        onclick="return confirm('"+ungettext('Confirm to delete this group: %s with %s repository','Confirm to delete this group: %s with %s repositories',gr_count) % (repo_group_name, gr_count)+"');")}
 
    ${h.end_form()}
 
 </div>
 
</%def>
 

	
 
<%def name="user_name(user_id, username)">
 
    ${h.link_to(username,h.url('edit_user', id=user_id))}
 
</%def>
 

	
 
<%def name="repo_group_name(repo_group_name, children_groups)">
 
  <div style="white-space: nowrap">
 
  <a href="${h.url('repos_group_home',group_name=repo_group_name)}">
 
    <i class="icon-folder" title="${_('Repository group')}"></i> ${h.literal(' &raquo; '.join(children_groups))}</a>
 
  </div>
 
</%def>
 

	
 
<%def name="user_group_name(user_group_id, user_group_name)">
 
  <div style="white-space: nowrap">
 
  <a href="${h.url('edit_users_group', id=user_group_id)}">
 
    <i class="icon-users" title="${_('User group')}"></i> ${user_group_name}</a>
 
  </div>
 
</%def>
 

	
 
<%def name="toggle_follow(repo_id)">
 
  <span id="follow_toggle_${repo_id}" class="following" title="${_('Stop following this repository')}"
 
        onclick="toggleFollowingRepo(this, ${repo_id})">
 
  </span>
 
</%def>
kallithea/tests/functional/test_admin_repos.py
Show inline comments
 
@@ -264,263 +264,263 @@ class _BaseTestCase(TestController):
 
        new_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == repo_name_full).one()
 
        new_repo_id = new_repo.repo_id
 

	
 
        assert new_repo.repo_name == repo_name_full
 
        assert new_repo.description == description
 

	
 
        # test if the repository is visible in the list ?
 
        response = self.app.get(url('summary_home', repo_name=repo_name_full))
 
        response.mustcontain(repo_name_full)
 
        response.mustcontain(self.REPO_TYPE)
 

	
 
        # test if the repository was created on filesystem
 
        try:
 
            vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_full)))
 
        except vcs.exceptions.VCSError:
 
            RepoGroupModel().delete(group_name)
 
            Session().commit()
 
            pytest.fail('no repo %s in filesystem' % repo_name)
 

	
 
        #check if inherited permissiona are applied
 
        inherited_perms = UserRepoToPerm.query() \
 
            .filter(UserRepoToPerm.repository_id == new_repo_id).all()
 
        assert len(inherited_perms) == 2
 

	
 
        assert TEST_USER_REGULAR_LOGIN in [x.user.username
 
                                                    for x in inherited_perms]
 
        assert 'repository.write' in [x.permission.permission_name
 
                                               for x in inherited_perms]
 

	
 
        RepoModel().delete(repo_name_full)
 
        RepoGroupModel().delete(group_name)
 
        Session().commit()
 

	
 
    def test_create_remote_repo_wrong_clone_uri(self):
 
        self.log_user()
 
        repo_name = self.NEW_REPO
 
        description = u'description for newly created repo'
 
        response = self.app.post(url('repos'),
 
                        fixture._get_repo_create_params(repo_private=False,
 
                                                repo_name=repo_name,
 
                                                repo_type=self.REPO_TYPE,
 
                                                repo_description=description,
 
                                                clone_uri='http://127.0.0.1/repo',
 
                                                _authentication_token=self.authentication_token()))
 
        response.mustcontain('Invalid repository URL')
 

	
 

	
 
    def test_create_remote_repo_wrong_clone_uri_hg_svn(self):
 
        self.log_user()
 
        repo_name = self.NEW_REPO
 
        description = u'description for newly created repo'
 
        response = self.app.post(url('repos'),
 
                        fixture._get_repo_create_params(repo_private=False,
 
                                                repo_name=repo_name,
 
                                                repo_type=self.REPO_TYPE,
 
                                                repo_description=description,
 
                                                clone_uri='svn+http://127.0.0.1/repo',
 
                                                _authentication_token=self.authentication_token()))
 
        response.mustcontain('Invalid repository URL')
 

	
 

	
 
    def test_delete(self):
 
        self.log_user()
 
        repo_name = u'vcs_test_new_to_delete_%s' % self.REPO_TYPE
 
        description = u'description for newly created repo'
 
        response = self.app.post(url('repos'),
 
                        fixture._get_repo_create_params(repo_private=False,
 
                                                repo_type=self.REPO_TYPE,
 
                                                repo_name=repo_name,
 
                                                repo_description=description,
 
                                                _authentication_token=self.authentication_token()))
 
        ## run the check page that triggers the flash message
 
        response = self.app.get(url('repo_check_home', repo_name=repo_name))
 
        self.checkSessionFlash(response,
 
                               'Created repository <a href="/%s">%s</a>'
 
                               % (repo_name, repo_name))
 
        # test if the repo was created in the database
 
        new_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == repo_name).one()
 

	
 
        assert new_repo.repo_name == repo_name
 
        assert new_repo.description == description
 

	
 
        # test if the repository is visible in the list ?
 
        response = self.app.get(url('summary_home', repo_name=repo_name))
 
        response.mustcontain(repo_name)
 
        response.mustcontain(self.REPO_TYPE)
 

	
 
        # test if the repository was created on filesystem
 
        try:
 
            vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)))
 
        except vcs.exceptions.VCSError:
 
            pytest.fail('no repo %s in filesystem' % repo_name)
 

	
 
        response = self.app.post(url('delete_repo', repo_name=repo_name),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 
            params={'_authentication_token': self.authentication_token()})
 

	
 
        self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name))
 

	
 
        response.follow()
 

	
 
        #check if repo was deleted from db
 
        deleted_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == repo_name).scalar()
 

	
 
        assert deleted_repo == None
 

	
 
        assert os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)) == False
 

	
 
    def test_delete_non_ascii(self):
 
        self.log_user()
 
        non_ascii = "ąęł"
 
        repo_name = "%s%s" % (safe_str(self.NEW_REPO), non_ascii)
 
        repo_name_unicode = safe_unicode(repo_name)
 
        description = 'description for newly created repo' + non_ascii
 
        description_unicode = safe_unicode(description)
 
        response = self.app.post(url('repos'),
 
                        fixture._get_repo_create_params(repo_private=False,
 
                                                repo_name=repo_name,
 
                                                repo_type=self.REPO_TYPE,
 
                                                repo_description=description,
 
                                                _authentication_token=self.authentication_token()))
 
        ## run the check page that triggers the flash message
 
        response = self.app.get(url('repo_check_home', repo_name=repo_name))
 
        assert response.json == {u'result': True}
 
        self.checkSessionFlash(response,
 
                               u'Created repository <a href="/%s">%s</a>'
 
                               % (urllib.quote(repo_name), repo_name_unicode))
 
        # test if the repo was created in the database
 
        new_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == repo_name_unicode).one()
 

	
 
        assert new_repo.repo_name == repo_name_unicode
 
        assert new_repo.description == description_unicode
 

	
 
        # test if the repository is visible in the list ?
 
        response = self.app.get(url('summary_home', repo_name=repo_name))
 
        response.mustcontain(repo_name)
 
        response.mustcontain(self.REPO_TYPE)
 

	
 
        # test if the repository was created on filesystem
 
        try:
 
            vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_unicode)))
 
        except vcs.exceptions.VCSError:
 
            pytest.fail('no repo %s in filesystem' % repo_name)
 

	
 
        response = self.app.post(url('delete_repo', repo_name=repo_name),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 
            params={'_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name_unicode))
 
        response.follow()
 

	
 
        #check if repo was deleted from db
 
        deleted_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == repo_name_unicode).scalar()
 

	
 
        assert deleted_repo == None
 

	
 
        assert os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_unicode)) == False
 

	
 
    def test_delete_repo_with_group(self):
 
        #TODO:
 
        pass
 

	
 
    def test_delete_browser_fakeout(self):
 
        response = self.app.post(url('delete_repo', repo_name=self.REPO),
 
                                 params=dict(_method='delete', _authentication_token=self.authentication_token()))
 
                                 params=dict(_authentication_token=self.authentication_token()))
 

	
 
    def test_show(self):
 
        self.log_user()
 
        response = self.app.get(url('summary_home', repo_name=self.REPO))
 

	
 
    def test_edit(self):
 
        response = self.app.get(url('edit_repo', repo_name=self.REPO))
 

	
 
    def test_set_private_flag_sets_default_to_none(self):
 
        self.log_user()
 
        #initially repository perm should be read
 
        perm = _get_permission_for_user(user='default', repo=self.REPO)
 
        assert len(perm), 1
 
        assert perm[0].permission.permission_name == 'repository.read'
 
        assert Repository.get_by_repo_name(self.REPO).private == False
 

	
 
        response = self.app.put(url('put_repo', repo_name=self.REPO),
 
                        fixture._get_repo_create_params(repo_private=1,
 
                                                repo_name=self.REPO,
 
                                                repo_type=self.REPO_TYPE,
 
                                                user=TEST_USER_ADMIN_LOGIN,
 
                                                _authentication_token=self.authentication_token()))
 
        self.checkSessionFlash(response,
 
                               msg='Repository %s updated successfully' % (self.REPO))
 
        assert Repository.get_by_repo_name(self.REPO).private == True
 

	
 
        #now the repo default permission should be None
 
        perm = _get_permission_for_user(user='default', repo=self.REPO)
 
        assert len(perm), 1
 
        assert perm[0].permission.permission_name == 'repository.none'
 

	
 
        response = self.app.put(url('put_repo', repo_name=self.REPO),
 
                        fixture._get_repo_create_params(repo_private=False,
 
                                                repo_name=self.REPO,
 
                                                repo_type=self.REPO_TYPE,
 
                                                user=TEST_USER_ADMIN_LOGIN,
 
                                                _authentication_token=self.authentication_token()))
 
        self.checkSessionFlash(response,
 
                               msg='Repository %s updated successfully' % (self.REPO))
 
        assert Repository.get_by_repo_name(self.REPO).private == False
 

	
 
        #we turn off private now the repo default permission should stay None
 
        perm = _get_permission_for_user(user='default', repo=self.REPO)
 
        assert len(perm), 1
 
        assert perm[0].permission.permission_name == 'repository.none'
 

	
 
        #update this permission back
 
        perm[0].permission = Permission.get_by_key('repository.read')
 
        Session().add(perm[0])
 
        Session().commit()
 

	
 
    def test_set_repo_fork_has_no_self_id(self):
 
        self.log_user()
 
        repo = Repository.get_by_repo_name(self.REPO)
 
        response = self.app.get(url('edit_repo_advanced', repo_name=self.REPO))
 
        opt = """<option value="%s">%s</option>""" % (repo.repo_id, self.REPO)
 
        response.mustcontain(no=[opt])
 

	
 
    def test_set_fork_of_other_repo(self):
 
        self.log_user()
 
        other_repo = u'other_%s' % self.REPO_TYPE
 
        fixture.create_repo(other_repo, repo_type=self.REPO_TYPE)
 
        repo = Repository.get_by_repo_name(self.REPO)
 
        repo2 = Repository.get_by_repo_name(other_repo)
 
        response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
 
                                params=dict(id_fork_of=repo2.repo_id, _authentication_token=self.authentication_token()))
 
        repo = Repository.get_by_repo_name(self.REPO)
 
        repo2 = Repository.get_by_repo_name(other_repo)
 
        self.checkSessionFlash(response,
 
            'Marked repository %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)
 

	
 
        fixture.destroy_repo(other_repo, forks='detach')
 

	
 
    def test_set_fork_of_other_type_repo(self):
 
        self.log_user()
 
        repo = Repository.get_by_repo_name(self.REPO)
 
        repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
 
        response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
 
                                params=dict(id_fork_of=repo2.repo_id, _authentication_token=self.authentication_token()))
 
        repo = Repository.get_by_repo_name(self.REPO)
 
        repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
 
        self.checkSessionFlash(response,
 
            'Cannot set repository as fork of repository with other type')
 

	
 
    def test_set_fork_of_none(self):
 
        self.log_user()
 
        ## mark it as None
 
        response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
kallithea/tests/functional/test_admin_users.py
Show inline comments
 
@@ -101,193 +101,193 @@ class TestAdminUsersController(TestContr
 
        response.mustcontain("""<span class="error-message">%s</span>""" % msg)
 
        response.mustcontain("""<span class="error-message">Please enter a value</span>""")
 
        response.mustcontain("""<span class="error-message">An email address must contain a single @</span>""")
 

	
 
        def get_user():
 
            Session().query(User).filter(User.username == username).one()
 

	
 
        with pytest.raises(NoResultFound):
 
            get_user(), 'found user in database'
 

	
 
    def test_new(self):
 
        self.log_user()
 
        response = self.app.get(url('new_user'))
 

	
 
    @parametrize('name,attrs',
 
        [('firstname', {'firstname': 'new_username'}),
 
         ('lastname', {'lastname': 'new_username'}),
 
         ('admin', {'admin': True}),
 
         ('admin', {'admin': False}),
 
         ('extern_type', {'extern_type': 'ldap'}),
 
         ('extern_type', {'extern_type': None}),
 
         ('extern_name', {'extern_name': 'test'}),
 
         ('extern_name', {'extern_name': None}),
 
         ('active', {'active': False}),
 
         ('active', {'active': True}),
 
         ('email', {'email': 'someemail@example.com'}),
 
        # ('new_password', {'new_password': 'foobar123',
 
        #                   'password_confirmation': 'foobar123'})
 
        ])
 
    def test_update(self, name, attrs):
 
        self.log_user()
 
        usr = fixture.create_user(self.test_user_1, password='qweqwe',
 
                                  email='testme@example.com',
 
                                  extern_type='internal',
 
                                  extern_name=self.test_user_1,
 
                                  skip_if_exists=True)
 
        Session().commit()
 
        params = usr.get_api_data(True)
 
        params.update({'password_confirmation': ''})
 
        params.update({'new_password': ''})
 
        params.update(attrs)
 
        if name == 'email':
 
            params['emails'] = [attrs['email']]
 
        if name == 'extern_type':
 
            #cannot update this via form, expected value is original one
 
            params['extern_type'] = "internal"
 
        if name == 'extern_name':
 
            #cannot update this via form, expected value is original one
 
            params['extern_name'] = self.test_user_1
 
            # special case since this user is not
 
                                          # logged in yet his data is not filled
 
                                          # so we use creation data
 

	
 
        params.update({'_authentication_token': self.authentication_token()})
 
        response = self.app.put(url('user', id=usr.user_id), params)
 
        self.checkSessionFlash(response, 'User updated successfully')
 
        params.pop('_authentication_token')
 

	
 
        updated_user = User.get_by_username(self.test_user_1)
 
        updated_params = updated_user.get_api_data(True)
 
        updated_params.update({'password_confirmation': ''})
 
        updated_params.update({'new_password': ''})
 

	
 
        assert params == updated_params
 

	
 
    def test_delete(self):
 
        self.log_user()
 
        username = 'newtestuserdeleteme'
 

	
 
        fixture.create_user(name=username)
 

	
 
        new_user = Session().query(User) \
 
            .filter(User.username == username).one()
 
        response = self.app.post(url('delete_user', id=new_user.user_id),
 
            params={'_authentication_token': self.authentication_token()})
 

	
 
        self.checkSessionFlash(response, 'Successfully deleted user')
 

	
 
    def test_delete_repo_err(self):
 
        self.log_user()
 
        username = 'repoerr'
 
        reponame = u'repoerr_fail'
 

	
 
        fixture.create_user(name=username)
 
        fixture.create_repo(name=reponame, cur_user=username)
 

	
 
        new_user = Session().query(User) \
 
            .filter(User.username == username).one()
 
        response = self.app.post(url('delete_user', id=new_user.user_id),
 
            params={'_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'User "%s" still '
 
                               'owns 1 repositories and cannot be removed. '
 
                               'Switch owners or remove those repositories: '
 
                               '%s' % (username, reponame))
 

	
 
        response = self.app.post(url('delete_repo', repo_name=reponame),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 
            params={'_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'Deleted repository %s' % reponame)
 

	
 
        response = self.app.post(url('delete_user', id=new_user.user_id),
 
            params={'_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'Successfully deleted user')
 

	
 
    def test_delete_repo_group_err(self):
 
        self.log_user()
 
        username = 'repogrouperr'
 
        groupname = u'repogroup_fail'
 

	
 
        fixture.create_user(name=username)
 
        fixture.create_repo_group(name=groupname, cur_user=username)
 

	
 
        new_user = Session().query(User) \
 
            .filter(User.username == username).one()
 
        response = self.app.post(url('delete_user', id=new_user.user_id),
 
            params={'_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'User "%s" still '
 
                               'owns 1 repository groups and cannot be removed. '
 
                               'Switch owners or remove those repository groups: '
 
                               '%s' % (username, groupname))
 

	
 
        # Relevant _if_ the user deletion succeeded to make sure we can render groups without owner
 
        # rg = RepoGroup.get_by_group_name(group_name=groupname)
 
        # response = self.app.get(url('repos_groups', id=rg.group_id))
 

	
 
        response = self.app.post(url('delete_repo_group', group_name=groupname),
 
            params={'_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'Removed repository group %s' % groupname)
 

	
 
        response = self.app.post(url('delete_user', id=new_user.user_id),
 
            params={'_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'Successfully deleted user')
 

	
 
    def test_delete_user_group_err(self):
 
        self.log_user()
 
        username = 'usergrouperr'
 
        groupname = u'usergroup_fail'
 

	
 
        fixture.create_user(name=username)
 
        ug = fixture.create_user_group(name=groupname, cur_user=username)
 

	
 
        new_user = Session().query(User) \
 
            .filter(User.username == username).one()
 
        response = self.app.post(url('delete_user', id=new_user.user_id),
 
            params={'_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'User "%s" still '
 
                               'owns 1 user groups and cannot be removed. '
 
                               'Switch owners or remove those user groups: '
 
                               '%s' % (username, groupname))
 

	
 
        # TODO: why do this fail?
 
        #response = self.app.delete(url('delete_users_group', id=groupname))
 
        #self.checkSessionFlash(response, 'Removed user group %s' % groupname)
 

	
 
        fixture.destroy_user_group(ug.users_group_id)
 

	
 
        response = self.app.post(url('delete_user', id=new_user.user_id),
 
            params={'_authentication_token': self.authentication_token()})
 
        self.checkSessionFlash(response, 'Successfully deleted user')
 

	
 
    def test_edit(self):
 
        self.log_user()
 
        user = User.get_by_username(TEST_USER_ADMIN_LOGIN)
 
        response = self.app.get(url('edit_user', id=user.user_id))
 

	
 
    def test_add_perm_create_repo(self):
 
        self.log_user()
 
        perm_none = Permission.get_by_key('hg.create.none')
 
        perm_create = Permission.get_by_key('hg.create.repository')
 

	
 
        user = UserModel().create_or_update(username='dummy', password='qwe',
 
                                            email='dummy', firstname=u'a',
 
                                            lastname=u'b')
 
        Session().commit()
 
        uid = user.user_id
 

	
 
        try:
 
            #User should have None permission on creation repository
 
            assert UserModel().has_perm(user, perm_none) == False
 
            assert UserModel().has_perm(user, perm_create) == False
 

	
 
            response = self.app.post(url('edit_user_perms', id=uid),
 
                                     params=dict(_method='put',
 
                                                 create_repo_perm=True,
 
                                                 _authentication_token=self.authentication_token()))
 

	
 
            perm_none = Permission.get_by_key('hg.create.none')
 
            perm_create = Permission.get_by_key('hg.create.repository')
 

	
 
            #User should have None permission on creation repository
 
            assert UserModel().has_perm(uid, perm_none) == False
 
            assert UserModel().has_perm(uid, perm_create) == True
 
        finally:
 
            UserModel().delete(uid)
kallithea/tests/functional/test_forks.py
Show inline comments
 
# -*- coding: utf-8 -*-
 

	
 
import unittest
 

	
 
from kallithea.tests import *
 
from kallithea.tests.fixture import Fixture
 

	
 
from kallithea.model.db import Repository
 
from kallithea.model.repo import RepoModel
 
from kallithea.model.user import UserModel
 
from kallithea.model.meta import Session
 

	
 
fixture = Fixture()
 

	
 
class _BaseTestCase(TestController):
 
    """
 
    Write all tests here
 
    """
 
    REPO = None
 
    REPO_TYPE = None
 
    NEW_REPO = None
 
    REPO_FORK = None
 

	
 
    def setup_method(self, method):
 
        self.username = u'forkuser'
 
        self.password = u'qweqwe'
 
        self.u1 = fixture.create_user(self.username, password=self.password,
 
                                      email=u'fork_king@example.com')
 
        Session().commit()
 

	
 
    def teardown_method(self, method):
 
        Session().delete(self.u1)
 
        Session().commit()
 

	
 

	
 
    def test_index(self):
 
        self.log_user()
 
        repo_name = self.REPO
 
        response = self.app.get(url(controller='forks', action='forks',
 
                                    repo_name=repo_name))
 

	
 
        response.mustcontain("""There are no forks yet""")
 

	
 
    def test_no_permissions_to_fork(self):
 
        usr = self.log_user(TEST_USER_REGULAR_LOGIN,
 
                            TEST_USER_REGULAR_PASS)['user_id']
 
        user_model = UserModel()
 
        user_model.revoke_perm(usr, 'hg.fork.repository')
 
        user_model.grant_perm(usr, 'hg.fork.none')
 
        u = UserModel().get(usr)
 
        u.inherit_default_permissions = False
 
        Session().commit()
 
        # try create a fork
 
        repo_name = self.REPO
 
        self.app.post(url(controller='forks', action='fork_create',
 
                          repo_name=repo_name), {'_authentication_token': self.authentication_token()}, status=403)
 

	
 
    def test_index_with_fork(self):
 
        self.log_user()
 

	
 
        # create a fork
 
        fork_name = self.REPO_FORK
 
        description = 'fork of vcs test'
 
        repo_name = self.REPO
 
        org_repo = Repository.get_by_repo_name(repo_name)
 
        creation_args = {
 
            'repo_name': fork_name,
 
            'repo_group': u'-1',
 
            'fork_parent_id': org_repo.repo_id,
 
            'repo_type': self.REPO_TYPE,
 
            'description': description,
 
            'private': 'False',
 
            'landing_rev': 'rev:tip',
 
            '_authentication_token': self.authentication_token()}
 

	
 
        self.app.post(url(controller='forks', action='fork_create',
 
                          repo_name=repo_name), creation_args)
 

	
 
        response = self.app.get(url(controller='forks', action='forks',
 
                                    repo_name=repo_name))
 

	
 
        response.mustcontain(
 
            """<a href="/%s">%s</a>""" % (fork_name, fork_name)
 
        )
 

	
 
        # remove this fork
 
        response = self.app.post(url('delete_repo', repo_name=fork_name),
 
            params={'_method': 'delete', '_authentication_token': self.authentication_token()})
 
            params={'_authentication_token': self.authentication_token()})
 

	
 
    def test_fork_create_into_group(self):
 
        self.log_user()
 
        group = fixture.create_repo_group(u'vc')
 
        group_id = group.group_id
 
        fork_name = self.REPO_FORK
 
        fork_name_full = 'vc/%s' % fork_name
 
        description = 'fork of vcs test'
 
        repo_name = self.REPO
 
        org_repo = Repository.get_by_repo_name(repo_name)
 
        creation_args = {
 
            'repo_name': fork_name,
 
            'repo_group': group_id,
 
            'fork_parent_id': org_repo.repo_id,
 
            'repo_type': self.REPO_TYPE,
 
            'description': description,
 
            'private': 'False',
 
            'landing_rev': 'rev:tip',
 
            '_authentication_token': self.authentication_token()}
 
        self.app.post(url(controller='forks', action='fork_create',
 
                          repo_name=repo_name), creation_args)
 
        repo = Repository.get_by_repo_name(fork_name_full)
 
        assert repo.fork.repo_name == self.REPO
 

	
 
        ## run the check page that triggers the flash message
 
        response = self.app.get(url('repo_check_home', repo_name=fork_name_full))
 
        #test if we have a message that fork is ok
 
        self.checkSessionFlash(response,
 
                'Forked repository %s as <a href="/%s">%s</a>'
 
                % (repo_name, fork_name_full, fork_name_full))
 

	
 
        #test if the fork was created in the database
 
        fork_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == fork_name_full).one()
 

	
 
        assert fork_repo.repo_name == fork_name_full
 
        assert fork_repo.fork.repo_name == repo_name
 

	
 
        # test if the repository is visible in the list ?
 
        response = self.app.get(url('summary_home', repo_name=fork_name_full))
 
        response.mustcontain(fork_name_full)
 
        response.mustcontain(self.REPO_TYPE)
 
        response.mustcontain('Fork of "<a href="/%s">%s</a>"' % (repo_name, repo_name))
 

	
 
        fixture.destroy_repo(fork_name_full)
 
        fixture.destroy_repo_group(group_id)
 

	
 
    def test_z_fork_create(self):
 
        self.log_user()
 
        fork_name = self.REPO_FORK
 
        description = 'fork of vcs test'
 
        repo_name = self.REPO
 
        org_repo = Repository.get_by_repo_name(repo_name)
 
        creation_args = {
 
            'repo_name': fork_name,
 
            'repo_group': u'-1',
 
            'fork_parent_id': org_repo.repo_id,
 
            'repo_type': self.REPO_TYPE,
 
            'description': description,
 
            'private': 'False',
 
            'landing_rev': 'rev:tip',
 
            '_authentication_token': self.authentication_token()}
 
        self.app.post(url(controller='forks', action='fork_create',
 
                          repo_name=repo_name), creation_args)
 
        repo = Repository.get_by_repo_name(self.REPO_FORK)
 
        assert repo.fork.repo_name == self.REPO
 

	
 
        ## run the check page that triggers the flash message
 
        response = self.app.get(url('repo_check_home', repo_name=fork_name))
 
        #test if we have a message that fork is ok
 
        self.checkSessionFlash(response,
 
                'Forked repository %s as <a href="/%s">%s</a>'
 
                % (repo_name, fork_name, fork_name))
 

	
 
        #test if the fork was created in the database
 
        fork_repo = Session().query(Repository) \
 
            .filter(Repository.repo_name == fork_name).one()
 

	
 
        assert fork_repo.repo_name == fork_name
 
        assert fork_repo.fork.repo_name == repo_name
 

	
 
        # test if the repository is visible in the list ?
 
        response = self.app.get(url('summary_home', repo_name=fork_name))
 
        response.mustcontain(fork_name)
 
        response.mustcontain(self.REPO_TYPE)
 
        response.mustcontain('Fork of "<a href="/%s">%s</a>"' % (repo_name, repo_name))
 

	
 
    def test_zz_fork_permission_page(self):
 
        usr = self.log_user(self.username, self.password)['user_id']
 
        repo_name = self.REPO
 

	
 
        forks = Repository.query() \
 
            .filter(Repository.repo_type == self.REPO_TYPE) \
 
            .filter(Repository.fork_id != None).all()
 
        assert 1 == len(forks)
 

	
0 comments (0 inline, 0 general)