"""Checkouts and working trees (working copies)."""
import bzrlib, bzrlib.add
-from bzrlib import urlutils
+from bzrlib import osutils, urlutils, ignores
from bzrlib.branch import PullResult
from bzrlib.bzrdir import BzrDirFormat, BzrDir
from bzrlib.errors import (InvalidRevisionId, NotBranchError, NoSuchFile,
NoRepositoryPresent, BzrError, UninitializableFormat,
- OutOfDateTree)
+ OutOfDateTree, UnsupportedFormatError)
from bzrlib.inventory import Inventory, InventoryFile, InventoryLink
from bzrlib.lockable_files import TransportLock, LockableFiles
from bzrlib.lockdir import LockDir
-from bzrlib.osutils import file_kind, fingerprint_file, supports_executable
from bzrlib.revision import NULL_REVISION
from bzrlib.trace import mutter
from bzrlib.revisiontree import RevisionTree
from bzrlib.transport.local import LocalTransport
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat
+from bzrlib.plugins.svn import constants
from bzrlib.plugins.svn.branch import SvnBranch
from bzrlib.plugins.svn.commit import _revision_id_to_svk_feature
from bzrlib.plugins.svn.convert import SvnConverter
from bzrlib.plugins.svn.repository import SvnRepository
from bzrlib.plugins.svn.svk import SVN_PROP_SVK_MERGE, parse_svk_features, serialize_svk_features
from bzrlib.plugins.svn.mapping import escape_svn_path
-from bzrlib.plugins.svn.transport import (SvnRaTransport, bzr_to_svn_url, create_svn_client,
- svn_config)
+from bzrlib.plugins.svn.transport import (SvnRaTransport, bzr_to_svn_url, svn_config)
from bzrlib.plugins.svn.tree import SvnBasisTree
import os
import urllib
-import svn.core, svn.wc
-from svn.core import SubversionException
+import core, wc
+from core import SubversionException, time_to_cstring
-from bzrlib.plugins.svn.errors import NoCheckoutSupport
from bzrlib.plugins.svn.format import get_rich_root_format
def generate_ignore_list(ignore_map):
class SvnWorkingTree(WorkingTree):
"""WorkingTree implementation that uses a Subversion Working Copy for storage."""
def __init__(self, bzrdir, local_path, branch):
- self._format = SvnWorkingTreeFormat()
+ version = wc.check_wc(local_path)
+ self._format = SvnWorkingTreeFormat(version)
self.basedir = local_path
assert isinstance(self.basedir, unicode)
self.bzrdir = bzrdir
self._branch = branch
- self.base_revnum = 0
- self.client_ctx = create_svn_client(bzrdir.svn_url)
- self.client_ctx.log_msg_func2 = \
- svn.client.svn_swig_py_get_commit_log_func
-
self._get_wc()
- status = svn.wc.revision_status(self.basedir, None, True, None, None)
- self.base_revnum = status.max_rev
+ (min_rev, max_rev, switch, modified) = wc.revision_status(self.basedir, None, True, None)
+ self.base_revnum = max_rev
self.base_tree = SvnBasisTree(self)
self.base_revid = branch.generate_revision_id(self.base_revnum)
self.read_working_inventory()
- self.controldir = os.path.join(self.basedir, svn.wc.get_adm_dir(),
- 'bzr')
+ self.controldir = os.path.join(self.basedir, wc.get_adm_dir(), 'bzr')
try:
os.makedirs(self.controldir)
os.makedirs(os.path.join(self.controldir, 'lock'))
except OSError:
pass
control_transport = bzrdir.transport.clone(urlutils.join(
- svn.wc.get_adm_dir(), 'bzr'))
+ wc.get_adm_dir(), 'bzr'))
self._control_files = LockableFiles(control_transport, 'lock', LockDir)
def get_ignore_list(self):
- ignores = set([svn.wc.get_adm_dir()])
- ignores.update(svn.wc.get_default_ignores(svn_config))
+ ignore_globs = set([wc.get_adm_dir()])
+ ignore_globs.update(ignores.get_runtime_ignores())
+ ignore_globs.update(ignores.get_user_ignores())
- def dir_add(wc, prefix, patprefix):
- ignorestr = svn.wc.prop_get(svn.core.SVN_PROP_IGNORE,
- self.abspath(prefix).rstrip("/"), wc)
+ def dir_add(adm, prefix, patprefix):
+ ignorestr = adm.prop_get(constants.PROP_IGNORE,
+ self.abspath(prefix).rstrip("/"))
if ignorestr is not None:
for pat in ignorestr.splitlines():
- ignores.add(urlutils.joinpath(patprefix, pat))
+ ignore_globs.add(urlutils.joinpath(patprefix, pat))
- entries = svn.wc.entries_read(wc, False)
+ entries = adm.entries_read(False)
for entry in entries:
if entry == "":
continue
# Ignore ignores on things that aren't directories
- if entries[entry].kind != svn.core.svn_node_dir:
+ if entries[entry].kind != core.NODE_DIR:
continue
subprefix = os.path.join(prefix, entry)
- subwc = svn.wc.adm_open3(wc, self.abspath(subprefix), False,
+ subwc = wc.WorkingCopy(adm, self.abspath(subprefix), False,
0, None)
try:
dir_add(subwc, subprefix, urlutils.joinpath(patprefix, entry))
finally:
- svn.wc.adm_close(subwc)
+ subwc.close()
- wc = self._get_wc()
+ adm = self._get_wc()
try:
- dir_add(wc, "", ".")
+ dir_add(adm, "", ".")
finally:
- svn.wc.adm_close(wc)
+ adm.close()
- return ignores
+ return ignore_globs
def is_control_filename(self, path):
- return svn.wc.is_adm_dir(path)
+ return wc.is_adm_dir(path)
def apply_inventory_delta(self, changes):
raise NotImplementedError(self.apply_inventory_delta)
- def update(self, change_reporter=None):
- rev = svn.core.svn_opt_revision_t()
- rev.kind = svn.core.svn_opt_revision_head
- svn.client.update(self.basedir, rev, True, self.client_ctx)
+ def _update(self, revnum=None):
+ if revnum is None:
+ # FIXME: should be able to use -1 here
+ revnum = self.branch.get_revnum()
+ adm = self._get_wc()
+ # FIXME: honor SVN_CONFIG_SECTION_HELPERS:SVN_CONFIG_OPTION_DIFF3_CMD
+ # FIXME: honor SVN_CONFIG_SECTION_MISCELLANY:SVN_CONFIG_OPTION_USE_COMMIT_TIMES
+ # FIXME: honor SVN_CONFIG_SECTION_MISCELLANY:SVN_CONFIG_OPTION_PRESERVED_CF_EXTS
+ try:
+ editor = adm.get_update_editor(self.basedir, use_commit_times=False, recurse=True)
+ assert editor is not None
+ reporter = self.branch.repository.transport.do_update(revnum, True, editor)
+ adm.crawl_revisions(self.basedir, reporter, restore_files=False, recurse=True,
+ use_commit_times=False)
+ # FIXME: handle externals
+ finally:
+ adm.close()
+ return revnum
+
+ def update(self, change_reporter=None, possible_transports=None, revnum=None):
+ orig_revnum = self.base_revnum
+ self.base_revnum = self._update(revnum)
+ self.base_revid = self.branch.generate_revision_id(self.base_revnum)
+ self.base_tree = None
+ self.read_working_inventory()
+ return self.base_revnum - orig_revnum
def remove(self, files, verbose=False, to_file=None):
# FIXME: Use to_file argument
# FIXME: Use verbose argument
assert isinstance(files, list)
- wc = self._get_wc(write_lock=True)
+ adm = self._get_wc(write_lock=True)
try:
for file in files:
- svn.wc.delete2(self.abspath(file), wc, None, None, None)
+ adm.delete(self.abspath(file))
finally:
- svn.wc.adm_close(wc)
+ adm.close()
for file in files:
self._change_fileid_mapping(None, file)
self.read_working_inventory()
def _get_wc(self, relpath="", write_lock=False):
- return svn.wc.adm_open3(None, self.abspath(relpath).rstrip("/"),
+ return wc.WorkingCopy(None, self.abspath(relpath).rstrip("/"),
write_lock, 0, None)
def _get_rel_wc(self, relpath, write_lock=False):
def move(self, from_paths, to_dir=None, after=False, **kwargs):
# FIXME: Use after argument
assert after != True
- revt = svn.core.svn_opt_revision_t()
- revt.kind = svn.core.svn_opt_revision_working
for entry in from_paths:
try:
to_wc = self._get_wc(to_dir, write_lock=True)
- svn.wc.copy(self.abspath(entry), to_wc,
+ to_wc.copy(self.abspath(entry),
os.path.basename(entry), None, None)
finally:
- svn.wc.adm_close(to_wc)
+ to_wc.close()
try:
from_wc = self._get_wc(write_lock=True)
- svn.wc.delete2(self.abspath(entry), from_wc, None, None, None)
+ from_wc.delete(self.abspath(entry))
finally:
- svn.wc.adm_close(from_wc)
+ from_wc.close()
new_name = urlutils.join(to_dir, os.path.basename(entry))
self._change_fileid_mapping(self.inventory.path2id(entry), new_name)
self._change_fileid_mapping(None, entry)
def rename_one(self, from_rel, to_rel, after=False):
# FIXME: Use after
assert after != True
- revt = svn.core.svn_opt_revision_t()
- revt.kind = svn.core.svn_opt_revision_unspecified
(to_wc, to_file) = self._get_rel_wc(to_rel, write_lock=True)
if os.path.dirname(from_rel) == os.path.dirname(to_rel):
# Prevent lock contention
(from_wc, _) = self._get_rel_wc(from_rel, write_lock=True)
from_id = self.inventory.path2id(from_rel)
try:
- svn.wc.copy(self.abspath(from_rel), to_wc, to_file, None, None)
- svn.wc.delete2(self.abspath(from_rel), from_wc, None, None, None)
+ to_wc.copy(self.abspath(from_rel), to_file)
+ from_wc.delete(self.abspath(from_rel))
finally:
- svn.wc.adm_close(to_wc)
+ to_wc.close()
self._change_fileid_mapping(None, from_rel)
self._change_fileid_mapping(from_id, to_rel)
self.read_working_inventory()
assert isinstance(path, str)
rp = self.branch.unprefix(path)
- entry = self.base_tree.id_map[rp]
+ entry = self.basis_tree().id_map[rp]
assert entry[0] is not None
assert isinstance(entry[0], str), "fileid %r for %r is not a string" % (entry[0], path)
return entry
file = InventoryFile(id, os.path.basename(relpath), parent_id)
file.revision = revid
try:
- data = fingerprint_file(open(self.abspath(relpath)))
+ data = osutils.fingerprint_file(open(self.abspath(relpath)))
file.text_sha1 = data['sha1']
file.text_size = data['size']
file.executable = self.is_executable(id, relpath)
pass
def find_copies(url, relpath=""):
- wc = self._get_wc(relpath)
- entries = svn.wc.entries_read(wc, False)
+ adm = self._get_wc(relpath)
+ entries = adm.entries_read(False)
for entry in entries.values():
subrelpath = os.path.join(relpath, entry.name)
if entry.name == "" or entry.kind != 'directory':
if ((entry.copyfrom_url == url or entry.url == url) and
- not (entry.schedule in (svn.wc.schedule_delete,
- svn.wc.schedule_replace))):
+ not (entry.schedule in (wc.SCHEDULE_DELETE,
+ wc.SCHEDULE_REPLACE))):
yield os.path.join(
self.branch.get_branch_path().strip("/"),
subrelpath)
else:
find_copies(subrelpath)
- svn.wc.adm_close(wc)
+ adm.close()
def find_ids(entry, rootwc):
relpath = urllib.unquote(entry.url[len(entry.repos):].strip("/"))
- assert entry.schedule in (svn.wc.schedule_normal,
- svn.wc.schedule_delete,
- svn.wc.schedule_add,
- svn.wc.schedule_replace)
- if entry.schedule == svn.wc.schedule_normal:
+ assert entry.schedule in (wc.SCHEDULE_NORMAL,
+ wc.SCHEDULE_DELETE,
+ wc.SCHEDULE_ADD,
+ wc.SCHEDULE_REPLACE)
+ if entry.schedule == wc.SCHEDULE_NORMAL:
assert entry.revision >= 0
# Keep old id
return self.path_to_file_id(entry.cmt_rev, entry.revision,
relpath)
- elif entry.schedule == svn.wc.schedule_delete:
+ elif entry.schedule == wc.SCHEDULE_DELETE:
return (None, None)
- elif (entry.schedule == svn.wc.schedule_add or
- entry.schedule == svn.wc.schedule_replace):
+ elif (entry.schedule == wc.SCHEDULE_ADD or
+ entry.schedule == wc.SCHEDULE_REPLACE):
# See if the file this file was copied from disappeared
# and has no other copies -> in that case, take id of other file
if (entry.copyfrom_url and
# FIXME: Generate more random file ids
return ("NEW-" + escape_svn_path(entry.url[len(entry.repos):].strip("/")), None)
- def add_dir_to_inv(relpath, wc, parent_id):
+ def add_dir_to_inv(relpath, adm, parent_id):
assert isinstance(relpath, unicode)
- entries = svn.wc.entries_read(wc, False)
+ entries = adm.entries_read(False)
entry = entries[""]
assert parent_id is None or isinstance(parent_id, str), \
"%r is not a string" % parent_id
entry = entries[name]
assert entry
- if entry.kind == svn.core.svn_node_dir:
- subwc = svn.wc.adm_open3(wc, self.abspath(subrelpath),
+ if entry.kind == core.NODE_DIR:
+ subwc = wc.WorkingCopy(adm, self.abspath(subrelpath),
False, 0, None)
try:
add_dir_to_inv(subrelpath, subwc, id)
finally:
- svn.wc.adm_close(subwc)
+ subwc.close()
else:
(subid, subrevid) = find_ids(entry, rootwc)
if subid:
try:
add_dir_to_inv(u"", rootwc, None)
finally:
- svn.wc.adm_close(rootwc)
+ rootwc.close()
self._set_inventory(inv, dirty=False)
return inv
if revid is None or revid == NULL_REVISION:
self.base_revid = revid
self.base_revnum = 0
- self.base_tree = RevisionTree(self, Inventory(), revid)
+ self.base_tree = None
return
rev = self.branch.lookup_revision_id(revid)
self.base_revnum = rev
self.base_revid = revid
- self.base_tree = SvnBasisTree(self)
+ self.base_tree = None
# TODO: Implement more efficient version
newrev = self.branch.repository.get_revision(revid)
newrevtree = self.branch.repository.revision_tree(revid)
- def update_settings(wc, path):
+ def update_settings(adm, path):
id = newrevtree.inventory.path2id(path)
mutter("Updating settings for %r" % id)
revnum = self.branch.lookup_revision_id(
newrevtree.inventory[id].revision)
- svn.wc.process_committed2(self.abspath(path).rstrip("/"), wc,
+ adm.process_committed(self.abspath(path).rstrip("/"),
False, revnum,
- svn.core.svn_time_to_cstring(newrev.timestamp),
+ svn_time_to_cstring(newrev.timestamp),
newrev.committer, None, False)
if newrevtree.inventory[id].kind != 'directory':
return
- entries = svn.wc.entries_read(wc, True)
+ entries = adm.entries_read(True)
for entry in entries:
if entry == "":
continue
- subwc = svn.wc.adm_open3(wc, os.path.join(self.basedir, path, entry), False, 0, None)
+ subwc = wc.WorkingCopy(adm, os.path.join(self.basedir, path, entry), False, 0, None)
try:
update_settings(subwc, os.path.join(path, entry))
finally:
- svn.wc.adm_close(subwc)
+ subwc.close()
# Set proper version for all files in the wc
- wc = self._get_wc(write_lock=True)
+ adm = self._get_wc(write_lock=True)
try:
- update_settings(wc, "")
+ update_settings(adm, "")
finally:
- svn.wc.adm_close(wc)
+ adm.close()
self.base_revid = revid
- def commit(self, message=None, message_callback=None, revprops=None,
- timestamp=None, timezone=None, committer=None, rev_id=None,
- allow_pointless=True, strict=False, verbose=False, local=False,
- reporter=None, config=None, specific_files=None, author=None):
- if author is not None:
- revprops['author'] = author
- # FIXME: Use allow_pointless
- # FIXME: Use verbose
- # FIXME: Use reporter
- # FIXME: Use strict
- if local:
- raise LocalCommitsUnsupported()
-
- if specific_files:
- specific_files = [self.abspath(x).encode('utf8') for x in specific_files]
- else:
- specific_files = [self.basedir.encode('utf8')]
-
- if message_callback is not None:
- def log_message_func(items, pool):
- """ Simple log message provider for unit tests. """
- return message_callback(self).encode("utf-8")
- else:
- assert isinstance(message, basestring)
- def log_message_func(items, pool):
- """ Simple log message provider for unit tests. """
- return message.encode("utf-8")
-
- self.client_ctx.log_msg_baton2 = log_message_func
- if rev_id is not None:
- extra = "%d %s\n" % (self.branch.revno()+1, rev_id)
- else:
- extra = ""
- wc = self._get_wc(write_lock=True)
- try:
- svn.wc.prop_set(SVN_PROP_BZR_REVISION_ID+str(self.branch.mapping.scheme),
- self._get_bzr_revids(self._get_base_branch_props()) + extra,
- self.basedir, wc)
- svn.wc.prop_set(SVN_PROP_BZR_REVISION_INFO,
- generate_revision_metadata(timestamp,
- timezone,
- committer,
- revprops),
- self.basedir, wc)
- finally:
- svn.wc.adm_close(wc)
-
- try:
- try:
- commit_info = svn.client.commit3(specific_files, True, False,
- self.client_ctx)
- except SubversionException, (_, num):
- if num == svn.core.SVN_ERR_FS_TXN_OUT_OF_DATE:
- raise OutOfDateTree(self)
- raise
- except:
- # Reset properties so the next subversion commit won't
- # accidently set these properties.
- wc = self._get_wc(write_lock=True)
- base_branch_props = self._get_base_branch_props()
- svn.wc.prop_set(SVN_PROP_BZR_REVISION_ID+str(self.branch.mapping.scheme),
- self._get_bzr_revids(base_branch_props), self.basedir, wc)
- svn.wc.prop_set(SVN_PROP_BZR_REVISION_INFO,
- base_branch_props.get(SVN_PROP_BZR_REVISION_INFO, ""),
- self.basedir, wc)
- svn.wc.adm_close(wc)
- raise
-
- self.client_ctx.log_msg_baton2 = None
-
- revid = self.branch.generate_revision_id(commit_info.revision)
-
- self.base_revid = revid
- self.base_revnum = commit_info.revision
- self.base_tree = SvnBasisTree(self)
-
- return revid
-
def smart_add(self, file_list, recurse=True, action=None, save=True):
assert isinstance(recurse, bool)
if action is None:
todo = []
file_path = os.path.abspath(file_path)
f = self.relpath(file_path)
- wc = self._get_wc(os.path.dirname(f), write_lock=True)
+ adm = self._get_wc(os.path.dirname(f), write_lock=True)
try:
if not self.inventory.has_filename(f):
if save:
mutter('adding %r' % file_path)
- svn.wc.add2(file_path, wc, None, 0, None, None, None)
+ adm.add(file_path, None, 0, None, None, None)
added.append(file_path)
- if recurse and file_kind(file_path) == 'directory':
+ if recurse and osutils.file_kind(file_path) == 'directory':
# Filter out ignored files and update ignored
for c in os.listdir(file_path):
if self.is_control_filename(c):
ignored.setdefault(ignore_glob, []).append(c_path)
todo.append(c_path)
finally:
- svn.wc.adm_close(wc)
+ adm.close()
if todo != []:
cadded, cignored = self.smart_add(todo, recurse, action, save)
added.extend(cadded)
ids = iter(ids)
assert isinstance(files, list)
for f in files:
- wc = self._get_wc(os.path.dirname(f), write_lock=True)
+ adm = self._get_wc(os.path.dirname(f), write_lock=True)
try:
try:
- svn.wc.add2(os.path.join(self.basedir, f), wc, None, 0,
- None, None, None)
+ adm.add(os.path.join(self.basedir, f))
if ids is not None:
- self._change_fileid_mapping(ids.next(), f, wc)
+ self._change_fileid_mapping(ids.next(), f, adm)
except SubversionException, (_, num):
- if num == svn.core.SVN_ERR_ENTRY_EXISTS:
+ if num == constants.ERR_ENTRY_EXISTS:
continue
- elif num == svn.core.SVN_ERR_WC_PATH_NOT_FOUND:
+ elif num == constants.ERR_WC_PATH_NOT_FOUND:
raise NoSuchFile(path=f)
raise
finally:
- svn.wc.adm_close(wc)
+ adm.close()
self.read_working_inventory()
def basis_tree(self):
if self.base_revid is None or self.base_revid == NULL_REVISION:
return self.branch.repository.revision_tree(self.base_revid)
+ if self.base_tree is None:
+ self.base_tree = SvnBasisTree(self)
+
return self.base_tree
def pull(self, source, overwrite=False, stop_revision=None,
delta_reporter=None, possible_transports=None):
# FIXME: Use delta_reporter
+ # FIXME: Use source
# FIXME: Use overwrite
result = PullResult()
result.source_branch = source
(result.old_revno, result.old_revid) = self.branch.last_revision_info()
if stop_revision is None:
stop_revision = self.branch.last_revision()
- rev = svn.core.svn_opt_revision_t()
- rev.kind = svn.core.svn_opt_revision_number
- rev.value.number = self.branch.lookup_revision_id(stop_revision)
- fetched = svn.client.update(self.basedir, rev, True, self.client_ctx)
+ revnumber = self.branch.lookup_revision_id(stop_revision)
+ fetched = self._update(revnum)
+ self.base_revnum = fetched
self.base_revid = self.branch.generate_revision_id(fetched)
+ self.base_tree = None
+ self.read_working_inventory()
result.new_revid = self.base_revid
result.new_revno = self.branch.revision_id_to_revno(result.new_revid)
return result
def get_file_sha1(self, file_id, path=None, stat_value=None):
if not path:
path = self._inventory.id2path(file_id)
- return fingerprint_file(open(self.abspath(path)))['sha1']
+ return osutils.fingerprint_file(open(self.abspath(path)))['sha1']
- def _change_fileid_mapping(self, id, path, wc=None):
- if wc is None:
+ def _change_fileid_mapping(self, id, path, adm=None):
+ if adm is None:
subwc = self._get_wc(write_lock=True)
else:
- subwc = wc
+ subwc = adm
new_entries = self._get_new_file_ids(subwc)
if id is None:
if new_entries.has_key(path):
new_entries[path] = id
existing = "".join(map(lambda (path, id): "%s\t%s\n" % (path, id), new_entries.items()))
if existing != "":
- svn.wc.prop_set(SVN_PROP_BZR_FILEIDS, existing.encode("utf-8"), self.basedir, subwc)
- if wc is None:
- svn.wc.adm_close(subwc)
+ subwc.prop_set(SVN_PROP_BZR_FILEIDS, existing.encode("utf-8"), self.basedir)
+ if adm is None:
+ subwc.close()
def _get_base_branch_props(self):
return self.branch.repository.branchprop_list.get_properties(
self.branch.get_branch_path(self.base_revnum), self.base_revnum)
- def _get_new_file_ids(self, wc):
+ def _get_new_file_ids(self, adm):
committed = self._get_base_branch_props().get(SVN_PROP_BZR_FILEIDS, "")
- existing = svn.wc.prop_get(SVN_PROP_BZR_FILEIDS, self.basedir, wc)
+ existing = adm.prop_get(SVN_PROP_BZR_FILEIDS, self.basedir)
if existing is None or committed == existing:
return {}
return dict(map(lambda x: str(x).split("\t"),
def set_pending_merges(self, merges):
"""See MutableTree.set_pending_merges()."""
- wc = self._get_wc(write_lock=True)
+ adm = self._get_wc(write_lock=True)
try:
# Set bzr:merge
if len(merges) > 0:
else:
bzr_merge = ""
- svn.wc.prop_set(SVN_PROP_BZR_ANCESTRY+str(self.branch.mapping.scheme),
+ adm.prop_set(SVN_PROP_BZR_ANCESTRY+str(self.branch.mapping.scheme),
self._get_bzr_merges(self._get_base_branch_props()) + bzr_merge,
- self.basedir, wc)
+ self.basedir)
svk_merges = parse_svk_features(self._get_svk_merges(self._get_base_branch_props()))
except InvalidRevisionId:
pass
- svn.wc.prop_set2(SVN_PROP_SVK_MERGE,
+ adm.prop_set(SVN_PROP_SVK_MERGE,
serialize_svk_features(svk_merges), self.basedir,
- wc, False)
+ False)
finally:
- svn.wc.adm_close(wc)
+ adm.close()
def add_pending_merge(self, revid):
merges = self.pending_merges()
merges.append(revid)
self.set_pending_merges(merges)
+ def get_parent_ids(self):
+ return [self.base_revid] + self.pending_merges()
+
def pending_merges(self):
merged = self._get_bzr_merges(self._get_base_branch_props()).splitlines()
- wc = self._get_wc()
+ adm = self._get_wc()
try:
- merged_data = svn.wc.prop_get(
- SVN_PROP_BZR_ANCESTRY+str(self.branch.mapping.scheme), self.basedir, wc)
+ merged_data = adm.prop_get(
+ SVN_PROP_BZR_ANCESTRY+str(self.branch.mapping.scheme), self.basedir)
if merged_data is None:
set_merged = []
else:
set_merged = merged_data.splitlines()
finally:
- svn.wc.adm_close(wc)
+ adm.close()
assert (len(merged) == len(set_merged) or
len(merged)+1 == len(set_merged))
return []
+ def path_content_summary(self, path, _lstat=os.lstat,
+ _mapper=osutils.file_kind_from_stat_mode):
+ """See Tree.path_content_summary."""
+ abspath = self.abspath(path)
+ try:
+ stat_result = _lstat(abspath)
+ except OSError, e:
+ if getattr(e, 'errno', None) == errno.ENOENT:
+ # no file.
+ return ('missing', None, None, None)
+ # propagate other errors
+ raise
+ kind = _mapper(stat_result.st_mode)
+ if kind == 'file':
+ size = stat_result.st_size
+ # try for a stat cache lookup
+ executable = self._is_executable_from_path_and_stat(path, stat_result)
+ return (kind, size, executable, self._sha_from_stat(
+ path, stat_result))
+ elif kind == 'directory':
+ return kind, None, None, None
+ elif kind == 'symlink':
+ return ('symlink', None, None, os.readlink(abspath))
+ else:
+ return (kind, None, None, None)
+
def _reset_data(self):
pass
finally:
self.branch.unlock()
- if not supports_executable():
+ if not osutils.supports_executable():
def is_executable(self, file_id, path=None):
inv = self.basis_tree()._inventory
if file_id in inv:
class SvnWorkingTreeFormat(WorkingTreeFormat):
"""Subversion working copy format."""
+ def __init__(self, version):
+ self.version = version
+
def __get_matchingbzrdir(self):
return SvnWorkingTreeDirFormat()
_matchingbzrdir = property(__get_matchingbzrdir)
def get_format_description(self):
- return "Subversion Working Copy"
+ return "Subversion Working Copy Version %d" % self.version
def get_format_string(self):
- return "Subversion Working Copy Format"
+ raise NotImplementedError
def initialize(self, a_bzrdir, revision_id=None):
raise NotImplementedError(self.initialize)
self.local_path = transport.local_abspath(".")
# Open related remote repository + branch
- wc = svn.wc.adm_open3(None, self.local_path, False, 0, None)
try:
- self.svn_url = svn.wc.entry(self.local_path, wc, True).url
+ adm = wc.WorkingCopy(None, self.local_path, False, 0, None)
+ except SubversionException, (msg, constants.ERR_WC_UNSUPPORTED_FORMAT):
+ raise UnsupportedFormatError(msg, kind='workingtree')
+ try:
+ self.svn_url = adm.entry(self.local_path, True).url
finally:
- svn.wc.adm_close(wc)
+ adm.close()
self.remote_transport = SvnRaTransport(self.svn_url)
self.remote_bzrdir = SvnRemoteAccess(self.remote_transport)
branch = SvnBranch(self.remote_transport.base, repos,
self.remote_bzrdir.branch_path)
except SubversionException, (_, num):
- if num == svn.core.SVN_ERR_WC_NOT_DIRECTORY:
+ if num == constants.ERR_WC_NOT_DIRECTORY:
raise NotBranchError(path=self.base)
raise
git.samba.org / jelmer / subvertpy.git / blobdiff