~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2005-2010 Canonical Ltd

    controldir,

    transport,

from bzrlib.lock import LogicalLockResult

# TODO: Modifying the conflict objects or their type is currently nearly

# impossible as there is no clear relationship between the working tree format

# and the conflict list file format.

class WorkingTree(bzrlib.mutabletree.MutableTree,

    controldir.ControlComponent):

 

    :ivar basedir: The root of the tree on disk. This is a unicode path object

        (as opposed to a URL).

    @property

    def user_transport(self):

        return self.bzrdir.user_transport

 

    @property

    def control_transport(self):

        return self._transport

 

    def _get_check_refs(self):

        """Return the references needed to perform a check of this tree.

        

        The default implementation returns no refs, and is only suitable for

        trees that have no local caching and can commit on ghosts at any time.

 

        :seealso: bzrlib.check for details about check_refs.

        """

        return []

 

    def open_containing_paths(file_list, default_directory='.',

        canonicalize=True, apply_view=True):

        """Open the WorkingTree that contains a set of paths.

 

        Fail if the paths given are not all in a single tree.

 

        This is used for the many command-line interfaces that take a list of

        any number of files and that require they all be in the same tree.

        """

        # recommended replacement for builtins.internal_tree_files

        if file_list is None or len(file_list) == 0:

            tree = WorkingTree.open_containing(default_directory)[0]

            # XXX: doesn't really belong here, and seems to have the strange

            # side effect of making it return a bunch of files, not the whole

            # tree -- mbp 20100716

            if tree.supports_views() and apply_view:

                view_files = tree.views.lookup_view()

                if view_files:

                    file_list = view_files

                    view_str = views.view_display_str(view_files)

                    note("Ignoring files outside view. View is %s" % view_str)

            return tree, file_list

        tree = WorkingTree.open_containing(file_list[0])[0]

        return tree, tree.safe_relpath_files(file_list, canonicalize,

            apply_view=apply_view)

 

    def safe_relpath_files(self, file_list, canonicalize=True, apply_view=True):

        """Convert file_list into a list of relpaths in tree.

 

        :param self: A tree to operate on.

        :param file_list: A list of user provided paths or None.

        :param apply_view: if True and a view is set, apply it or check that

            specified files are within it

        :return: A list of relative paths.

        :raises errors.PathNotChild: When a provided path is in a different self

            than self.

        """

        if file_list is None:

            return None

        if self.supports_views() and apply_view:

            view_files = self.views.lookup_view()

        else:

            view_files = []

        new_list = []

        # self.relpath exists as a "thunk" to osutils, but canonical_relpath

        # doesn't - fix that up here before we enter the loop.

        if canonicalize:

            fixer = lambda p: osutils.canonical_relpath(self.basedir, p)

        else:

            fixer = self.relpath

        for filename in file_list:

            relpath = fixer(osutils.dereference_path(filename))

            if view_files and not osutils.is_inside_any(view_files, relpath):

                raise errors.FileOutsideView(filename, view_files)

            new_list.append(relpath)

        return new_list

 

    @staticmethod

        t = transport.get_transport(location)

        iterator = bzrdir.BzrDir.find_bzrdirs(t, evaluate=evaluate,

        return [tr for tr in iterator if tr is not None]

        my_file = self.get_file(file_id, path=path, filtered=filtered)

        try:

            return my_file.read()

        finally:

            my_file.close()

        text = self.get_file_text(file_id)

            merges_bytes = self._transport.get_bytes('pending-merges')

            for l in osutils.split_lines(merges_bytes):

        # XXX: this returns the on-disk size; it should probably return the

        # canonical size

        if file_id is None:

            # For unversioned files on win32, we just assume they are not

            # executable

            return False

            return self._file_content_summary(path, stat_result)

    def _file_content_summary(self, path, stat_result):

        size = stat_result.st_size

        executable = self._is_executable_from_path_and_stat(path, stat_result)

        # try for a stat cache lookup

        return ('file', size, executable, self._sha_from_stat(

            path, stat_result))

 

                          merge_type=None, force=False):

        merger = Merger(self.branch, this_tree=self)

        # check that there are no local alterations

        if not force and self.has_changes():

            raise errors.UncommittedChanges(self)

        if to_revision is None:

            to_revision = _mod_revision.ensure_null(branch.last_revision())

        merger.other_rev_id = to_revision

        if _mod_revision.is_null(merger.other_rev_id):

            raise errors.NoCommits(branch)

        self.branch.fetch(branch, last_revision=merger.other_rev_id)

        merger.other_basis = merger.other_rev_id

        merger.other_tree = self.branch.repository.revision_tree(

            merger.other_rev_id)

        merger.other_branch = branch

        if from_revision is None:

            merger.find_base()

        else:

            merger.set_base_revision(from_revision, branch)

        if merger.base_rev_id == merger.other_rev_id:

            raise errors.PointlessMerge

        merger.backup_files = False

        if merge_type is None:

            merger.merge_type = Merge3Merger

        else:

            merger.merge_type = merge_type

        merger.set_interesting_files(None)

        merger.show_base = False

        merger.reprocess = False

        conflicts = merger.do_merge()

        merger.set_pending()

            branch.BranchReferenceFormat().initialize(tree_bzrdir,

                target_branch=new_branch)

        :param include_root: if True, return an entry for the root

                dir_ie = inv[from_dir_id]

                if dir_ie.kind == 'directory':

                    f_ie = dir_ie.children.get(f)

                else:

                    f_ie = None

                    # we may not have found this file, because of a unicode

                    # issue, or because the directory was actually a symlink.

    def move(self, from_paths, to_dir=None, after=False):

            raise TypeError('You must supply a target directory')

            # if file is missing in the inventory maybe it's in the basis_tree

            basis_tree = self.branch.basis_tree()

            from_id = basis_tree.path2id(from_rel)

            if from_id is None:

                raise errors.BzrRenameFailedError(from_rel,to_rel,

                    errors.NotVersionedError(path=str(from_rel)))

            # put entry back in the inventory so we can rename it

            from_entry = basis_tree.inventory[from_id].copy()

            inv.add(from_entry)

        else:

            from_entry = inv[from_id]

             change_reporter=None, possible_transports=None, local=False,

             show_base=False):

                                pb=None,

                                change_reporter=change_reporter,

                                show_base=show_base)

                    basis_root_id = basis_tree.get_root_id()

                    new_root_id = new_basis_tree.get_root_id()

                    if basis_root_id != new_root_id:

                        self.set_root_id(new_root_id)

        others match against only the last component.  Patterns starting

        with '!' are ignore exceptions.  Exceptions take precedence

        over regular patterns and cause the filename to not be ignored.

            self._ignoreglobster = globbing.ExceptionGlobster(self.get_ignore_list())

        """Lock the tree for reading.

 

        This also locks the branch, and can be unlocked via self.unlock().

 

        :return: A bzrlib.lock.LogicalLockResult.

        """

            self._control_files.lock_read()

            return LogicalLockResult(self.unlock)

        """See MutableTree.lock_tree_write, and WorkingTree.unlock.

 

        :return: A bzrlib.lock.LogicalLockResult.

        """

            self._control_files.lock_write()

            return LogicalLockResult(self.unlock)

        """See MutableTree.lock_write, and WorkingTree.unlock.

 

        :return: A bzrlib.lock.LogicalLockResult.

        """

            self._control_files.lock_write()

            return LogicalLockResult(self.unlock)

        f = self._transport.get('inventory')

        try:

            result = self._deserialize(f)

        finally:

            f.close()

            xml = self.branch.repository._get_inventory_xml(new_revision)

                inv = self.branch.repository._serializer.read_inventory_from_string(

                    xml, new_revision)

        f = self._transport.get('inventory')

        try:

            result = self._deserialize(f)

        finally:

            f.close()

        all_files = set() # specified and nested files 

        if to_file is None:

            to_file = sys.stdout

 

        files_to_backup = []

            for parent_info, file_infos in self.walkdirs(directory):

                    if self.path2id(relpath):

                        all_files.add(relpath)

                        # Files which are not versioned

                        files_to_backup.append(relpath)

                all_files.add(filename)

        files = list(all_files)

            for (file_id, path, content_change, versioned, parent_id, name,

                 kind, executable) in self.iter_changes(self.basis_tree(),

                     include_unchanged=True, require_versioned=False,

                     want_unversioned=True, specific_files=files):

                if versioned[0] == False:

                    # The record is unknown or newly added

                    files_to_backup.append(path[1])

                elif (content_change and (kind[1] is not None) and

                        osutils.is_inside_any(files, path[1])):

                    # Versioned and changed, but not deleted, and still

                    # in one of the dirs to be deleted.

                    files_to_backup.append(path[1])

        def backup(file_to_backup):

            backup_name = self.bzrdir.generate_backup_name(file_to_backup)

            osutils.rename(abs_path, self.abspath(backup_name))

            return "removed %s (but kept a copy: %s)" % (file_to_backup, backup_name)

                    # XXX: Really should be a more abstract reporter interface

                    kind_ch = osutils.kind_marker(self.kind(fid))

                    to_file.write(new_status + '       ' + f + kind_ch + '\n')

                            message = "deleted %s" % (f,)

                            message = backup(f)

                        if f in files_to_backup:

                            message = backup(f)

                        else:

                            osutils.delete_any(abs_path)

                            message = "deleted %s" % (f,)

               pb=None, report_changes=False):

    _marker = object()

 

    def update(self, change_reporter=None, possible_transports=None,

               revision=None, old_tip=_marker, show_base=False):

 

        :param revision: The target revision to update to. Must be in the

            revision history.

        :param old_tip: If branch.update() has already been run, the value it

            returned (old tip of the branch or None). _marker is used

            otherwise.

            update_branch = (old_tip is self._marker)

                if old_tip is self._marker:

                    old_tip = None

            return self._update_tree(old_tip, change_reporter, revision, show_base)

    def _update_tree(self, old_tip=None, change_reporter=None, revision=None,

                     show_base=False):

        nb_conflicts = 0

        if revision is None:

            revision = self.branch.last_revision()

 

        old_tip = old_tip or _mod_revision.NULL_REVISION

 

        if not _mod_revision.is_null(old_tip) and old_tip != last_rev:

            # the branch we are bound to was updated

            # merge those changes in first

            base_tree  = self.basis_tree()

            other_tree = self.branch.repository.revision_tree(old_tip)

            nb_conflicts = merge.merge_inner(self.branch, other_tree,

                                             base_tree, this_tree=self,

                                             change_reporter=change_reporter,

                                             show_base=show_base)

            if nb_conflicts:

                self.add_parent_tree((old_tip, other_tree))

                trace.note('Rerun update after fixing the conflicts.')

                return nb_conflicts

 

        if last_rev != _mod_revision.ensure_null(revision):

            # the working tree is up to date with the branch

            # we can merge the specified revision from master

            to_tree = self.branch.repository.revision_tree(revision)

            to_root_id = to_tree.get_root_id()

 

                if (basis.inventory.root is None

                    or basis.inventory.root.file_id != to_root_id):

                    self.set_root_id(to_root_id)

 

            # determine the branch point

            graph = self.branch.repository.get_graph()

            base_rev_id = graph.find_unique_lca(self.branch.last_revision(),

                                                last_rev)

            base_tree = self.branch.repository.revision_tree(base_rev_id)

 

            nb_conflicts = merge.merge_inner(self.branch, to_tree, base_tree,

                                             this_tree=self,

                                             change_reporter=change_reporter,

                                             show_base=show_base)

            self.set_last_revision(revision)

            parent_trees = [(revision, to_tree)]

            if not _mod_revision.is_null(old_tip):

        return nb_conflicts

    def _check(self, references):

        """Check the tree for consistency.

 

        :param references: A dict with keys matching the items returned by

            self._get_check_refs(), and values from looking those keys up in

            the repository.

        """

            repo_basis = references[('trees', self.last_revision())]

    def _get_check_refs(self):

        """Return the references needed to perform a check of this tree."""

        return [('trees', self.last_revision())]