Mercurial > kallithea
changeset 691:7486da5f0628 beta
Refactor codes for scm model
Some test updates, added test for admin user controller
line wrap: on
line diff
--- a/rhodecode/config/environment.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/config/environment.py Sun Nov 14 22:54:16 2010 +0100 @@ -6,7 +6,7 @@ from rhodecode.lib.auth import set_available_permissions, set_base_path from rhodecode.lib.utils import repo2db_mapper, make_ui, set_rhodecode_config from rhodecode.model import init_model -from rhodecode.model.hg import HgModel +from rhodecode.model.scm import ScmModel from sqlalchemy import engine_from_config import logging import os @@ -71,7 +71,7 @@ config['pylons.app_globals'].baseui = make_ui('db') g = config['pylons.app_globals'] - repo2db_mapper(HgModel().repo_scan(g.paths[0][1], g.baseui, initial)) + repo2db_mapper(ScmModel().repo_scan(g.paths[0][1], g.baseui, initial)) set_available_permissions(config) set_base_path(config) set_rhodecode_config(config)
--- a/rhodecode/controllers/admin/repos.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/controllers/admin/repos.py Sun Nov 14 22:54:16 2010 +0100 @@ -35,7 +35,7 @@ from rhodecode.lib.utils import invalidate_cache, action_logger from rhodecode.model.db import User from rhodecode.model.forms import RepoForm -from rhodecode.model.hg import HgModel +from rhodecode.model.scm import ScmModel from rhodecode.model.repo import RepoModel import formencode import logging @@ -60,7 +60,7 @@ def index(self, format='html'): """GET /repos: All items in the collection""" # url('repos') - cached_repo_list = HgModel().get_repos() + cached_repo_list = ScmModel().get_repos() c.repos_list = sorted(cached_repo_list, key=itemgetter('name_sort')) return render('admin/repos/repos.html')
--- a/rhodecode/controllers/admin/settings.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/controllers/admin/settings.py Sun Nov 14 22:54:16 2010 +0100 @@ -36,7 +36,7 @@ from rhodecode.model.db import RhodeCodeSettings, RhodeCodeUi, Repository from rhodecode.model.forms import UserForm, ApplicationSettingsForm, \ ApplicationUiSettingsForm -from rhodecode.model.hg import HgModel +from rhodecode.model.scm import ScmModel from rhodecode.model.user import UserModel from rhodecode.lib.celerylib import tasks, run_task from sqlalchemy import func @@ -99,7 +99,7 @@ rm_obsolete = request.POST.get('destroy', False) log.debug('Rescanning directories with destroy=%s', rm_obsolete) - initial = HgModel().repo_scan(g.paths[0][1], g.baseui) + initial = ScmModel().repo_scan(g.paths[0][1], g.baseui) for repo_name in initial.keys(): invalidate_cache('get_repo_cached_%s' % repo_name) @@ -253,7 +253,7 @@ .filter(Repository.user_id == c.user.user_id)\ .order_by(func.lower(Repository.repo_name))\ .all() - c.user_repos = HgModel().get_repos(all_repos) + c.user_repos = ScmModel().get_repos(all_repos) if c.user.username == 'default': h.flash(_("You can't edit this user since it's" @@ -294,7 +294,7 @@ .filter(Repository.user_id == c.user.user_id)\ .order_by(func.lower(Repository.repo_name))\ .all() - c.user_repos = HgModel().get_repos(all_repos) + c.user_repos = ScmModel().get_repos(all_repos) return htmlfill.render( render('admin/users/user_edit_my_account.html'),
--- a/rhodecode/controllers/branches.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/controllers/branches.py Sun Nov 14 22:54:16 2010 +0100 @@ -26,7 +26,7 @@ from rhodecode.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator from rhodecode.lib.base import BaseController, render from rhodecode.lib.utils import OrderedDict -from rhodecode.model.hg import HgModel +from rhodecode.model.scm import ScmModel import logging log = logging.getLogger(__name__) @@ -38,7 +38,7 @@ super(BranchesController, self).__before__() def index(self): - hg_model = HgModel() + hg_model = ScmModel() c.repo_info = hg_model.get_repo(c.repo_name) c.repo_branches = OrderedDict() for name, hash_ in c.repo_info.branches.items():
--- a/rhodecode/controllers/changelog.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/controllers/changelog.py Sun Nov 14 22:54:16 2010 +0100 @@ -32,7 +32,7 @@ from pylons import request, session, tmpl_context as c from rhodecode.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator from rhodecode.lib.base import BaseController, render -from rhodecode.model.hg import HgModel +from rhodecode.model.scm import ScmModel from webhelpers.paginate import Page import logging log = logging.getLogger(__name__) @@ -60,7 +60,7 @@ else: c.size = int(session.get('changelog_size', default)) - changesets = HgModel().get_repo(c.repo_name) + changesets = ScmModel().get_repo(c.repo_name) p = int(request.params.get('page', 1)) c.total_cs = len(changesets)
--- a/rhodecode/controllers/changeset.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/controllers/changeset.py Sun Nov 14 22:54:16 2010 +0100 @@ -28,7 +28,7 @@ from rhodecode.lib.base import BaseController, render from rhodecode.lib.utils import EmptyChangeset import rhodecode.lib.helpers as h -from rhodecode.model.hg import HgModel +from rhodecode.model.scm import ScmModel from vcs.exceptions import RepositoryError, ChangesetError from vcs.nodes import FileNode from vcs.utils import diffs as differ @@ -46,7 +46,7 @@ super(ChangesetController, self).__before__() def index(self, revision): - hg_model = HgModel() + hg_model = ScmModel() cut_off_limit = 1024 * 250 def wrap_to_table(str): @@ -132,7 +132,7 @@ def raw_changeset(self, revision): - hg_model = HgModel() + hg_model = ScmModel() method = request.GET.get('diff', 'show') try: r = hg_model.get_repo(c.repo_name)
--- a/rhodecode/controllers/feed.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/controllers/feed.py Sun Nov 14 22:54:16 2010 +0100 @@ -24,7 +24,7 @@ """ from pylons import tmpl_context as c, url, response from rhodecode.lib.base import BaseController, render -from rhodecode.model.hg import HgModel +from rhodecode.model.scm import ScmModel from webhelpers.feedgenerator import Atom1Feed, Rss201rev2Feed import logging log = logging.getLogger(__name__) @@ -49,7 +49,7 @@ language=self.language, ttl=self.ttl) - changesets = HgModel().get_repo(repo_name) + changesets = ScmModel().get_repo(repo_name) for cs in changesets[:self.feed_nr]: feed.add_item(title=cs.message, @@ -69,7 +69,7 @@ language=self.language, ttl=self.ttl) - changesets = HgModel().get_repo(repo_name) + changesets = ScmModel().get_repo(repo_name) for cs in changesets[:self.feed_nr]: feed.add_item(title=cs.message, link=url('changeset_home', repo_name=repo_name,
--- a/rhodecode/controllers/files.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/controllers/files.py Sun Nov 14 22:54:16 2010 +0100 @@ -29,7 +29,7 @@ from rhodecode.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator from rhodecode.lib.base import BaseController, render from rhodecode.lib.utils import EmptyChangeset -from rhodecode.model.hg import HgModel +from rhodecode.model.scm import ScmModel from vcs.exceptions import RepositoryError, ChangesetError from vcs.nodes import FileNode from vcs.utils import diffs as differ @@ -49,7 +49,7 @@ c.file_size_limit = 250 * 1024 #limit of file size to display def index(self, repo_name, revision, f_path): - hg_model = HgModel() + hg_model = ScmModel() c.repo = hg_model.get_repo(c.repo_name) revision = request.POST.get('at_rev', None) or revision @@ -95,7 +95,7 @@ return render('files/files.html') def rawfile(self, repo_name, revision, f_path): - hg_model = HgModel() + hg_model = ScmModel() c.repo = hg_model.get_repo(c.repo_name) file_node = c.repo.get_changeset(revision).get_node(f_path) response.content_type = file_node.mimetype @@ -104,7 +104,7 @@ return file_node.content def raw(self, repo_name, revision, f_path): - hg_model = HgModel() + hg_model = ScmModel() c.repo = hg_model.get_repo(c.repo_name) file_node = c.repo.get_changeset(revision).get_node(f_path) response.content_type = 'text/plain' @@ -112,7 +112,7 @@ return file_node.content def annotate(self, repo_name, revision, f_path): - hg_model = HgModel() + hg_model = ScmModel() c.repo = hg_model.get_repo(c.repo_name) c.cs = c.repo.get_changeset(revision) c.file = c.cs.get_node(f_path) @@ -141,7 +141,7 @@ yield data archive = tempfile.TemporaryFile() - repo = HgModel().get_repo(repo_name).repo + repo = ScmModel().get_repo(repo_name).repo fname = '%s-%s%s' % (repo_name, revision, fileformat) archival.archive(repo, archive, revision, archive_specs[fileformat][1], prefix='%s-%s' % (repo_name, revision)) @@ -151,7 +151,7 @@ return read_in_chunks(archive) def diff(self, repo_name, f_path): - hg_model = HgModel() + hg_model = ScmModel() diff1 = request.GET.get('diff1') diff2 = request.GET.get('diff2') c.action = request.GET.get('diff')
--- a/rhodecode/controllers/home.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/controllers/home.py Sun Nov 14 22:54:16 2010 +0100 @@ -26,7 +26,7 @@ from pylons import tmpl_context as c, request from rhodecode.lib.auth import LoginRequired from rhodecode.lib.base import BaseController, render -from rhodecode.model.hg import HgModel +from rhodecode.model.scm import ScmModel import logging log = logging.getLogger(__name__) @@ -47,7 +47,7 @@ else: c.sort_by = current_sort c.sort_slug = current_sort_slug - cached_repo_list = HgModel().get_repos() + cached_repo_list = ScmModel().get_repos() sort_key = current_sort_slug + '_sort' if c.sort_by.startswith('-'):
--- a/rhodecode/controllers/shortlog.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/controllers/shortlog.py Sun Nov 14 22:54:16 2010 +0100 @@ -25,7 +25,7 @@ from pylons import tmpl_context as c, request from rhodecode.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator from rhodecode.lib.base import BaseController, render -from rhodecode.model.hg import HgModel +from rhodecode.model.scm import ScmModel from webhelpers.paginate import Page import logging log = logging.getLogger(__name__) @@ -40,7 +40,7 @@ def index(self): p = int(request.params.get('page', 1)) - repo = HgModel().get_repo(c.repo_name) + repo = ScmModel().get_repo(c.repo_name) c.repo_changesets = Page(repo, page=p, items_per_page=20) c.shortlog_data = render('shortlog/shortlog_data.html') if request.params.get('partial'):
--- a/rhodecode/controllers/summary.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/controllers/summary.py Sun Nov 14 22:54:16 2010 +0100 @@ -27,7 +27,7 @@ from rhodecode.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator from rhodecode.lib.base import BaseController, render from rhodecode.lib.utils import OrderedDict, EmptyChangeset -from rhodecode.model.hg import HgModel +from rhodecode.model.scm import ScmModel from rhodecode.model.db import Statistics from webhelpers.paginate import Page from rhodecode.lib.celerylib import run_task @@ -52,7 +52,7 @@ super(SummaryController, self).__before__() def index(self): - hg_model = HgModel() + hg_model = ScmModel() c.repo_info = hg_model.get_repo(c.repo_name) def url_generator(**kw): return url('shortlog_home', repo_name=c.repo_name, **kw)
--- a/rhodecode/controllers/tags.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/controllers/tags.py Sun Nov 14 22:54:16 2010 +0100 @@ -26,7 +26,7 @@ from rhodecode.lib.auth import LoginRequired, HasRepoPermissionAnyDecorator from rhodecode.lib.base import BaseController, render from rhodecode.lib.utils import OrderedDict -from rhodecode.model.hg import HgModel +from rhodecode.model.scm import ScmModel import logging log = logging.getLogger(__name__) @@ -38,7 +38,7 @@ super(TagsController, self).__before__() def index(self): - hg_model = HgModel() + hg_model = ScmModel() c.repo_info = hg_model.get_repo(c.repo_name) c.repo_tags = OrderedDict() for name, hash_ in c.repo_info.tags.items():
--- a/rhodecode/lib/base.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/lib/base.py Sun Nov 14 22:54:16 2010 +0100 @@ -9,7 +9,7 @@ from rhodecode.lib import auth from rhodecode.lib.utils import get_repo_slug from rhodecode.model import meta -from rhodecode.model.hg import HgModel +from rhodecode.model.scm import ScmModel from vcs import BACKENDS class BaseController(WSGIController): @@ -18,11 +18,11 @@ c.rhodecode_version = __version__ c.rhodecode_name = config['rhodecode_title'] c.repo_name = get_repo_slug(request) - c.cached_repo_list = HgModel().get_repos() + c.cached_repo_list = ScmModel().get_repos() c.backends = BACKENDS.keys() if c.repo_name: - cached_repo = HgModel().get(c.repo_name) + cached_repo = ScmModel().get(c.repo_name) if cached_repo: c.repository_tags = cached_repo.tags
--- a/rhodecode/lib/celerylib/tasks.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/lib/celerylib/tasks.py Sun Nov 14 22:54:16 2010 +0100 @@ -10,7 +10,7 @@ import os import traceback from vcs.backends import get_repo -from rhodecode.model.hg import HgModel +from rhodecode.model.scm import ScmModel try: import json except ImportError: @@ -62,7 +62,7 @@ commits_by_day_author_aggregate = {} commits_by_day_aggregate = {} - repos_path = HgModel().repos_path + repos_path = ScmModel().repos_path p = os.path.join(repos_path, repo_name) repo = get_repo(p) @@ -274,7 +274,7 @@ repo_model = RepoModel(get_session()) repo_model.create(form_data, cur_user, just_db=True, fork=True) repo_name = form_data['repo_name'] - repos_path = HgModel().repos_path + repos_path = ScmModel().repos_path repo_path = os.path.join(repos_path, repo_name) repo_fork_path = os.path.join(repos_path, form_data['fork_name']) alias = form_data['repo_type'] @@ -291,7 +291,7 @@ 's', 'sh', 'tpl', 'txt', 'vim', 'wss', 'xhtml', 'xml', 'xsl', 'xslt', 'yaws'] - repos_path = HgModel().repos_path + repos_path = ScmModel().repos_path p = os.path.join(repos_path, repo_name) repo = get_repo(p) tip = repo.get_changeset()
--- a/rhodecode/lib/indexers/__init__.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/lib/indexers/__init__.py Sun Nov 14 22:54:16 2010 +0100 @@ -6,7 +6,7 @@ sys.path.append(dn(dn(dn(os.path.realpath(__file__))))) from rhodecode.config.environment import load_environment -from rhodecode.model.hg import HgModel +from rhodecode.model.scm import ScmModel from shutil import rmtree from webhelpers.html.builder import escape from vcs.utils.lazy import LazyProperty
--- a/rhodecode/lib/indexers/daemon.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/lib/indexers/daemon.py Sun Nov 14 22:54:16 2010 +0100 @@ -33,7 +33,7 @@ sys.path.append(project_path) -from rhodecode.model.hg import HgModel +from rhodecode.model.scm import ScmModel from rhodecode.lib.helpers import safe_unicode from whoosh.index import create_in, open_dir from shutil import rmtree @@ -78,7 +78,7 @@ if not repo_location: raise Exception('You have to provide repositories location') - self.repo_paths = HgModel().repo_scan(self.repo_location, None, True) + self.repo_paths = ScmModel().repo_scan(self.repo_location, None, True) self.initial = False if not os.path.isdir(self.index_location): os.mkdir(self.index_location)
--- a/rhodecode/model/hg.py Sun Nov 14 22:18:51 2010 +0100 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,177 +0,0 @@ -#!/usr/bin/env python -# encoding: utf-8 -# Model for RhodeCode -# Copyright (C) 2009-2010 Marcin Kuzminski <marcin@python-works.com> -# -# This program is free software; you can redistribute it and/or -# modify it under the terms of the GNU General Public License -# as published by the Free Software Foundation; version 2 -# of the License or (at your opinion) any later version of the license. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program; if not, write to the Free Software -# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, -# MA 02110-1301, USA. -""" -Created on April 9, 2010 -Model for RhodeCode -@author: marcink -""" -from beaker.cache import cache_region, region_invalidate -from mercurial import ui -from rhodecode.lib import helpers as h -from rhodecode.lib.auth import HasRepoPermissionAny -from rhodecode.lib.utils import get_repos -from rhodecode.model import meta -from rhodecode.model.db import Repository, User, RhodeCodeUi -from sqlalchemy.orm import joinedload -from vcs import get_backend -from vcs.utils.helpers import get_scm -from vcs.exceptions import RepositoryError, VCSError -from vcs.utils.lazy import LazyProperty -import logging -import os -import time - -log = logging.getLogger(__name__) - -class HgModel(object): - """ - Mercurial Model - """ - - def __init__(self, sa=None): - if not sa: - self.sa = meta.Session() - else: - self.sa = sa - - - @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, baseui, initial=False): - """ - Listing of repositories in given path. This path should not be a - repository itself. Return a dictionary of repository objects - - :param repos_path: path to directory containing repositories - :param baseui - :param initial: initial scan - """ - log.info('scanning for repositories in %s', repos_path) - - if not isinstance(baseui, ui.ui): - baseui = ui.ui() - repos_list = {} - - for name, path in get_repos(repos_path): - try: - if repos_list.has_key(name): - raise RepositoryError('Duplicate repository name %s ' - 'found in %s' % (name, path)) - else: - - klass = get_backend(path[0]) - - if path[0] == 'hg': - repos_list[name] = klass(path[1], baseui=baseui) - - if path[0] == 'git': - repos_list[name] = klass(path[1]) - except OSError: - continue - - return repos_list - - def get_repos(self, all_repos=None): - """ - 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: give specific repositories list, good for filtering - """ - if not all_repos: - all_repos = self.sa.query(Repository).all() - - for r in all_repos: - - repo = self.get(r.repo_name) - - if repo is not None: - last_change = repo.last_change - tip = h.get_changeset_safe(repo, 'tip') - - tmp_d = {} - tmp_d['name'] = repo.name - tmp_d['name_sort'] = tmp_d['name'].lower() - tmp_d['description'] = repo.dbrepo.description - tmp_d['description_sort'] = tmp_d['description'] - 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'] = repo.dbrepo.user.full_contact - tmp_d['contact_sort'] = tmp_d['contact'] - tmp_d['repo_archives'] = list(repo._get_archives()) - tmp_d['last_msg'] = tip.message - tmp_d['repo'] = repo - yield tmp_d - - def get_repo(self, repo_name): - return self.get(repo_name) - - def get(self, repo_name): - """ - Get's repository from given name, creates BackendInstance and - propagates it's data from database with all additional information - :param repo_name: - """ - if not HasRepoPermissionAny('repository.read', 'repository.write', - 'repository.admin')(repo_name, 'get repo check'): - return - - @cache_region('long_term', 'get_repo_cached_%s' % repo_name) - def _get_repo(repo_name): - - repo_path = os.path.join(self.repos_path, repo_name) - alias = get_scm(repo_path)[0] - - log.debug('Creating instance of %s repository', alias) - backend = get_backend(alias) - - if alias == 'hg': - repo = backend(repo_path, create=False, baseui=None) - #skip hidden web repository - if repo._get_hidden(): - return - else: - repo = backend(repo_path, create=False) - - dbrepo = self.sa.query(Repository)\ - .options(joinedload(Repository.fork))\ - .options(joinedload(Repository.user))\ - .filter(Repository.repo_name == repo_name)\ - .scalar() - repo.dbrepo = dbrepo - return repo - - invalidate = False - if invalidate: - log.info('INVALIDATING CACHE FOR %s', repo_name) - region_invalidate(_get_repo, None, repo_name) - - return _get_repo(repo_name) -
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/rhodecode/model/scm.py Sun Nov 14 22:54:16 2010 +0100 @@ -0,0 +1,177 @@ +#!/usr/bin/env python +# encoding: utf-8 +# Model for RhodeCode +# Copyright (C) 2009-2010 Marcin Kuzminski <marcin@python-works.com> +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License +# as published by the Free Software Foundation; version 2 +# of the License or (at your opinion) any later version of the license. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, +# MA 02110-1301, USA. +""" +Created on April 9, 2010 +Model for RhodeCode +@author: marcink +""" +from beaker.cache import cache_region, region_invalidate +from mercurial import ui +from rhodecode.lib import helpers as h +from rhodecode.lib.auth import HasRepoPermissionAny +from rhodecode.lib.utils import get_repos +from rhodecode.model import meta +from rhodecode.model.db import Repository, User, RhodeCodeUi +from sqlalchemy.orm import joinedload +from vcs import get_backend +from vcs.utils.helpers import get_scm +from vcs.exceptions import RepositoryError, VCSError +from vcs.utils.lazy import LazyProperty +import logging +import os +import time + +log = logging.getLogger(__name__) + +class ScmModel(object): + """ + Mercurial Model + """ + + def __init__(self, sa=None): + if not sa: + self.sa = meta.Session() + else: + self.sa = sa + + + @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, baseui, initial=False): + """ + Listing of repositories in given path. This path should not be a + repository itself. Return a dictionary of repository objects + + :param repos_path: path to directory containing repositories + :param baseui + :param initial: initial scan + """ + log.info('scanning for repositories in %s', repos_path) + + if not isinstance(baseui, ui.ui): + baseui = ui.ui() + repos_list = {} + + for name, path in get_repos(repos_path): + try: + if repos_list.has_key(name): + raise RepositoryError('Duplicate repository name %s ' + 'found in %s' % (name, path)) + else: + + klass = get_backend(path[0]) + + if path[0] == 'hg': + repos_list[name] = klass(path[1], baseui=baseui) + + if path[0] == 'git': + repos_list[name] = klass(path[1]) + except OSError: + continue + + return repos_list + + def get_repos(self, all_repos=None): + """ + 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: give specific repositories list, good for filtering + """ + if not all_repos: + all_repos = self.sa.query(Repository).all() + + for r in all_repos: + + repo = self.get(r.repo_name) + + if repo is not None: + last_change = repo.last_change + tip = h.get_changeset_safe(repo, 'tip') + + tmp_d = {} + tmp_d['name'] = repo.name + tmp_d['name_sort'] = tmp_d['name'].lower() + tmp_d['description'] = repo.dbrepo.description + tmp_d['description_sort'] = tmp_d['description'] + 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'] = repo.dbrepo.user.full_contact + tmp_d['contact_sort'] = tmp_d['contact'] + tmp_d['repo_archives'] = list(repo._get_archives()) + tmp_d['last_msg'] = tip.message + tmp_d['repo'] = repo + yield tmp_d + + def get_repo(self, repo_name): + return self.get(repo_name) + + def get(self, repo_name): + """ + Get's repository from given name, creates BackendInstance and + propagates it's data from database with all additional information + :param repo_name: + """ + if not HasRepoPermissionAny('repository.read', 'repository.write', + 'repository.admin')(repo_name, 'get repo check'): + return + + @cache_region('long_term', 'get_repo_cached_%s' % repo_name) + def _get_repo(repo_name): + + repo_path = os.path.join(self.repos_path, repo_name) + alias = get_scm(repo_path)[0] + + log.debug('Creating instance of %s repository', alias) + backend = get_backend(alias) + + if alias == 'hg': + repo = backend(repo_path, create=False, baseui=None) + #skip hidden web repository + if repo._get_hidden(): + return + else: + repo = backend(repo_path, create=False) + + dbrepo = self.sa.query(Repository)\ + .options(joinedload(Repository.fork))\ + .options(joinedload(Repository.user))\ + .filter(Repository.repo_name == repo_name)\ + .scalar() + repo.dbrepo = dbrepo + return repo + + invalidate = False + if invalidate: + log.info('INVALIDATING CACHE FOR %s', repo_name) + region_invalidate(_get_repo, None, repo_name) + + return _get_repo(repo_name) +
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/rhodecode/tests/functional/test_admin_permissions.py Sun Nov 14 22:54:16 2010 +0100 @@ -0,0 +1,43 @@ +from rhodecode.tests import * + +class TestAdminPermissionsController(TestController): + + def test_index(self): + response = self.app.get(url('permissions')) + # Test response... + + def test_index_as_xml(self): + response = self.app.get(url('formatted_permissions', format='xml')) + + def test_create(self): + response = self.app.post(url('permissions')) + + def test_new(self): + response = self.app.get(url('new_permission')) + + def test_new_as_xml(self): + response = self.app.get(url('formatted_new_permission', format='xml')) + + def test_update(self): + response = self.app.put(url('permission', id=1)) + + def test_update_browser_fakeout(self): + response = self.app.post(url('permission', id=1), params=dict(_method='put')) + + def test_delete(self): + response = self.app.delete(url('permission', id=1)) + + def test_delete_browser_fakeout(self): + response = self.app.post(url('permission', id=1), params=dict(_method='delete')) + + def test_show(self): + response = self.app.get(url('permission', id=1)) + + def test_show_as_xml(self): + response = self.app.get(url('formatted_permission', id=1, format='xml')) + + def test_edit(self): + response = self.app.get(url('edit_permission', id=1)) + + def test_edit_as_xml(self): + response = self.app.get(url('formatted_edit_permission', id=1, format='xml'))
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/rhodecode/tests/functional/test_admin_repos.py Sun Nov 14 22:54:16 2010 +0100 @@ -0,0 +1,130 @@ +from rhodecode.model.db import Repository +from rhodecode.tests import * + +class TestAdminReposController(TestController): + + def test_index(self): + self.log_user() + response = self.app.get(url('repos')) + # Test response... + + def test_index_as_xml(self): + response = self.app.get(url('formatted_repos', format='xml')) + + def test_create_hg(self): + self.log_user() + repo_name = NEW_HG_REPO + description = 'description for newly created repo' + private = False + response = self.app.post(url('repos'), {'repo_name':repo_name, + 'repo_type':'hg', + 'description':description, + 'private':private}) + + + #test if we have a message for that repository + assert '''created repository %s''' % (repo_name) in response.session['flash'][0], 'No flash message about new repo' + + #test if the fork was created in the database + new_repo = self.sa.query(Repository).filter(Repository.repo_name == repo_name).one() + + assert new_repo.repo_name == repo_name, 'wrong name of repo name in db' + assert new_repo.description == description, 'wrong description' + + #test if repository is visible in the list ? + response = response.follow() + + assert repo_name in response.body, 'missing new repo from the main repos list' + + def test_create_git(self): + self.log_user() + repo_name = NEW_GIT_REPO + description = 'description for newly created repo' + private = False + response = self.app.post(url('repos'), {'repo_name':repo_name, + 'repo_type':'git', + 'description':description, + 'private':private}) + + + #test if we have a message for that repository + assert '''created repository %s''' % (repo_name) in response.session['flash'][0], 'No flash message about new repo' + + #test if the fork was created in the database + new_repo = self.sa.query(Repository).filter(Repository.repo_name == repo_name).one() + + assert new_repo.repo_name == repo_name, 'wrong name of repo name in db' + assert new_repo.description == description, 'wrong description' + + #test if repository is visible in the list ? + response = response.follow() + + assert repo_name in response.body, 'missing new repo from the main repos list' + + + def test_new(self): + self.log_user() + response = self.app.get(url('new_repo')) + + def test_new_as_xml(self): + response = self.app.get(url('formatted_new_repo', format='xml')) + + def test_update(self): + response = self.app.put(url('repo', repo_name=HG_REPO)) + + def test_update_browser_fakeout(self): + response = self.app.post(url('repo', repo_name=HG_REPO), params=dict(_method='put')) + + def test_delete(self): + self.log_user() + repo_name = 'vcs_test_new_to_delete' + description = 'description for newly created repo' + private = False + response = self.app.post(url('repos'), {'repo_name':repo_name, + 'repo_type':'hg', + 'description':description, + 'private':private}) + + + #test if we have a message for that repository + assert '''created repository %s''' % (repo_name) in response.session['flash'][0], 'No flash message about new repo' + + #test if the repo was created in the database + new_repo = self.sa.query(Repository).filter(Repository.repo_name == repo_name).one() + + assert new_repo.repo_name == repo_name, 'wrong name of repo name in db' + assert new_repo.description == description, 'wrong description' + + #test if repository is visible in the list ? + response = response.follow() + + assert repo_name in response.body, 'missing new repo from the main repos list' + + + response = self.app.delete(url('repo', repo_name=repo_name)) + + assert '''deleted repository %s''' % (repo_name) in response.session['flash'][0], 'No flash message about delete repo' + + response.follow() + + #check if repo was deleted from db + deleted_repo = self.sa.query(Repository).filter(Repository.repo_name == repo_name).scalar() + + assert deleted_repo is None, 'Deleted repository was found in db' + + + def test_delete_browser_fakeout(self): + response = self.app.post(url('repo', repo_name=HG_REPO), params=dict(_method='delete')) + + def test_show(self): + self.log_user() + response = self.app.get(url('repo', repo_name=HG_REPO)) + + def test_show_as_xml(self): + response = self.app.get(url('formatted_repo', repo_name=HG_REPO, format='xml')) + + def test_edit(self): + response = self.app.get(url('edit_repo', repo_name=HG_REPO)) + + def test_edit_as_xml(self): + response = self.app.get(url('formatted_edit_repo', repo_name=HG_REPO, format='xml'))
--- a/rhodecode/tests/functional/test_admin_settings.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/tests/functional/test_admin_settings.py Sun Nov 14 22:54:16 2010 +0100 @@ -1,7 +1,7 @@ from rhodecode.tests import * from rhodecode.model.db import User -class TestSettingsController(TestController): +class TestAdminSettingsController(TestController): def test_index(self): response = self.app.get(url('admin_settings')) @@ -48,9 +48,9 @@ response = self.app.get(url('admin_settings_my_account')) print response assert 'value="test_admin' in response.body - - - + + + def test_my_account_update(self): self.log_user() new_email = 'new@mail.pl' @@ -64,16 +64,16 @@ email=new_email,)) response.follow() print response - + print 'x' * 100 print response.session assert 'Your account was updated succesfully' in response.session['flash'][0][1], 'no flash message about success of change' user = self.sa.query(User).filter(User.username == 'test_admin').one() assert user.email == new_email , 'incorrect user email after update got %s vs %s' % (user.email, new_email) - + def test_my_account_update_own_email_ok(self): self.log_user() - + new_email = 'new@mail.pl' response = self.app.post(url('admin_settings_my_account_update'), params=dict( _method='put', @@ -83,10 +83,10 @@ lastname='NewLastname', email=new_email,)) print response - + def test_my_account_update_err_email_exists(self): self.log_user() - + new_email = 'test_regular@mail.com'#already exisitn email response = self.app.post(url('admin_settings_my_account_update'), params=dict( _method='put', @@ -96,13 +96,13 @@ lastname='NewLastname', email=new_email,)) print response - + assert 'That e-mail address is already taken' in response.body, 'Missing error message about existing email' - - + + 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',
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/rhodecode/tests/functional/test_admin_users.py Sun Nov 14 22:54:16 2010 +0100 @@ -0,0 +1,119 @@ +from rhodecode.tests import * +from rhodecode.model.db import User +from rhodecode.lib.auth import check_password +from sqlalchemy.orm.exc import NoResultFound + +class TestAdminUsersController(TestController): + + def test_index(self): + response = self.app.get(url('users')) + # Test response... + + def test_index_as_xml(self): + response = self.app.get(url('formatted_users', format='xml')) + + def test_create(self): + self.log_user() + username = 'newtestuser' + password = 'test12' + name = 'name' + lastname = 'lastname' + email = 'mail@mail.com' + + response = self.app.post(url('users'), {'username':username, + 'password':password, + 'name':name, + 'active':True, + 'lastname':lastname, + 'email':email}) + + + assert '''created user %s''' % (username) in response.session['flash'][0], 'No flash message about new user' + + new_user = self.sa.query(User).filter(User.username == username).one() + + + assert new_user.username == username, 'wrong info about username' + assert check_password(password, new_user.password) == True , 'wrong info about password' + assert new_user.name == name, 'wrong info about name' + assert new_user.lastname == lastname, 'wrong info about lastname' + assert new_user.email == email, 'wrong info about email' + + + response.follow() + response = response.follow() + assert """edit">newtestuser</a>""" in response.body + + def test_create_err(self): + self.log_user() + username = 'new_user' + password = '' + name = 'name' + lastname = 'lastname' + email = 'errmail.com' + + response = self.app.post(url('users'), {'username':username, + 'password':password, + 'name':name, + 'active':False, + 'lastname':lastname, + 'email':email}) + + assert """<span class="error-message">Invalid username</span>""" in response.body + assert """<span class="error-message">Please enter a value</span>""" in response.body + assert """<span class="error-message">An email address must contain a single @</span>""" in response.body + + def get_user(): + self.sa.query(User).filter(User.username == username).one() + + self.assertRaises(NoResultFound, get_user), 'found user in database' + + def test_new(self): + response = self.app.get(url('new_user')) + + def test_new_as_xml(self): + response = self.app.get(url('formatted_new_user', format='xml')) + + def test_update(self): + response = self.app.put(url('user', id=1)) + + def test_update_browser_fakeout(self): + response = self.app.post(url('user', id=1), params=dict(_method='put')) + + def test_delete(self): + self.log_user() + username = 'newtestuserdeleteme' + password = 'test12' + name = 'name' + lastname = 'lastname' + email = 'todeletemail@mail.com' + + response = self.app.post(url('users'), {'username':username, + 'password':password, + 'name':name, + 'active':True, + 'lastname':lastname, + 'email':email}) + + response = response.follow() + + new_user = self.sa.query(User).filter(User.username == username).one() + response = self.app.delete(url('user', id=new_user.user_id)) + + assert """sucessfully deleted user""" in response.session['flash'][0], 'No info about user deletion' + + + def test_delete_browser_fakeout(self): + response = self.app.post(url('user', id=1), params=dict(_method='delete')) + + def test_show(self): + response = self.app.get(url('user', id=1)) + + def test_show_as_xml(self): + response = self.app.get(url('formatted_user', id=1, format='xml')) + + def test_edit(self): + response = self.app.get(url('edit_user', id=1)) + + def test_edit_as_xml(self): + response = self.app.get(url('formatted_edit_user', id=1, format='xml'))
--- a/rhodecode/tests/functional/test_hg.py Sun Nov 14 22:18:51 2010 +0100 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,11 +0,0 @@ -from rhodecode.tests import * - -class TestAdminController(TestController): - - def test_index(self): - self.log_user() - response = self.app.get(url(controller='home', action='index')) - #if global permission is set - assert 'ADD NEW REPOSITORY' in response.body, 'Wrong main page' - assert 'href="/%s/summary"' % HG_REPO in response.body, ' mising repository in list' - # Test response...
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/rhodecode/tests/functional/test_home.py Sun Nov 14 22:54:16 2010 +0100 @@ -0,0 +1,11 @@ +from rhodecode.tests import * + +class TestHomeController(TestController): + + def test_index(self): + self.log_user() + response = self.app.get(url(controller='home', action='index')) + #if global permission is set + assert 'ADD NEW REPOSITORY' in response.body, 'Wrong main page' + assert 'href="/%s/summary"' % HG_REPO in response.body, ' mising repository in list' + # Test response...
--- a/rhodecode/tests/functional/test_login.py Sun Nov 14 22:18:51 2010 +0100 +++ b/rhodecode/tests/functional/test_login.py Sun Nov 14 22:54:16 2010 +0100 @@ -102,7 +102,6 @@ 'email':email, 'name':name, 'lastname':lastname}) - print response.body assert response.status == '302 Found', 'Wrong response from register page got %s' % response.status assert 'You have successfully registered into rhodecode' in response.session['flash'][0], 'No flash message about user registration'
--- a/rhodecode/tests/functional/test_permissions.py Sun Nov 14 22:18:51 2010 +0100 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,43 +0,0 @@ -from rhodecode.tests import * - -class TestPermissionsController(TestController): - - def test_index(self): - response = self.app.get(url('permissions')) - # Test response... - - def test_index_as_xml(self): - response = self.app.get(url('formatted_permissions', format='xml')) - - def test_create(self): - response = self.app.post(url('permissions')) - - def test_new(self): - response = self.app.get(url('new_permission')) - - def test_new_as_xml(self): - response = self.app.get(url('formatted_new_permission', format='xml')) - - def test_update(self): - response = self.app.put(url('permission', id=1)) - - def test_update_browser_fakeout(self): - response = self.app.post(url('permission', id=1), params=dict(_method='put')) - - def test_delete(self): - response = self.app.delete(url('permission', id=1)) - - def test_delete_browser_fakeout(self): - response = self.app.post(url('permission', id=1), params=dict(_method='delete')) - - def test_show(self): - response = self.app.get(url('permission', id=1)) - - def test_show_as_xml(self): - response = self.app.get(url('formatted_permission', id=1, format='xml')) - - def test_edit(self): - response = self.app.get(url('edit_permission', id=1)) - - def test_edit_as_xml(self): - response = self.app.get(url('formatted_edit_permission', id=1, format='xml'))
--- a/rhodecode/tests/functional/test_repos.py Sun Nov 14 22:18:51 2010 +0100 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,141 +0,0 @@ -from rhodecode.model.db import Repository -from rhodecode.tests import * - -class TestReposController(TestController): - - def test_index(self): - self.log_user() - response = self.app.get(url('repos')) - # Test response... - - def test_index_as_xml(self): - response = self.app.get(url('formatted_repos', format='xml')) - - def test_create_hg(self): - self.log_user() - repo_name = NEW_HG_REPO - description = 'description for newly created repo' - private = False - response = self.app.post(url('repos'), {'repo_name':repo_name, - 'repo_type':'hg', - 'description':description, - 'private':private}) - - print response - - #test if we have a message for that repository - print '-' * 100 - print response.session - assert '''created repository %s''' % (repo_name) in response.session['flash'][0], 'No flash message about new repo' - - #test if the fork was created in the database - new_repo = self.sa.query(Repository).filter(Repository.repo_name == repo_name).one() - - assert new_repo.repo_name == repo_name, 'wrong name of repo name in db' - assert new_repo.description == description, 'wrong description' - - #test if repository is visible in the list ? - response = response.follow() - - assert repo_name in response.body, 'missing new repo from the main repos list' - - def test_create_git(self): - self.log_user() - repo_name = NEW_GIT_REPO - description = 'description for newly created repo' - private = False - response = self.app.post(url('repos'), {'repo_name':repo_name, - 'repo_type':'git', - 'description':description, - 'private':private}) - - print response - - #test if we have a message for that repository - print '-' * 100 - print response.session - assert '''created repository %s''' % (repo_name) in response.session['flash'][0], 'No flash message about new repo' - - #test if the fork was created in the database - new_repo = self.sa.query(Repository).filter(Repository.repo_name == repo_name).one() - - assert new_repo.repo_name == repo_name, 'wrong name of repo name in db' - assert new_repo.description == description, 'wrong description' - - #test if repository is visible in the list ? - response = response.follow() - - assert repo_name in response.body, 'missing new repo from the main repos list' - - - def test_new(self): - self.log_user() - response = self.app.get(url('new_repo')) - - def test_new_as_xml(self): - response = self.app.get(url('formatted_new_repo', format='xml')) - - def test_update(self): - response = self.app.put(url('repo', repo_name=HG_REPO)) - - def test_update_browser_fakeout(self): - response = self.app.post(url('repo', repo_name=HG_REPO), params=dict(_method='put')) - - def test_delete(self): - self.log_user() - repo_name = 'vcs_test_new_to_delete' - description = 'description for newly created repo' - private = False - response = self.app.post(url('repos'), {'repo_name':repo_name, - 'repo_type':'hg', - 'description':description, - 'private':private}) - - print response - - #test if we have a message for that repository - print '-' * 100 - print response.session - assert '''created repository %s''' % (repo_name) in response.session['flash'][0], 'No flash message about new repo' - - #test if the repo was created in the database - new_repo = self.sa.query(Repository).filter(Repository.repo_name == repo_name).one() - - assert new_repo.repo_name == repo_name, 'wrong name of repo name in db' - assert new_repo.description == description, 'wrong description' - - #test if repository is visible in the list ? - response = response.follow() - - assert repo_name in response.body, 'missing new repo from the main repos list' - - - response = self.app.delete(url('repo', repo_name=repo_name)) - - print '-' * 100 - print response.session - assert '''deleted repository %s''' % (repo_name) in response.session['flash'][0], 'No flash message about delete repo' - - response.follow() - - #check if repo was deleted from db - deleted_repo = self.sa.query(Repository).filter(Repository.repo_name == repo_name).scalar() - - assert deleted_repo is None, 'Deleted repository was found in db' - - - def test_delete_browser_fakeout(self): - response = self.app.post(url('repo', repo_name=HG_REPO), params=dict(_method='delete')) - - def test_show(self): - self.log_user() - response = self.app.get(url('repo', repo_name=HG_REPO)) - - def test_show_as_xml(self): - response = self.app.get(url('formatted_repo', repo_name=HG_REPO, format='xml')) - - def test_edit(self): - response = self.app.get(url('edit_repo', repo_name=HG_REPO)) - - def test_edit_as_xml(self): - response = self.app.get(url('formatted_edit_repo', repo_name=HG_REPO, format='xml'))
--- a/rhodecode/tests/functional/test_users.py Sun Nov 14 22:18:51 2010 +0100 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,49 +0,0 @@ -from rhodecode.tests import * - -class TestUsersController(TestController): - - def test_index(self): - response = self.app.get(url('users')) - # Test response... - - def test_index_as_xml(self): - response = self.app.get(url('formatted_users', format='xml')) - - def test_create(self): - self.log_user() -# user_name = 'new_user' -# response = self.app.post(url('users'),{'repo_name':user_name, -# 'repo_type':'hg', -# 'description':description, -# 'private':private}) - - - def test_new(self): - response = self.app.get(url('new_user')) - - def test_new_as_xml(self): - response = self.app.get(url('formatted_new_user', format='xml')) - - def test_update(self): - response = self.app.put(url('user', id=1)) - - def test_update_browser_fakeout(self): - response = self.app.post(url('user', id=1), params=dict(_method='put')) - - def test_delete(self): - response = self.app.delete(url('user', id=1)) - - def test_delete_browser_fakeout(self): - response = self.app.post(url('user', id=1), params=dict(_method='delete')) - - def test_show(self): - response = self.app.get(url('user', id=1)) - - def test_show_as_xml(self): - response = self.app.get(url('formatted_user', id=1, format='xml')) - - def test_edit(self): - response = self.app.get(url('edit_user', id=1)) - - def test_edit_as_xml(self): - response = self.app.get(url('formatted_edit_user', id=1, format='xml'))