##// END OF EJS Templates
release: Finish preparation for 4.14.1
release: Finish preparation for 4.14.1

File last commit:

r3078:0a44452a default
r3204:a47eeac5 v4.14.1 stable
Show More
343 lines | 11.7 KiB | text/x-python | PythonLexer
# -*- coding: utf-8 -*-
# Copyright (C) 2014-2018 RhodeCode GmbH
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
SVN repository module
import logging
import os
import urllib
from zope.cachedescriptors.property import Lazy as LazyProperty
from rhodecode.lib.compat import OrderedDict
from rhodecode.lib.datelib import date_astimestamp
from rhodecode.lib.utils import safe_str, safe_unicode
from rhodecode.lib.vcs import connection, path as vcspath
from rhodecode.lib.vcs.backends import base
from rhodecode.lib.vcs.backends.svn.commit import (
SubversionCommit, _date_from_svn_properties)
from rhodecode.lib.vcs.backends.svn.diff import SubversionDiff
from rhodecode.lib.vcs.backends.svn.inmemory import SubversionInMemoryCommit
from rhodecode.lib.vcs.conf import settings
from rhodecode.lib.vcs.exceptions import (
CommitDoesNotExistError, EmptyRepositoryError, RepositoryError,
VCSError, NodeDoesNotExistError)
log = logging.getLogger(__name__)
class SubversionRepository(base.BaseRepository):
Subversion backend implementation
.. important::
It is very important to distinguish the commit index and the commit id
which is assigned by Subversion. The first one is always handled as an
`int` by this implementation. The commit id assigned by Subversion on
the other side will always be a `str`.
There is a specific trap since the first commit will have the index
``0`` but the svn id will be ``"1"``.
# Note: Subversion does not really have a default branch name.
contact = base.BaseRepository.DEFAULT_CONTACT
description = base.BaseRepository.DEFAULT_DESCRIPTION
def __init__(self, repo_path, config=None, create=False, src_url=None, bare=False,
self.path = safe_str(os.path.abspath(repo_path))
self.config = config if config else self.get_default_config()
self._init_repo(create, src_url)
def _remote(self):
return connection.Svn(self.path, self.config)
def _init_repo(self, create, src_url):
if create and os.path.exists(self.path):
raise RepositoryError(
"Cannot create repository at %s, location already exist"
% self.path)
if create:
if src_url:
src_url = _sanitize_url(src_url)
def commit_ids(self):
head = self._remote.lookup(None)
return [str(r) for r in xrange(1, head + 1)]
def branches(self):
return self._tags_or_branches('vcs_svn_branch')
def branches_closed(self):
return {}
def bookmarks(self):
return {}
def branches_all(self):
# TODO: johbo: Implement proper branch support
all_branches = {}
return all_branches
def tags(self):
return self._tags_or_branches('vcs_svn_tag')
def _tags_or_branches(self, config_section):
found_items = {}
if self.is_empty():
return {}
for pattern in self._patterns_from_section(config_section):
pattern = vcspath.sanitize(pattern)
tip = self.get_commit()
if pattern.endswith('*'):
basedir = tip.get_node(vcspath.dirname(pattern))
directories = basedir.dirs
directories = (tip.get_node(pattern), )
except NodeDoesNotExistError:
for n in directories)
def get_name(item):
return item[0]
return OrderedDict(sorted(found_items.items(), key=get_name))
def _patterns_from_section(self, section):
return (pattern for key, pattern in self.config.items(section))
def get_common_ancestor(self, commit_id1, commit_id2, repo2):
if self != repo2:
raise ValueError(
"Subversion does not support getting common ancestor of"
" different repositories.")
if int(commit_id1) < int(commit_id2):
return commit_id1
return commit_id2
def verify(self):
verify = self._remote.verify()
return verify
def compare(self, commit_id1, commit_id2, repo2, merge, pre_load=None):
# TODO: johbo: Implement better comparison, this is a very naive
# version which does not allow to compare branches, tags or folders
# at all.
if repo2 != self:
raise ValueError(
"Subversion does not support comparison of of different "
if commit_id1 == commit_id2:
return []
commit_idx1 = self._get_commit_idx(commit_id1)
commit_idx2 = self._get_commit_idx(commit_id2)
commits = [
for idx in range(commit_idx1 + 1, commit_idx2 + 1)]
return commits
def _get_commit_idx(self, commit_id):
svn_rev = int(commit_id)
# TODO: johbo: this might be only one case, HEAD, check this
svn_rev = self._remote.lookup(commit_id)
commit_idx = svn_rev - 1
if commit_idx >= len(self.commit_ids):
raise CommitDoesNotExistError(
"Commit at index %s does not exist." % (commit_idx, ))
return commit_idx
def check_url(url, config):
Check if `url` is a valid source to import a Subversion repository.
# convert to URL if it's a local directory
if os.path.isdir(url):
url = 'file://' + urllib.pathname2url(url)
return connection.Svn.check_url(url, config.serialize())
def is_valid_repository(path):
return True
except VCSError:
return False
def _check_path(self):
if not os.path.exists(self.path):
raise VCSError('Path "%s" does not exist!' % (self.path, ))
if not self._remote.is_path_valid_repository(self.path):
raise VCSError(
'Path "%s" does not contain a Subversion repository' %
(self.path, ))
def last_change(self):
Returns last change made on this repository as
`datetime.datetime` object.
# Subversion always has a first commit which has id "0" and contains
# what we are looking for.
last_id = len(self.commit_ids)
properties = self._remote.revision_properties(last_id)
return _date_from_svn_properties(properties)
def in_memory_commit(self):
return SubversionInMemoryCommit(self)
def get_hook_location(self):
returns absolute path to location where hooks are stored
return os.path.join(self.path, 'hooks')
def get_commit(self, commit_id=None, commit_idx=None, pre_load=None):
if self.is_empty():
raise EmptyRepositoryError("There are no commits yet")
if commit_id is not None:
elif commit_idx is not None:
commit_id = self.commit_ids[commit_idx]
except IndexError:
raise CommitDoesNotExistError
commit_id = self._sanitize_commit_id(commit_id)
commit = SubversionCommit(repository=self, commit_id=commit_id)
return commit
def get_commits(
self, start_id=None, end_id=None, start_date=None, end_date=None,
branch_name=None, show_hidden=False, pre_load=None):
if self.is_empty():
raise EmptyRepositoryError("There are no commit_ids yet")
if start_id is not None:
if end_id is not None:
start_raw_id = self._sanitize_commit_id(start_id)
start_pos = self.commit_ids.index(start_raw_id) if start_id else None
end_raw_id = self._sanitize_commit_id(end_id)
end_pos = max(0, self.commit_ids.index(end_raw_id)) if end_id else None
if None not in [start_id, end_id] and start_pos > end_pos:
raise RepositoryError(
"Start commit '%s' cannot be after end commit '%s'" %
(start_id, end_id))
if end_pos is not None:
end_pos += 1
# Date based filtering
if start_date or end_date:
start_raw_id, end_raw_id = self._remote.lookup_interval(
date_astimestamp(start_date) if start_date else None,
date_astimestamp(end_date) if end_date else None)
start_pos = start_raw_id - 1
end_pos = end_raw_id
commit_ids = self.commit_ids
# TODO: johbo: Reconsider impact of DEFAULT_BRANCH_NAME here
if branch_name not in [None, self.DEFAULT_BRANCH_NAME]:
svn_rev = long(self.commit_ids[-1])
commit_ids = self._remote.node_history(
path=branch_name, revision=svn_rev, limit=None)
commit_ids = [str(i) for i in reversed(commit_ids)]
if start_pos or end_pos:
commit_ids = commit_ids[start_pos:end_pos]
return base.CollectionGenerator(self, commit_ids, pre_load=pre_load)
def _sanitize_commit_id(self, commit_id):
if commit_id and commit_id.isdigit():
if int(commit_id) <= len(self.commit_ids):
return commit_id
raise CommitDoesNotExistError(
"Commit %s does not exist." % (commit_id, ))
if commit_id not in [
None, 'HEAD', 'tip', self.DEFAULT_BRANCH_NAME]:
raise CommitDoesNotExistError(
"Commit id %s not understood." % (commit_id, ))
svn_rev = self._remote.lookup('HEAD')
return str(svn_rev)
def get_diff(
self, commit1, commit2, path=None, ignore_whitespace=False,
context=3, path1=None):
self._validate_diff_commits(commit1, commit2)
svn_rev1 = long(commit1.raw_id)
svn_rev2 = long(commit2.raw_id)
diff = self._remote.diff(
svn_rev1, svn_rev2, path1=path1, path2=path,
ignore_whitespace=ignore_whitespace, context=context)
return SubversionDiff(diff)
def _sanitize_url(url):
if '://' not in url:
url = 'file://' + urllib.pathname2url(url)
return url