##// END OF EJS Templates
release: Bump version 5.1.0 to 5.1.1
release: Bump version 5.1.0 to 5.1.1

File last commit:

r1261:0f8db01d default
r1269:20bfc127 v5.1.1 stable
Show More
svn_remote.py
959 lines | 33.9 KiB | text/x-python | PythonLexer
# RhodeCode VCSServer provides access to different vcs backends via network.
# Copyright (C) 2014-2023 RhodeCode GmbH
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
import os
import subprocess
from urllib.error import URLError
import urllib.parse
import logging
import posixpath as vcspath
import io
import urllib.request
import urllib.parse
import urllib.error
import traceback
import svn.client # noqa
import svn.core # noqa
import svn.delta # noqa
import svn.diff # noqa
import svn.fs # noqa
import svn.repos # noqa
import rhodecode
from vcsserver import svn_diff, exceptions, subprocessio, settings
from vcsserver.base import (
RepoFactory,
raise_from_original,
ArchiveNode,
store_archive_in_cache,
BytesEnvelope,
BinaryEnvelope,
)
from vcsserver.exceptions import NoContentException
from vcsserver.vcs_base import RemoteBase
from vcsserver.lib.str_utils import safe_str, safe_bytes
from vcsserver.lib.type_utils import assert_bytes
from vcsserver.lib.svnremoterepo import svnremoterepo
from vcsserver.lib.svn_txn_utils import store_txn_id_data
log = logging.getLogger(__name__)
svn_compatible_versions_map = {
'pre-1.4-compatible': '1.3',
'pre-1.5-compatible': '1.4',
'pre-1.6-compatible': '1.5',
'pre-1.8-compatible': '1.7',
'pre-1.9-compatible': '1.8',
}
current_compatible_version = '1.14'
def reraise_safe_exceptions(func):
"""Decorator for converting svn exceptions to something neutral."""
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
if not hasattr(e, '_vcs_kind'):
log.exception("Unhandled exception in svn remote call")
raise_from_original(exceptions.UnhandledException(e), e)
raise
return wrapper
class SubversionFactory(RepoFactory):
repo_type = 'svn'
def _create_repo(self, wire, create, compatible_version):
path = svn.core.svn_path_canonicalize(wire['path'])
if create:
fs_config = {'compatible-version': current_compatible_version}
if compatible_version:
compatible_version_string = \
svn_compatible_versions_map.get(compatible_version) \
or compatible_version
fs_config['compatible-version'] = compatible_version_string
log.debug('Create SVN repo with config `%s`', fs_config)
repo = svn.repos.create(path, "", "", None, fs_config)
else:
repo = svn.repos.open(path)
log.debug('repository created: got SVN object: %s', repo)
return repo
def repo(self, wire, create=False, compatible_version=None):
"""
Get a repository instance for the given path.
"""
return self._create_repo(wire, create, compatible_version)
NODE_TYPE_MAPPING = {
svn.core.svn_node_file: 'file',
svn.core.svn_node_dir: 'dir',
}
class SvnRemote(RemoteBase):
def __init__(self, factory, hg_factory=None):
self._factory = factory
self._bulk_methods = {
# NOT supported in SVN ATM...
}
self._bulk_file_methods = {
"size": self.get_file_size,
"data": self.get_file_content,
"flags": self.get_node_type,
"is_binary": self.is_binary,
"md5": self.md5_hash
}
@reraise_safe_exceptions
def bulk_file_request(self, wire, commit_id, path, pre_load):
cache_on, context_uid, repo_id = self._cache_on(wire)
region = self._region(wire)
# since we use unified API, we need to cast from str to in for SVN
commit_id = int(commit_id)
@region.conditional_cache_on_arguments(condition=cache_on)
def _bulk_file_request(_repo_id, _commit_id, _path, _pre_load):
result = {}
for attr in pre_load:
try:
method = self._bulk_file_methods[attr]
wire.update({'cache': False}) # disable cache for bulk calls so we don't double cache
result[attr] = method(wire, _commit_id, _path)
except KeyError as e:
raise exceptions.VcsException(e)(f'Unknown bulk attribute: "{attr}"')
return result
return BinaryEnvelope(_bulk_file_request(repo_id, commit_id, path, sorted(pre_load)))
@reraise_safe_exceptions
def discover_svn_version(self):
try:
import svn.core
svn_ver = svn.core.SVN_VERSION
except ImportError:
svn_ver = None
return safe_str(svn_ver)
@reraise_safe_exceptions
def is_empty(self, wire):
try:
return self.lookup(wire, -1) == 0
except Exception:
log.exception("failed to read object_store")
return False
def check_url(self, url, config):
# uuid function gets only valid UUID from proper repo, else
# throws exception
username, password, src_url = self.get_url_and_credentials(url)
try:
svnremoterepo(safe_bytes(username), safe_bytes(password), safe_bytes(src_url)).svn().uuid
except Exception:
tb = traceback.format_exc()
log.debug("Invalid Subversion url: `%s`, tb: %s", url, tb)
raise URLError(f'"{url}" is not a valid Subversion source url.')
return True
def is_path_valid_repository(self, wire, path):
# NOTE(marcink): short circuit the check for SVN repo
# the repos.open might be expensive to check, but we have one cheap
# pre-condition that we can use, to check for 'format' file
if not os.path.isfile(os.path.join(path, 'format')):
return False
cache_on, context_uid, repo_id = self._cache_on(wire)
region = self._region(wire)
@region.conditional_cache_on_arguments(condition=cache_on)
def _assert_correct_path(_context_uid, _repo_id, fast_check):
try:
svn.repos.open(path)
except svn.core.SubversionException:
tb = traceback.format_exc()
log.debug("Invalid Subversion path `%s`, tb: %s", path, tb)
return False
return True
return _assert_correct_path(context_uid, repo_id, True)
@reraise_safe_exceptions
def verify(self, wire,):
repo_path = wire['path']
if not self.is_path_valid_repository(wire, repo_path):
raise Exception(
f"Path {repo_path} is not a valid Subversion repository.")
cmd = ['svnadmin', 'info', repo_path]
stdout, stderr = subprocessio.run_command(cmd)
return stdout
@reraise_safe_exceptions
def lookup(self, wire, revision):
if revision not in [-1, None, 'HEAD']:
raise NotImplementedError
repo = self._factory.repo(wire)
fs_ptr = svn.repos.fs(repo)
head = svn.fs.youngest_rev(fs_ptr)
return head
@reraise_safe_exceptions
def lookup_interval(self, wire, start_ts, end_ts):
repo = self._factory.repo(wire)
fsobj = svn.repos.fs(repo)
start_rev = None
end_rev = None
if start_ts:
start_ts_svn = apr_time_t(start_ts)
start_rev = svn.repos.dated_revision(repo, start_ts_svn) + 1
else:
start_rev = 1
if end_ts:
end_ts_svn = apr_time_t(end_ts)
end_rev = svn.repos.dated_revision(repo, end_ts_svn)
else:
end_rev = svn.fs.youngest_rev(fsobj)
return start_rev, end_rev
@reraise_safe_exceptions
def revision_properties(self, wire, revision):
cache_on, context_uid, repo_id = self._cache_on(wire)
region = self._region(wire)
@region.conditional_cache_on_arguments(condition=cache_on)
def _revision_properties(_repo_id, _revision):
repo = self._factory.repo(wire)
fs_ptr = svn.repos.fs(repo)
return svn.fs.revision_proplist(fs_ptr, revision)
return _revision_properties(repo_id, revision)
def revision_changes(self, wire, revision):
repo = self._factory.repo(wire)
fsobj = svn.repos.fs(repo)
rev_root = svn.fs.revision_root(fsobj, revision)
editor = svn.repos.ChangeCollector(fsobj, rev_root)
editor_ptr, editor_baton = svn.delta.make_editor(editor)
base_dir = ""
send_deltas = False
svn.repos.replay2(
rev_root, base_dir, svn.core.SVN_INVALID_REVNUM, send_deltas,
editor_ptr, editor_baton, None)
added = []
changed = []
removed = []
# TODO: CHANGE_ACTION_REPLACE: Figure out where it belongs
for path, change in editor.changes.items():
# TODO: Decide what to do with directory nodes. Subversion can add
# empty directories.
if change.item_kind == svn.core.svn_node_dir:
continue
if change.action in [svn.repos.CHANGE_ACTION_ADD]:
added.append(path)
elif change.action in [svn.repos.CHANGE_ACTION_MODIFY,
svn.repos.CHANGE_ACTION_REPLACE]:
changed.append(path)
elif change.action in [svn.repos.CHANGE_ACTION_DELETE]:
removed.append(path)
else:
raise NotImplementedError(
"Action {} not supported on path {}".format(
change.action, path))
changes = {
'added': added,
'changed': changed,
'removed': removed,
}
return changes
@reraise_safe_exceptions
def node_history(self, wire, path, revision, limit):
cache_on, context_uid, repo_id = self._cache_on(wire)
region = self._region(wire)
@region.conditional_cache_on_arguments(condition=cache_on)
def _assert_correct_path(_context_uid, _repo_id, _path, _revision, _limit):
cross_copies = False
repo = self._factory.repo(wire)
fsobj = svn.repos.fs(repo)
rev_root = svn.fs.revision_root(fsobj, revision)
history_revisions = []
history = svn.fs.node_history(rev_root, path)
history = svn.fs.history_prev(history, cross_copies)
while history:
__, node_revision = svn.fs.history_location(history)
history_revisions.append(node_revision)
if limit and len(history_revisions) >= limit:
break
history = svn.fs.history_prev(history, cross_copies)
return history_revisions
return _assert_correct_path(context_uid, repo_id, path, revision, limit)
@reraise_safe_exceptions
def node_properties(self, wire, path, revision):
cache_on, context_uid, repo_id = self._cache_on(wire)
region = self._region(wire)
@region.conditional_cache_on_arguments(condition=cache_on)
def _node_properties(_repo_id, _path, _revision):
repo = self._factory.repo(wire)
fsobj = svn.repos.fs(repo)
rev_root = svn.fs.revision_root(fsobj, revision)
return svn.fs.node_proplist(rev_root, path)
return _node_properties(repo_id, path, revision)
def file_annotate(self, wire, path, revision):
abs_path = 'file://' + urllib.request.pathname2url(
vcspath.join(wire['path'], path))
file_uri = svn.core.svn_path_canonicalize(abs_path)
start_rev = svn_opt_revision_value_t(0)
peg_rev = svn_opt_revision_value_t(revision)
end_rev = peg_rev
annotations = []
def receiver(line_no, revision, author, date, line, pool):
annotations.append((line_no, revision, line))
# TODO: Cannot use blame5, missing typemap function in the swig code
try:
svn.client.blame2(
file_uri, peg_rev, start_rev, end_rev,
receiver, svn.client.create_context())
except svn.core.SubversionException as exc:
log.exception("Error during blame operation.")
raise Exception(
f"Blame not supported or file does not exist at path {path}. "
f"Error {exc}.")
return BinaryEnvelope(annotations)
@reraise_safe_exceptions
def get_node_type(self, wire, revision=None, path=''):
cache_on, context_uid, repo_id = self._cache_on(wire)
region = self._region(wire)
@region.conditional_cache_on_arguments(condition=cache_on)
def _get_node_type(_repo_id, _revision, _path):
repo = self._factory.repo(wire)
fs_ptr = svn.repos.fs(repo)
if _revision is None:
_revision = svn.fs.youngest_rev(fs_ptr)
root = svn.fs.revision_root(fs_ptr, _revision)
node = svn.fs.check_path(root, path)
return NODE_TYPE_MAPPING.get(node, None)
return _get_node_type(repo_id, revision, path)
@reraise_safe_exceptions
def get_nodes(self, wire, revision=None, path=''):
cache_on, context_uid, repo_id = self._cache_on(wire)
region = self._region(wire)
@region.conditional_cache_on_arguments(condition=cache_on)
def _get_nodes(_repo_id, _path, _revision):
repo = self._factory.repo(wire)
fsobj = svn.repos.fs(repo)
if _revision is None:
_revision = svn.fs.youngest_rev(fsobj)
root = svn.fs.revision_root(fsobj, _revision)
entries = svn.fs.dir_entries(root, path)
result = []
for entry_path, entry_info in entries.items():
result.append(
(entry_path, NODE_TYPE_MAPPING.get(entry_info.kind, None)))
return result
return _get_nodes(repo_id, path, revision)
@reraise_safe_exceptions
def get_file_content(self, wire, rev=None, path=''):
repo = self._factory.repo(wire)
fsobj = svn.repos.fs(repo)
if rev is None:
rev = svn.fs.youngest_rev(fsobj)
root = svn.fs.revision_root(fsobj, rev)
content = svn.core.Stream(svn.fs.file_contents(root, path))
return BytesEnvelope(content.read())
@reraise_safe_exceptions
def get_file_size(self, wire, revision=None, path=''):
cache_on, context_uid, repo_id = self._cache_on(wire)
region = self._region(wire)
@region.conditional_cache_on_arguments(condition=cache_on)
def _get_file_size(_repo_id, _revision, _path):
repo = self._factory.repo(wire)
fsobj = svn.repos.fs(repo)
if _revision is None:
_revision = svn.fs.youngest_revision(fsobj)
root = svn.fs.revision_root(fsobj, _revision)
size = svn.fs.file_length(root, path)
return size
return _get_file_size(repo_id, revision, path)
def create_repository(self, wire, compatible_version=None):
log.info('Creating Subversion repository in path "%s"', wire['path'])
self._factory.repo(wire, create=True,
compatible_version=compatible_version)
def get_url_and_credentials(self, src_url) -> tuple[str, str, str]:
obj = urllib.parse.urlparse(src_url)
username = obj.username or ''
password = obj.password or ''
return username, password, src_url
def import_remote_repository(self, wire, src_url):
repo_path = wire['path']
if not self.is_path_valid_repository(wire, repo_path):
raise Exception(
f"Path {repo_path} is not a valid Subversion repository.")
username, password, src_url = self.get_url_and_credentials(src_url)
rdump_cmd = ['svnrdump', 'dump', '--non-interactive',
'--trust-server-cert-failures=unknown-ca']
if username and password:
rdump_cmd += ['--username', username, '--password', password]
rdump_cmd += [src_url]
rdump = subprocess.Popen(
rdump_cmd,
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
load = subprocess.Popen(
['svnadmin', 'load', repo_path], stdin=rdump.stdout)
# TODO: johbo: This can be a very long operation, might be better
# to track some kind of status and provide an api to check if the
# import is done.
rdump.wait()
load.wait()
log.debug('Return process ended with code: %s', rdump.returncode)
if rdump.returncode != 0:
errors = rdump.stderr.read()
log.error('svnrdump dump failed: statuscode %s: message: %s', rdump.returncode, errors)
reason = 'UNKNOWN'
if b'svnrdump: E230001:' in errors:
reason = 'INVALID_CERTIFICATE'
if reason == 'UNKNOWN':
reason = f'UNKNOWN:{safe_str(errors)}'
raise Exception(
'Failed to dump the remote repository from {}. Reason:{}'.format(
src_url, reason))
if load.returncode != 0:
raise Exception(
f'Failed to load the dump of remote repository from {src_url}.')
def commit(self, wire, message, author, timestamp, updated, removed):
message = safe_bytes(message)
author = safe_bytes(author)
repo = self._factory.repo(wire)
fsobj = svn.repos.fs(repo)
rev = svn.fs.youngest_rev(fsobj)
txn = svn.repos.fs_begin_txn_for_commit(repo, rev, author, message)
txn_root = svn.fs.txn_root(txn)
for node in updated:
TxnNodeProcessor(node, txn_root).update()
for node in removed:
TxnNodeProcessor(node, txn_root).remove()
svn_txn_id = safe_str(svn.fs.svn_fs_txn_name(txn))
full_repo_path = wire['path']
txn_id_data = {'svn_txn_id': svn_txn_id, 'rc_internal_commit': True}
store_txn_id_data(full_repo_path, svn_txn_id, txn_id_data)
commit_id = svn.repos.fs_commit_txn(repo, txn)
if timestamp:
apr_time = apr_time_t(timestamp)
ts_formatted = svn.core.svn_time_to_cstring(apr_time)
svn.fs.change_rev_prop(fsobj, commit_id, 'svn:date', ts_formatted)
log.debug('Committed revision "%s" to "%s".', commit_id, wire['path'])
return commit_id
@reraise_safe_exceptions
def diff(self, wire, rev1, rev2, path1=None, path2=None,
ignore_whitespace=False, context=3):
wire.update(cache=False)
repo = self._factory.repo(wire)
diff_creator = SvnDiffer(
repo, rev1, path1, rev2, path2, ignore_whitespace, context)
try:
return BytesEnvelope(diff_creator.generate_diff())
except svn.core.SubversionException as e:
log.exception(
"Error during diff operation operation. "
"Path might not exist %s, %s", path1, path2)
return BytesEnvelope(b'')
@reraise_safe_exceptions
def is_large_file(self, wire, path):
return False
@reraise_safe_exceptions
def is_binary(self, wire, rev, path):
cache_on, context_uid, repo_id = self._cache_on(wire)
region = self._region(wire)
@region.conditional_cache_on_arguments(condition=cache_on)
def _is_binary(_repo_id, _rev, _path):
raw_bytes = self.get_file_content(wire, rev, path)
if not raw_bytes:
return False
return b'\0' in raw_bytes
return _is_binary(repo_id, rev, path)
@reraise_safe_exceptions
def md5_hash(self, wire, rev, path):
cache_on, context_uid, repo_id = self._cache_on(wire)
region = self._region(wire)
@region.conditional_cache_on_arguments(condition=cache_on)
def _md5_hash(_repo_id, _rev, _path):
return ''
return _md5_hash(repo_id, rev, path)
@reraise_safe_exceptions
def run_svn_command(self, wire, cmd, **opts):
path = wire.get('path', None)
debug_mode = rhodecode.ConfigGet().get_bool('debug')
if path and os.path.isdir(path):
opts['cwd'] = path
safe_call = opts.pop('_safe', False)
svnenv = os.environ.copy()
svnenv.update(opts.pop('extra_env', {}))
_opts = {'env': svnenv, 'shell': False}
try:
_opts.update(opts)
proc = subprocessio.SubprocessIOChunker(cmd, **_opts)
return b''.join(proc), b''.join(proc.stderr)
except OSError as err:
if safe_call:
return '', safe_str(err).strip()
else:
cmd = ' '.join(map(safe_str, cmd)) # human friendly CMD
call_opts = {}
if debug_mode:
call_opts = _opts
tb_err = ("Couldn't run svn command ({}).\n"
"Original error was:{}\n"
"Call options:{}\n"
.format(cmd, err, call_opts))
log.exception(tb_err)
raise exceptions.VcsException()(tb_err)
@reraise_safe_exceptions
def install_hooks(self, wire, force=False):
from vcsserver.hook_utils import install_svn_hooks
repo_path = wire['path']
binary_dir = settings.BINARY_DIR
executable = None
if binary_dir:
executable = os.path.join(binary_dir, 'python3')
return install_svn_hooks(repo_path, force_create=force)
@reraise_safe_exceptions
def get_hooks_info(self, wire):
from vcsserver.hook_utils import (
get_svn_pre_hook_version, get_svn_post_hook_version)
repo_path = wire['path']
return {
'pre_version': get_svn_pre_hook_version(repo_path),
'post_version': get_svn_post_hook_version(repo_path),
}
@reraise_safe_exceptions
def set_head_ref(self, wire, head_name):
pass
@reraise_safe_exceptions
def archive_repo(self, wire, archive_name_key, kind, mtime, archive_at_path,
archive_dir_name, commit_id, cache_config):
def walk_tree(root, root_dir, _commit_id):
"""
Special recursive svn repo walker
"""
root_dir = safe_bytes(root_dir)
filemode_default = 0o100644
filemode_executable = 0o100755
file_iter = svn.fs.dir_entries(root, root_dir)
for f_name in file_iter:
f_type = NODE_TYPE_MAPPING.get(file_iter[f_name].kind, None)
if f_type == 'dir':
# return only DIR, and then all entries in that dir
yield os.path.join(root_dir, f_name), {'mode': filemode_default}, f_type
new_root = os.path.join(root_dir, f_name)
yield from walk_tree(root, new_root, _commit_id)
else:
f_path = os.path.join(root_dir, f_name).rstrip(b'/')
prop_list = svn.fs.node_proplist(root, f_path)
f_mode = filemode_default
if prop_list.get('svn:executable'):
f_mode = filemode_executable
f_is_link = False
if prop_list.get('svn:special'):
f_is_link = True
data = {
'is_link': f_is_link,
'mode': f_mode,
'content_stream': svn.core.Stream(svn.fs.file_contents(root, f_path)).read
}
yield f_path, data, f_type
def file_walker(_commit_id, path):
repo = self._factory.repo(wire)
root = svn.fs.revision_root(svn.repos.fs(repo), int(commit_id))
def no_content():
raise NoContentException()
for f_name, f_data, f_type in walk_tree(root, path, _commit_id):
file_path = f_name
if f_type == 'dir':
mode = f_data['mode']
yield ArchiveNode(file_path, mode, False, no_content)
else:
mode = f_data['mode']
is_link = f_data['is_link']
data_stream = f_data['content_stream']
yield ArchiveNode(file_path, mode, is_link, data_stream)
return store_archive_in_cache(
file_walker, archive_name_key, kind, mtime, archive_at_path, archive_dir_name, commit_id, cache_config=cache_config)
class SvnDiffer:
"""
Utility to create diffs based on difflib and the Subversion api
"""
binary_content = False
def __init__(
self, repo, src_rev, src_path, tgt_rev, tgt_path,
ignore_whitespace, context):
self.repo = repo
self.ignore_whitespace = ignore_whitespace
self.context = context
fsobj = svn.repos.fs(repo)
self.tgt_rev = tgt_rev
self.tgt_path = tgt_path or ''
self.tgt_root = svn.fs.revision_root(fsobj, tgt_rev)
self.tgt_kind = svn.fs.check_path(self.tgt_root, self.tgt_path)
self.src_rev = src_rev
self.src_path = src_path or self.tgt_path
self.src_root = svn.fs.revision_root(fsobj, src_rev)
self.src_kind = svn.fs.check_path(self.src_root, self.src_path)
self._validate()
def _validate(self):
if (self.tgt_kind != svn.core.svn_node_none and
self.src_kind != svn.core.svn_node_none and
self.src_kind != self.tgt_kind):
# TODO: johbo: proper error handling
raise Exception(
"Source and target are not compatible for diff generation. "
"Source type: %s, target type: %s" %
(self.src_kind, self.tgt_kind))
def generate_diff(self) -> bytes:
buf = io.BytesIO()
if self.tgt_kind == svn.core.svn_node_dir:
self._generate_dir_diff(buf)
else:
self._generate_file_diff(buf)
return buf.getvalue()
def _generate_dir_diff(self, buf: io.BytesIO):
editor = DiffChangeEditor()
editor_ptr, editor_baton = svn.delta.make_editor(editor)
svn.repos.dir_delta2(
self.src_root,
self.src_path,
'', # src_entry
self.tgt_root,
self.tgt_path,
editor_ptr, editor_baton,
authorization_callback_allow_all,
False, # text_deltas
svn.core.svn_depth_infinity, # depth
False, # entry_props
False, # ignore_ancestry
)
for path, __, change in sorted(editor.changes):
self._generate_node_diff(
buf, change, path, self.tgt_path, path, self.src_path)
def _generate_file_diff(self, buf: io.BytesIO):
change = None
if self.src_kind == svn.core.svn_node_none:
change = "add"
elif self.tgt_kind == svn.core.svn_node_none:
change = "delete"
tgt_base, tgt_path = vcspath.split(self.tgt_path)
src_base, src_path = vcspath.split(self.src_path)
self._generate_node_diff(
buf, change, tgt_path, tgt_base, src_path, src_base)
def _generate_node_diff(
self, buf: io.BytesIO, change, tgt_path, tgt_base, src_path, src_base):
tgt_path_bytes = safe_bytes(tgt_path)
tgt_path = safe_str(tgt_path)
src_path_bytes = safe_bytes(src_path)
src_path = safe_str(src_path)
if self.src_rev == self.tgt_rev and tgt_base == src_base:
# makes consistent behaviour with git/hg to return empty diff if
# we compare same revisions
return
tgt_full_path = vcspath.join(tgt_base, tgt_path)
src_full_path = vcspath.join(src_base, src_path)
self.binary_content = False
mime_type = self._get_mime_type(tgt_full_path)
if mime_type and not mime_type.startswith(b'text'):
self.binary_content = True
buf.write(b"=" * 67 + b'\n')
buf.write(b"Cannot display: file marked as a binary type.\n")
buf.write(b"svn:mime-type = %s\n" % mime_type)
buf.write(b"Index: %b\n" % tgt_path_bytes)
buf.write(b"=" * 67 + b'\n')
buf.write(b"diff --git a/%b b/%b\n" % (tgt_path_bytes, tgt_path_bytes))
if change == 'add':
# TODO: johbo: SVN is missing a zero here compared to git
buf.write(b"new file mode 10644\n")
# TODO(marcink): intro to binary detection of svn patches
# if self.binary_content:
# buf.write(b'GIT binary patch\n')
buf.write(b"--- /dev/null\t(revision 0)\n")
src_lines = []
else:
if change == 'delete':
buf.write(b"deleted file mode 10644\n")
# TODO(marcink): intro to binary detection of svn patches
# if self.binary_content:
# buf.write('GIT binary patch\n')
buf.write(b"--- a/%b\t(revision %d)\n" % (src_path_bytes, self.src_rev))
src_lines = self._svn_readlines(self.src_root, src_full_path)
if change == 'delete':
buf.write(b"+++ /dev/null\t(revision %d)\n" % self.tgt_rev)
tgt_lines = []
else:
buf.write(b"+++ b/%b\t(revision %d)\n" % (tgt_path_bytes, self.tgt_rev))
tgt_lines = self._svn_readlines(self.tgt_root, tgt_full_path)
# we made our diff header, time to generate the diff content into our buffer
if not self.binary_content:
udiff = svn_diff.unified_diff(
src_lines, tgt_lines, context=self.context,
ignore_blank_lines=self.ignore_whitespace,
ignore_case=False,
ignore_space_changes=self.ignore_whitespace)
buf.writelines(udiff)
def _get_mime_type(self, path) -> bytes:
try:
mime_type = svn.fs.node_prop(
self.tgt_root, path, svn.core.SVN_PROP_MIME_TYPE)
except svn.core.SubversionException:
mime_type = svn.fs.node_prop(
self.src_root, path, svn.core.SVN_PROP_MIME_TYPE)
return mime_type
def _svn_readlines(self, fs_root, node_path):
if self.binary_content:
return []
node_kind = svn.fs.check_path(fs_root, node_path)
if node_kind not in (
svn.core.svn_node_file, svn.core.svn_node_symlink):
return []
content = svn.core.Stream(
svn.fs.file_contents(fs_root, node_path)).read()
return content.splitlines(True)
class DiffChangeEditor(svn.delta.Editor):
"""
Records changes between two given revisions
"""
def __init__(self):
self.changes = []
def delete_entry(self, path, revision, parent_baton, pool=None):
self.changes.append((path, None, 'delete'))
def add_file(
self, path, parent_baton, copyfrom_path, copyfrom_revision,
file_pool=None):
self.changes.append((path, 'file', 'add'))
def open_file(self, path, parent_baton, base_revision, file_pool=None):
self.changes.append((path, 'file', 'change'))
def authorization_callback_allow_all(root, path, pool):
return True
class TxnNodeProcessor:
"""
Utility to process the change of one node within a transaction root.
It encapsulates the knowledge of how to add, update or remove
a node for a given transaction root. The purpose is to support the method
`SvnRemote.commit`.
"""
def __init__(self, node, txn_root):
assert_bytes(node['path'])
self.node = node
self.txn_root = txn_root
def update(self):
self._ensure_parent_dirs()
self._add_file_if_node_does_not_exist()
self._update_file_content()
self._update_file_properties()
def remove(self):
svn.fs.delete(self.txn_root, self.node['path'])
# TODO: Clean up directory if empty
def _ensure_parent_dirs(self):
curdir = vcspath.dirname(self.node['path'])
dirs_to_create = []
while not self._svn_path_exists(curdir):
dirs_to_create.append(curdir)
curdir = vcspath.dirname(curdir)
for curdir in reversed(dirs_to_create):
log.debug('Creating missing directory "%s"', curdir)
svn.fs.make_dir(self.txn_root, curdir)
def _svn_path_exists(self, path):
path_status = svn.fs.check_path(self.txn_root, path)
return path_status != svn.core.svn_node_none
def _add_file_if_node_does_not_exist(self):
kind = svn.fs.check_path(self.txn_root, self.node['path'])
if kind == svn.core.svn_node_none:
svn.fs.make_file(self.txn_root, self.node['path'])
def _update_file_content(self):
assert_bytes(self.node['content'])
handler, baton = svn.fs.apply_textdelta(
self.txn_root, self.node['path'], None, None)
svn.delta.svn_txdelta_send_string(self.node['content'], handler, baton)
def _update_file_properties(self):
properties = self.node.get('properties', {})
for key, value in properties.items():
svn.fs.change_node_prop(
self.txn_root, self.node['path'], safe_bytes(key), safe_bytes(value))
def apr_time_t(timestamp):
"""
Convert a Python timestamp into APR timestamp type apr_time_t
"""
return int(timestamp * 1E6)
def svn_opt_revision_value_t(num):
"""
Put `num` into a `svn_opt_revision_value_t` structure.
"""
value = svn.core.svn_opt_revision_value_t()
value.number = num
revision = svn.core.svn_opt_revision_t()
revision.kind = svn.core.svn_opt_revision_number
revision.value = value
return revision